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 } 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(entries: Record<string, string>): Promise<void>;65  /** Apply incoming tombstones; vault retains the latest deletion ts. */66  applyTombstones(tombstones: Record<string, number>): Promise<void>;67  /**68   * Optional — v2 brief §4.3 device manifest exchange. The extension (as69   * the trust anchor for the device graph) returns a signed manifest70   * enumerating every paired device it knows about; the peer verifies it71   * against the extension's known Ed25519 signing pubkey and upserts any72   * newly-vouched devices into its own registry.73   *74   * When this method is undefined, the coordinator skips the manifest75   * step entirely. Both ends must opt in for the exchange to occur.76   */77  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;78  /**79   * Handle a manifest received from the peer. The transport-authenticated80   * Noise static key is supplied separately so a self-asserted issuer id81   * cannot select a different paired-device verification key.82   */83  applyIncomingDeviceManifest?(84    manifest: SignedDeviceManifest,85    authenticatedRemoteStaticPublicKey: Uint8Array,86  ): Promise<void>;87}8889export interface SyncOutcome {90  status: "completed" | "vault_mismatch" | "errored" | "timeout";91  entriesReceived: number;92  entriesSent: number;93  tombstonesReceived: number;94  tombstonesSent: number;95  error?: string;96}9798/**99 * Run one sync round as the lower-key initiator. Device type does not decide100 * the role; the higher-key peer runs [runSyncResponder].101 */102export async function runSyncInitiator(103  session: SyncSession,104  vault: SyncVaultAdapter,105): Promise<SyncOutcome> {106  const result: SyncOutcome = {107    status: "errored",108    entriesReceived: 0,109    entriesSent: 0,110    tombstonesReceived: 0,111    tombstonesSent: 0,112  };113114  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);115  // Privacy-safe progress only: stages contain no device identifiers, relay116  // capabilities, keys, entry ids/counts, or vault content. Keeping the last117  // stage turns an otherwise opaque 30-second timeout into an actionable118  // cross-device diagnostic.119  let stage: SyncStage = "preparing_hello";120  try {121    const ourVaultId = await vault.computeVaultId();122    const ourDeviceId = await vault.getOurDeviceId();123    const localVersion = await vault.getLocalVersion();124    const localTime = Date.now();125    const supportsDeviceManifest =126      vault.buildOutgoingDeviceManifest !== undefined &&127      vault.applyIncomingDeviceManifest !== undefined;128129    await sendMessage(session, {130      type: "SYNC_HELLO",131      vaultId: ourVaultId,132      deviceId: ourDeviceId,133      version: localVersion,134      deviceTime: localTime,135      deviceManifestVersion: supportsDeviceManifest136        ? DEVICE_MANIFEST_VERSION137        : undefined,138    });139    stage = "waiting_for_device_hello";140    const helloIn = await receiveMessage(session, timer);141    if (helloIn.type !== "SYNC_HELLO") {142      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);143    }144    if (helloIn.vaultId !== ourVaultId) {145      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });146      result.status = "vault_mismatch";147      return result;148    }149150    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {151      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);152    }153154    // Equal entry versions do not imply equal device graphs: paired-device155    // introductions and revocations live outside the vault version counter.156    // Exchange the signed graph when both peers advertise support, while an157    // older peer that omits the capability retains the legacy DONE fast path.158    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {159      if (160        supportsDeviceManifest &&161        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION162      ) {163        stage = "building_device_manifest";164        const ours = await vault.buildOutgoingDeviceManifest!();165        await sendMessage(session, {166          type: "SYNC_DEVICE_MANIFEST",167          manifest: ours,168        });169        stage = "waiting_for_device_graph";170        const graphIn = await receiveMessage(session, timer);171        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {172          return errorOutcome(173            result,174            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,175          );176        }177        if (graphIn.manifest) {178          await vault.applyIncomingDeviceManifest!(179            graphIn.manifest,180            session.remoteStaticPublicKey,181          );182        }183      }184      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });185      stage = "waiting_for_device_done";186      const closing = await awaitClosingDone(session, timer);187      if (closing.error) return errorOutcome(result, closing.error);188      result.status = "completed";189      return result;190    }191192    stage = "building_local_manifest";193    const ourEntries = await vault.buildEntryManifest();194    const ourTombstones = await vault.buildTombstoneManifest();195    await sendMessage(session, {196      type: "SYNC_MANIFEST",197      entries: ourEntries,198      tombstones: ourTombstones,199    });200    stage = "waiting_for_device_manifest";201    const manifestIn = await receiveMessage(session, timer);202    if (manifestIn.type !== "SYNC_MANIFEST") {203      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);204    }205206    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);207    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });208209    stage = "waiting_for_device_request";210    const requestIn = await receiveMessage(session, timer);211    if (requestIn.type !== "SYNC_REQUEST") {212      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);213    }214    // Sanity: the peer's request must be a subset of what its manifest215    // declared as newer. A malformed/over-eager request is a protocol216    // violation but not a data-safety problem — we just send what they ask.217218    stage = "building_requested_entries";219    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);220    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });221    result.entriesSent = Object.keys(ourPayload).length;222223    if (wantFromPeer.length > 0) {224      stage = "waiting_for_device_entries";225      const entriesIn = await receiveMessage(session, timer);226      if (entriesIn.type !== "SYNC_ENTRIES") {227        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);228      }229      await vault.applyEntries(entriesIn.entries);230      result.entriesReceived = Object.keys(entriesIn.entries).length;231    }232233    // Tombstones: each side ships the tombstones that the other does not234    // yet have a newer record for.235    const newerTombstonesForPeer: Record<string, number> = {};236    for (const [id, ts] of Object.entries(ourTombstones)) {237      const peerTs = manifestIn.tombstones[id] ?? 0;238      if (ts > peerTs) newerTombstonesForPeer[id] = ts;239    }240    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });241    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;242243    stage = "waiting_for_device_tombstones";244    const tombstonesIn = await receiveMessage(session, timer);245    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {246      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);247    }248    await vault.applyTombstones(tombstonesIn.tombstones);249    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;250251    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends252    // must expose the adapter methods. Disabling on either side cleanly253    // bypasses the step. The exchange is symmetric on the wire; the254    // production adapter accepts only a directly paired issuer whose stored255    // long-term key matches the Noise-authenticated remote static key.256    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {257      stage = "building_device_manifest";258      const ours = await vault.buildOutgoingDeviceManifest();259      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });260      stage = "waiting_for_device_graph";261      const manifestIn = await receiveMessage(session, timer);262      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {263        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);264      }265      if (manifestIn.manifest) {266        await vault.applyIncomingDeviceManifest(267          manifestIn.manifest,268          session.remoteStaticPublicKey,269        );270      }271    }272273    await sendMessage(session, {274      type: "SYNC_DONE",275      entriesReceived: result.entriesReceived,276      entriesSent: result.entriesSent,277    });278    stage = "waiting_for_device_done";279    const closing = await awaitClosingDone(session, timer);280    if (closing.error) return errorOutcome(result, closing.error);281    result.status = "completed";282    return result;283  } catch (err) {284    if ((err as Error).message === "sync_timeout") {285      result.status = "timeout";286      result.error = `sync_timeout:${stage}`;287      return result;288    }289    return errorOutcome(result, (err as Error).message);290  } finally {291    // Session lifetime is owned by the runner (sync-runner.ts), which292    // closes it in its own finally block. The coordinator must not close it293    // here — the runner still needs to record the outcome after we return.294    timer.cancel();295  }296}297298/**299 * Exact complement of [runSyncInitiator] for the higher-key peer holding a300 * relay waiter room. Keeping this in the browser lets two extensions form a301 * direct mesh edge; the device that introduced them is not in the data path.302 */303export async function runSyncResponder(304  session: SyncSession,305  vault: SyncVaultAdapter,306): Promise<SyncOutcome> {307  const result: SyncOutcome = {308    status: "errored",309    entriesReceived: 0,310    entriesSent: 0,311    tombstonesReceived: 0,312    tombstonesSent: 0,313  };314  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);315  let stage: SyncStage = "waiting_for_device_hello";316  try {317    const ourVaultId = await vault.computeVaultId();318    const ourDeviceId = await vault.getOurDeviceId();319    const localVersion = await vault.getLocalVersion();320    const localTime = Date.now();321    const supportsDeviceManifest =322      vault.buildOutgoingDeviceManifest !== undefined &&323      vault.applyIncomingDeviceManifest !== undefined;324325    const helloIn = await receiveMessage(session, timer);326    if (helloIn.type !== "SYNC_HELLO") {327      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);328    }329    await sendMessage(session, {330      type: "SYNC_HELLO",331      vaultId: ourVaultId,332      deviceId: ourDeviceId,333      version: localVersion,334      deviceTime: localTime,335      deviceManifestVersion: supportsDeviceManifest336        ? DEVICE_MANIFEST_VERSION337        : undefined,338    });339    if (helloIn.vaultId !== ourVaultId) {340      result.status = "vault_mismatch";341      return result;342    }343    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {344      console.warn(345        `[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`,346      );347    }348349    if (helloIn.version === localVersion) {350      if (351        supportsDeviceManifest &&352        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION353      ) {354        stage = "waiting_for_device_graph";355        const graphIn = await receiveMessage(session, timer);356        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {357          return errorOutcome(358            result,359            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,360          );361        }362        const ours = await vault.buildOutgoingDeviceManifest!();363        await sendMessage(session, {364          type: "SYNC_DEVICE_MANIFEST",365          manifest: ours,366        });367        if (graphIn.manifest) {368          await vault.applyIncomingDeviceManifest!(369            graphIn.manifest,370            session.remoteStaticPublicKey,371          );372        }373      }374      stage = "waiting_for_device_done";375      const done = await receiveMessage(session, timer);376      if (done.type !== "SYNC_DONE") {377        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);378      }379      await sendMessage(session, {380        type: "SYNC_DONE",381        entriesReceived: 0,382        entriesSent: 0,383      });384      result.status = "completed";385      return result;386    }387388    const ourEntries = await vault.buildEntryManifest();389    const ourTombstones = await vault.buildTombstoneManifest();390    stage = "waiting_for_device_manifest";391    const manifestIn = await receiveMessage(session, timer);392    if (manifestIn.type !== "SYNC_MANIFEST") {393      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);394    }395    await sendMessage(session, {396      type: "SYNC_MANIFEST",397      entries: ourEntries,398      tombstones: ourTombstones,399    });400401    stage = "waiting_for_device_request";402    const requestIn = await receiveMessage(session, timer);403    if (requestIn.type !== "SYNC_REQUEST") {404      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);405    }406    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);407    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });408409    stage = "waiting_for_device_entries";410    const entriesIn = await receiveMessage(session, timer);411    if (entriesIn.type !== "SYNC_ENTRIES") {412      return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);413    }414    if (wantFromPeer.length > 0) {415      await vault.applyEntries(entriesIn.entries);416      result.entriesReceived = Object.keys(entriesIn.entries).length;417    }418    if (requestIn.entryIds.length > 0) {419      const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);420      await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });421      result.entriesSent = Object.keys(ourPayload).length;422    }423424    stage = "waiting_for_device_tombstones";425    const tombstonesIn = await receiveMessage(session, timer);426    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {427      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);428    }429    await vault.applyTombstones(tombstonesIn.tombstones);430    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;431432    const newerTombstonesForPeer: Record<string, number> = {};433    for (const [id, timestamp] of Object.entries(ourTombstones)) {434      if (timestamp > (manifestIn.tombstones[id] ?? 0)) {435        newerTombstonesForPeer[id] = timestamp;436      }437    }438    await sendMessage(session, {439      type: "SYNC_TOMBSTONES",440      tombstones: newerTombstonesForPeer,441    });442    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;443444    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {445      stage = "waiting_for_device_graph";446      const graphIn = await receiveMessage(session, timer);447      if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {448        return errorOutcome(449          result,450          `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,451        );452      }453      const ours = await vault.buildOutgoingDeviceManifest();454      await sendMessage(session, {455        type: "SYNC_DEVICE_MANIFEST",456        manifest: ours,457      });458      if (graphIn.manifest) {459        await vault.applyIncomingDeviceManifest(460          graphIn.manifest,461          session.remoteStaticPublicKey,462        );463      }464    }465466    stage = "waiting_for_device_done";467    const done = await receiveMessage(session, timer);468    if (done.type !== "SYNC_DONE") {469      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);470    }471    await sendMessage(session, {472      type: "SYNC_DONE",473      entriesReceived: result.entriesReceived,474      entriesSent: result.entriesSent,475    });476    result.status = "completed";477    return result;478  } catch (error) {479    if ((error as Error).message === "sync_timeout") {480      result.status = "timeout";481      result.error = `sync_timeout:${stage}`;482      return result;483    }484    return errorOutcome(result, (error as Error).message);485  } finally {486    timer.cancel();487  }488}489490/**491 * Wait for the peer's closing SYNC_DONE echo, treating a hang-up as success.492 *493 * By the time we send our own SYNC_DONE, every entry and tombstone has already494 * been exchanged and applied on both sides — the echo is a courtesy, not a495 * commit. Android closes the socket as soon as it has our DONE, so the parked496 * receive() rejects with "WebSocket closed"; counting that as `run_failed`497 * reported a red indicator for syncs that had fully succeeded and whose data498 * had demonstrably landed on the peer.499 *500 * A wrong *message* here is still a real protocol error — only a closed501 * transport is forgiven.502 */503async function awaitClosingDone(504  session: SyncSession,505  timer: Timer,506): Promise<{ error?: string }> {507  try {508    const done = await receiveMessage(session, timer);509    if (done.type !== "SYNC_DONE") {510      return { error: `expected SYNC_DONE, got ${done.type}` };511    }512    return {};513  } catch (err) {514    if ((err as Error).message === "sync_timeout") throw err;515    if (isConnectionClosedError(err)) return {};516    return { error: (err as Error).message };517  }518}519520function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {521  const want: string[] = [];522  for (const [id, ts] of Object.entries(remote)) {523    const ours = local[id];524    if (ours === undefined || ts > ours) want.push(id);525  }526  return want;527}528529interface Timer {530  signal: AbortSignal;531  cancel(): void;532}533534function scheduleTimeout(ms: number): Timer {535  const ctrl = new AbortController();536  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);537  return {538    signal: ctrl.signal,539    cancel: () => clearTimeout(t),540  };541}542543async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {544  session.send(new TextEncoder().encode(JSON.stringify(msg)));545}546547async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {548  const pending = session.receive();549  const aborted = new Promise<never>((_, reject) => {550    if (timer.signal.aborted) reject(new Error("sync_timeout"));551    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));552  });553  const bytes = await Promise.race([pending, aborted]);554  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS555  // transport may not have enforced anything) — re-check before JSON.parse556  // materialises a multi-GB string/object graph.557  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {558    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);559  }560  const parsed = JSON.parse(new TextDecoder().decode(bytes));561  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");562  return parsed as SyncMessage;563}564565function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {566  outcome.status = "errored";567  outcome.error = error;568  return outcome;569}