From 1b04c71e045093a0f93bf630d1bdf4d4a3e252a3 Mon Sep 17 00:00:00 2001 From: Gregor MacLennan Date: Thu, 5 Nov 2026 03:12:34 +0000 Subject: [PATCH 1/2] test(backend): pin the server_disconnect sync metric on stop, disconnect and autostop --- backend/lib/sync-observer.test.mjs | 164 ++++++++++++++++++++++++++++- 1 file changed, 163 insertions(+), 1 deletion(-) diff --git a/backend/lib/sync-observer.test.mjs b/backend/lib/sync-observer.test.mjs index 4ed5ca43..ffa54f66 100644 --- a/backend/lib/sync-observer.test.mjs +++ b/backend/lib/sync-observer.test.mjs @@ -68,7 +68,7 @@ function initMetrics(sdk, applicationUsageData) { function fakeSyncApi(initialState) { const emitter = new EventEmitter(); let state = initialState; - return { + const api = { on: emitter.on.bind(emitter), off: emitter.off.bind(emitter), getState: () => state, @@ -76,7 +76,37 @@ function fakeSyncApi(initialState) { state = next; emitter.emit("sync-state", next); }, + // Like core, stop() flips data sync off and emits synchronously. + stop() { + api.setState({ ...state, data: { isSyncEnabled: false } }); + }, + disconnectServers() {}, }; + return api; +} + +const SERVER_ID = "fakeserverid"; + +/** State with one connected server peer (plus any other `peers`). */ +function serverState({ dataEnabled, want, peers = {} }) { + return syncState({ + dataEnabled, + devices: { + ...peers, + [SERVER_ID]: { + initial: { isSyncEnabled: true, want: 0, wanted: 0 }, + data: { isSyncEnabled: true, want, wanted: 0 }, + }, + }, + }); +} + +const getServerDeviceIds = async () => [SERVER_ID]; + +const tick = () => new Promise((resolve) => setImmediate(resolve)); + +function serverDisconnects(rec) { + return rec.counts.filter((c) => c.name === "comapeo.sync.server_disconnect"); } function deviceStates(n, { want = 0, wanted = 0 } = {}) { @@ -309,3 +339,135 @@ test("watcher seeded from getState picks up a session already running", async () await Sentry.close(); }); + +test("disconnectServers() while the server still wants blocks records one server_disconnect count", async () => { + initSentry({ ...baseArgv, applicationUsageData: true }); + const rec = recordingMetricsSdk(); + initMetrics(rec.sdk, true); + setSink(() => {}); + + const api = fakeSyncApi(idleState); + const detach = watchSyncApi(api, { getServerDeviceIds }); + // A phone peer with a bigger backlog sits alongside the server; only the + // server's outstanding blocks count. + api.setState( + serverState({ + dataEnabled: true, + want: 50, + peers: deviceStates(1, { want: 500 }), + }), + ); + await tick(); + api.disconnectServers(); + detach(); + + const events = serverDisconnects(rec); + assert.equal(events.length, 1, "expected exactly one server_disconnect"); + const { attributes } = events[0]; + assert.equal(attributes.reason, "disconnect"); + assert.equal(attributes.server_want_bucket, "11-100"); + assert.equal(attributes.data_enabled, true); + const serialized = JSON.stringify(attributes); + assert.ok( + !serialized.includes(SERVER_ID) && !serialized.includes("fakepeerid"), + "server_disconnect leaked a peer id", + ); + + await Sentry.close(); +}); + +test("stop() mid-upload records reason=stop once; the following disconnectServers() does not repeat it", async () => { + initSentry({ ...baseArgv, applicationUsageData: true }); + const rec = recordingMetricsSdk(); + initMetrics(rec.sdk, true); + setSink(() => {}); + + const api = fakeSyncApi(idleState); + const detach = watchSyncApi(api, { getServerDeviceIds }); + api.setState(serverState({ dataEnabled: true, want: 5 })); + await tick(); + api.stop(); + api.disconnectServers(); + + let events = serverDisconnects(rec); + assert.equal(events.length, 1, "expected exactly one server_disconnect"); + assert.equal(events[0].attributes.reason, "stop"); + assert.equal(events[0].attributes.server_want_bucket, "1-10"); + + // The server leaves and comes back: a later disconnect is recorded again. + api.setState(syncState({ dataEnabled: false, devices: {} })); + api.setState(serverState({ dataEnabled: false, want: 5 })); + api.disconnectServers(); + detach(); + + events = serverDisconnects(rec); + assert.equal(events.length, 2, "expected a second server_disconnect"); + assert.equal(events[1].attributes.reason, "disconnect"); + + await Sentry.close(); +}); + +test("no server peer connected: disconnectServers() and stop() record nothing", async () => { + initSentry({ ...baseArgv, applicationUsageData: true }); + const rec = recordingMetricsSdk(); + initMetrics(rec.sdk, true); + setSink(() => {}); + + const api = fakeSyncApi(idleState); + const detach = watchSyncApi(api, { getServerDeviceIds }); + api.setState( + syncState({ dataEnabled: true, devices: deviceStates(1, { want: 50 }) }), + ); + await tick(); + api.disconnectServers(); + api.stop(); + detach(); + + assert.equal(serverDisconnects(rec).length, 0); + + await Sentry.close(); +}); + +test("autostop with a server connected records reason=autostop and bucket 0", async () => { + initSentry({ ...baseArgv, applicationUsageData: true }); + const rec = recordingMetricsSdk(); + initMetrics(rec.sdk, true); + setSink(() => {}); + + const api = fakeSyncApi(idleState); + const detach = watchSyncApi(api, { getServerDeviceIds }); + api.setState(serverState({ dataEnabled: true, want: 0 })); + await tick(); + api.setState(serverState({ dataEnabled: false, want: 0 })); + detach(); + + const events = serverDisconnects(rec); + assert.equal(events.length, 1, "expected exactly one server_disconnect"); + assert.equal(events[0].attributes.reason, "autostop"); + assert.equal(events[0].attributes.server_want_bucket, "0"); + assert.equal(events[0].attributes.data_enabled, true); + + await Sentry.close(); +}); + +test("usage OFF: server_disconnect still records, without the want bucket", async () => { + initSentry({ ...baseArgv, applicationUsageData: false }); + const rec = recordingMetricsSdk(); + initMetrics(rec.sdk, false); + setSink(() => {}); + + const api = fakeSyncApi(idleState); + const detach = watchSyncApi(api, { getServerDeviceIds }); + api.setState(serverState({ dataEnabled: false, want: 20 })); + await tick(); + api.disconnectServers(); + detach(); + + const events = serverDisconnects(rec); + assert.equal(events.length, 1, "expected exactly one server_disconnect"); + assert.equal(events[0].attributes.reason, "disconnect"); + assert.equal(events[0].attributes.data_enabled, false); + assert.equal(events[0].attributes.server_want_bucket, undefined); + + await Sentry.close(); +}); From 844d9895cebba7e63a59fc3c0eb758a2d10e2f2f Mon Sep 17 00:00:00 2001 From: Gregor MacLennan Date: Thu, 5 Nov 2026 03:14:35 +0000 Subject: [PATCH 2/2] feat(backend): record server_disconnect metric when a server peer is left mid-sync --- backend/lib/metrics.js | 25 ++++++++ backend/lib/sync-observer.js | 113 ++++++++++++++++++++++++++++++++++- 2 files changed, 135 insertions(+), 3 deletions(-) diff --git a/backend/lib/metrics.js b/backend/lib/metrics.js index 983dc285..59025081 100644 --- a/backend/lib/metrics.js +++ b/backend/lib/metrics.js @@ -208,6 +208,22 @@ export function syncSession(outcome, ms, peersBucket, bytesBucket) { } } +/** + * The phone stopped serving a connected server peer (`disconnectServers()`, + * `stop()` or autostop). `server_want_bucket` is a data-volume bucket, so it + * is usage-gated like `bytes_bucket`. + * + * @param {"disconnect" | "stop" | "autostop"} reason + * @param {string} serverWantBucket + * @param {boolean} dataEnabled + */ +export function syncServerDisconnect(reason, serverWantBucket, dataEnabled) { + /** @type {Record} */ + const attrs = { reason, data_enabled: dataEnabled }; + if (config?.applicationUsageData) attrs.server_want_bucket = serverWantBucket; + count("comapeo.sync.server_disconnect", attrs); +} + // ── Backend health (60s sampler) ──────────────────────────────── /** @@ -267,6 +283,15 @@ export function peersBucket(peers) { return "10+"; } +/** @param {number} blocks @returns {string} */ +export function serverWantBucket(blocks) { + if (blocks <= 0) return "0"; + if (blocks <= 10) return "1-10"; + if (blocks <= 100) return "11-100"; + if (blocks <= 1000) return "101-1000"; + return "1000+"; +} + /** @param {number} bytes @returns {string} */ export function bytesBucket(bytes) { if (bytes < 1_000_000) return "<1M"; diff --git a/backend/lib/sync-observer.js b/backend/lib/sync-observer.js index 0725adfc..69a01cd7 100644 --- a/backend/lib/sync-observer.js +++ b/backend/lib/sync-observer.js @@ -27,7 +27,12 @@ import * as sentry from "./sentry.js"; * getState: () => SyncApiState, * on: (event: "sync-state", listener: (state: SyncApiState) => void) => unknown, * off: (event: "sync-state", listener: (state: SyncApiState) => void) => unknown, + * stop: () => void, + * disconnectServers: () => void, * }} SyncApiLike + * @typedef {{ + * getServerDeviceIds?: () => Promise, + * }} WatchOptions */ // The sync API exposes remaining block counts, not byte counters, so a @@ -52,12 +57,22 @@ export function observeSyncSessions(manager) { const project = await getProject(projectPublicId); if (!observed.has(project.$sync)) { observed.add(project.$sync); - watchSyncApi(project.$sync); + watchSyncApi(project.$sync, { + getServerDeviceIds: () => getServerDeviceIds(project.$member), + }); } return project; }; } +/** @param {import("@comapeo/core").MapeoProject["$member"]} memberApi */ +async function getServerDeviceIds(memberApi) { + const members = await memberApi.getMany(); + return members + .filter((m) => m.deviceType === "selfHostedServer") + .map((m) => m.deviceId); +} + /** * Track sync sessions on one project's `$sync`. Session start is the * `data.isSyncEnabled` false→true edge; end is the true→false edge @@ -66,9 +81,15 @@ export function observeSyncSessions(manager) { * is not derivable — the noise/protomux handshake is internal to core. * Returns a detach function (used by tests). * + * Also wraps `stop()` / `disconnectServers()` so that when the phone stops + * serving a connected server peer (either call, or the autostop edge) one + * `comapeo.sync.server_disconnect` metric records how much the server still + * wanted — once per data-sync session on a server connection. + * * @param {SyncApiLike} syncApi + * @param {WatchOptions} [options] */ -export function watchSyncApi(syncApi) { +export function watchSyncApi(syncApi, { getServerDeviceIds } = {}) { /** * @type {{ * startedAt: number, @@ -80,11 +101,73 @@ export function watchSyncApi(syncApi) { */ let session = null; + // The wrappers run synchronously but membership is async, so cache the + // server ids and refresh whenever a peer we have not seen before appears. + /** @type {Set} */ + let serverIds = new Set(); + /** @type {Set} */ + const seenPeers = new Set(); + let serverReported = false; + + const refreshServerIds = () => { + getServerDeviceIds?.().then( + (ids) => { + serverIds = new Set(ids); + }, + () => {}, + ); + }; + + /** + * Blocks the connected server peers still want from us, or `null` when + * no server peer is connected. + * + * @param {SyncApiState} state + */ + const serverWant = (state) => { + /** @type {number | null} */ + let want = null; + for (const [peerId, peer] of Object.entries(state.remoteDeviceSyncState)) { + if (serverIds.has(peerId)) want = (want ?? 0) + peer.data.want; + } + return want; + }; + + /** + * @param {"disconnect" | "stop" | "autostop"} reason + * @param {SyncApiState} state + * @param {boolean} [dataEnabled] + */ + const reportServerDisconnect = ( + reason, + state, + dataEnabled = state.data.isSyncEnabled, + ) => { + if (serverReported) return; + const want = serverWant(state); + if (want === null) return; + serverReported = true; + metrics.syncServerDisconnect( + reason, + metrics.serverWantBucket(want), + dataEnabled, + ); + }; + /** @param {SyncApiState} state */ const onSyncState = (state) => { - const peers = Object.keys(state.remoteDeviceSyncState).length; + const peerIds = Object.keys(state.remoteDeviceSyncState); + const peers = peerIds.length; + if (peerIds.some((id) => !seenPeers.has(id))) { + for (const id of peerIds) seenPeers.add(id); + refreshServerIds(); + } + if (serverWant(state) === null) serverReported = false; if (!session) { if (!state.data.isSyncEnabled) return; + serverReported = false; + // A server's deviceType may not have synced when it first appeared. + refreshServerIds(); session = { startedAt: performance.now(), maxPeers: 0, @@ -105,6 +188,8 @@ export function watchSyncApi(syncApi) { } } if (!state.data.isSyncEnabled) { + // Any edge not already reported by the stop() wrapper is the autostop. + reportServerDisconnect("autostop", state, true); const outcome = session.synced ? "completed" : "stopped"; const durationMs = performance.now() - session.startedAt; const peersBucket = metrics.peersBucket(session.maxPeers); @@ -123,12 +208,34 @@ export function watchSyncApi(syncApi) { } }; + // Report before delegating: core's stop() emits the data-disabled edge + // synchronously, which would otherwise read as an autostop. + const { stop, disconnectServers } = syncApi; + /** @param {"disconnect" | "stop"} reason */ + const reportBeforeCall = (reason) => { + try { + reportServerDisconnect(reason, syncApi.getState()); + } catch { + // Telemetry must never block the real call. + } + }; + syncApi.stop = () => { + reportBeforeCall("stop"); + return stop.call(syncApi); + }; + syncApi.disconnectServers = () => { + reportBeforeCall("disconnect"); + return disconnectServers.call(syncApi); + }; + syncApi.on("sync-state", onSyncState); // Seed from the current state in case data sync was already running // when the watcher attached. onSyncState(syncApi.getState()); return () => { syncApi.off("sync-state", onSyncState); + syncApi.stop = stop; + syncApi.disconnectServers = disconnectServers; }; }