feat(sentry): wire the diagnostic metric emitters for shutdown, IPC errors, and telemetry failures

backend/index.js

@@ -126,13 +126,25 @@ const controlIpcServer = new SimpleRpcServer({
       Promise.resolve(p).catch((e) =>
         console.error(`shutdown: ${label} close failed`, e),
       );
-    await Promise.all([
-      settle("control", controlIpcServer.close()),
-      settle("fastify", fastify.close()),
-      ...(comapeoRpcServer ? [settle("rpc", comapeoRpcServer.close())] : []),
-    ]);
-    if (comapeoManager) await settle("manager", comapeoManager.close());
-    if (mapServer) await settle("map-server", mapServer.close());
+    // withSpan records each phase to `comapeo.shutdown.phase_duration_ms`
+    // even when traces are off, mirroring the boot phases.
+    await sentry.withSpan("shutdown.close-servers", () =>
+      Promise.all([
+        settle("control", controlIpcServer.close()),
+        settle("fastify", fastify.close()),
+        ...(comapeoRpcServer ? [settle("rpc", comapeoRpcServer.close())] : []),
+      ]),
+    );
+    if (comapeoManager) {
+      await sentry.withSpan("shutdown.close-manager", () =>
+        settle("manager", comapeoManager.close()),
+      );
+    }
+    if (mapServer) {
+      await sentry.withSpan("shutdown.close-map-server", () =>
+        settle("map-server", mapServer.close()),
+      );
+    }
   },
   /**
    * Android-only attribution channel: FGS-local failures (rootkey load,

backend/lib/comapeo-rpc.js

@@ -1,5 +1,6 @@
 import { SocketMessagePort } from "./message-port.js";
 import { ServerHelper } from "./server-helper.js";
+import * as metrics from "./metrics.js";
 import {
   createComapeoCoreServer,
   createComapeoServicesServer,
@@ -33,6 +34,7 @@ export class ComapeoRpc extends ServerHelper {
 
       messagePort.addEventListener("messageerror", (event) => {
         console.error("Client sent invalid message", event.data);
+        metrics.ipcError(event.data?.name);
       });
 
       messagePort.addEventListener("close", () => {

backend/lib/comapeo-rpc.test.mjs

@@ -1,13 +1,16 @@
 import { test } from "node:test";
 import assert from "node:assert/strict";
+import { Buffer } from "node:buffer";
+import FramedStream from "framed-stream";
 import {
   createComapeoServicesClient,
   closeComapeoServicesClient,
 } from "@comapeo/ipc/client.js";
 
 import { ComapeoRpc } from "./comapeo-rpc.js";
 import { SocketMessagePort } from "./message-port.js";
-import { connectSocket, socketPath } from "./test-helpers.mjs";
+import * as metrics from "./metrics.js";
+import { connectSocket, socketPath, waitFor } from "./test-helpers.mjs";
 
 // Core requests aren't exercised here, so a bare object stands in for the
 // manager — createComapeoCoreServer only forwards calls it never receives.
@@ -83,3 +86,33 @@ test("a client disconnecting does not stop the server serving the next client",
   const second = await connectServicesClient(t, path);
   assert.equal(await second.mapServer.getBaseUrl(), "http://127.0.0.1:1001");
 });
+
+test("a malformed frame on the comapeo socket records comapeo.ipc.errors", async (t) => {
+  /** @type {Array<{ name: string, attributes: Record<string, unknown> }>} */
+  const counts = [];
+  metrics.init({
+    Sentry: /** @type {any} */ ({
+      metrics: {
+        count: (name, value, data) => counts.push({ name, ...data }),
+      },
+    }),
+    platform: "android",
+    deviceClass: "mid",
+    osMajor: "android.14",
+    applicationUsageData: false,
+  });
+  t.after(() => metrics.resetForTests());
+
+  const { path } = await startRpc(t, {
+    mapServer: { getBaseUrl: async () => "http://127.0.0.1:9999" },
+  });
+  const socket = await connectSocket(t, path);
+  const raw = new FramedStream(socket);
+  raw.write(Buffer.from("garbage not json"));
+
+  await waitFor(() => counts.some((c) => c.name === "comapeo.ipc.errors"), {
+    message: "ipc error metric recorded",
+  });
+  const err = counts.find((c) => c.name === "comapeo.ipc.errors");
+  assert.equal(err?.attributes.error_class, "SyntaxError");
+});

backend/lib/sentry.js

@@ -84,10 +84,15 @@ const forwardingTransport = () => ({
     if (!envelopeToFrame) return {};
     const frame = envelopeToFrame(envelope);
     if (sink) {
-      sink(frame);
+      try {
+        sink(frame);
+      } catch {
+        metrics.telemetryForwardingFailure();
+      }
     } else {
       if (preListenQueue.length >= PRE_LISTEN_QUEUE_MAX) {
         preListenQueue.shift();
+        metrics.telemetryForwardingFailure();
       }
       preListenQueue.push(frame);
     }
@@ -265,12 +270,21 @@ export async function withBootTrace(args, loadIndex) {
  * @returns {Promise<T>}
  */
 export async function withSpan(op, fn) {
-  // Always record the boot-phase duration as a metric, even
+  // Always record the phase duration as a metric, even
   // when traces are off (`debug=false`). The span only materialises
   // under `debug` via `tracesSampleRate`, but the metric is always-on.
-  const phase = op.startsWith("boot.") ? op.slice("boot.".length) : op;
   const start = performance.now();
-  const recordPhase = () => metrics.bootPhase(phase, performance.now() - start);
+  const recordPhase = () => {
+    const ms = performance.now() - start;
+    if (op.startsWith("shutdown.")) {
+      metrics.shutdownPhase(op.slice("shutdown.".length), ms);
+    } else {
+      metrics.bootPhase(
+        op.startsWith("boot.") ? op.slice("boot.".length) : op,
+        ms,
+      );
+    }
+  };
   if (!Sentry) {
     try {
       return await fn();
@@ -335,6 +349,7 @@ export function setSink(realSink) {
       realSink(frame);
     } catch {
       // Sink threw — drop the rest rather than retrying forever.
+      metrics.telemetryForwardingFailure();
       break;
     }
   }

backend/lib/sentry.test.mjs

@@ -4,7 +4,7 @@ import assert from "node:assert/strict";
 import * as Sentry from "@sentry/node-core";
 
 import { initSentry } from "./sentry-init.js";
-import { rpcHook, setSink, flush } from "./sentry.js";
+import { rpcHook, setSink, flush, withSpan } from "./sentry.js";
 import * as metrics from "./metrics.js";
 
 /**
@@ -38,16 +38,18 @@ const baseArgv = {
   platformTag: "android",
 };
 
-/** Fake metrics SDK that records distribution calls instead of emitting. */
+/** Fake metrics SDK that records distribution/count calls instead of emitting. */
 function recordingMetricsSdk() {
   const distributions = [];
+  const counts = [];
   return {
     distributions,
+    counts,
     sdk: {
       metrics: {
         distribution: (name, value, data) =>
           distributions.push({ name, value, ...data }),
-        count: () => {},
+        count: (name, value, data) => counts.push({ name, value, ...data }),
         gauge: () => {},
       },
     },
@@ -196,6 +198,65 @@ test("debug OFF: a rejecting RPC records the duration metric but captures no iss
   await Sentry.close();
 });
 
+test("withSpan on a shutdown op records the shutdown phase metric", async () => {
+  initSentry({ ...baseArgv, debug: false });
+  const rec = recordingMetricsSdk();
+  metrics.init({
+    Sentry: rec.sdk,
+    platform: "android",
+    deviceClass: "mid",
+    osMajor: "android.14",
+    applicationUsageData: true,
+  });
+
+  await withSpan("shutdown.close-servers", async () => {});
+  await withSpan("boot.manager-init", async () => {});
+
+  const shutdown = rec.distributions.find(
+    (d) => d.name === "comapeo.shutdown.phase_duration_ms",
+  );
+  assert.ok(shutdown, "shutdown phase metric not recorded via withSpan");
+  assert.equal(shutdown.attributes.phase, "close-servers");
+  assert.equal(shutdown.unit, "millisecond");
+  assert.ok(shutdown.value >= 0);
+
+  // Boot ops still route to the boot metric with the prefix stripped.
+  const boot = rec.distributions.find(
+    (d) => d.name === "comapeo.boot.phase_duration_ms",
+  );
+  assert.ok(boot, "boot phase metric not recorded via withSpan");
+  assert.equal(boot.attributes.phase, "manager-init");
+
+  await Sentry.close();
+});
+
+test("a throwing envelope sink records the telemetry forwarding-failure metric", async () => {
+  initSentry({ ...baseArgv, debug: false });
+  const rec = recordingMetricsSdk();
+  metrics.init({
+    Sentry: rec.sdk,
+    platform: "android",
+    deviceClass: "mid",
+    osMajor: "android.14",
+    applicationUsageData: true,
+  });
+
+  setSink(() => {
+    throw new Error("sink boom");
+  });
+
+  // Real call path: capture → forwardingTransport.send → sink throws.
+  Sentry.captureMessage("forwarding failure smoke");
+  await flush(2000);
+
+  assert.ok(
+    rec.counts.some((c) => c.name === "comapeo.telemetry.forwarding_failures"),
+    "sink throw must record comapeo.telemetry.forwarding_failures",
+  );
+
+  await Sentry.close();
+});
+
 test("initialScope carries the native-derived user.id on outgoing events", async () => {
   initSentry({ ...baseArgv, debug: false, sentryUserId: "e15e7255ae360358" });
 

backend/lib/simple-rpc.js

@@ -1,5 +1,6 @@
 import { ServerHelper } from "./server-helper.js";
 import { SocketMessagePort } from "./message-port.js";
+import * as metrics from "./metrics.js";
 
 /**
  * @typedef {{ type: "stopping" } | { type: "error", phase: string, message: string, stack?: string }} TerminalFrame
@@ -48,6 +49,7 @@ export class SimpleRpcServer extends ServerHelper {
     messagePort.addEventListener("message", this.#handleMessageEvent);
     messagePort.addEventListener("messageerror", (event) => {
       console.error("Client sent invalid message", event.data);
+      metrics.ipcError(event.data?.name);
     });
     messagePort.addEventListener("close", () => {
       this.#clients.delete(messagePort);
@@ -138,6 +140,7 @@ export class SimpleRpcServer extends ServerHelper {
         client.postMessage(message);
       } catch (e) {
         console.error("broadcast: client postMessage threw", e);
+        metrics.ipcError(e instanceof Error ? e.name : undefined);
       }
     }
   }

backend/lib/simple-rpc.test.mjs

@@ -5,6 +5,7 @@ import FramedStream from "framed-stream";
 
 import { SimpleRpcServer } from "./simple-rpc.js";
 import { SocketMessagePort } from "./message-port.js";
+import * as metrics from "./metrics.js";
 import { connectSocket, socketPath, waitFor } from "./test-helpers.mjs";
 
 /**
@@ -158,3 +159,31 @@ test("a malformed frame does not crash the server", async (t) => {
   await waitFor(() => received.length === 1, { message: "survived bad frame" });
   assert.deepEqual(received[0], { type: "init", ok: true });
 });
+
+test("a malformed frame on the control socket records comapeo.ipc.errors", async (t) => {
+  /** @type {Array<{ name: string, attributes: Record<string, unknown> }>} */
+  const counts = [];
+  metrics.init({
+    Sentry: /** @type {any} */ ({
+      metrics: {
+        count: (name, value, data) => counts.push({ name, ...data }),
+      },
+    }),
+    platform: "android",
+    deviceClass: "mid",
+    osMajor: "android.14",
+    applicationUsageData: false,
+  });
+  t.after(() => metrics.resetForTests());
+
+  const { path } = await startServer(t, {});
+  const socket = await connectSocket(t, path);
+  const raw = new FramedStream(socket);
+  raw.write(Buffer.from("garbage not json"));
+
+  await waitFor(() => counts.some((c) => c.name === "comapeo.ipc.errors"), {
+    message: "ipc error metric recorded",
+  });
+  const err = counts.find((c) => c.name === "comapeo.ipc.errors");
+  assert.equal(err?.attributes.error_class, "SyntaxError");
+});

docs/sentry-integration.md

@@ -1391,7 +1391,7 @@ The diagnostic tier carries aggregate, low-cardinality operational signal;
 | --- | --- | --- |
 | RPC latency, aggregate (`rpc.{client,server}.duration_ms` with `status` + device tags) | Diagnostics | Latency by status and device bucket is pure performance; no per-operation detail. |
 | RPC `method` attribute on the same metrics (+ `rpc.client.send_ms{method}`) | applicationUsageData | The set and frequency of `@comapeo/core` methods a user invokes reveals what they do (create vs view vs sync) — usage behaviour, not perf. |
-| Boot / shutdown phase timings + outcome | Diagnostics | Startup/teardown performance; no user-specific content. *Shutdown emitter not yet wired ([#190](https://github.com/digidem/comapeo-core-react-native/issues/190)).* |
+| Boot / shutdown phase timings + outcome | Diagnostics | Startup/teardown performance; no user-specific content. |
 | Backend health gauges (memory, heap, uptime, event-loop delay) | Diagnostics | Process resource health; independent of user activity. |
 | Sync session duration + outcome | Diagnostics | Sync performance/reliability is core-function health; that a sync ran is inherent to a P2P app. *Not yet wired ([#80](https://github.com/digidem/comapeo-core-react-native/issues/80)).* |
 | Sync `peers_bucket` | applicationUsageData | How many devices a user syncs with is a proxy for their collaboration/social-graph size. *Not yet wired ([#80](https://github.com/digidem/comapeo-core-react-native/issues/80)).* |
@@ -1401,7 +1401,7 @@ The diagnostic tier carries aggregate, low-cardinality operational signal;
 | App-exit coarse buckets (`uptime_bucket`, `bg_duration_bucket`, OEM-kill flags) | Diagnostics | Aggregate stability signal ("which OEMs kill us"); low-resolution. |
 | App-exit exact-ms (`alive_for_ms`, `backgrounded_for_ms`) | applicationUsageData | Millisecond session/foreground durations are fine-grained usage-shape data. |
 | `device_class` / `os_major` / `platform` tags | Diagnostics | Low-cardinality device-capability buckets; not user-identifying. |
-| `ipc.errors`, `telemetry.forwarding_failures` | Diagnostics | Internal transport health. *Not yet wired ([#190](https://github.com/digidem/comapeo-core-react-native/issues/190)).* |
+| `ipc.errors`, `telemetry.forwarding_failures` | Diagnostics | Internal transport health. |
 | Per-RPC traces / OTel spans / `rpc.args` | `debug` (separate) | Investigation-only; behind the 72h auto-off `debug` toggle, not `applicationUsageData`. |
 
 #### Why restart-to-activate