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 }7//   <-  SYNC_HELLO { vaultId, deviceId }       (mismatch → SYNC_ERROR)8//   <-> SYNC_MANIFEST { entries, tombstones }  (each side sends one)9//   ->  SYNC_REQUEST { entryIds }              (entries newer on the peer)10//   <-  SYNC_ENTRIES { entries }11//   <-  SYNC_REQUEST { entryIds }              (entries newer on us)12//   ->  SYNC_ENTRIES { entries }13//   <-> SYNC_TOMBSTONES { tombstones }14//   <-> SYNC_DONE15//16// Entries are opaque base64 ciphertext — the coordinator never decrypts.17// Last-write-wins at the entry level by mtime; tombstones override updates18// with a later deletion timestamp.1920import type { SyncMessage, SignedDeviceManifest } from "../shared/sync-types";21import type { SyncSession } from "../shared/sync-transport";22import { SYNC_LIMITS } from "../shared/sync-limits";2324const SESSION_TIMEOUT_MS = 30_000;2526/**27 * Vault-side adapter — the coordinator's dependency on the vault layer.28 * The vault holds the canonical entry/tombstone tables; the coordinator29 * pulls a manifest, requests entries by id, applies remote entries, and30 * applies remote tombstones.31 */32export interface SyncVaultAdapter {33  computeVaultId(): Promise<string>;34  getOurDeviceId(): Promise<string>;35  /** Returns the vault's monotonic version (`peach_last_modified_at`). */36  getLocalVersion(): Promise<number>;37  /** Returns the per-entry mtimes for our active entries. */38  buildEntryManifest(): Promise<Record<string, number>>;39  /** Returns the per-entry deletion timestamps for our tombstones. */40  buildTombstoneManifest(): Promise<Record<string, number>>;41  /** Returns base64 ciphertext for each requested entry id. */42  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;43  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */44  applyEntries(entries: Record<string, string>): Promise<void>;45  /** Apply incoming tombstones; vault retains the latest deletion ts. */46  applyTombstones(tombstones: Record<string, number>): Promise<void>;47  /**48   * Optional — v2 brief §4.3 device manifest exchange. The extension (as49   * the trust anchor for the device graph) returns a signed manifest50   * enumerating every paired device it knows about; the peer verifies it51   * against the extension's known Ed25519 signing pubkey and upserts any52   * newly-vouched devices into its own registry.53   *54   * When this method is undefined, the coordinator skips the manifest55   * step entirely. Both ends must opt in for the exchange to occur.56   */57  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;58  /**59   * Handle a manifest received from the peer. The transport-authenticated60   * Noise static key is supplied separately so a self-asserted issuer id61   * cannot select a different paired-device verification key.62   */63  applyIncomingDeviceManifest?(64    manifest: SignedDeviceManifest,65    authenticatedRemoteStaticPublicKey: Uint8Array,66  ): Promise<void>;67}6869export interface SyncOutcome {70  status: "completed" | "vault_mismatch" | "errored" | "timeout";71  entriesReceived: number;72  entriesSent: number;73  tombstonesReceived: number;74  tombstonesSent: number;75  error?: string;76}7778/**79 * Run one sync round as the initiator. The extension is always the80 * initiator (it is the WebSocket client). The peer's role is symmetric81 * after SYNC_HELLO, so a corresponding responder helper exists for the82 * mobile-side coordinator that calls this same protocol the other way.83 */84export async function runSyncInitiator(85  session: SyncSession,86  vault: SyncVaultAdapter,87): Promise<SyncOutcome> {88  const result: SyncOutcome = {89    status: "errored",90    entriesReceived: 0,91    entriesSent: 0,92    tombstonesReceived: 0,93    tombstonesSent: 0,94  };9596  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);97  try {98    const ourVaultId = await vault.computeVaultId();99    const ourDeviceId = await vault.getOurDeviceId();100    const localVersion = await vault.getLocalVersion();101    const localTime = Date.now();102103    await sendMessage(session, {104      type: "SYNC_HELLO",105      vaultId: ourVaultId,106      deviceId: ourDeviceId,107      version: localVersion,108      deviceTime: localTime,109    });110    const helloIn = await receiveMessage(session, timer);111    if (helloIn.type !== "SYNC_HELLO") {112      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);113    }114    if (helloIn.vaultId !== ourVaultId) {115      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });116      result.status = "vault_mismatch";117      return result;118    }119120    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {121      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);122    }123124    // Equal versions — nothing to exchange.125    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {126      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });127      const done = await receiveMessage(session, timer);128      if (done.type !== "SYNC_DONE") {129        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);130      }131      result.status = "completed";132      return result;133    }134135    const ourEntries = await vault.buildEntryManifest();136    const ourTombstones = await vault.buildTombstoneManifest();137    await sendMessage(session, {138      type: "SYNC_MANIFEST",139      entries: ourEntries,140      tombstones: ourTombstones,141    });142    const manifestIn = await receiveMessage(session, timer);143    if (manifestIn.type !== "SYNC_MANIFEST") {144      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);145    }146147    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);148    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });149150    const requestIn = await receiveMessage(session, timer);151    if (requestIn.type !== "SYNC_REQUEST") {152      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);153    }154    // Sanity: the peer's request must be a subset of what its manifest155    // declared as newer. A malformed/over-eager request is a protocol156    // violation but not a data-safety problem — we just send what they ask.157158    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);159    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });160    result.entriesSent = Object.keys(ourPayload).length;161162    if (wantFromPeer.length > 0) {163      const entriesIn = await receiveMessage(session, timer);164      if (entriesIn.type !== "SYNC_ENTRIES") {165        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);166      }167      await vault.applyEntries(entriesIn.entries);168      result.entriesReceived = Object.keys(entriesIn.entries).length;169    }170171    // Tombstones: each side ships the tombstones that the other does not172    // yet have a newer record for.173    const newerTombstonesForPeer: Record<string, number> = {};174    for (const [id, ts] of Object.entries(ourTombstones)) {175      const peerTs = manifestIn.tombstones[id] ?? 0;176      if (ts > peerTs) newerTombstonesForPeer[id] = ts;177    }178    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });179    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;180181    const tombstonesIn = await receiveMessage(session, timer);182    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {183      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);184    }185    await vault.applyTombstones(tombstonesIn.tombstones);186    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;187188    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends189    // must expose the adapter methods. Disabling on either side cleanly190    // bypasses the step. The exchange is symmetric on the wire; the191    // production adapter accepts only a directly paired issuer whose stored192    // long-term key matches the Noise-authenticated remote static key.193    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {194      const ours = await vault.buildOutgoingDeviceManifest();195      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });196      const manifestIn = await receiveMessage(session, timer);197      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {198        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);199      }200      if (manifestIn.manifest) {201        await vault.applyIncomingDeviceManifest(202          manifestIn.manifest,203          session.remoteStaticPublicKey,204        );205      }206    }207208    await sendMessage(session, {209      type: "SYNC_DONE",210      entriesReceived: result.entriesReceived,211      entriesSent: result.entriesSent,212    });213    const done = await receiveMessage(session, timer);214    if (done.type !== "SYNC_DONE") {215      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);216    }217    result.status = "completed";218    return result;219  } catch (err) {220    if ((err as Error).message === "sync_timeout") {221      result.status = "timeout";222      result.error = "session timed out after 30s";223      return result;224    }225    return errorOutcome(result, (err as Error).message);226  } finally {227    // Session lifetime is owned by the runner (sync-runner.ts), which228    // closes it in its own finally block. The coordinator must not close it229    // here — the runner still needs to record the outcome after we return.230    timer.cancel();231  }232}233234function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {235  const want: string[] = [];236  for (const [id, ts] of Object.entries(remote)) {237    const ours = local[id];238    if (ours === undefined || ts > ours) want.push(id);239  }240  return want;241}242243interface Timer {244  signal: AbortSignal;245  cancel(): void;246}247248function scheduleTimeout(ms: number): Timer {249  const ctrl = new AbortController();250  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);251  return {252    signal: ctrl.signal,253    cancel: () => clearTimeout(t),254  };255}256257async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {258  session.send(new TextEncoder().encode(JSON.stringify(msg)));259}260261async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {262  const pending = session.receive();263  const aborted = new Promise<never>((_, reject) => {264    if (timer.signal.aborted) reject(new Error("sync_timeout"));265    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));266  });267  const bytes = await Promise.race([pending, aborted]);268  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS269  // transport may not have enforced anything) — re-check before JSON.parse270  // materialises a multi-GB string/object graph.271  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {272    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);273  }274  const parsed = JSON.parse(new TextDecoder().decode(bytes));275  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");276  return parsed as SyncMessage;277}278279function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {280  outcome.status = "errored";281  outcome.error = error;282  return outcome;283}