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  type SyncMessage,25  type SignedDeviceManifest,26} from "../shared/sync-types";27import type { SyncSession } from "../shared/sync-transport";28import { SYNC_LIMITS, validateSyncManifestCounts } from "../shared/sync-limits";29import { isConnectionClosedError } from "../shared/sync-outcome";3031const SESSION_TIMEOUT_MS = 30_000;3233type SyncStage =34  | "preparing_hello"35  | "waiting_for_device_hello"36  | "building_local_manifest"37  | "waiting_for_device_manifest"38  | "waiting_for_device_request"39  | "building_requested_entries"40  | "waiting_for_device_entries"41  | "waiting_for_device_tombstones"42  | "building_device_manifest"43  | "waiting_for_device_graph"44  | "waiting_for_device_done";4546/**47 * Vault-side adapter — the coordinator's dependency on the vault layer.48 * The vault holds the canonical entry/tombstone tables; the coordinator49 * pulls a manifest, requests entries by id, applies remote entries, and50 * applies remote tombstones.51 */52export interface SyncVaultAdapter {53  computeVaultId(): Promise<string>;54  getOurDeviceId(): Promise<string>;55  /** Returns the vault's monotonic version (`peach_last_modified_at`). */56  getLocalVersion(): Promise<number>;57  /** Returns the per-entry mtimes for our active entries. */58  buildEntryManifest(): Promise<Record<string, number>>;59  /** Returns the per-entry deletion timestamps for our tombstones. */60  buildTombstoneManifest(): Promise<Record<string, number>>;61  /** Returns base64 ciphertext for each requested entry id. */62  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;63  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */64  applyEntries(65    entries: Record<string, string>,66    remoteManifest?: Record<string, number>,67  ): Promise<void>;68  /** Apply incoming tombstones; vault retains the latest deletion ts. */69  applyTombstones(tombstones: Record<string, number>): Promise<void>;70  /**71   * Optional — v2 brief §4.3 device manifest exchange. The extension (as72   * the trust anchor for the device graph) returns a signed manifest73   * enumerating every paired device it knows about; the peer verifies it74   * against the extension's known Ed25519 signing pubkey and upserts any75   * newly-vouched devices into its own registry.76   *77   * When this method is undefined, the coordinator skips the manifest78   * step entirely. Both ends must opt in for the exchange to occur.79   */80  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;81  /**82   * Handle a manifest received from the peer. The transport-authenticated83   * Noise static key is supplied separately so a self-asserted issuer id84   * cannot select a different paired-device verification key.85   */86  applyIncomingDeviceManifest?(87    manifest: SignedDeviceManifest,88    authenticatedRemoteStaticPublicKey: Uint8Array,89  ): Promise<void>;90}9192export interface SyncOutcome {93  status: "completed" | "vault_mismatch" | "errored" | "timeout";94  entriesReceived: number;95  entriesSent: number;96  tombstonesReceived: number;97  tombstonesSent: number;98  error?: string;99}100101/**102 * Run one sync round as the lower-key initiator. Device type does not decide103 * the role; the higher-key peer runs [runSyncResponder].104 */105export async function runSyncInitiator(106  session: SyncSession,107  vault: SyncVaultAdapter,108): Promise<SyncOutcome> {109  const result: SyncOutcome = {110    status: "errored",111    entriesReceived: 0,112    entriesSent: 0,113    tombstonesReceived: 0,114    tombstonesSent: 0,115  };116117  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);118  // Privacy-safe progress only: stages contain no device identifiers, relay119  // capabilities, keys, entry ids/counts, or vault content. Keeping the last120  // stage turns an otherwise opaque 30-second timeout into an actionable121  // cross-device diagnostic.122  let stage: SyncStage = "preparing_hello";123  try {124    const ourVaultId = await vault.computeVaultId();125    const ourDeviceId = await vault.getOurDeviceId();126    const localVersion = await vault.getLocalVersion();127    const localTime = Date.now();128    const supportsDeviceManifest =129      vault.buildOutgoingDeviceManifest !== undefined &&130      vault.applyIncomingDeviceManifest !== undefined;131132    await sendMessage(session, {133      type: "SYNC_HELLO",134      vaultId: ourVaultId,135      deviceId: ourDeviceId,136      version: localVersion,137      deviceTime: localTime,138      deviceManifestVersion: supportsDeviceManifest139        ? DEVICE_MANIFEST_VERSION140        : undefined,141    });142    stage = "waiting_for_device_hello";143    const helloIn = await receiveMessage(session, timer);144    if (helloIn.type !== "SYNC_HELLO") {145      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);146    }147    if (helloIn.vaultId !== ourVaultId) {148      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });149      result.status = "vault_mismatch";150      return result;151    }152153    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {154      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);155    }156157    // Equal entry versions do not imply equal device graphs: paired-device158    // introductions and revocations live outside the vault version counter.159    // Exchange the signed graph when both peers advertise support, while an160    // older peer that omits the capability retains the legacy DONE fast path.161    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {162      if (163        supportsDeviceManifest &&164        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION165      ) {166        stage = "building_device_manifest";167        const ours = await vault.buildOutgoingDeviceManifest!();168        await sendMessage(session, {169          type: "SYNC_DEVICE_MANIFEST",170          manifest: ours,171        });172        stage = "waiting_for_device_graph";173        const graphIn = await receiveMessage(session, timer);174        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {175          return errorOutcome(176            result,177            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,178          );179        }180        if (graphIn.manifest) {181          await vault.applyIncomingDeviceManifest!(182            graphIn.manifest,183            session.remoteStaticPublicKey,184          );185        }186      }187      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });188      stage = "waiting_for_device_done";189      const closing = await awaitClosingDone(session, timer);190      if (closing.error) return errorOutcome(result, closing.error);191      result.status = "completed";192      return result;193    }194195    stage = "building_local_manifest";196    const ourEntries = await vault.buildEntryManifest();197    const ourTombstones = await vault.buildTombstoneManifest();198    await sendMessage(session, {199      type: "SYNC_MANIFEST",200      entries: ourEntries,201      tombstones: ourTombstones,202    });203    stage = "waiting_for_device_manifest";204    const manifestIn = await receiveMessage(session, timer);205    if (manifestIn.type !== "SYNC_MANIFEST") {206      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);207    }208    validateSyncManifestCounts(manifestIn.entries, manifestIn.tombstones);209210    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);211    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });212213    stage = "waiting_for_device_request";214    const requestIn = await receiveMessage(session, timer);215    if (requestIn.type !== "SYNC_REQUEST") {216      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);217    }218    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {219      return errorOutcome(result, "sync request entry count exceeds limit");220    }221    // Sanity: the peer's request must be a subset of what its manifest222    // declared as newer. A malformed/over-eager request is a protocol223    // violation but not a data-safety problem — we just send what they ask.224225    stage = "building_requested_entries";226    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);227    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });228    result.entriesSent = Object.keys(ourPayload).length;229230    if (wantFromPeer.length > 0) {231      stage = "waiting_for_device_entries";232      const entriesIn = await receiveMessage(session, timer);233      if (entriesIn.type !== "SYNC_ENTRIES") {234        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);235      }236      await vault.applyEntries(entriesIn.entries, manifestIn.entries);237      result.entriesReceived = Object.keys(entriesIn.entries).length;238    }239240    // Tombstones: each side ships the tombstones that the other does not241    // yet have a newer record for.242    const newerTombstonesForPeer: Record<string, number> = {};243    for (const [id, ts] of Object.entries(ourTombstones)) {244      const peerTs = manifestIn.tombstones[id] ?? 0;245      if (ts > peerTs) newerTombstonesForPeer[id] = ts;246    }247    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });248    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;249250    stage = "waiting_for_device_tombstones";251    const tombstonesIn = await receiveMessage(session, timer);252    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {253      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);254    }255    await vault.applyTombstones(tombstonesIn.tombstones);256    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;257258    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends259    // must expose the adapter methods. Disabling on either side cleanly260    // bypasses the step. The exchange is symmetric on the wire; the261    // production adapter accepts only a directly paired issuer whose stored262    // long-term key matches the Noise-authenticated remote static key.263    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {264      stage = "building_device_manifest";265      const ours = await vault.buildOutgoingDeviceManifest();266      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });267      stage = "waiting_for_device_graph";268      const manifestIn = await receiveMessage(session, timer);269      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {270        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);271      }272      if (manifestIn.manifest) {273        await vault.applyIncomingDeviceManifest(274          manifestIn.manifest,275          session.remoteStaticPublicKey,276        );277      }278    }279280    await sendMessage(session, {281      type: "SYNC_DONE",282      entriesReceived: result.entriesReceived,283      entriesSent: result.entriesSent,284    });285    stage = "waiting_for_device_done";286    const closing = await awaitClosingDone(session, timer);287    if (closing.error) return errorOutcome(result, closing.error);288    result.status = "completed";289    return result;290  } catch (err) {291    if ((err as Error).message === "sync_timeout") {292      result.status = "timeout";293      result.error = `sync_timeout:${stage}`;294      return result;295    }296    return errorOutcome(result, (err as Error).message);297  } finally {298    // Session lifetime is owned by the runner (sync-runner.ts), which299    // closes it in its own finally block. The coordinator must not close it300    // here — the runner still needs to record the outcome after we return.301    timer.cancel();302  }303}304305/**306 * Exact complement of [runSyncInitiator] for the higher-key peer holding a307 * relay waiter room. Keeping this in the browser lets two extensions form a308 * direct mesh edge; the device that introduced them is not in the data path.309 */310export async function runSyncResponder(311  session: SyncSession,312  vault: SyncVaultAdapter,313): Promise<SyncOutcome> {314  const result: SyncOutcome = {315    status: "errored",316    entriesReceived: 0,317    entriesSent: 0,318    tombstonesReceived: 0,319    tombstonesSent: 0,320  };321  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);322  let stage: SyncStage = "waiting_for_device_hello";323  try {324    const ourVaultId = await vault.computeVaultId();325    const ourDeviceId = await vault.getOurDeviceId();326    const localVersion = await vault.getLocalVersion();327    const localTime = Date.now();328    const supportsDeviceManifest =329      vault.buildOutgoingDeviceManifest !== undefined &&330      vault.applyIncomingDeviceManifest !== undefined;331332    const helloIn = await receiveMessage(session, timer);333    if (helloIn.type !== "SYNC_HELLO") {334      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);335    }336    await sendMessage(session, {337      type: "SYNC_HELLO",338      vaultId: ourVaultId,339      deviceId: ourDeviceId,340      version: localVersion,341      deviceTime: localTime,342      deviceManifestVersion: supportsDeviceManifest343        ? DEVICE_MANIFEST_VERSION344        : undefined,345    });346    if (helloIn.vaultId !== ourVaultId) {347      result.status = "vault_mismatch";348      return result;349    }350    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {351      console.warn(352        `[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`,353      );354    }355356    if (helloIn.version === localVersion) {357      if (358        supportsDeviceManifest &&359        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION360      ) {361        stage = "waiting_for_device_graph";362        const graphIn = await receiveMessage(session, timer);363        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {364          return errorOutcome(365            result,366            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,367          );368        }369        const ours = await vault.buildOutgoingDeviceManifest!();370        await sendMessage(session, {371          type: "SYNC_DEVICE_MANIFEST",372          manifest: ours,373        });374        if (graphIn.manifest) {375          await vault.applyIncomingDeviceManifest!(376            graphIn.manifest,377            session.remoteStaticPublicKey,378          );379        }380      }381      stage = "waiting_for_device_done";382      const done = await receiveMessage(session, timer);383      if (done.type !== "SYNC_DONE") {384        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);385      }386      await sendMessage(session, {387        type: "SYNC_DONE",388        entriesReceived: 0,389        entriesSent: 0,390      });391      result.status = "completed";392      return result;393    }394395    const ourEntries = await vault.buildEntryManifest();396    const ourTombstones = await vault.buildTombstoneManifest();397    stage = "waiting_for_device_manifest";398    const manifestIn = await receiveMessage(session, timer);399    if (manifestIn.type !== "SYNC_MANIFEST") {400      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);401    }402    validateSyncManifestCounts(manifestIn.entries, manifestIn.tombstones);403    await sendMessage(session, {404      type: "SYNC_MANIFEST",405      entries: ourEntries,406      tombstones: ourTombstones,407    });408409    stage = "waiting_for_device_request";410    const requestIn = await receiveMessage(session, timer);411    if (requestIn.type !== "SYNC_REQUEST") {412      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);413    }414    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {415      return errorOutcome(result, "sync request entry count exceeds limit");416    }417    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);418    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });419420    stage = "waiting_for_device_entries";421    const entriesIn = await receiveMessage(session, timer);422    if (entriesIn.type !== "SYNC_ENTRIES") {423      return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);424    }425    if (wantFromPeer.length > 0) {426      await vault.applyEntries(entriesIn.entries, manifestIn.entries);427      result.entriesReceived = Object.keys(entriesIn.entries).length;428    }429    if (requestIn.entryIds.length > 0) {430      const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);431      await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });432      result.entriesSent = Object.keys(ourPayload).length;433    }434435    stage = "waiting_for_device_tombstones";436    const tombstonesIn = await receiveMessage(session, timer);437    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {438      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);439    }440    await vault.applyTombstones(tombstonesIn.tombstones);441    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;442443    const newerTombstonesForPeer: Record<string, number> = {};444    for (const [id, timestamp] of Object.entries(ourTombstones)) {445      if (timestamp > (manifestIn.tombstones[id] ?? 0)) {446        newerTombstonesForPeer[id] = timestamp;447      }448    }449    await sendMessage(session, {450      type: "SYNC_TOMBSTONES",451      tombstones: newerTombstonesForPeer,452    });453    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;454455    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {456      stage = "waiting_for_device_graph";457      const graphIn = await receiveMessage(session, timer);458      if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {459        return errorOutcome(460          result,461          `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,462        );463      }464      const ours = await vault.buildOutgoingDeviceManifest();465      await sendMessage(session, {466        type: "SYNC_DEVICE_MANIFEST",467        manifest: ours,468      });469      if (graphIn.manifest) {470        await vault.applyIncomingDeviceManifest(471          graphIn.manifest,472          session.remoteStaticPublicKey,473        );474      }475    }476477    stage = "waiting_for_device_done";478    const done = await receiveMessage(session, timer);479    if (done.type !== "SYNC_DONE") {480      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);481    }482    await sendMessage(session, {483      type: "SYNC_DONE",484      entriesReceived: result.entriesReceived,485      entriesSent: result.entriesSent,486    });487    result.status = "completed";488    return result;489  } catch (error) {490    if ((error as Error).message === "sync_timeout") {491      result.status = "timeout";492      result.error = `sync_timeout:${stage}`;493      return result;494    }495    return errorOutcome(result, (error as Error).message);496  } finally {497    timer.cancel();498  }499}500501/**502 * Wait for the peer's closing SYNC_DONE echo, treating a hang-up as success.503 *504 * By the time we send our own SYNC_DONE, every entry and tombstone has already505 * been exchanged and applied on both sides — the echo is a courtesy, not a506 * commit. Android closes the socket as soon as it has our DONE, so the parked507 * receive() rejects with "WebSocket closed"; counting that as `run_failed`508 * reported a red indicator for syncs that had fully succeeded and whose data509 * had demonstrably landed on the peer.510 *511 * A wrong *message* here is still a real protocol error — only a closed512 * transport is forgiven.513 */514async function awaitClosingDone(515  session: SyncSession,516  timer: Timer,517): Promise<{ error?: string }> {518  try {519    const done = await receiveMessage(session, timer);520    if (done.type !== "SYNC_DONE") {521      return { error: `expected SYNC_DONE, got ${done.type}` };522    }523    return {};524  } catch (err) {525    if ((err as Error).message === "sync_timeout") throw err;526    if (isConnectionClosedError(err)) return {};527    return { error: (err as Error).message };528  }529}530531function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {532  const want: string[] = [];533  for (const [id, ts] of Object.entries(remote)) {534    const ours = local[id];535    if (ours === undefined || ts > ours) want.push(id);536  }537  return want;538}539540interface Timer {541  signal: AbortSignal;542  cancel(): void;543}544545function scheduleTimeout(ms: number): Timer {546  const ctrl = new AbortController();547  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);548  return {549    signal: ctrl.signal,550    cancel: () => clearTimeout(t),551  };552}553554async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {555  session.send(new TextEncoder().encode(JSON.stringify(msg)));556}557558async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {559  const pending = session.receive();560  const aborted = new Promise<never>((_, reject) => {561    if (timer.signal.aborted) reject(new Error("sync_timeout"));562    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));563  });564  const bytes = await Promise.race([pending, aborted]);565  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS566  // transport may not have enforced anything) — re-check before JSON.parse567  // materialises a multi-GB string/object graph.568  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {569    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);570  }571  const parsed = JSON.parse(new TextDecoder().decode(bytes));572  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");573  return parsed as SyncMessage;574}575576function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {577  outcome.status = "errored";578  outcome.error = error;579  return outcome;580}