chirp-sync-coordinator.ts

1// Chirp Sync — Phase 2 manifest-diff coordinator.2//3// After Noise XX is open and the responder's static key has been verified4// against the paired-device registry, this coordinator drives the5// 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 { ChirpSyncMessage, SignedDeviceManifest } from "../shared/sync-types";22import type { ChirpSyncSession } from "../shared/chirp-sync-transport";23import { SYNC_LIMITS } from "../shared/sync-limits";2425const SESSION_TIMEOUT_MS = 30_000;2627// Sentinel signature used when an adapter can't build a real manifest (e.g.28// the master key isn't accessible from the coordinator context). The peer29// recognises this exact value and skips signature verification. 64 hex30// chars of zero is impossible to produce as a real Ed25519 signature for a31// fresh body, so collisions are negligible.32const EMPTY_MANIFEST_PLACEHOLDER: SignedDeviceManifest = {33  body: { manifestVersion: 1, issuerDeviceId: "", issuedAt: 0, devices: [] },34  signatureHex: "0".repeat(128),35};3637/**38 * Vault-side adapter — the coordinator's dependency on the vault layer.39 * The vault holds the canonical entry/tombstone tables; the coordinator40 * pulls a manifest, requests entries by id, applies remote entries, and41 * applies remote tombstones.42 */43export interface ChirpSyncVaultAdapter {44  computeVaultId(): Promise<string>;45  getOurDeviceId(): Promise<string>;46  /** Returns the vault's monotonic version (`peach_last_modified_at`). */47  getLocalVersion(): Promise<number>;48  /** Returns the per-entry mtimes for our active entries. */49  buildEntryManifest(): Promise<Record<string, number>>;50  /** Returns the per-entry deletion timestamps for our tombstones. */51  buildTombstoneManifest(): Promise<Record<string, number>>;52  /** Returns base64 ciphertext for each requested entry id. */53  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;54  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */55  applyEntries(entries: Record<string, string>): Promise<void>;56  /** Apply incoming tombstones; vault retains the latest deletion ts. */57  applyTombstones(tombstones: Record<string, number>): Promise<void>;58  /**59   * Optional — v2 brief §4.3 device manifest exchange. The extension (as60   * the trust anchor for the device graph) returns a signed manifest61   * enumerating every paired device it knows about; the peer verifies it62   * against the extension's known Ed25519 signing pubkey and upserts any63   * newly-vouched devices into its own registry.64   *65   * When this method is undefined, the coordinator skips the manifest66   * step entirely. Both ends must opt in for the exchange to occur.67   */68  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;69  /** v2 brief §4.3 — handle a manifest received from the peer. */70  applyIncomingDeviceManifest?(manifest: SignedDeviceManifest): Promise<void>;71}7273export interface SyncOutcome {74  status: "completed" | "vault_mismatch" | "errored" | "timeout";75  entriesReceived: number;76  entriesSent: number;77  tombstonesReceived: number;78  tombstonesSent: number;79  error?: string;80}8182/**83 * Run one sync round as the initiator. The extension is always the84 * initiator (it is the WebSocket client). The peer's role is symmetric85 * after SYNC_HELLO, so a corresponding responder helper exists for the86 * mobile-side coordinator that calls this same protocol the other way.87 */88export async function runChirpSyncInitiator(89  session: ChirpSyncSession,90  vault: ChirpSyncVaultAdapter,91): Promise<SyncOutcome> {92  const result: SyncOutcome = {93    status: "errored",94    entriesReceived: 0,95    entriesSent: 0,96    tombstonesReceived: 0,97    tombstonesSent: 0,98  };99100  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);101  try {102    const ourVaultId = await vault.computeVaultId();103    const ourDeviceId = await vault.getOurDeviceId();104    const localVersion = await vault.getLocalVersion();105    const localTime = Date.now();106107    await sendMessage(session, {108      type: "SYNC_HELLO",109      vaultId: ourVaultId,110      deviceId: ourDeviceId,111      version: localVersion,112      deviceTime: localTime,113    });114    const helloIn = await receiveMessage(session, timer);115    if (helloIn.type !== "SYNC_HELLO") {116      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);117    }118    if (helloIn.vaultId !== ourVaultId) {119      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });120      result.status = "vault_mismatch";121      return result;122    }123124    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {125      console.warn(`[chirp-sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);126    }127128    // Equal versions — nothing to exchange.129    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {130      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });131      const done = await receiveMessage(session, timer);132      if (done.type !== "SYNC_DONE") {133        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);134      }135      result.status = "completed";136      return result;137    }138139    const ourEntries = await vault.buildEntryManifest();140    const ourTombstones = await vault.buildTombstoneManifest();141    await sendMessage(session, {142      type: "SYNC_MANIFEST",143      entries: ourEntries,144      tombstones: ourTombstones,145    });146    const manifestIn = await receiveMessage(session, timer);147    if (manifestIn.type !== "SYNC_MANIFEST") {148      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);149    }150151    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);152    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });153154    const requestIn = await receiveMessage(session, timer);155    if (requestIn.type !== "SYNC_REQUEST") {156      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);157    }158    // Sanity: the peer's request must be a subset of what its manifest159    // declared as newer. A malformed/over-eager request is a protocol160    // violation but not a data-safety problem — we just send what they ask.161162    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);163    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });164    result.entriesSent = Object.keys(ourPayload).length;165166    if (wantFromPeer.length > 0) {167      const entriesIn = await receiveMessage(session, timer);168      if (entriesIn.type !== "SYNC_ENTRIES") {169        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);170      }171      await vault.applyEntries(entriesIn.entries);172      result.entriesReceived = Object.keys(entriesIn.entries).length;173    }174175    // Tombstones: each side ships the tombstones that the other does not176    // yet have a newer record for.177    const newerTombstonesForPeer: Record<string, number> = {};178    for (const [id, ts] of Object.entries(ourTombstones)) {179      const peerTs = manifestIn.tombstones[id] ?? 0;180      if (ts > peerTs) newerTombstonesForPeer[id] = ts;181    }182    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });183    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;184185    const tombstonesIn = await receiveMessage(session, timer);186    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {187      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);188    }189    await vault.applyTombstones(tombstonesIn.tombstones);190    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;191192    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends193    // must expose the adapter methods. Disabling on either side cleanly194    // bypasses the step. The exchange is symmetric on the wire; the195    // production adapter still treats the extension as the trust anchor and196    // only applies extension-issued manifests.197    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {198      const ours = await vault.buildOutgoingDeviceManifest();199      if (ours) {200        await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });201      } else {202        // No manifest to send (e.g. master key unavailable) — still emit a203        // protocol frame so the peer's recv doesn't hang. Empty body is204        // serialised by the adapter when present; if it returned null we205        // synthesise a no-op shape the receiver can ignore.206        await sendMessage(session, {207          type: "SYNC_DEVICE_MANIFEST",208          manifest: EMPTY_MANIFEST_PLACEHOLDER,209        });210      }211      const manifestIn = await receiveMessage(session, timer);212      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {213        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);214      }215      if (manifestIn.manifest && manifestIn.manifest.signatureHex !== EMPTY_MANIFEST_PLACEHOLDER.signatureHex) {216        await vault.applyIncomingDeviceManifest(manifestIn.manifest);217      }218    }219220    await sendMessage(session, {221      type: "SYNC_DONE",222      entriesReceived: result.entriesReceived,223      entriesSent: result.entriesSent,224    });225    const done = await receiveMessage(session, timer);226    if (done.type !== "SYNC_DONE") {227      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);228    }229    result.status = "completed";230    return result;231  } catch (err) {232    if ((err as Error).message === "chirp_sync_timeout") {233      result.status = "timeout";234      result.error = "session timed out after 30s";235      return result;236    }237    return errorOutcome(result, (err as Error).message);238  } finally {239    // Session lifetime is owned by the runner (chirp-sync-runner.ts), which240    // closes it in its own finally block. The coordinator must not close it241    // here — the runner still needs to record the outcome after we return.242    timer.cancel();243  }244}245246function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {247  const want: string[] = [];248  for (const [id, ts] of Object.entries(remote)) {249    const ours = local[id];250    if (ours === undefined || ts > ours) want.push(id);251  }252  return want;253}254255interface Timer {256  signal: AbortSignal;257  cancel(): void;258}259260function scheduleTimeout(ms: number): Timer {261  const ctrl = new AbortController();262  const t = setTimeout(() => ctrl.abort(new Error("chirp_sync_timeout")), ms);263  return {264    signal: ctrl.signal,265    cancel: () => clearTimeout(t),266  };267}268269async function sendMessage(session: ChirpSyncSession, msg: ChirpSyncMessage): Promise<void> {270  session.send(new TextEncoder().encode(JSON.stringify(msg)));271}272273async function receiveMessage(session: ChirpSyncSession, timer: Timer): Promise<ChirpSyncMessage> {274  const pending = session.receive();275  const aborted = new Promise<never>((_, reject) => {276    if (timer.signal.aborted) reject(new Error("chirp_sync_timeout"));277    timer.signal.addEventListener("abort", () => reject(new Error("chirp_sync_timeout")));278  });279  const bytes = await Promise.race([pending, aborted]);280  // MV-3: decrypted plaintext can exceed the wire frame (and a non-WS281  // transport may not have enforced anything) — re-check before JSON.parse282  // materialises a multi-GB string/object graph.283  if (bytes.length > SYNC_LIMITS.MAX_FRAME_BYTES) {284    throw new Error(`sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes)`);285  }286  const parsed = JSON.parse(new TextDecoder().decode(bytes));287  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");288  return parsed as ChirpSyncMessage;289}290291function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {292  outcome.status = "errored";293  outcome.error = error;294  return outcome;295}