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  SYNC_PROTOCOL_VERSION,25  type SyncMessage,26  type SignedDeviceManifest,27} from "../shared/sync-types";28import type { SyncSession } from "../shared/sync-transport";29import { SYNC_LIMITS, validateSyncManifestCounts } from "../shared/sync-limits";30import { isConnectionClosedError } from "../shared/sync-outcome";3132const SESSION_TIMEOUT_MS = 30_000;3334type SyncStage =35  | "preparing_hello"36  | "waiting_for_device_hello"37  | "building_local_manifest"38  | "waiting_for_device_manifest"39  | "waiting_for_device_request"40  | "building_requested_entries"41  | "waiting_for_device_entries"42  | "waiting_for_device_tombstones"43  | "building_device_manifest"44  | "waiting_for_device_graph"45  | "waiting_for_device_done";4647/**48 * Vault-side adapter — the coordinator's dependency on the vault layer.49 * The vault holds the canonical entry/tombstone tables; the coordinator50 * pulls a manifest, requests entries by id, applies remote entries, and51 * applies remote tombstones.52 */53export interface SyncVaultAdapter {54  computeVaultId(): Promise<string>;55  getOurDeviceId(): Promise<string>;56  /** Returns the vault's monotonic version (`peach_last_modified_at`). */57  getLocalVersion(): Promise<number>;58  /** Returns the per-entry mtimes for our active entries. */59  buildEntryManifest(): Promise<Record<string, number>>;60  /** Returns the per-entry deletion timestamps for our tombstones. */61  buildTombstoneManifest(): Promise<Record<string, number>>;62  /** Returns base64 ciphertext for each requested entry id. */63  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;64  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */65  applyEntries(66    entries: Record<string, string>,67    remoteManifest?: Record<string, number>,68  ): Promise<void>;69  /** Apply incoming tombstones; vault retains the latest deletion ts. */70  applyTombstones(tombstones: Record<string, number>): Promise<void>;71  /**72   * Optional — v2 brief §4.3 device manifest exchange. The extension (as73   * the trust anchor for the device graph) returns a signed manifest74   * enumerating every paired device it knows about; the peer verifies it75   * against the extension's known Ed25519 signing pubkey and upserts any76   * newly-vouched devices into its own registry.77   *78   * When this method is undefined, the coordinator skips the manifest79   * step entirely. Both ends must opt in for the exchange to occur.80   */81  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;82  /**83   * Handle a manifest received from the peer. The transport-authenticated84   * Noise static key is supplied separately so a self-asserted issuer id85   * cannot select a different paired-device verification key.86   */87  applyIncomingDeviceManifest?(88    manifest: SignedDeviceManifest,89    authenticatedRemoteStaticPublicKey: Uint8Array,90  ): Promise<void>;91}9293export interface SyncOutcome {94  status: "completed" | "vault_mismatch" | "errored" | "timeout";95  entriesReceived: number;96  entriesSent: number;97  tombstonesReceived: number;98  tombstonesSent: number;99  error?: string;100}101102/**103 * Run one sync round as the lower-key initiator. Device type does not decide104 * the role; the higher-key peer runs [runSyncResponder].105 */106export async function runSyncInitiator(107  session: SyncSession,108  vault: SyncVaultAdapter,109): Promise<SyncOutcome> {110  const result: SyncOutcome = {111    status: "errored",112    entriesReceived: 0,113    entriesSent: 0,114    tombstonesReceived: 0,115    tombstonesSent: 0,116  };117118  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);119  // Privacy-safe progress only: stages contain no device identifiers, relay120  // capabilities, keys, entry ids/counts, or vault content. Keeping the last121  // stage turns an otherwise opaque 30-second timeout into an actionable122  // cross-device diagnostic.123  let stage: SyncStage = "preparing_hello";124  try {125    const ourVaultId = await vault.computeVaultId();126    const ourDeviceId = await vault.getOurDeviceId();127    const localVersion = await vault.getLocalVersion();128    const localTime = Date.now();129    const supportsDeviceManifest =130      vault.buildOutgoingDeviceManifest !== undefined &&131      vault.applyIncomingDeviceManifest !== undefined;132133    await sendMessage(session, {134      type: "SYNC_HELLO",135      vaultId: ourVaultId,136      deviceId: ourDeviceId,137      version: localVersion,138      deviceTime: localTime,139      protocolVersion: SYNC_PROTOCOL_VERSION,140      deviceManifestVersion: supportsDeviceManifest141        ? DEVICE_MANIFEST_VERSION142        : undefined,143    });144    stage = "waiting_for_device_hello";145    const helloIn = await receiveMessage(session, timer);146    if (helloIn.type !== "SYNC_HELLO") {147      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);148    }149    if ((helloIn.protocolVersion ?? 1) < SYNC_PROTOCOL_VERSION) {150      await sendMessage(session, { type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" });151      return errorOutcome(result, "PEER_PROTOCOL_OUTDATED");152    }153    if (helloIn.vaultId !== ourVaultId) {154      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });155      result.status = "vault_mismatch";156      return result;157    }158159    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {160      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);161    }162163    // Equal entry versions do not imply equal device graphs: paired-device164    // introductions and revocations live outside the vault version counter.165    // Exchange the signed graph when both peers advertise support, while an166    // older peer that omits the capability retains the legacy DONE fast path.167    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {168      if (169        supportsDeviceManifest &&170        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION171      ) {172        stage = "building_device_manifest";173        const ours = await vault.buildOutgoingDeviceManifest!();174        await sendMessage(session, {175          type: "SYNC_DEVICE_MANIFEST",176          manifest: ours,177        });178        stage = "waiting_for_device_graph";179        const graphIn = await receiveMessage(session, timer);180        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {181          return errorOutcome(182            result,183            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,184          );185        }186        if (graphIn.manifest) {187          await vault.applyIncomingDeviceManifest!(188            graphIn.manifest,189            session.remoteStaticPublicKey,190          );191        }192      }193      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });194      stage = "waiting_for_device_done";195      const closing = await awaitClosingDone(session, timer);196      if (closing.error) return errorOutcome(result, closing.error);197      result.status = "completed";198      return result;199    }200201    stage = "building_local_manifest";202    const ourEntries = await vault.buildEntryManifest();203    const ourTombstones = await vault.buildTombstoneManifest();204    await sendMessage(session, {205      type: "SYNC_MANIFEST",206      entries: ourEntries,207      tombstones: ourTombstones,208    });209    stage = "waiting_for_device_manifest";210    const manifestIn = await receiveMessage(session, timer);211    if (manifestIn.type !== "SYNC_MANIFEST") {212      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);213    }214    validateSyncManifestCounts(manifestIn.entries, manifestIn.tombstones);215216    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);217    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });218219    stage = "waiting_for_device_request";220    const requestIn = await receiveMessage(session, timer);221    if (requestIn.type !== "SYNC_REQUEST") {222      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);223    }224    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {225      return errorOutcome(result, "sync request entry count exceeds limit");226    }227    // Sanity: the peer's request must be a subset of what its manifest228    // declared as newer. A malformed/over-eager request is a protocol229    // violation but not a data-safety problem — we just send what they ask.230231    stage = "building_requested_entries";232    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);233    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });234    result.entriesSent = Object.keys(ourPayload).length;235236    if (wantFromPeer.length > 0) {237      stage = "waiting_for_device_entries";238      const entriesIn = await receiveMessage(session, timer);239      if (entriesIn.type !== "SYNC_ENTRIES") {240        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);241      }242      await vault.applyEntries(entriesIn.entries, manifestIn.entries);243      result.entriesReceived = Object.keys(entriesIn.entries).length;244    }245246    // Tombstones: each side ships the tombstones that the other does not247    // yet have a newer record for.248    const newerTombstonesForPeer: Record<string, number> = {};249    for (const [id, ts] of Object.entries(ourTombstones)) {250      const peerTs = manifestIn.tombstones[id] ?? 0;251      if (ts > peerTs) newerTombstonesForPeer[id] = ts;252    }253    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });254    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;255256    stage = "waiting_for_device_tombstones";257    const tombstonesIn = await receiveMessage(session, timer);258    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {259      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);260    }261    await vault.applyTombstones(tombstonesIn.tombstones);262    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;263264    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends265    // must expose the adapter methods. Disabling on either side cleanly266    // bypasses the step. The exchange is symmetric on the wire; the267    // production adapter accepts only a directly paired issuer whose stored268    // long-term key matches the Noise-authenticated remote static key.269    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {270      stage = "building_device_manifest";271      const ours = await vault.buildOutgoingDeviceManifest();272      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });273      stage = "waiting_for_device_graph";274      const manifestIn = await receiveMessage(session, timer);275      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {276        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);277      }278      if (manifestIn.manifest) {279        await vault.applyIncomingDeviceManifest(280          manifestIn.manifest,281          session.remoteStaticPublicKey,282        );283      }284    }285286    await sendMessage(session, {287      type: "SYNC_DONE",288      entriesReceived: result.entriesReceived,289      entriesSent: result.entriesSent,290    });291    stage = "waiting_for_device_done";292    const closing = await awaitClosingDone(session, timer);293    if (closing.error) return errorOutcome(result, closing.error);294    result.status = "completed";295    return result;296  } catch (err) {297    if ((err as Error).message === "sync_timeout") {298      result.status = "timeout";299      result.error = `sync_timeout:${stage}`;300      return result;301    }302    return errorOutcome(result, (err as Error).message);303  } finally {304    // Session lifetime is owned by the runner (sync-runner.ts), which305    // closes it in its own finally block. The coordinator must not close it306    // here — the runner still needs to record the outcome after we return.307    timer.cancel();308  }309}310311/**312 * Exact complement of [runSyncInitiator] for the higher-key peer holding a313 * relay waiter room. Keeping this in the browser lets two extensions form a314 * direct mesh edge; the device that introduced them is not in the data path.315 */316export async function runSyncResponder(317  session: SyncSession,318  vault: SyncVaultAdapter,319): Promise<SyncOutcome> {320  const result: SyncOutcome = {321    status: "errored",322    entriesReceived: 0,323    entriesSent: 0,324    tombstonesReceived: 0,325    tombstonesSent: 0,326  };327  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);328  let stage: SyncStage = "waiting_for_device_hello";329  try {330    const ourVaultId = await vault.computeVaultId();331    const ourDeviceId = await vault.getOurDeviceId();332    const localVersion = await vault.getLocalVersion();333    const localTime = Date.now();334    const supportsDeviceManifest =335      vault.buildOutgoingDeviceManifest !== undefined &&336      vault.applyIncomingDeviceManifest !== undefined;337338    const helloIn = await receiveMessage(session, timer);339    if (helloIn.type !== "SYNC_HELLO") {340      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);341    }342    if ((helloIn.protocolVersion ?? 1) < SYNC_PROTOCOL_VERSION) {343      await sendMessage(session, { type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" });344      return errorOutcome(result, "PEER_PROTOCOL_OUTDATED");345    }346    await sendMessage(session, {347      type: "SYNC_HELLO",348      vaultId: ourVaultId,349      deviceId: ourDeviceId,350      version: localVersion,351      deviceTime: localTime,352      protocolVersion: SYNC_PROTOCOL_VERSION,353      deviceManifestVersion: supportsDeviceManifest354        ? DEVICE_MANIFEST_VERSION355        : undefined,356    });357    if (helloIn.vaultId !== ourVaultId) {358      result.status = "vault_mismatch";359      return result;360    }361    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {362      console.warn(363        `[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`,364      );365    }366367    if (helloIn.version === localVersion) {368      if (369        supportsDeviceManifest &&370        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION371      ) {372        stage = "waiting_for_device_graph";373        const graphIn = await receiveMessage(session, timer);374        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {375          return errorOutcome(376            result,377            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,378          );379        }380        const ours = await vault.buildOutgoingDeviceManifest!();381        await sendMessage(session, {382          type: "SYNC_DEVICE_MANIFEST",383          manifest: ours,384        });385        if (graphIn.manifest) {386          await vault.applyIncomingDeviceManifest!(387            graphIn.manifest,388            session.remoteStaticPublicKey,389          );390        }391      }392      stage = "waiting_for_device_done";393      const done = await receiveMessage(session, timer);394      if (done.type !== "SYNC_DONE") {395        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);396      }397      await sendMessage(session, {398        type: "SYNC_DONE",399        entriesReceived: 0,400        entriesSent: 0,401      });402      result.status = "completed";403      return result;404    }405406    const ourEntries = await vault.buildEntryManifest();407    const ourTombstones = await vault.buildTombstoneManifest();408    stage = "waiting_for_device_manifest";409    const manifestIn = await receiveMessage(session, timer);410    if (manifestIn.type !== "SYNC_MANIFEST") {411      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);412    }413    validateSyncManifestCounts(manifestIn.entries, manifestIn.tombstones);414    await sendMessage(session, {415      type: "SYNC_MANIFEST",416      entries: ourEntries,417      tombstones: ourTombstones,418    });419420    stage = "waiting_for_device_request";421    const requestIn = await receiveMessage(session, timer);422    if (requestIn.type !== "SYNC_REQUEST") {423      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);424    }425    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {426      return errorOutcome(result, "sync request entry count exceeds limit");427    }428    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);429    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });430431    stage = "waiting_for_device_entries";432    const entriesIn = await receiveMessage(session, timer);433    if (entriesIn.type !== "SYNC_ENTRIES") {434      return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);435    }436    if (wantFromPeer.length > 0) {437      await vault.applyEntries(entriesIn.entries, manifestIn.entries);438      result.entriesReceived = Object.keys(entriesIn.entries).length;439    }440    if (requestIn.entryIds.length > 0) {441      const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);442      await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });443      result.entriesSent = Object.keys(ourPayload).length;444    }445446    stage = "waiting_for_device_tombstones";447    const tombstonesIn = await receiveMessage(session, timer);448    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {449      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);450    }451    await vault.applyTombstones(tombstonesIn.tombstones);452    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;453454    const newerTombstonesForPeer: Record<string, number> = {};455    for (const [id, timestamp] of Object.entries(ourTombstones)) {456      if (timestamp > (manifestIn.tombstones[id] ?? 0)) {457        newerTombstonesForPeer[id] = timestamp;458      }459    }460    await sendMessage(session, {461      type: "SYNC_TOMBSTONES",462      tombstones: newerTombstonesForPeer,463    });464    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;465466    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {467      stage = "waiting_for_device_graph";468      const graphIn = await receiveMessage(session, timer);469      if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {470        return errorOutcome(471          result,472          `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,473        );474      }475      const ours = await vault.buildOutgoingDeviceManifest();476      await sendMessage(session, {477        type: "SYNC_DEVICE_MANIFEST",478        manifest: ours,479      });480      if (graphIn.manifest) {481        await vault.applyIncomingDeviceManifest(482          graphIn.manifest,483          session.remoteStaticPublicKey,484        );485      }486    }487488    stage = "waiting_for_device_done";489    const done = await receiveMessage(session, timer);490    if (done.type !== "SYNC_DONE") {491      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);492    }493    await sendMessage(session, {494      type: "SYNC_DONE",495      entriesReceived: result.entriesReceived,496      entriesSent: result.entriesSent,497    });498    result.status = "completed";499    return result;500  } catch (error) {501    if ((error as Error).message === "sync_timeout") {502      result.status = "timeout";503      result.error = `sync_timeout:${stage}`;504      return result;505    }506    return errorOutcome(result, (error as Error).message);507  } finally {508    timer.cancel();509  }510}511512/**513 * Wait for the peer's closing SYNC_DONE echo, treating a hang-up as success.514 *515 * By the time we send our own SYNC_DONE, every entry and tombstone has already516 * been exchanged and applied on both sides — the echo is a courtesy, not a517 * commit. Android closes the socket as soon as it has our DONE, so the parked518 * receive() rejects with "WebSocket closed"; counting that as `run_failed`519 * reported a red indicator for syncs that had fully succeeded and whose data520 * had demonstrably landed on the peer.521 *522 * A wrong *message* here is still a real protocol error — only a closed523 * transport is forgiven.524 */525async function awaitClosingDone(526  session: SyncSession,527  timer: Timer,528): Promise<{ error?: string }> {529  try {530    const done = await receiveMessage(session, timer);531    if (done.type !== "SYNC_DONE") {532      return { error: `expected SYNC_DONE, got ${done.type}` };533    }534    return {};535  } catch (err) {536    if ((err as Error).message === "sync_timeout") throw err;537    if (isConnectionClosedError(err)) return {};538    return { error: (err as Error).message };539  }540}541542function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {543  const want: string[] = [];544  for (const [id, ts] of Object.entries(remote)) {545    const ours = local[id];546    if (ours === undefined || ts > ours) want.push(id);547  }548  return want;549}550551interface Timer {552  signal: AbortSignal;553  cancel(): void;554}555556function scheduleTimeout(ms: number): Timer {557  const ctrl = new AbortController();558  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);559  return {560    signal: ctrl.signal,561    cancel: () => clearTimeout(t),562  };563}564565async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {566  session.send(new TextEncoder().encode(JSON.stringify(msg)));567}568569async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {570  const pending = session.receive();571  const aborted = new Promise<never>((_, reject) => {572    if (timer.signal.aborted) reject(new Error("sync_timeout"));573    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));574  });575  const bytes = await Promise.race([pending, aborted]);576  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS577  // transport may not have enforced anything) — re-check before JSON.parse578  // materialises a multi-GB string/object graph.579  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {580    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);581  }582  const parsed = JSON.parse(new TextDecoder().decode(bytes));583  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");584  return parsed as SyncMessage;585}586587function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {588  outcome.status = "errored";589  outcome.error = error;590  return outcome;591}