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";23import { isConnectionClosedError } from "../shared/sync-outcome";2425const SESSION_TIMEOUT_MS = 30_000;2627type SyncStage =28  | "preparing_hello"29  | "waiting_for_device_hello"30  | "building_local_manifest"31  | "waiting_for_device_manifest"32  | "waiting_for_device_request"33  | "building_requested_entries"34  | "waiting_for_device_entries"35  | "waiting_for_device_tombstones"36  | "building_device_manifest"37  | "waiting_for_device_graph"38  | "waiting_for_device_done";3940/**41 * Vault-side adapter — the coordinator's dependency on the vault layer.42 * The vault holds the canonical entry/tombstone tables; the coordinator43 * pulls a manifest, requests entries by id, applies remote entries, and44 * applies remote tombstones.45 */46export interface SyncVaultAdapter {47  computeVaultId(): Promise<string>;48  getOurDeviceId(): Promise<string>;49  /** Returns the vault's monotonic version (`peach_last_modified_at`). */50  getLocalVersion(): Promise<number>;51  /** Returns the per-entry mtimes for our active entries. */52  buildEntryManifest(): Promise<Record<string, number>>;53  /** Returns the per-entry deletion timestamps for our tombstones. */54  buildTombstoneManifest(): Promise<Record<string, number>>;55  /** Returns base64 ciphertext for each requested entry id. */56  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;57  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */58  applyEntries(entries: Record<string, string>): Promise<void>;59  /** Apply incoming tombstones; vault retains the latest deletion ts. */60  applyTombstones(tombstones: Record<string, number>): Promise<void>;61  /**62   * Optional — v2 brief §4.3 device manifest exchange. The extension (as63   * the trust anchor for the device graph) returns a signed manifest64   * enumerating every paired device it knows about; the peer verifies it65   * against the extension's known Ed25519 signing pubkey and upserts any66   * newly-vouched devices into its own registry.67   *68   * When this method is undefined, the coordinator skips the manifest69   * step entirely. Both ends must opt in for the exchange to occur.70   */71  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;72  /**73   * Handle a manifest received from the peer. The transport-authenticated74   * Noise static key is supplied separately so a self-asserted issuer id75   * cannot select a different paired-device verification key.76   */77  applyIncomingDeviceManifest?(78    manifest: SignedDeviceManifest,79    authenticatedRemoteStaticPublicKey: Uint8Array,80  ): Promise<void>;81}8283export interface SyncOutcome {84  status: "completed" | "vault_mismatch" | "errored" | "timeout";85  entriesReceived: number;86  entriesSent: number;87  tombstonesReceived: number;88  tombstonesSent: number;89  error?: string;90}9192/**93 * Run one sync round as the lower-key initiator. Device type does not decide94 * the role; the higher-key peer runs [runSyncResponder].95 */96export async function runSyncInitiator(97  session: SyncSession,98  vault: SyncVaultAdapter,99): Promise<SyncOutcome> {100  const result: SyncOutcome = {101    status: "errored",102    entriesReceived: 0,103    entriesSent: 0,104    tombstonesReceived: 0,105    tombstonesSent: 0,106  };107108  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);109  // Privacy-safe progress only: stages contain no device identifiers, relay110  // capabilities, keys, entry ids/counts, or vault content. Keeping the last111  // stage turns an otherwise opaque 30-second timeout into an actionable112  // cross-device diagnostic.113  let stage: SyncStage = "preparing_hello";114  try {115    const ourVaultId = await vault.computeVaultId();116    const ourDeviceId = await vault.getOurDeviceId();117    const localVersion = await vault.getLocalVersion();118    const localTime = Date.now();119120    await sendMessage(session, {121      type: "SYNC_HELLO",122      vaultId: ourVaultId,123      deviceId: ourDeviceId,124      version: localVersion,125      deviceTime: localTime,126    });127    stage = "waiting_for_device_hello";128    const helloIn = await receiveMessage(session, timer);129    if (helloIn.type !== "SYNC_HELLO") {130      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);131    }132    if (helloIn.vaultId !== ourVaultId) {133      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });134      result.status = "vault_mismatch";135      return result;136    }137138    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {139      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);140    }141142    // Equal versions — nothing to exchange.143    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {144      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });145      stage = "waiting_for_device_done";146      const closing = await awaitClosingDone(session, timer);147      if (closing.error) return errorOutcome(result, closing.error);148      result.status = "completed";149      return result;150    }151152    stage = "building_local_manifest";153    const ourEntries = await vault.buildEntryManifest();154    const ourTombstones = await vault.buildTombstoneManifest();155    await sendMessage(session, {156      type: "SYNC_MANIFEST",157      entries: ourEntries,158      tombstones: ourTombstones,159    });160    stage = "waiting_for_device_manifest";161    const manifestIn = await receiveMessage(session, timer);162    if (manifestIn.type !== "SYNC_MANIFEST") {163      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);164    }165166    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);167    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });168169    stage = "waiting_for_device_request";170    const requestIn = await receiveMessage(session, timer);171    if (requestIn.type !== "SYNC_REQUEST") {172      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);173    }174    // Sanity: the peer's request must be a subset of what its manifest175    // declared as newer. A malformed/over-eager request is a protocol176    // violation but not a data-safety problem — we just send what they ask.177178    stage = "building_requested_entries";179    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);180    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });181    result.entriesSent = Object.keys(ourPayload).length;182183    if (wantFromPeer.length > 0) {184      stage = "waiting_for_device_entries";185      const entriesIn = await receiveMessage(session, timer);186      if (entriesIn.type !== "SYNC_ENTRIES") {187        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);188      }189      await vault.applyEntries(entriesIn.entries);190      result.entriesReceived = Object.keys(entriesIn.entries).length;191    }192193    // Tombstones: each side ships the tombstones that the other does not194    // yet have a newer record for.195    const newerTombstonesForPeer: Record<string, number> = {};196    for (const [id, ts] of Object.entries(ourTombstones)) {197      const peerTs = manifestIn.tombstones[id] ?? 0;198      if (ts > peerTs) newerTombstonesForPeer[id] = ts;199    }200    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });201    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;202203    stage = "waiting_for_device_tombstones";204    const tombstonesIn = await receiveMessage(session, timer);205    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {206      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);207    }208    await vault.applyTombstones(tombstonesIn.tombstones);209    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;210211    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends212    // must expose the adapter methods. Disabling on either side cleanly213    // bypasses the step. The exchange is symmetric on the wire; the214    // production adapter accepts only a directly paired issuer whose stored215    // long-term key matches the Noise-authenticated remote static key.216    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {217      stage = "building_device_manifest";218      const ours = await vault.buildOutgoingDeviceManifest();219      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });220      stage = "waiting_for_device_graph";221      const manifestIn = await receiveMessage(session, timer);222      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {223        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);224      }225      if (manifestIn.manifest) {226        await vault.applyIncomingDeviceManifest(227          manifestIn.manifest,228          session.remoteStaticPublicKey,229        );230      }231    }232233    await sendMessage(session, {234      type: "SYNC_DONE",235      entriesReceived: result.entriesReceived,236      entriesSent: result.entriesSent,237    });238    stage = "waiting_for_device_done";239    const closing = await awaitClosingDone(session, timer);240    if (closing.error) return errorOutcome(result, closing.error);241    result.status = "completed";242    return result;243  } catch (err) {244    if ((err as Error).message === "sync_timeout") {245      result.status = "timeout";246      result.error = `sync_timeout:${stage}`;247      return result;248    }249    return errorOutcome(result, (err as Error).message);250  } finally {251    // Session lifetime is owned by the runner (sync-runner.ts), which252    // closes it in its own finally block. The coordinator must not close it253    // here — the runner still needs to record the outcome after we return.254    timer.cancel();255  }256}257258/**259 * Exact complement of [runSyncInitiator] for the higher-key peer holding a260 * relay waiter room. Keeping this in the browser lets two extensions form a261 * direct mesh edge; the device that introduced them is not in the data path.262 */263export async function runSyncResponder(264  session: SyncSession,265  vault: SyncVaultAdapter,266): Promise<SyncOutcome> {267  const result: SyncOutcome = {268    status: "errored",269    entriesReceived: 0,270    entriesSent: 0,271    tombstonesReceived: 0,272    tombstonesSent: 0,273  };274  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);275  let stage: SyncStage = "waiting_for_device_hello";276  try {277    const ourVaultId = await vault.computeVaultId();278    const ourDeviceId = await vault.getOurDeviceId();279    const localVersion = await vault.getLocalVersion();280    const localTime = Date.now();281282    const helloIn = await receiveMessage(session, timer);283    if (helloIn.type !== "SYNC_HELLO") {284      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);285    }286    await sendMessage(session, {287      type: "SYNC_HELLO",288      vaultId: ourVaultId,289      deviceId: ourDeviceId,290      version: localVersion,291      deviceTime: localTime,292    });293    if (helloIn.vaultId !== ourVaultId) {294      result.status = "vault_mismatch";295      return result;296    }297    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {298      console.warn(299        `[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`,300      );301    }302303    if (helloIn.version === localVersion) {304      stage = "waiting_for_device_done";305      const done = await receiveMessage(session, timer);306      if (done.type !== "SYNC_DONE") {307        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);308      }309      await sendMessage(session, {310        type: "SYNC_DONE",311        entriesReceived: 0,312        entriesSent: 0,313      });314      result.status = "completed";315      return result;316    }317318    const ourEntries = await vault.buildEntryManifest();319    const ourTombstones = await vault.buildTombstoneManifest();320    stage = "waiting_for_device_manifest";321    const manifestIn = await receiveMessage(session, timer);322    if (manifestIn.type !== "SYNC_MANIFEST") {323      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);324    }325    await sendMessage(session, {326      type: "SYNC_MANIFEST",327      entries: ourEntries,328      tombstones: ourTombstones,329    });330331    stage = "waiting_for_device_request";332    const requestIn = await receiveMessage(session, timer);333    if (requestIn.type !== "SYNC_REQUEST") {334      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);335    }336    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);337    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });338339    stage = "waiting_for_device_entries";340    const entriesIn = await receiveMessage(session, timer);341    if (entriesIn.type !== "SYNC_ENTRIES") {342      return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);343    }344    if (wantFromPeer.length > 0) {345      await vault.applyEntries(entriesIn.entries);346      result.entriesReceived = Object.keys(entriesIn.entries).length;347    }348    if (requestIn.entryIds.length > 0) {349      const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);350      await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });351      result.entriesSent = Object.keys(ourPayload).length;352    }353354    stage = "waiting_for_device_tombstones";355    const tombstonesIn = await receiveMessage(session, timer);356    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {357      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);358    }359    await vault.applyTombstones(tombstonesIn.tombstones);360    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;361362    const newerTombstonesForPeer: Record<string, number> = {};363    for (const [id, timestamp] of Object.entries(ourTombstones)) {364      if (timestamp > (manifestIn.tombstones[id] ?? 0)) {365        newerTombstonesForPeer[id] = timestamp;366      }367    }368    await sendMessage(session, {369      type: "SYNC_TOMBSTONES",370      tombstones: newerTombstonesForPeer,371    });372    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;373374    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {375      stage = "waiting_for_device_graph";376      const graphIn = await receiveMessage(session, timer);377      if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {378        return errorOutcome(379          result,380          `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,381        );382      }383      const ours = await vault.buildOutgoingDeviceManifest();384      await sendMessage(session, {385        type: "SYNC_DEVICE_MANIFEST",386        manifest: ours,387      });388      if (graphIn.manifest) {389        await vault.applyIncomingDeviceManifest(390          graphIn.manifest,391          session.remoteStaticPublicKey,392        );393      }394    }395396    stage = "waiting_for_device_done";397    const done = await receiveMessage(session, timer);398    if (done.type !== "SYNC_DONE") {399      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);400    }401    await sendMessage(session, {402      type: "SYNC_DONE",403      entriesReceived: result.entriesReceived,404      entriesSent: result.entriesSent,405    });406    result.status = "completed";407    return result;408  } catch (error) {409    if ((error as Error).message === "sync_timeout") {410      result.status = "timeout";411      result.error = `sync_timeout:${stage}`;412      return result;413    }414    return errorOutcome(result, (error as Error).message);415  } finally {416    timer.cancel();417  }418}419420/**421 * Wait for the peer's closing SYNC_DONE echo, treating a hang-up as success.422 *423 * By the time we send our own SYNC_DONE, every entry and tombstone has already424 * been exchanged and applied on both sides — the echo is a courtesy, not a425 * commit. Android closes the socket as soon as it has our DONE, so the parked426 * receive() rejects with "WebSocket closed"; counting that as `run_failed`427 * reported a red indicator for syncs that had fully succeeded and whose data428 * had demonstrably landed on the peer.429 *430 * A wrong *message* here is still a real protocol error — only a closed431 * transport is forgiven.432 */433async function awaitClosingDone(434  session: SyncSession,435  timer: Timer,436): Promise<{ error?: string }> {437  try {438    const done = await receiveMessage(session, timer);439    if (done.type !== "SYNC_DONE") {440      return { error: `expected SYNC_DONE, got ${done.type}` };441    }442    return {};443  } catch (err) {444    if ((err as Error).message === "sync_timeout") throw err;445    if (isConnectionClosedError(err)) return {};446    return { error: (err as Error).message };447  }448}449450function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {451  const want: string[] = [];452  for (const [id, ts] of Object.entries(remote)) {453    const ours = local[id];454    if (ours === undefined || ts > ours) want.push(id);455  }456  return want;457}458459interface Timer {460  signal: AbortSignal;461  cancel(): void;462}463464function scheduleTimeout(ms: number): Timer {465  const ctrl = new AbortController();466  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);467  return {468    signal: ctrl.signal,469    cancel: () => clearTimeout(t),470  };471}472473async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {474  session.send(new TextEncoder().encode(JSON.stringify(msg)));475}476477async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {478  const pending = session.receive();479  const aborted = new Promise<never>((_, reject) => {480    if (timer.signal.aborted) reject(new Error("sync_timeout"));481    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));482  });483  const bytes = await Promise.race([pending, aborted]);484  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS485  // transport may not have enforced anything) — re-check before JSON.parse486  // materialises a multi-GB string/object graph.487  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {488    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);489  }490  const parsed = JSON.parse(new TextDecoder().decode(bytes));491  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");492  return parsed as SyncMessage;493}494495function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {496  outcome.status = "errored";497  outcome.error = error;498  return outcome;499}