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(65    entries: Record<string, string>,66    remoteManifest?: Record<string, number>,67  ): Promise<void>;68  /** Apply incoming tombstones; vault retains the latest deletion ts. */69  applyTombstones(tombstones: Record<string, number>): Promise<void>;70  /**71   * Optional — v2 brief §4.3 device manifest exchange. The extension (as72   * the trust anchor for the device graph) returns a signed manifest73   * enumerating every paired device it knows about; the peer verifies it74   * against the extension's known Ed25519 signing pubkey and upserts any75   * newly-vouched devices into its own registry.76   *77   * When this method is undefined, the coordinator skips the manifest78   * step entirely. Both ends must opt in for the exchange to occur.79   */80  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;81  /**82   * Handle a manifest received from the peer. The transport-authenticated83   * Noise static key is supplied separately so a self-asserted issuer id84   * cannot select a different paired-device verification key.85   */86  applyIncomingDeviceManifest?(87    manifest: SignedDeviceManifest,88    authenticatedRemoteStaticPublicKey: Uint8Array,89  ): Promise<void>;90}9192export interface SyncOutcome {93  status: "completed" | "vault_mismatch" | "errored" | "timeout";94  entriesReceived: number;95  entriesSent: number;96  tombstonesReceived: number;97  tombstonesSent: number;98  error?: string;99}100101/**102 * Run one sync round as the lower-key initiator. Device type does not decide103 * the role; the higher-key peer runs [runSyncResponder].104 */105export async function runSyncInitiator(106  session: SyncSession,107  vault: SyncVaultAdapter,108): Promise<SyncOutcome> {109  const result: SyncOutcome = {110    status: "errored",111    entriesReceived: 0,112    entriesSent: 0,113    tombstonesReceived: 0,114    tombstonesSent: 0,115  };116117  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);118  // Privacy-safe progress only: stages contain no device identifiers, relay119  // capabilities, keys, entry ids/counts, or vault content. Keeping the last120  // stage turns an otherwise opaque 30-second timeout into an actionable121  // cross-device diagnostic.122  let stage: SyncStage = "preparing_hello";123  try {124    const ourVaultId = await vault.computeVaultId();125    const ourDeviceId = await vault.getOurDeviceId();126    const localVersion = await vault.getLocalVersion();127    const localTime = Date.now();128    const supportsDeviceManifest =129      vault.buildOutgoingDeviceManifest !== undefined &&130      vault.applyIncomingDeviceManifest !== undefined;131132    await sendMessage(session, {133      type: "SYNC_HELLO",134      vaultId: ourVaultId,135      deviceId: ourDeviceId,136      version: localVersion,137      deviceTime: localTime,138      deviceManifestVersion: supportsDeviceManifest139        ? DEVICE_MANIFEST_VERSION140        : undefined,141    });142    stage = "waiting_for_device_hello";143    const helloIn = await receiveMessage(session, timer);144    if (helloIn.type !== "SYNC_HELLO") {145      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);146    }147    if (helloIn.vaultId !== ourVaultId) {148      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });149      result.status = "vault_mismatch";150      return result;151    }152153    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {154      console.warn(`[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);155    }156157    // Equal entry versions do not imply equal device graphs: paired-device158    // introductions and revocations live outside the vault version counter.159    // Exchange the signed graph when both peers advertise support, while an160    // older peer that omits the capability retains the legacy DONE fast path.161    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {162      if (163        supportsDeviceManifest &&164        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION165      ) {166        stage = "building_device_manifest";167        const ours = await vault.buildOutgoingDeviceManifest!();168        await sendMessage(session, {169          type: "SYNC_DEVICE_MANIFEST",170          manifest: ours,171        });172        stage = "waiting_for_device_graph";173        const graphIn = await receiveMessage(session, timer);174        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {175          return errorOutcome(176            result,177            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,178          );179        }180        if (graphIn.manifest) {181          await vault.applyIncomingDeviceManifest!(182            graphIn.manifest,183            session.remoteStaticPublicKey,184          );185        }186      }187      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });188      stage = "waiting_for_device_done";189      const closing = await awaitClosingDone(session, timer);190      if (closing.error) return errorOutcome(result, closing.error);191      result.status = "completed";192      return result;193    }194195    stage = "building_local_manifest";196    const ourEntries = await vault.buildEntryManifest();197    const ourTombstones = await vault.buildTombstoneManifest();198    await sendMessage(session, {199      type: "SYNC_MANIFEST",200      entries: ourEntries,201      tombstones: ourTombstones,202    });203    stage = "waiting_for_device_manifest";204    const manifestIn = await receiveMessage(session, timer);205    if (manifestIn.type !== "SYNC_MANIFEST") {206      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);207    }208209    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);210    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });211212    stage = "waiting_for_device_request";213    const requestIn = await receiveMessage(session, timer);214    if (requestIn.type !== "SYNC_REQUEST") {215      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);216    }217    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {218      return errorOutcome(result, "sync request entry count exceeds limit");219    }220    // Sanity: the peer's request must be a subset of what its manifest221    // declared as newer. A malformed/over-eager request is a protocol222    // violation but not a data-safety problem — we just send what they ask.223224    stage = "building_requested_entries";225    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);226    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });227    result.entriesSent = Object.keys(ourPayload).length;228229    if (wantFromPeer.length > 0) {230      stage = "waiting_for_device_entries";231      const entriesIn = await receiveMessage(session, timer);232      if (entriesIn.type !== "SYNC_ENTRIES") {233        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);234      }235      await vault.applyEntries(entriesIn.entries, manifestIn.entries);236      result.entriesReceived = Object.keys(entriesIn.entries).length;237    }238239    // Tombstones: each side ships the tombstones that the other does not240    // yet have a newer record for.241    const newerTombstonesForPeer: Record<string, number> = {};242    for (const [id, ts] of Object.entries(ourTombstones)) {243      const peerTs = manifestIn.tombstones[id] ?? 0;244      if (ts > peerTs) newerTombstonesForPeer[id] = ts;245    }246    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });247    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;248249    stage = "waiting_for_device_tombstones";250    const tombstonesIn = await receiveMessage(session, timer);251    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {252      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);253    }254    await vault.applyTombstones(tombstonesIn.tombstones);255    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;256257    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends258    // must expose the adapter methods. Disabling on either side cleanly259    // bypasses the step. The exchange is symmetric on the wire; the260    // production adapter accepts only a directly paired issuer whose stored261    // long-term key matches the Noise-authenticated remote static key.262    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {263      stage = "building_device_manifest";264      const ours = await vault.buildOutgoingDeviceManifest();265      await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });266      stage = "waiting_for_device_graph";267      const manifestIn = await receiveMessage(session, timer);268      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {269        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);270      }271      if (manifestIn.manifest) {272        await vault.applyIncomingDeviceManifest(273          manifestIn.manifest,274          session.remoteStaticPublicKey,275        );276      }277    }278279    await sendMessage(session, {280      type: "SYNC_DONE",281      entriesReceived: result.entriesReceived,282      entriesSent: result.entriesSent,283    });284    stage = "waiting_for_device_done";285    const closing = await awaitClosingDone(session, timer);286    if (closing.error) return errorOutcome(result, closing.error);287    result.status = "completed";288    return result;289  } catch (err) {290    if ((err as Error).message === "sync_timeout") {291      result.status = "timeout";292      result.error = `sync_timeout:${stage}`;293      return result;294    }295    return errorOutcome(result, (err as Error).message);296  } finally {297    // Session lifetime is owned by the runner (sync-runner.ts), which298    // closes it in its own finally block. The coordinator must not close it299    // here — the runner still needs to record the outcome after we return.300    timer.cancel();301  }302}303304/**305 * Exact complement of [runSyncInitiator] for the higher-key peer holding a306 * relay waiter room. Keeping this in the browser lets two extensions form a307 * direct mesh edge; the device that introduced them is not in the data path.308 */309export async function runSyncResponder(310  session: SyncSession,311  vault: SyncVaultAdapter,312): Promise<SyncOutcome> {313  const result: SyncOutcome = {314    status: "errored",315    entriesReceived: 0,316    entriesSent: 0,317    tombstonesReceived: 0,318    tombstonesSent: 0,319  };320  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);321  let stage: SyncStage = "waiting_for_device_hello";322  try {323    const ourVaultId = await vault.computeVaultId();324    const ourDeviceId = await vault.getOurDeviceId();325    const localVersion = await vault.getLocalVersion();326    const localTime = Date.now();327    const supportsDeviceManifest =328      vault.buildOutgoingDeviceManifest !== undefined &&329      vault.applyIncomingDeviceManifest !== undefined;330331    const helloIn = await receiveMessage(session, timer);332    if (helloIn.type !== "SYNC_HELLO") {333      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);334    }335    await sendMessage(session, {336      type: "SYNC_HELLO",337      vaultId: ourVaultId,338      deviceId: ourDeviceId,339      version: localVersion,340      deviceTime: localTime,341      deviceManifestVersion: supportsDeviceManifest342        ? DEVICE_MANIFEST_VERSION343        : undefined,344    });345    if (helloIn.vaultId !== ourVaultId) {346      result.status = "vault_mismatch";347      return result;348    }349    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {350      console.warn(351        `[sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`,352      );353    }354355    if (helloIn.version === localVersion) {356      if (357        supportsDeviceManifest &&358        helloIn.deviceManifestVersion === DEVICE_MANIFEST_VERSION359      ) {360        stage = "waiting_for_device_graph";361        const graphIn = await receiveMessage(session, timer);362        if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {363          return errorOutcome(364            result,365            `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,366          );367        }368        const ours = await vault.buildOutgoingDeviceManifest!();369        await sendMessage(session, {370          type: "SYNC_DEVICE_MANIFEST",371          manifest: ours,372        });373        if (graphIn.manifest) {374          await vault.applyIncomingDeviceManifest!(375            graphIn.manifest,376            session.remoteStaticPublicKey,377          );378        }379      }380      stage = "waiting_for_device_done";381      const done = await receiveMessage(session, timer);382      if (done.type !== "SYNC_DONE") {383        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);384      }385      await sendMessage(session, {386        type: "SYNC_DONE",387        entriesReceived: 0,388        entriesSent: 0,389      });390      result.status = "completed";391      return result;392    }393394    const ourEntries = await vault.buildEntryManifest();395    const ourTombstones = await vault.buildTombstoneManifest();396    stage = "waiting_for_device_manifest";397    const manifestIn = await receiveMessage(session, timer);398    if (manifestIn.type !== "SYNC_MANIFEST") {399      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);400    }401    await sendMessage(session, {402      type: "SYNC_MANIFEST",403      entries: ourEntries,404      tombstones: ourTombstones,405    });406407    stage = "waiting_for_device_request";408    const requestIn = await receiveMessage(session, timer);409    if (requestIn.type !== "SYNC_REQUEST") {410      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);411    }412    if (requestIn.entryIds.length > SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT) {413      return errorOutcome(result, "sync request entry count exceeds limit");414    }415    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);416    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });417418    stage = "waiting_for_device_entries";419    const entriesIn = await receiveMessage(session, timer);420    if (entriesIn.type !== "SYNC_ENTRIES") {421      return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);422    }423    if (wantFromPeer.length > 0) {424      await vault.applyEntries(entriesIn.entries, manifestIn.entries);425      result.entriesReceived = Object.keys(entriesIn.entries).length;426    }427    if (requestIn.entryIds.length > 0) {428      const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);429      await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });430      result.entriesSent = Object.keys(ourPayload).length;431    }432433    stage = "waiting_for_device_tombstones";434    const tombstonesIn = await receiveMessage(session, timer);435    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {436      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);437    }438    await vault.applyTombstones(tombstonesIn.tombstones);439    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;440441    const newerTombstonesForPeer: Record<string, number> = {};442    for (const [id, timestamp] of Object.entries(ourTombstones)) {443      if (timestamp > (manifestIn.tombstones[id] ?? 0)) {444        newerTombstonesForPeer[id] = timestamp;445      }446    }447    await sendMessage(session, {448      type: "SYNC_TOMBSTONES",449      tombstones: newerTombstonesForPeer,450    });451    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;452453    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {454      stage = "waiting_for_device_graph";455      const graphIn = await receiveMessage(session, timer);456      if (graphIn.type !== "SYNC_DEVICE_MANIFEST") {457        return errorOutcome(458          result,459          `expected SYNC_DEVICE_MANIFEST, got ${graphIn.type}`,460        );461      }462      const ours = await vault.buildOutgoingDeviceManifest();463      await sendMessage(session, {464        type: "SYNC_DEVICE_MANIFEST",465        manifest: ours,466      });467      if (graphIn.manifest) {468        await vault.applyIncomingDeviceManifest(469          graphIn.manifest,470          session.remoteStaticPublicKey,471        );472      }473    }474475    stage = "waiting_for_device_done";476    const done = await receiveMessage(session, timer);477    if (done.type !== "SYNC_DONE") {478      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);479    }480    await sendMessage(session, {481      type: "SYNC_DONE",482      entriesReceived: result.entriesReceived,483      entriesSent: result.entriesSent,484    });485    result.status = "completed";486    return result;487  } catch (error) {488    if ((error as Error).message === "sync_timeout") {489      result.status = "timeout";490      result.error = `sync_timeout:${stage}`;491      return result;492    }493    return errorOutcome(result, (error as Error).message);494  } finally {495    timer.cancel();496  }497}498499/**500 * Wait for the peer's closing SYNC_DONE echo, treating a hang-up as success.501 *502 * By the time we send our own SYNC_DONE, every entry and tombstone has already503 * been exchanged and applied on both sides — the echo is a courtesy, not a504 * commit. Android closes the socket as soon as it has our DONE, so the parked505 * receive() rejects with "WebSocket closed"; counting that as `run_failed`506 * reported a red indicator for syncs that had fully succeeded and whose data507 * had demonstrably landed on the peer.508 *509 * A wrong *message* here is still a real protocol error — only a closed510 * transport is forgiven.511 */512async function awaitClosingDone(513  session: SyncSession,514  timer: Timer,515): Promise<{ error?: string }> {516  try {517    const done = await receiveMessage(session, timer);518    if (done.type !== "SYNC_DONE") {519      return { error: `expected SYNC_DONE, got ${done.type}` };520    }521    return {};522  } catch (err) {523    if ((err as Error).message === "sync_timeout") throw err;524    if (isConnectionClosedError(err)) return {};525    return { error: (err as Error).message };526  }527}528529function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {530  const want: string[] = [];531  for (const [id, ts] of Object.entries(remote)) {532    const ours = local[id];533    if (ours === undefined || ts > ours) want.push(id);534  }535  return want;536}537538interface Timer {539  signal: AbortSignal;540  cancel(): void;541}542543function scheduleTimeout(ms: number): Timer {544  const ctrl = new AbortController();545  const t = setTimeout(() => ctrl.abort(new Error("sync_timeout")), ms);546  return {547    signal: ctrl.signal,548    cancel: () => clearTimeout(t),549  };550}551552async function sendMessage(session: SyncSession, msg: SyncMessage): Promise<void> {553  session.send(new TextEncoder().encode(JSON.stringify(msg)));554}555556async function receiveMessage(session: SyncSession, timer: Timer): Promise<SyncMessage> {557  const pending = session.receive();558  const aborted = new Promise<never>((_, reject) => {559    if (timer.signal.aborted) reject(new Error("sync_timeout"));560    timer.signal.addEventListener("abort", () => reject(new Error("sync_timeout")));561  });562  const bytes = await Promise.race([pending, aborted]);563  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS564  // transport may not have enforced anything) — re-check before JSON.parse565  // materialises a multi-GB string/object graph.566  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {567    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);568  }569  const parsed = JSON.parse(new TextDecoder().decode(bytes));570  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");571  return parsed as SyncMessage;572}573574function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {575  outcome.status = "errored";576  outcome.error = error;577  return outcome;578}