feat(backend): record server_disconnect metric when a server peer is left mid-sync

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<string, string | boolean>} */
+  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";

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<string[]>,
+ * }} 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<string>} */
+  let serverIds = new Set();
+  /** @type {Set<string>} */
+  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;
   };
 }
 

backend/lib/sync-observer.test.mjs

@@ -68,15 +68,45 @@ 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,
     setState(next) {
       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();
+});