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;2526type SyncStage =27  | "preparing_hello"28  | "waiting_for_device_hello"29  | "building_local_manifest"30  | "waiting_for_device_manifest"31  | "waiting_for_device_request"32  | "building_requested_entries"33  | "waiting_for_device_entries"34  | "waiting_for_device_tombstones"35  | "building_device_manifest"36  | "waiting_for_device_graph"37  | "waiting_for_device_done";3839/**40 * Vault-side adapter — the coordinator's dependency on the vault layer.41 * The vault holds the canonical entry/tombstone tables; the coordinator42 * pulls a manifest, requests entries by id, applies remote entries, and43 * applies remote tombstones.44 */45export interface SyncVaultAdapter {46  computeVaultId(): Promise<string>;47  getOurDeviceId(): Promise<string>;48  /** Returns the vault's monotonic version (`peach_last_modified_at`). */49  getLocalVersion(): Promise<number>;50  /** Returns the per-entry mtimes for our active entries. */51  buildEntryManifest(): Promise<Record<string, number>>;52  /** Returns the per-entry deletion timestamps for our tombstones. */53  buildTombstoneManifest(): Promise<Record<string, number>>;54  /** Returns base64 ciphertext for each requested entry id. */55  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;56  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */57  applyEntries(entries: Record<string, string>): Promise<void>;58  /** Apply incoming tombstones; vault retains the latest deletion ts. */59  applyTombstones(tombstones: Record<string, number>): Promise<void>;60  /**61   * Optional — v2 brief §4.3 device manifest exchange. The extension (as62   * the trust anchor for the device graph) returns a signed manifest63   * enumerating every paired device it knows about; the peer verifies it64   * against the extension's known Ed25519 signing pubkey and upserts any65   * newly-vouched devices into its own registry.66   *67   * When this method is undefined, the coordinator skips the manifest68   * step entirely. Both ends must opt in for the exchange to occur.69   */70  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;71  /**72   * Handle a manifest received from the peer. The transport-authenticated73   * Noise static key is supplied separately so a self-asserted issuer id74   * cannot select a different paired-device verification key.75   */76  applyIncomingDeviceManifest?(77    manifest: SignedDeviceManifest,78    authenticatedRemoteStaticPublicKey: Uint8Array,79  ): Promise<void>;80}8182export interface SyncOutcome {83  status: "completed" | "vault_mismatch" | "errored" | "timeout";84  entriesReceived: number;85  entriesSent: number;86  tombstonesReceived: number;87  tombstonesSent: number;88  error?: string;89}9091/**92 * Run one sync round as the initiator. The extension is always the93 * initiator (it is the WebSocket client). The peer's role is symmetric94 * after SYNC_HELLO, so a corresponding responder helper exists for the95 * mobile-side coordinator that calls this same protocol the other way.96 */97export async function runSyncInitiator(98  session: SyncSession,99  vault: SyncVaultAdapter,100): Promise<SyncOutcome> {101  const result: SyncOutcome = {102    status: "errored",103    entriesReceived: 0,104    entriesSent: 0,105    tombstonesReceived: 0,106    tombstonesSent: 0,107  };108109  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);110  // Privacy-safe progress only: stages contain no device identifiers, relay111  // capabilities, keys, entry ids/counts, or vault content. Keeping the last112  // stage turns an otherwise opaque 30-second timeout into an actionable113  // cross-device diagnostic.114  let stage: SyncStage = "preparing_hello";115  try {116    const ourVaultId = await vault.computeVaultId();117    const ourDeviceId = await vault.getOurDeviceId();118    const localVersion = await vault.getLocalVersion();119    const localTime = Date.now();120121    await sendMessage(session, {122      type: "SYNC_HELLO",123      vaultId: ourVaultId,124      deviceId: ourDeviceId,125      version: localVersion,126      deviceTime: localTime,127    });128    stage = "waiting_for_device_hello";129    const helloIn = await receiveMessage(session, timer);130    if (helloIn.type !== "SYNC_HELLO") {131      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);132    }133    if (helloIn.vaultId !== ourVaultId) {134      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });135      result.status = "vault_mismatch";136      return result;137    }138139    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {140      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);141    }142143    // Equal versions — nothing to exchange.144    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {145      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });146      stage = "waiting_for_device_done";147      const done = await receiveMessage(session, timer);148      if (done.type !== "SYNC_DONE") {149        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);150      }151      result.status = "completed";152      return result;153    }154155    stage = "building_local_manifest";156    const ourEntries = await vault.buildEntryManifest();157    const ourTombstones = await vault.buildTombstoneManifest();158    await sendMessage(session, {159      type: "SYNC_MANIFEST",160      entries: ourEntries,161      tombstones: ourTombstones,162    });163    stage = "waiting_for_device_manifest";164    const manifestIn = await receiveMessage(session, timer);165    if (manifestIn.type !== "SYNC_MANIFEST") {166      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);167    }168169    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);170    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });171172    stage = "waiting_for_device_request";173    const requestIn = await receiveMessage(session, timer);174    if (requestIn.type !== "SYNC_REQUEST") {175      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);176    }177    // Sanity: the peer's request must be a subset of what its manifest178    // declared as newer. A malformed/over-eager request is a protocol179    // violation but not a data-safety problem — we just send what they ask.180181    stage = "building_requested_entries";182    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);183    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });184    result.entriesSent = Object.keys(ourPayload).length;185186    if (wantFromPeer.length > 0) {187      stage = "waiting_for_device_entries";188      const entriesIn = await receiveMessage(session, timer);189      if (entriesIn.type !== "SYNC_ENTRIES") {190        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);191      }192      await vault.applyEntries(entriesIn.entries);193      result.entriesReceived = Object.keys(entriesIn.entries).length;194    }195196    // Tombstones: each side ships the tombstones that the other does not197    // yet have a newer record for.198    const newerTombstonesForPeer: Record<string, number> = {};199    for (const [id, ts] of Object.entries(ourTombstones)) {200      const peerTs = manifestIn.tombstones[id] ?? 0;201      if (ts > peerTs) newerTombstonesForPeer[id] = ts;202    }203    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });204    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;205206    stage = "waiting_for_device_tombstones";207    const tombstonesIn = await receiveMessage(session, timer);208    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {209      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);210    }211    await vault.applyTombstones(tombstonesIn.tombstones);212    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;213214    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends215    // must expose the adapter methods. Disabling on either side cleanly216    // bypasses the step. The exchange is symmetric on the wire; the217    // production adapter accepts only a directly paired issuer whose stored218    // long-term key matches the Noise-authenticated remote static key.219    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {220      stage = "building_device_manifest";221      const ours = await vault.buildOutgoingDeviceManifest();222      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });223      stage = "waiting_for_device_graph";224      const manifestIn = await receiveMessage(session, timer);225      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {226        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);227      }228      if (manifestIn.manifest) {229        await vault.applyIncomingDeviceManifest(230          manifestIn.manifest,231          session.remoteStaticPublicKey,232        );233      }234    }235236    await sendMessage(session, {237      type: "SYNC_DONE",238      entriesReceived: result.entriesReceived,239      entriesSent: result.entriesSent,240    });241    stage = "waiting_for_device_done";242    const done = await receiveMessage(session, timer);243    if (done.type !== "SYNC_DONE") {244      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);245    }246    result.status = "completed";247    return result;248  } catch (err) {249    if ((err as Error).message === "sync_timeout") {250      result.status = "timeout";251      result.error = `sync_timeout:${stage}`;252      return result;253    }254    return errorOutcome(result, (err as Error).message);255  } finally {256    // Session lifetime is owned by the runner (sync-runner.ts), which257    // closes it in its own finally block. The coordinator must not close it258    // here — the runner still needs to record the outcome after we return.259    timer.cancel();260  }261}262263function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {264  const want: string[] = [];265  for (const [id, ts] of Object.entries(remote)) {266    const ours = local[id];267    if (ours === undefined || ts > ours) want.push(id);268  }269  return want;270}271272interface Timer {273  signal: AbortSignal;274  cancel(): void;275}276277function scheduleTimeout(ms: number): Timer {278  const ctrl = new AbortController();279  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);280  return {281    signal: ctrl.signal,282    cancel: () => clearTimeout(t),283  };284}285286async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {287  session.send(new TextEncoder().encode(JSON.stringify(msg)));288}289290async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {291  const pending = session.receive();292  const aborted = new Promise<never>((_, reject) => {293    if (timer.signal.aborted) reject(new Error("sync_timeout"));294    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));295  });296  const bytes = await Promise.race([pending, aborted]);297  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS298  // transport may not have enforced anything) — re-check before JSON.parse299  // materialises a multi-GB string/object graph.300  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {301    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);302  }303  const parsed = JSON.parse(new TextDecoder().decode(bytes));304  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");305  return parsed as SyncMessage;306}307308function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {309  outcome.status = "errored";310  outcome.error = error;311  return outcome;312}