sync-coordinator.ts

1// Vault sync — Phase 2 manifest-diff coordinator.2//3// After Noise NK is open and PAIR_IDENTITY has been exchanged4// (`shared/sync-transport.ts`, HANDSHAKE_STANDARDIZE_ON_NK.md), this5// coordinator drives the manifest exchange:6//7//   ->  SYNC_HELLO { vaultId, deviceId }8//   <-  SYNC_HELLO { vaultId, deviceId }       (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_DONE16//17// Entries are opaque base64 ciphertext — the coordinator never decrypts.18// Last-write-wins at the entry level by mtime; tombstones override updates19// with a later deletion timestamp.2021import type { SyncMessage, SignedDeviceManifest } from "../shared/sync-types";22import type { SyncSession } from "../shared/sync-transport";23import { SYNC_LIMITS } from "../shared/sync-limits";2425const SESSION_TIMEOUT_MS = 30_000;2627/**28 * Vault-side adapter — the coordinator's dependency on the vault layer.29 * The vault holds the canonical entry/tombstone tables; the coordinator30 * pulls a manifest, requests entries by id, applies remote entries, and31 * applies remote tombstones.32 */33export interface SyncVaultAdapter {34  computeVaultId(): Promise<string>;35  getOurDeviceId(): Promise<string>;36  /** Returns the vault's monotonic version (`peach_last_modified_at`). */37  getLocalVersion(): Promise<number>;38  /** Returns the per-entry mtimes for our active entries. */39  buildEntryManifest(): Promise<Record<string, number>>;40  /** Returns the per-entry deletion timestamps for our tombstones. */41  buildTombstoneManifest(): Promise<Record<string, number>>;42  /** Returns base64 ciphertext for each requested entry id. */43  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;44  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */45  applyEntries(entries: Record<string, string>): Promise<void>;46  /** Apply incoming tombstones; vault retains the latest deletion ts. */47  applyTombstones(tombstones: Record<string, number>): Promise<void>;48  /**49   * Optional — v2 brief §4.3 device manifest exchange. The extension (as50   * the trust anchor for the device graph) returns a signed manifest51   * enumerating every paired device it knows about; the peer verifies it52   * against the extension's known Ed25519 signing pubkey and upserts any53   * newly-vouched devices into its own registry.54   *55   * When this method is undefined, the coordinator skips the manifest56   * step entirely. Both ends must opt in for the exchange to occur.57   */58  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;59  /**60   * Handle a manifest received from the peer. The transport-authenticated61   * Noise static key is supplied separately so a self-asserted issuer id62   * cannot select a different paired-device verification key.63   */64  applyIncomingDeviceManifest?(65    manifest: SignedDeviceManifest,66    authenticatedRemoteStaticPublicKey: Uint8Array,67  ): Promise<void>;68}6970export interface SyncOutcome {71  status: "completed" | "vault_mismatch" | "errored" | "timeout";72  entriesReceived: number;73  entriesSent: number;74  tombstonesReceived: number;75  tombstonesSent: number;76  error?: string;77}7879/**80 * Run one sync round as the initiator. The extension is always the81 * initiator (it is the WebSocket client). The peer's role is symmetric82 * after SYNC_HELLO, so a corresponding responder helper exists for the83 * mobile-side coordinator that calls this same protocol the other way.84 */85export async function runSyncInitiator(86  session: SyncSession,87  vault: SyncVaultAdapter,88): Promise<SyncOutcome> {89  const result: SyncOutcome = {90    status: "errored",91    entriesReceived: 0,92    entriesSent: 0,93    tombstonesReceived: 0,94    tombstonesSent: 0,95  };9697  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);98  try {99    const ourVaultId = await vault.computeVaultId();100    const ourDeviceId = await vault.getOurDeviceId();101    const localVersion = await vault.getLocalVersion();102    const localTime = Date.now();103104    await sendMessage(session, {105      type: "SYNC_HELLO",106      vaultId: ourVaultId,107      deviceId: ourDeviceId,108      version: localVersion,109      deviceTime: localTime,110    });111    const helloIn = await receiveMessage(session, timer);112    if (helloIn.type !== "SYNC_HELLO") {113      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);114    }115    if (helloIn.vaultId !== ourVaultId) {116      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });117      result.status = "vault_mismatch";118      return result;119    }120121    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {122      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);123    }124125    // Equal versions — nothing to exchange.126    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {127      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });128      const done = await receiveMessage(session, timer);129      if (done.type !== "SYNC_DONE") {130        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);131      }132      result.status = "completed";133      return result;134    }135136    const ourEntries = await vault.buildEntryManifest();137    const ourTombstones = await vault.buildTombstoneManifest();138    await sendMessage(session, {139      type: "SYNC_MANIFEST",140      entries: ourEntries,141      tombstones: ourTombstones,142    });143    const manifestIn = await receiveMessage(session, timer);144    if (manifestIn.type !== "SYNC_MANIFEST") {145      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);146    }147148    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);149    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });150151    const requestIn = await receiveMessage(session, timer);152    if (requestIn.type !== "SYNC_REQUEST") {153      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);154    }155    // Sanity: the peer's request must be a subset of what its manifest156    // declared as newer. A malformed/over-eager request is a protocol157    // violation but not a data-safety problem — we just send what they ask.158159    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);160    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });161    result.entriesSent = Object.keys(ourPayload).length;162163    if (wantFromPeer.length > 0) {164      const entriesIn = await receiveMessage(session, timer);165      if (entriesIn.type !== "SYNC_ENTRIES") {166        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);167      }168      await vault.applyEntries(entriesIn.entries);169      result.entriesReceived = Object.keys(entriesIn.entries).length;170    }171172    // Tombstones: each side ships the tombstones that the other does not173    // yet have a newer record for.174    const newerTombstonesForPeer: Record<string, number> = {};175    for (const [id, ts] of Object.entries(ourTombstones)) {176      const peerTs = manifestIn.tombstones[id] ?? 0;177      if (ts > peerTs) newerTombstonesForPeer[id] = ts;178    }179    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });180    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;181182    const tombstonesIn = await receiveMessage(session, timer);183    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {184      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);185    }186    await vault.applyTombstones(tombstonesIn.tombstones);187    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;188189    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends190    // must expose the adapter methods. Disabling on either side cleanly191    // bypasses the step. The exchange is symmetric on the wire; the192    // production adapter accepts only a directly paired issuer whose stored193    // long-term key matches the Noise-authenticated remote static key.194    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {195      const ours = await vault.buildOutgoingDeviceManifest();196      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });197      const manifestIn = await receiveMessage(session, timer);198      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {199        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);200      }201      if (manifestIn.manifest) {202        await vault.applyIncomingDeviceManifest(203          manifestIn.manifest,204          session.remoteStaticPublicKey,205        );206      }207    }208209    await sendMessage(session, {210      type: "SYNC_DONE",211      entriesReceived: result.entriesReceived,212      entriesSent: result.entriesSent,213    });214    const done = await receiveMessage(session, timer);215    if (done.type !== "SYNC_DONE") {216      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);217    }218    result.status = "completed";219    return result;220  } catch (err) {221    if ((err as Error).message === "sync_timeout") {222      result.status = "timeout";223      result.error = "session timed out after 30s";224      return result;225    }226    return errorOutcome(result, (err as Error).message);227  } finally {228    // Session lifetime is owned by the runner (sync-runner.ts), which229    // closes it in its own finally block. The coordinator must not close it230    // here — the runner still needs to record the outcome after we return.231    timer.cancel();232  }233}234235function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {236  const want: string[] = [];237  for (const [id, ts] of Object.entries(remote)) {238    const ours = local[id];239    if (ours === undefined || ts > ours) want.push(id);240  }241  return want;242}243244interface Timer {245  signal: AbortSignal;246  cancel(): void;247}248249function scheduleTimeout(ms: number): Timer {250  const ctrl = new AbortController();251  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);252  return {253    signal: ctrl.signal,254    cancel: () => clearTimeout(t),255  };256}257258async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {259  session.send(new TextEncoder().encode(JSON.stringify(msg)));260}261262async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {263  const pending = session.receive();264  const aborted = new Promise<never>((_, reject) => {265    if (timer.signal.aborted) reject(new Error("sync_timeout"));266    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));267  });268  const bytes = await Promise.race([pending, aborted]);269  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS270  // transport may not have enforced anything) — re-check before JSON.parse271  // materialises a multi-GB string/object graph.272  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {273    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);274  }275  const parsed = JSON.parse(new TextDecoder().decode(bytes));276  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");277  return parsed as SyncMessage;278}279280function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {281  outcome.status = "errored";282  outcome.error = error;283  return outcome;284}