From 1b2a8286071b49577538372550f183c3326b87eb Mon Sep 17 00:00:00 2001 From: Gregor MacLennan Date: Tue, 7 Jul 2026 21:40:39 +0100 Subject: [PATCH 1/3] feat(sentry): wire the diagnostic metric emitters for shutdown, IPC errors, and telemetry failures Closes #190 --- backend/index.js | 28 +++++++++---- backend/lib/comapeo-rpc.js | 2 + backend/lib/comapeo-rpc.test.mjs | 35 ++++++++++++++++- backend/lib/sentry.js | 23 +++++++++-- backend/lib/sentry.test.mjs | 67 ++++++++++++++++++++++++++++++-- backend/lib/simple-rpc.js | 3 ++ backend/lib/simple-rpc.test.mjs | 29 ++++++++++++++ docs/sentry-integration.md | 4 +- 8 files changed, 174 insertions(+), 17 deletions(-) diff --git a/backend/index.js b/backend/index.js index de72846a..abcd5559 100644 --- a/backend/index.js +++ b/backend/index.js @@ -126,13 +126,27 @@ 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())] : []), + ]), + ); + const manager = comapeoManager; + if (manager) { + await sentry.withSpan("shutdown.close-manager", () => + settle("manager", manager.close()), + ); + } + const maps = mapServer; + if (maps) { + await sentry.withSpan("shutdown.close-map-server", () => + settle("map-server", maps.close()), + ); + } }, /** * Android-only attribution channel: FGS-local failures (rootkey load, diff --git a/backend/lib/comapeo-rpc.js b/backend/lib/comapeo-rpc.js index 286b1d8d..4d86546d 100644 --- a/backend/lib/comapeo-rpc.js +++ b/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", () => { diff --git a/backend/lib/comapeo-rpc.test.mjs b/backend/lib/comapeo-rpc.test.mjs index fb77efee..4d8a5ce1 100644 --- a/backend/lib/comapeo-rpc.test.mjs +++ b/backend/lib/comapeo-rpc.test.mjs @@ -1,5 +1,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; +import { Buffer } from "node:buffer"; +import FramedStream from "framed-stream"; import { createComapeoServicesClient, closeComapeoServicesClient, @@ -7,7 +9,8 @@ import { 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 }>} */ + 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"); +}); diff --git a/backend/lib/sentry.js b/backend/lib/sentry.js index 05e26490..9ceab411 100644 --- a/backend/lib/sentry.js +++ b/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} */ 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; } } diff --git a/backend/lib/sentry.test.mjs b/backend/lib/sentry.test.mjs index c6f9a0ee..4da8687f 100644 --- a/backend/lib/sentry.test.mjs +++ b/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" }); diff --git a/backend/lib/simple-rpc.js b/backend/lib/simple-rpc.js index 1db15589..1d775768 100644 --- a/backend/lib/simple-rpc.js +++ b/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); } } } diff --git a/backend/lib/simple-rpc.test.mjs b/backend/lib/simple-rpc.test.mjs index 235c63e5..7afd2f24 100644 --- a/backend/lib/simple-rpc.test.mjs +++ b/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 }>} */ + 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"); +}); diff --git a/docs/sentry-integration.md b/docs/sentry-integration.md index 10fb7fec..a6226873 100644 --- a/docs/sentry-integration.md +++ b/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 From 396c66b8d5176dc22e79cd1373c9e3e131d7add7 Mon Sep 17 00:00:00 2001 From: Gregor MacLennan Date: Wed, 8 Jul 2026 11:45:51 +0100 Subject: [PATCH 2/3] remove unnecessary assignment --- backend/index.js | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/backend/index.js b/backend/index.js index abcd5559..af90542c 100644 --- a/backend/index.js +++ b/backend/index.js @@ -135,10 +135,9 @@ const controlIpcServer = new SimpleRpcServer({ ...(comapeoRpcServer ? [settle("rpc", comapeoRpcServer.close())] : []), ]), ); - const manager = comapeoManager; - if (manager) { + if (comapeoManager) { await sentry.withSpan("shutdown.close-manager", () => - settle("manager", manager.close()), + settle("manager", comapeoManager.close()), ); } const maps = mapServer; From f221dcce5764153404d2dc66a50dc17dcff31682 Mon Sep 17 00:00:00 2001 From: Gregor MacLennan Date: Wed, 8 Jul 2026 11:46:16 +0100 Subject: [PATCH 3/3] remove unnecessary assignment --- backend/index.js | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/backend/index.js b/backend/index.js index af90542c..8cb519e7 100644 --- a/backend/index.js +++ b/backend/index.js @@ -140,10 +140,9 @@ const controlIpcServer = new SimpleRpcServer({ settle("manager", comapeoManager.close()), ); } - const maps = mapServer; - if (maps) { + if (mapServer) { await sentry.withSpan("shutdown.close-map-server", () => - settle("map-server", maps.close()), + settle("map-server", mapServer.close()), ); } },