sync-coordinator.test.ts

1import { describe, expect, it, vi } from "vitest";2import { SYNC_LIMITS } from "../shared/sync-limits";3import {4  runSyncInitiator,5  runSyncResponder,6  type SyncVaultAdapter,7} from "./sync-coordinator";8import type { SyncMessage, SignedDeviceManifest } from "../shared/sync-types";9import type { SyncSession } from "../shared/sync-transport";10import { buildSignedDeviceManifest } from "../shared/device-manifest";11import { deriveSyncSigningKeypair } from "../shared/sync-identity";1213function inMemoryPair(): [SyncSession, FakePeer] {14  const a: Uint8Array[] = [];15  const b: Uint8Array[] = [];16  const aR: Array<(v: Uint8Array) => void> = [];17  const bR: Array<(v: Uint8Array) => void> = [];18  const aRej: Array<(e: Error) => void> = [];19  let closed = false;2021  const ourSession: SyncSession = {22    send(bytes) {23      const r = bR.shift();24      if (r) r(bytes); else b.push(bytes);25    },26    receive(): Promise<Uint8Array> {27      const v = a.shift();28      if (v) return Promise.resolve(v);29      if (closed) return Promise.reject(new Error("WebSocket closed"));30      return new Promise<Uint8Array>((res, rej) => { aR.push(res); aRej.push(rej); });31    },32    async close() { /* no-op */ },33    remoteStaticPublicKey: new Uint8Array(32),34  };3536  const peer: FakePeer = {37    async send(msg: SyncMessage) {38      const bytes = new TextEncoder().encode(JSON.stringify(msg));39      const r = aR.shift();40      if (r) r(bytes); else a.push(bytes);41    },42    async receive(): Promise<SyncMessage> {43      const v = b.shift();44      const bytes = v ?? await new Promise<Uint8Array>((res) => bR.push(res));45      return JSON.parse(new TextDecoder().decode(bytes)) as SyncMessage;46    },47    /**48     * Android's listener closes the socket the moment it has our final49     * SYNC_DONE — there is no courtesy echo. Surface that as a "WebSocket50     * closed" rejection on the initiator's parked receive().51     */52    async closeInitiatorReceive() {53      if (closed) return;54      closed = true;55      while (aRej.length > 0) aRej.shift()!(new Error("WebSocket closed"));56    },57  };58  return [ourSession, peer];59}6061function symmetricSessionPair(): [SyncSession, SyncSession] {62  const queues: [Uint8Array[], Uint8Array[]] = [[], []];63  const readers: [64    Array<(value: Uint8Array) => void>,65    Array<(value: Uint8Array) => void>,66  ] = [[], []];67  const make = (side: 0 | 1): SyncSession => ({68    send(bytes) {69      const other = side === 0 ? 1 : 0;70      const reader = readers[other].shift();71      if (reader) reader(bytes);72      else queues[other].push(bytes);73    },74    receive() {75      const queued = queues[side].shift();76      if (queued) return Promise.resolve(queued);77      return new Promise<Uint8Array>((resolve) => readers[side].push(resolve));78    },79    async close() {},80    remoteStaticPublicKey: new Uint8Array(32).fill(side === 0 ? 0x22 : 0x11),81  });82  return [make(0), make(1)];83}8485interface FakePeer {86  send(msg: SyncMessage): Promise<void>;87  receive(): Promise<SyncMessage>;88  /** Hang up on the initiator's parked receive(), as Android does at DONE. */89  closeInitiatorReceive(): Promise<void>;90}9192function makeVaultAdapter(opts: {93  vaultId?: string;94  deviceId?: string;95  localVersion?: number;96  entries?: Record<string, number>;97  tombstones?: Record<string, number>;98  ciphertext?: Record<string, string>;99}): SyncVaultAdapter & {100  appliedEntries: Record<string, string>[];101  appliedTombstones: Record<string, number>[];102} {103  const appliedEntries: Record<string, string>[] = [];104  const appliedTombstones: Record<string, number>[] = [];105  return {106    async computeVaultId() { return opts.vaultId ?? "vault-abc"; },107    async getOurDeviceId() { return opts.deviceId ?? "device-init"; },108    // Use a non-zero default so the equal-version short circuit doesn't fire in109    // tests that aren't explicitly testing that path.110    async getLocalVersion() { return opts.localVersion ?? 5; },111    async buildEntryManifest() { return { ...(opts.entries ?? {}) }; },112    async buildTombstoneManifest() { return { ...(opts.tombstones ?? {}) }; },113    async getEncryptedEntries(ids) {114      const out: Record<string, string> = {};115      const cipher = opts.ciphertext ?? {};116      for (const id of ids) if (cipher[id] !== undefined) out[id] = cipher[id];117      return out;118    },119    async applyEntries(entries) { appliedEntries.push(entries); },120    async applyTombstones(t) { appliedTombstones.push(t); },121    appliedEntries,122    appliedTombstones,123  };124}125126// PB-26-111: SYNC_REQUEST.entryIds was bounded only by the 24 MiB frame size —127// roughly a million UUIDs — while each requested id costs an HKDF derivation128// plus an AES-GCM encryption of the entry envelope. An authenticated peer could129// burn ~30s of service-worker CPU per dial, repeatably, with no data effect that130// would look like an attack.131describe("runSyncInitiator — bounded SYNC_REQUEST (PB-26-111)", () => {132  it("rejects an oversized id list before any per-entry work", async () => {133    const [session, peer] = inMemoryPair();134    const vault = makeVaultAdapter({ vaultId: "vault-abc", entries: { "id-1": 1 } });135    let encryptedFor: string[] | null = null;136    const spied = Object.assign({}, vault, {137      async getEncryptedEntries(ids: string[]) {138        encryptedFor = ids;139        return {};140      },141    });142143    const initPromise = runSyncInitiator(session, spied);144    void (async () => {145      await peer.receive(); // hello146      await peer.send({147        type: "SYNC_HELLO",148        vaultId: "vault-abc",149        deviceId: "p",150        version: 0,151        deviceTime: Date.now(),152      });153      await peer.receive(); // our manifest154      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });155      await peer.receive(); // our request156      // Ask for more ids than the protocol allows.157      await peer.send({158        type: "SYNC_REQUEST",159        entryIds: Array.from(160          { length: SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT + 1 },161          (_, i) => `id-${i}`,162        ),163      });164    })();165166    const outcome = await initPromise;167    expect(outcome.status).toBe("errored");168    expect(outcome.error).toMatch(/entry count exceeds limit/);169    // The whole point: the bound fires before the expensive per-id work.170    expect(encryptedFor).toBeNull();171  });172173  it("still serves a request at the limit", async () => {174    const [session, peer] = inMemoryPair();175    const ids = Array.from({ length: 3 }, (_, i) => `id-${i}`);176    const vault = makeVaultAdapter({177      vaultId: "vault-abc",178      entries: { "id-0": 1, "id-1": 1, "id-2": 1 },179      ciphertext: { "id-0": "AAAA", "id-1": "BBBB", "id-2": "CCCC" },180    });181182    const initPromise = runSyncInitiator(session, vault);183    void (async () => {184      await peer.receive(); // hello185      await peer.send({186        type: "SYNC_HELLO",187        vaultId: "vault-abc",188        deviceId: "p",189        version: 0,190        deviceTime: Date.now(),191      });192      await peer.receive(); // our manifest193      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });194      await peer.receive(); // our request195      await peer.send({ type: "SYNC_REQUEST", entryIds: ids });196      await peer.receive(); // SYNC_ENTRIES197      await peer.receive(); // SYNC_TOMBSTONES198      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });199      await peer.receive(); // SYNC_DONE200      await peer.send({ type: "SYNC_DONE" });201    })();202203    const outcome = await initPromise;204    expect(outcome.status).toBe("completed");205    expect(outcome.entriesSent).toBe(3);206  });207});208209describe("runSyncInitiator — HELLO fields and version optimisation", () => {210  it("sends HELLO with version and deviceTime", async () => {211    const [session, peer] = inMemoryPair();212    const vault = makeVaultAdapter({ localVersion: 7 });213214    const initPromise = runSyncInitiator(session, vault);215216    const sentHello = await peer.receive();217    expect(sentHello.type).toBe("SYNC_HELLO");218    if (sentHello.type === "SYNC_HELLO") {219      expect(sentHello.version).toBe(7);220      expect(typeof sentHello.deviceTime).toBe("number");221      expect(sentHello.deviceTime).toBeGreaterThan(0);222    }223    // Send a mismatched vaultId to abort early so the test doesn't hang.224    await peer.send({ type: "SYNC_HELLO", vaultId: "different-vault", deviceId: "p", version: 7, deviceTime: Date.now() });225    await peer.receive(); // SYNC_ERROR226    await initPromise;227  });228229  it("skips manifest exchange and returns completed when versions are equal", async () => {230    const [session, peer] = inMemoryPair();231    const vault = makeVaultAdapter({ localVersion: 10 });232233    const initPromise = runSyncInitiator(session, vault);234235    await peer.receive(); // SYNC_HELLO236    await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 10, deviceTime: Date.now() });237238    // Coordinator should send SYNC_DONE without any manifest exchange.239    const done = await peer.receive();240    expect(done.type).toBe("SYNC_DONE");241242    // No SYNC_MANIFEST was sent — peer's queue only has the DONE.243    await peer.send({ type: "SYNC_DONE" });244245    const outcome = await initPromise;246    expect(outcome.status).toBe("completed");247    expect(outcome.entriesReceived).toBe(0);248    expect(outcome.entriesSent).toBe(0);249  });250251  it("exchanges device graphs at equal entry versions when both peers advertise support", async () => {252    const [dialerSession, waiterSession] = symmetricSessionPair();253    const dialerManifest = {254      body: {255        manifestVersion: 3,256        issuerDeviceId: "dialer",257        issuedAt: 1,258        devices: [],259        revoked: [],260      },261      signatureHex: "11".repeat(64),262    } satisfies SignedDeviceManifest;263    const waiterManifest = {264      body: {265        manifestVersion: 3,266        issuerDeviceId: "waiter",267        issuedAt: 1,268        devices: [],269        revoked: [],270      },271      signatureHex: "22".repeat(64),272    } satisfies SignedDeviceManifest;273    const dialerApplied: SignedDeviceManifest[] = [];274    const waiterApplied: SignedDeviceManifest[] = [];275    const dialer: SyncVaultAdapter = {276      ...makeVaultAdapter({ deviceId: "dialer", localVersion: 7 }),277      async buildOutgoingDeviceManifest() { return dialerManifest; },278      async applyIncomingDeviceManifest(manifest) { dialerApplied.push(manifest); },279    };280    const waiter: SyncVaultAdapter = {281      ...makeVaultAdapter({ deviceId: "waiter", localVersion: 7 }),282      async buildOutgoingDeviceManifest() { return waiterManifest; },283      async applyIncomingDeviceManifest(manifest) { waiterApplied.push(manifest); },284    };285286    const [dialerOutcome, waiterOutcome] = await Promise.all([287      runSyncInitiator(dialerSession, dialer),288      runSyncResponder(waiterSession, waiter),289    ]);290291    expect(dialerOutcome.status).toBe("completed");292    expect(waiterOutcome.status).toBe("completed");293    expect(dialerApplied).toEqual([waiterManifest]);294    expect(waiterApplied).toEqual([dialerManifest]);295  });296297  it("logs a clock-skew warning but does not abort when clocks differ by > 120s", async () => {298    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);299    const [session, peer] = inMemoryPair();300    const vault = makeVaultAdapter({ localVersion: 5 });301302    const initPromise = runSyncInitiator(session, vault);303304    await peer.receive(); // SYNC_HELLO305    // Peer reports a deviceTime 200 s in the past — beyond the 120 s threshold.306    await peer.send({307      type: "SYNC_HELLO",308      vaultId: "vault-abc",309      deviceId: "p",310      version: 3, // different version so exchange proceeds311      deviceTime: Date.now() - 200_000,312    });313    // Yield so the coordinator's async continuation can run the clock-skew check.314    await new Promise((r) => setTimeout(r, 0));315    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("clock skew"));316317    // Drive the sync to completion so the test doesn't leak.318    await peer.receive(); // SYNC_MANIFEST319    await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });320    await peer.receive(); // SYNC_REQUEST321    await peer.send({ type: "SYNC_REQUEST", entryIds: [] });322    await peer.receive(); // SYNC_ENTRIES323    await peer.receive(); // SYNC_TOMBSTONES324    await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });325    await peer.receive(); // SYNC_DONE326    await peer.send({ type: "SYNC_DONE" });327328    const outcome = await initPromise;329    expect(outcome.status).toBe("completed");330    warnSpy.mockRestore();331  });332});333334describe("runSyncInitiator", () => {335  it("syncs two mesh peers directly with their introducer absent", async () => {336    const [dialerSession, waiterSession] = symmetricSessionPair();337    const dialer = makeVaultAdapter({338      deviceId: "laptop-a",339      localVersion: 10,340      entries: { a: 10 },341      ciphertext: { a: "cipher-a" },342    });343    const waiter = makeVaultAdapter({344      deviceId: "laptop-c",345      localVersion: 11,346      entries: { c: 11 },347      ciphertext: { c: "cipher-c" },348    });349350    const [dialerOutcome, waiterOutcome] = await Promise.all([351      runSyncInitiator(dialerSession, dialer),352      runSyncResponder(waiterSession, waiter),353    ]);354355    expect(dialerOutcome.status).toBe("completed");356    expect(waiterOutcome.status).toBe("completed");357    expect(dialer.appliedEntries).toEqual([{ c: "cipher-c" }]);358    expect(waiter.appliedEntries).toEqual([{ a: "cipher-a" }]);359  });360361  it("reports the privacy-safe protocol stage when a device stops answering", async () => {362    vi.useFakeTimers();363    try {364      const [session, peer] = inMemoryPair();365      const vault = makeVaultAdapter({});366      const initPromise = runSyncInitiator(session, vault);367368      const hello = await peer.receive();369      expect(hello.type).toBe("SYNC_HELLO");370371      await vi.advanceTimersByTimeAsync(30_000);372      await expect(initPromise).resolves.toMatchObject({373        status: "timeout",374        error: "sync_timeout:waiting_for_device_hello",375      });376    } finally {377      vi.useRealTimers();378    }379  });380381  it("completes a full sync with bi-directional updates", async () => {382    const [session, peer] = inMemoryPair();383    const vault = makeVaultAdapter({384      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50385      tombstones: { x: 70 },386      ciphertext: { a: "ct-a", b: "ct-b" },387    });388389    const initPromise = runSyncInitiator(session, vault);390391    const peerProtocol = (async () => {392      const hello = await peer.receive();393      expect(hello.type).toBe("SYNC_HELLO");394      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now() });395396      const ourManifest = await peer.receive();397      expect(ourManifest.type).toBe("SYNC_MANIFEST");398      await peer.send({399        type: "SYNC_MANIFEST",400        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'401        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'402      });403404      const ourRequest = await peer.receive();405      expect(ourRequest.type).toBe("SYNC_REQUEST");406      // We should request the entries the peer reported newer407      if (ourRequest.type === "SYNC_REQUEST") {408        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);409      }410      // Peer asks us for entries the peer claimed it had older versions of411      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });412413      const ourEntries = await peer.receive();414      expect(ourEntries.type).toBe("SYNC_ENTRIES");415      if (ourEntries.type === "SYNC_ENTRIES") {416        expect(ourEntries.entries).toEqual({ a: "ct-a" });417      }418      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });419420      const ourTombs = await peer.receive();421      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");422      if (ourTombs.type === "SYNC_TOMBSTONES") {423        // We had x@70, peer had x@60, so we ship x@70424        expect(ourTombs.tombstones).toEqual({ x: 70 });425      }426      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });427428      const ourDone = await peer.receive();429      expect(ourDone.type).toBe("SYNC_DONE");430      await peer.send({ type: "SYNC_DONE" });431    })();432433    const outcome = await initPromise;434    await peerProtocol;435436    expect(outcome.status).toBe("completed");437    expect(outcome.entriesSent).toBe(1);438    expect(outcome.entriesReceived).toBe(2);439    expect(outcome.tombstonesSent).toBe(1);440    expect(outcome.tombstonesReceived).toBe(1);441    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);442    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);443  });444445  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {446    const [session, peer] = inMemoryPair();447    const vault = makeVaultAdapter({ vaultId: "vault-A" });448449    const initPromise = runSyncInitiator(session, vault);450    const peerProtocol = (async () => {451      const hello = await peer.receive();452      expect(hello.type).toBe("SYNC_HELLO");453      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now() });454      // Peer should receive a SYNC_ERROR and we should stop.455      const err = await peer.receive();456      expect(err.type).toBe("SYNC_ERROR");457    })();458459    const outcome = await initPromise;460    await peerProtocol;461    expect(outcome.status).toBe("vault_mismatch");462  });463464  it("exchanges device manifests when both adapters opt in", async () => {465    const [session, peer] = inMemoryPair();466    const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us");467    const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them");468469    const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey);470    const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey);471    const applied: SignedDeviceManifest[] = [];472473    const base = makeVaultAdapter({});474    const vault: SyncVaultAdapter = {475      ...base,476      async buildOutgoingDeviceManifest() { return outgoing; },477      async applyIncomingDeviceManifest(m, authenticatedKey) {478        expect(authenticatedKey).toEqual(session.remoteStaticPublicKey);479        applied.push(m);480      },481    };482483    const initPromise = runSyncInitiator(session, vault);484    const peerProtocol = (async () => {485      await peer.receive(); // hello486      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });487      await peer.receive(); // manifest488      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });489      await peer.receive(); // request490      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });491      await peer.receive(); // entries492      await peer.receive(); // tombstones493      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });494      const ourManifest = await peer.receive();495      expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST");496      if (ourManifest.type === "SYNC_DEVICE_MANIFEST") {497        expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us");498      }499      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer });500      await peer.receive(); // done501      await peer.send({ type: "SYNC_DONE" });502    })();503504    const outcome = await initPromise;505    await peerProtocol;506    expect(outcome.status).toBe("completed");507    expect(applied).toHaveLength(1);508    expect(applied[0].body.issuerDeviceId).toBe("them");509  });510511  it("uses null to represent an absent device manifest", async () => {512    const [session, peer] = inMemoryPair();513    const applyIncoming = vi.fn();514    const vault: SyncVaultAdapter = {515      ...makeVaultAdapter({}),516      async buildOutgoingDeviceManifest() { return null; },517      applyIncomingDeviceManifest: applyIncoming,518    };519520    const initPromise = runSyncInitiator(session, vault);521    const peerProtocol = (async () => {522      await peer.receive();523      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });524      await peer.receive();525      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });526      await peer.receive();527      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });528      await peer.receive();529      await peer.receive();530      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });531      const manifest = await peer.receive();532      expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null });533      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null });534      await peer.receive();535      await peer.send({ type: "SYNC_DONE" });536    })();537538    const outcome = await initPromise;539    await peerProtocol;540    expect(outcome.status).toBe("completed");541    expect(applyIncoming).not.toHaveBeenCalled();542  });543544  it("returns errored when peer sends a wrong frame", async () => {545    const [session, peer] = inMemoryPair();546    const vault = makeVaultAdapter({ vaultId: "v" });547548    const initPromise = runSyncInitiator(session, vault);549    void (async () => {550      await peer.receive(); // hello551      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now() });552      await peer.receive(); // our manifest553      await peer.send({ type: "SYNC_DONE" }); // wrong frame554    })();555556    const outcome = await initPromise;557    expect(outcome.status).toBe("errored");558    expect(outcome.error).toMatch(/SYNC_MANIFEST/);559  });560561  it("reports completed when the peer closes the socket while we wait for its closing SYNC_DONE", async () => {562    // Bug: by the time we send our own SYNC_DONE, every entry and tombstone563    // has already been exchanged and applied on both sides. The peer's reply564    // is a courtesy, not a commit. Android closes the socket the moment it565    // has our DONE; previously the initiator's parked receive() rejected with566    // "WebSocket closed" and the run was reported as `errored` → run_failed567    // → red, even though the data had demonstrably landed.568    const [session, peer] = inMemoryPair();569    const vault = makeVaultAdapter({ localVersion: 5 });570571    const initPromise = runSyncInitiator(session, vault);572    void (async () => {573      await peer.receive(); // hello574      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 5, deviceTime: Date.now() });575      await peer.receive(); // our done576      // No SYNC_DONE echo — Android just hangs up.577      await peer.closeInitiatorReceive();578    })();579580    const outcome = await initPromise;581    expect(outcome.status).toBe("completed");582  });583});