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 initiator. The extension is always the94 * initiator (it is the WebSocket client). The peer's role is symmetric95 * after SYNC_HELLO, so a corresponding responder helper exists for the96 * mobile-side coordinator that calls this same protocol the other way.97 */98export async function runSyncInitiator(99  session: SyncSession,100  vault: SyncVaultAdapter,101): Promise<SyncOutcome> {102  const result: SyncOutcome = {103    status: "errored",104    entriesReceived: 0,105    entriesSent: 0,106    tombstonesReceived: 0,107    tombstonesSent: 0,108  };109110  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);111  // Privacy-safe progress only: stages contain no device identifiers, relay112  // capabilities, keys, entry ids/counts, or vault content. Keeping the last113  // stage turns an otherwise opaque 30-second timeout into an actionable114  // cross-device diagnostic.115  let stage: SyncStage = "preparing_hello";116  try {117    const ourVaultId = await vault.computeVaultId();118    const ourDeviceId = await vault.getOurDeviceId();119    const localVersion = await vault.getLocalVersion();120    const localTime = Date.now();121122    await sendMessage(session, {123      type: "SYNC_HELLO",124      vaultId: ourVaultId,125      deviceId: ourDeviceId,126      version: localVersion,127      deviceTime: localTime,128    });129    stage = "waiting_for_device_hello";130    const helloIn = await receiveMessage(session, timer);131    if (helloIn.type !== "SYNC_HELLO") {132      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);133    }134    if (helloIn.vaultId !== ourVaultId) {135      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });136      result.status = "vault_mismatch";137      return result;138    }139140    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {141      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);142    }143144    // Equal versions — nothing to exchange.145    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {146      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });147      stage = "waiting_for_device_done";148      const closing = await awaitClosingDone(session, timer);149      if (closing.error) return errorOutcome(result, closing.error);150      result.status = "completed";151      return result;152    }153154    stage = "building_local_manifest";155    const ourEntries = await vault.buildEntryManifest();156    const ourTombstones = await vault.buildTombstoneManifest();157    await sendMessage(session, {158      type: "SYNC_MANIFEST",159      entries: ourEntries,160      tombstones: ourTombstones,161    });162    stage = "waiting_for_device_manifest";163    const manifestIn = await receiveMessage(session, timer);164    if (manifestIn.type !== "SYNC_MANIFEST") {165      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);166    }167168    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);169    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });170171    stage = "waiting_for_device_request";172    const requestIn = await receiveMessage(session, timer);173    if (requestIn.type !== "SYNC_REQUEST") {174      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);175    }176    // Sanity: the peer's request must be a subset of what its manifest177    // declared as newer. A malformed/over-eager request is a protocol178    // violation but not a data-safety problem — we just send what they ask.179180    stage = "building_requested_entries";181    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);182    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });183    result.entriesSent = Object.keys(ourPayload).length;184185    if (wantFromPeer.length > 0) {186      stage = "waiting_for_device_entries";187      const entriesIn = await receiveMessage(session, timer);188      if (entriesIn.type !== "SYNC_ENTRIES") {189        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);190      }191      await vault.applyEntries(entriesIn.entries);192      result.entriesReceived = Object.keys(entriesIn.entries).length;193    }194195    // Tombstones: each side ships the tombstones that the other does not196    // yet have a newer record for.197    const newerTombstonesForPeer: Record<string, number> = {};198    for (const [id, ts] of Object.entries(ourTombstones)) {199      const peerTs = manifestIn.tombstones[id] ?? 0;200      if (ts > peerTs) newerTombstonesForPeer[id] = ts;201    }202    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });203    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;204205    stage = "waiting_for_device_tombstones";206    const tombstonesIn = await receiveMessage(session, timer);207    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {208      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);209    }210    await vault.applyTombstones(tombstonesIn.tombstones);211    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;212213    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends214    // must expose the adapter methods. Disabling on either side cleanly215    // bypasses the step. The exchange is symmetric on the wire; the216    // production adapter accepts only a directly paired issuer whose stored217    // long-term key matches the Noise-authenticated remote static key.218    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {219      stage = "building_device_manifest";220      const ours = await vault.buildOutgoingDeviceManifest();221      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });222      stage = "waiting_for_device_graph";223      const manifestIn = await receiveMessage(session, timer);224      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {225        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);226      }227      if (manifestIn.manifest) {228        await vault.applyIncomingDeviceManifest(229          manifestIn.manifest,230          session.remoteStaticPublicKey,231        );232      }233    }234235    await sendMessage(session, {236      type: "SYNC_DONE",237      entriesReceived: result.entriesReceived,238      entriesSent: result.entriesSent,239    });240    stage = "waiting_for_device_done";241    const closing = await awaitClosingDone(session, timer);242    if (closing.error) return errorOutcome(result, closing.error);243    result.status = "completed";244    return result;245  } catch (err) {246    if ((err as Error).message === "sync_timeout") {247      result.status = "timeout";248      result.error = `sync_timeout:${stage}`;249      return result;250    }251    return errorOutcome(result, (err as Error).message);252  } finally {253    // Session lifetime is owned by the runner (sync-runner.ts), which254    // closes it in its own finally block. The coordinator must not close it255    // here — the runner still needs to record the outcome after we return.256    timer.cancel();257  }258}259260/**261 * Wait for the peer's closing SYNC_DONE echo, treating a hang-up as success.262 *263 * By the time we send our own SYNC_DONE, every entry and tombstone has already264 * been exchanged and applied on both sides — the echo is a courtesy, not a265 * commit. Android closes the socket as soon as it has our DONE, so the parked266 * receive() rejects with "WebSocket closed"; counting that as `run_failed`267 * reported a red indicator for syncs that had fully succeeded and whose data268 * had demonstrably landed on the peer.269 *270 * A wrong *message* here is still a real protocol error — only a closed271 * transport is forgiven.272 */273async function awaitClosingDone(274  session: SyncSession,275  timer: Timer,276): Promise<{ error?: string }> {277  try {278    const done = await receiveMessage(session, timer);279    if (done.type !== "SYNC_DONE") {280      return { error: `expected SYNC_DONE, got ${done.type}` };281    }282    return {};283  } catch (err) {284    if ((err as Error).message === "sync_timeout") throw err;285    if (isConnectionClosedError(err)) return {};286    return { error: (err as Error).message };287  }288}289290function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {291  const want: string[] = [];292  for (const [id, ts] of Object.entries(remote)) {293    const ours = local[id];294    if (ours === undefined || ts > ours) want.push(id);295  }296  return want;297}298299interface Timer {300  signal: AbortSignal;301  cancel(): void;302}303304function scheduleTimeout(ms: number): Timer {305  const ctrl = new AbortController();306  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);307  return {308    signal: ctrl.signal,309    cancel: () => clearTimeout(t),310  };311}312313async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {314  session.send(new TextEncoder().encode(JSON.stringify(msg)));315}316317async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {318  const pending = session.receive();319  const aborted = new Promise<never>((_, reject) => {320    if (timer.signal.aborted) reject(new Error("sync_timeout"));321    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));322  });323  const bytes = await Promise.race([pending, aborted]);324  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS325  // transport may not have enforced anything) — re-check before JSON.parse326  // materialises a multi-GB string/object graph.327  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {328    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);329  }330  const parsed = JSON.parse(new TextDecoder().decode(bytes));331  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");332  return parsed as SyncMessage;333}334335function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {336  outcome.status = "errored";337  outcome.error = error;338  return outcome;339}