sync-coordinator.ts

1// Vault sync — Phase 2 manifest-diff coordinator.2//3// After reconnect Noise KK is open and PAIR_IDENTITY has been exchanged4// (`shared/sync-transport.ts`), this coordinator drives the manifest exchange:5//6//   ->  SYNC_HELLO { vaultId, deviceId, deviceManifestVersion? }7//   <-  SYNC_HELLO { vaultId, deviceId, deviceManifestVersion? }8//                                                (mismatch → SYNC_ERROR)9//   <-> SYNC_MANIFEST { entries, tombstones }  (each side sends one)10//   ->  SYNC_REQUEST { entryIds }              (entries newer on the peer)11//   <-  SYNC_ENTRIES { entries }12//   <-  SYNC_REQUEST { entryIds }              (entries newer on us)13//   ->  SYNC_ENTRIES { entries }14//   <-> SYNC_TOMBSTONES { tombstones }15//   <-> SYNC_DEVICE_MANIFEST                    (also runs on equal versions)16//   <-> SYNC_DONE17//18// Entries are opaque base64 ciphertext — the coordinator never decrypts.19// Last-write-wins at the entry level by mtime; tombstones override updates20// with a later deletion timestamp.2122import {23  DEVICE_MANIFEST_VERSION,24  SYNC_PROTOCOL_VERSION,25  type SyncMessage,26  type SignedDeviceManifest,27} from "../shared/sync-types";28import type { SyncSession } from "../shared/sync-transport";29import { SYNC_LIMITS, validateSyncManifestCounts } from "../shared/sync-limits";30import { isConnectionClosedError } from "../shared/sync-outcome";3132const SESSION_TIMEOUT_MS = 30_000;3334type SyncStage =35  | "preparing_hello"36  | "waiting_for_device_hello"37  | "building_local_manifest"38  | "waiting_for_device_manifest"39  | "waiting_for_device_request"40  | "building_requested_entries"41  | "waiting_for_device_entries"42  | "waiting_for_device_tombstones"43  | "building_device_manifest"44  | "waiting_for_device_graph"45  | "waiting_for_device_done";4647/**48 * Vault-side adapter — the coordinator's dependency on the vault layer.49 * The vault holds the canonical entry/tombstone tables; the coordinator50 * pulls a manifest, requests entries by id, applies remote entries, and51 * applies remote tombstones.52 */53export interface SyncVaultAdapter {54  computeVaultId(): Promise<string>;55  getOurDeviceId(): Promise<string>;56  /** Returns the vault's monotonic version (`peach_last_modified_at`). */57  getLocalVersion(): Promise<number>;58  /** Returns the per-entry mtimes for our active entries. */59  buildEntryManifest(): Promise<Record<string, number>>;60  /** Returns the per-entry deletion timestamps for our tombstones. */61  buildTombstoneManifest(): Promise<Record<string, number>>;62  /** Returns base64 ciphertext for each requested entry id. */63  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;64  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */65  applyEntries(66    entries: Record<string, string>,67    remoteManifest?: Record<string, number>,68  ): Promise<void>;69  /** Apply incoming tombstones; vault retains the latest deletion ts. */70  applyTombstones(tombstones: Record<string, number>): Promise<void>;71  /**72   * Optional — v2 brief §4.3 device manifest exchange. The extension (as73   * the trust anchor for the device graph) returns a signed manifest74   * enumerating every paired device it knows about; the peer verifies it75   * against the extension's known Ed25519 signing pubkey and upserts any76   * newly-vouched devices into its own registry.77   *78   * When this method is undefined, the coordinator skips the manifest79   * step entirely. Both ends must opt in for the exchange to occur.80   */81  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;82  /**83   * Handle a manifest received from the peer. The transport-authenticated84   * Noise static key is supplied separately so a self-asserted issuer id85   * cannot select a different paired-device verification key.86   */87  applyIncomingDeviceManifest?(88    manifest: SignedDeviceManifest,89    authenticatedRemoteStaticPublicKey: Uint8Array,90  ): Promise<void>;91}9293export interface SyncOutcome {94  status: "completed" | "vault_mismatch" | "errored" | "timeout";95  entriesReceived: number;96  entriesSent: number;97  tombstonesReceived: number;98  tombstonesSent: number;99  error?: string;100}101102/**103 * Run one sync round as the lower-key initiator. Device type does not decide104 * the role; the higher-key peer runs [runSyncResponder].105 */106export async function runSyncInitiator(107  session: SyncSession,108  vault: SyncVaultAdapter,109): Promise<SyncOutcome> {110  const result: SyncOutcome = {111    status: "errored",112    entriesReceived: 0,113    entriesSent: 0,114    tombstonesReceived: 0,115    tombstonesSent: 0,116  };117118  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);119  // Privacy-safe progress only: stages contain no device identifiers, relay120  // capabilities, keys, entry ids/counts, or vault content. Keeping the last121  // stage turns an otherwise opaque 30-second timeout into an actionable122  // cross-device diagnostic.123  let stage: SyncStage = "preparing_hello";124  try {125    const ourVaultId = await vault.computeVaultId();126    const ourDeviceId = await vault.getOurDeviceId();127    const localVersion = await vault.getLocalVersion();128    const localTime = Date.now();129    const supportsDeviceManifest =130      vault.buildOutgoingDeviceManifest !== undefined &&131      vault.applyIncomingDeviceManifest !== undefined;132133    await sendMessage(session, {134      type: "SYNC_HELLO",135      vaultId: ourVaultId,136      deviceId: ourDeviceId,137      version: localVersion,138      deviceTime: localTime,139      protocolVersion: SYNC_PROTOCOL_VERSION,140      deviceManifestVersion: supportsDeviceManifest141        ? DEVICE_MANIFEST_VERSION142        : undefined,143    });144    stage = "waiting_for_device_hello";145    const helloIn = await receiveMessage(session, timer);146    // The responder rejects an incompatible HELLO before sending its own, so an147    // initiator can receive that SYNC_ERROR first. Surface it as the same148    // repair-required outcome instead of a generic "expected SYNC_HELLO".149    if (helloIn.type === "SYNC_ERROR" && helloIn.error.includes("PEER_PROTOCOL_OUTDATED")) {150      return errorOutcome(result, "PEER_PROTOCOL_OUTDATED");151    }152    if (helloIn.type !== "SYNC_HELLO") {153      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);154    }155    // Exact equality: the breaking migration has one supported version.156    if ((helloIn.protocolVersion ?? 0) !== SYNC_PROTOCOL_VERSION) {157      await sendMessage(session, { type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" });158      return errorOutcome(result, "PEER_PROTOCOL_OUTDATED");159    }160    if (helloIn.vaultId !== ourVaultId) {161      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });162      result.status = "vault_mismatch";163      return result;164    }165166    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {167      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);168    }169170    const exchangeDeviceManifest =171      supportsDeviceManifest &&172      helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION;173174    stage = "building_local_manifest";175    const ourEntries = await vault.buildEntryManifest();176    const ourTombstones = await vault.buildTombstoneManifest();177    await sendMessage(session, {178      type: "SYNC_MANIFEST",179      entries: ourEntries,180      tombstones: ourTombstones,181    });182    stage = "waiting_for_device_manifest";183    const manifestIn = await receiveMessage(session, timer);184    if (manifestIn.type !== "SYNC_MANIFEST") {185      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);186    }187    validateSyncManifestCounts(manifestIn.entries, manifestIn.tombstones);188189    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);190    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });191192    stage = "waiting_for_device_request";193    const requestIn = await receiveMessage(session, timer);194    if (requestIn.type !== "SYNC_REQUEST") {195      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);196    }197    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {198      return errorOutcome(result, "sync request entry count exceeds limit");199    }200    // Sanity: the peer's request must be a subset of what its manifest201    // declared as newer. A malformed/over-eager request is a protocol202    // violation but not a data-safety problem — we just send what they ask.203204    stage = "building_requested_entries";205    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);206    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });207    result.entriesSent = Object.keys(ourPayload).length;208209    if (wantFromPeer.length > 0) {210      stage = "waiting_for_device_entries";211      const entriesIn = await receiveMessage(session, timer);212      if (entriesIn.type !== "SYNC_ENTRIES") {213        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);214      }215      await vault.applyEntries(entriesIn.entries, manifestIn.entries);216      result.entriesReceived = Object.keys(entriesIn.entries).length;217    }218219    // Tombstones: each side ships the tombstones that the other does not220    // yet have a newer record for.221    const newerTombstonesForPeer: Record<string, number> = {};222    for (const [id, ts] of Object.entries(ourTombstones)) {223      const peerTs = manifestIn.tombstones[id] ?? 0;224      if (ts > peerTs) newerTombstonesForPeer[id] = ts;225    }226    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });227    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;228229    stage = "waiting_for_device_tombstones";230    const tombstonesIn = await receiveMessage(session, timer);231    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {232      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);233    }234    await vault.applyTombstones(tombstonesIn.tombstones);235    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;236237    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends238    // must expose the adapter methods. Disabling on either side cleanly239    // bypasses the step. The exchange is symmetric on the wire; the240    // production adapter accepts only a directly paired issuer whose stored241    // long-term key matches the Noise-authenticated remote static key.242    if (exchangeDeviceManifest) {243      stage = "building_device_manifest";244      const ours = await vault.buildOutgoingDeviceManifest!();245      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });246      stage = "waiting_for_device_graph";247      const manifestIn = await receiveMessage(session, timer);248      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {249        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);250      }251      if (manifestIn.manifest) {252        await vault.applyIncomingDeviceManifest!(253          manifestIn.manifest,254          session.remoteStaticPublicKey,255        );256      }257    }258259    await sendMessage(session, {260      type: "SYNC_DONE",261      entriesReceived: result.entriesReceived,262      entriesSent: result.entriesSent,263    });264    stage = "waiting_for_device_done";265    const closing = await awaitClosingDone(session, timer);266    if (closing.error) return errorOutcome(result, closing.error);267    result.status = "completed";268    return result;269  } catch (err) {270    if ((err as Error).message === "sync_timeout") {271      result.status = "timeout";272      result.error = `sync_timeout:${stage}`;273      return result;274    }275    return errorOutcome(result, (err as Error).message);276  } finally {277    // Session lifetime is owned by the runner (sync-runner.ts), which278    // closes it in its own finally block. The coordinator must not close it279    // here — the runner still needs to record the outcome after we return.280    timer.cancel();281  }282}283284/**285 * Exact complement of [runSyncInitiator] for the higher-key peer holding a286 * relay waiter room. Keeping this in the browser lets two extensions form a287 * direct mesh edge; the device that introduced them is not in the data path.288 */289export async function runSyncResponder(290  session: SyncSession,291  vault: SyncVaultAdapter,292): Promise<SyncOutcome> {293  const result: SyncOutcome = {294    status: "errored",295    entriesReceived: 0,296    entriesSent: 0,297    tombstonesReceived: 0,298    tombstonesSent: 0,299  };300  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);301  let stage: SyncStage = "waiting_for_device_hello";302  try {303    const ourVaultId = await vault.computeVaultId();304    const ourDeviceId = await vault.getOurDeviceId();305    const localVersion = await vault.getLocalVersion();306    const localTime = Date.now();307    const supportsDeviceManifest =308      vault.buildOutgoingDeviceManifest !== undefined &&309      vault.applyIncomingDeviceManifest !== undefined;310311    const helloIn = await receiveMessage(session, timer);312    if (helloIn.type !== "SYNC_HELLO") {313      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);314    }315    // Exact equality: a lower or higher peer version is repair-required.316    if ((helloIn.protocolVersion ?? 0) !== SYNC_PROTOCOL_VERSION) {317      await sendMessage(session, { type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" });318      return errorOutcome(result, "PEER_PROTOCOL_OUTDATED");319    }320    await sendMessage(session, {321      type: "SYNC_HELLO",322      vaultId: ourVaultId,323      deviceId: ourDeviceId,324      version: localVersion,325      deviceTime: localTime,326      protocolVersion: SYNC_PROTOCOL_VERSION,327      deviceManifestVersion: supportsDeviceManifest328        ? DEVICE_MANIFEST_VERSION329        : undefined,330    });331    if (helloIn.vaultId !== ourVaultId) {332      result.status = "vault_mismatch";333      return result;334    }335    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {336      console.warn(337        `[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`,338      );339    }340341    const exchangeDeviceManifest =342      supportsDeviceManifest &&343      helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION;344345    const ourEntries = await vault.buildEntryManifest();346    const ourTombstones = await vault.buildTombstoneManifest();347    stage = "waiting_for_device_manifest";348    const manifestIn = await receiveMessage(session, timer);349    if (manifestIn.type !== "SYNC_MANIFEST") {350      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);351    }352    validateSyncManifestCounts(manifestIn.entries, manifestIn.tombstones);353    await sendMessage(session, {354      type: "SYNC_MANIFEST",355      entries: ourEntries,356      tombstones: ourTombstones,357    });358359    stage = "waiting_for_device_request";360    const requestIn = await receiveMessage(session, timer);361    if (requestIn.type !== "SYNC_REQUEST") {362      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);363    }364    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {365      return errorOutcome(result, "sync request entry count exceeds limit");366    }367    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);368    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });369370    stage = "waiting_for_device_entries";371    const entriesIn = await receiveMessage(session, timer);372    if (entriesIn.type !== "SYNC_ENTRIES") {373      return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);374    }375    if (wantFromPeer.length > 0) {376      await vault.applyEntries(entriesIn.entries, manifestIn.entries);377      result.entriesReceived = Object.keys(entriesIn.entries).length;378    }379    if (requestIn.entryIds.length > 0) {380      const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);381      await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });382      result.entriesSent = Object.keys(ourPayload).length;383    }384385    stage = "waiting_for_device_tombstones";386    const tombstonesIn = await receiveMessage(session, timer);387    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {388      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);389    }390    await vault.applyTombstones(tombstonesIn.tombstones);391    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;392393    const newerTombstonesForPeer: Record<string, number> = {};394    for (const [id, timestamp] of Object.entries(ourTombstones)) {395      if (timestamp > (manifestIn.tombstones[id] ?? 0)) {396        newerTombstonesForPeer[id] = timestamp;397      }398    }399    await sendMessage(session, {400      type: "SYNC_TOMBSTONES",401      tombstones: newerTombstonesForPeer,402    });403    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;404405    if (exchangeDeviceManifest) {406      stage = "waiting_for_device_graph";407      const graphIn = await receiveMessage(session, timer);408      if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {409        return errorOutcome(410          result,411          `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,412        );413      }414      const ours = await vault.buildOutgoingDeviceManifest!();415      await sendMessage(session, {416        type: "SYNC_DEVICE_MANIFEST",417        manifest: ours,418      });419      if (graphIn.manifest) {420        await vault.applyIncomingDeviceManifest!(421          graphIn.manifest,422          session.remoteStaticPublicKey,423        );424      }425    }426427    stage = "waiting_for_device_done";428    const done = await receiveMessage(session, timer);429    if (done.type !== "SYNC_DONE") {430      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);431    }432    await sendMessage(session, {433      type: "SYNC_DONE",434      entriesReceived: result.entriesReceived,435      entriesSent: result.entriesSent,436    });437    result.status = "completed";438    return result;439  } catch (error) {440    if ((error as Error).message === "sync_timeout") {441      result.status = "timeout";442      result.error = `sync_timeout:${stage}`;443      return result;444    }445    return errorOutcome(result, (error as Error).message);446  } finally {447    timer.cancel();448  }449}450451/**452 * Wait for the peer's closing SYNC_DONE echo, treating a hang-up as success.453 *454 * By the time we send our own SYNC_DONE, every entry and tombstone has already455 * been exchanged and applied on both sides — the echo is a courtesy, not a456 * commit. Android closes the socket as soon as it has our DONE, so the parked457 * receive() rejects with "WebSocket closed"; counting that as `run_failed`458 * reported a red indicator for syncs that had fully succeeded and whose data459 * had demonstrably landed on the peer.460 *461 * A wrong *message* here is still a real protocol error — only a closed462 * transport is forgiven.463 */464async function awaitClosingDone(465  session: SyncSession,466  timer: Timer,467): Promise<{ error?: string }> {468  try {469    const done = await receiveMessage(session, timer);470    if (done.type !== "SYNC_DONE") {471      return { error: `expected SYNC_DONE, got ${done.type}` };472    }473    return {};474  } catch (err) {475    if ((err as Error).message === "sync_timeout") throw err;476    if (isConnectionClosedError(err)) return {};477    return { error: (err as Error).message };478  }479}480481function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {482  const want: string[] = [];483  for (const [id, ts] of Object.entries(remote)) {484    const ours = local[id];485    if (ours === undefined || ts > ours) want.push(id);486  }487  return want;488}489490interface Timer {491  signal: AbortSignal;492  cancel(): void;493}494495function scheduleTimeout(ms: number): Timer {496  const ctrl = new AbortController();497  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);498  return {499    signal: ctrl.signal,500    cancel: () => clearTimeout(t),501  };502}503504async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {505  session.send(new TextEncoder().encode(JSON.stringify(msg)));506}507508async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {509  const pending = session.receive();510  const aborted = new Promise<never>((_, reject) => {511    if (timer.signal.aborted) reject(new Error("sync_timeout"));512    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));513  });514  const bytes = await Promise.race([pending, aborted]);515  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS516  // transport may not have enforced anything) — re-check before JSON.parse517  // materialises a multi-GB string/object graph.518  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {519    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);520  }521  const parsed = JSON.parse(new TextDecoder().decode(bytes));522  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");523  return parsed as SyncMessage;524}525526function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {527  outcome.status = "errored";528  outcome.error = error;529  return outcome;530}