sync-coordinator.test.ts

1import { describe, expect, it, vi } from "vitest";2import { runSyncInitiator, type SyncVaultAdapter } from "./sync-coordinator";3import type { SyncMessage, SignedDeviceManifest } from "../shared/sync-types";4import type { SyncSession } from "../shared/sync-transport";5import { buildSignedDeviceManifest } from "../shared/device-manifest";6import { deriveSyncSigningKeypair } from "../shared/sync-identity";78function inMemoryPair(): [SyncSession, FakePeer] {9  const a: Uint8Array[] = [];10  const b: Uint8Array[] = [];11  const aR: Array<(v: Uint8Array) => void> = [];12  const bR: Array<(v: Uint8Array) => void> = [];13  const aRej: Array<(e: Error) => void> = [];14  let closed = false;1516  const ourSession: SyncSession = {17    send(bytes) {18      const r = bR.shift();19      if (r) r(bytes); else b.push(bytes);20    },21    receive(): Promise<Uint8Array> {22      const v = a.shift();23      if (v) return Promise.resolve(v);24      if (closed) return Promise.reject(new Error("WebSocket closed"));25      return new Promise<Uint8Array>((res, rej) => { aR.push(res); aRej.push(rej); });26    },27    async close() { /* no-op */ },28    remoteStaticPublicKey: new Uint8Array(32),29  };3031  const peer: FakePeer = {32    async send(msg: SyncMessage) {33      const bytes = new TextEncoder().encode(JSON.stringify(msg));34      const r = aR.shift();35      if (r) r(bytes); else a.push(bytes);36    },37    async receive(): Promise<SyncMessage> {38      const v = b.shift();39      const bytes = v ?? await new Promise<Uint8Array>((res) => bR.push(res));40      return JSON.parse(new TextDecoder().decode(bytes)) as SyncMessage;41    },42    /**43     * Android's listener closes the socket the moment it has our final44     * SYNC_DONE — there is no courtesy echo. Surface that as a "WebSocket45     * closed" rejection on the initiator's parked receive().46     */47    async closeInitiatorReceive() {48      if (closed) return;49      closed = true;50      while (aRej.length > 0) aRej.shift()!(new Error("WebSocket closed"));51    },52  };53  return [ourSession, peer];54}5556interface FakePeer {57  send(msg: SyncMessage): Promise<void>;58  receive(): Promise<SyncMessage>;59  /** Hang up on the initiator's parked receive(), as Android does at DONE. */60  closeInitiatorReceive(): Promise<void>;61}6263function makeVaultAdapter(opts: {64  vaultId?: string;65  deviceId?: string;66  localVersion?: number;67  entries?: Record<string, number>;68  tombstones?: Record<string, number>;69  ciphertext?: Record<string, string>;70}): SyncVaultAdapter & {71  appliedEntries: Record<string, string>[];72  appliedTombstones: Record<string, number>[];73} {74  const appliedEntries: Record<string, string>[] = [];75  const appliedTombstones: Record<string, number>[] = [];76  return {77    async computeVaultId() { return opts.vaultId ?? "vault-abc"; },78    async getOurDeviceId() { return opts.deviceId ?? "device-init"; },79    // Use a non-zero default so the equal-version short circuit doesn't fire in80    // tests that aren't explicitly testing that path.81    async getLocalVersion() { return opts.localVersion ?? 5; },82    async buildEntryManifest() { return { ...(opts.entries ?? {}) }; },83    async buildTombstoneManifest() { return { ...(opts.tombstones ?? {}) }; },84    async getEncryptedEntries(ids) {85      const out: Record<string, string> = {};86      const cipher = opts.ciphertext ?? {};87      for (const id of ids) if (cipher[id] !== undefined) out[id] = cipher[id];88      return out;89    },90    async applyEntries(entries) { appliedEntries.push(entries); },91    async applyTombstones(t) { appliedTombstones.push(t); },92    appliedEntries,93    appliedTombstones,94  };95}9697describe("runSyncInitiator — HELLO fields and version optimisation", () => {98  it("sends HELLO with version and deviceTime", async () => {99    const [session, peer] = inMemoryPair();100    const vault = makeVaultAdapter({ localVersion: 7 });101102    const initPromise = runSyncInitiator(session, vault);103104    const sentHello = await peer.receive();105    expect(sentHello.type).toBe("SYNC_HELLO");106    if (sentHello.type === "SYNC_HELLO") {107      expect(sentHello.version).toBe(7);108      expect(typeof sentHello.deviceTime).toBe("number");109      expect(sentHello.deviceTime).toBeGreaterThan(0);110    }111    // Send a mismatched vaultId to abort early so the test doesn't hang.112    await peer.send({ type: "SYNC_HELLO", vaultId: "different-vault", deviceId: "p", version: 7, deviceTime: Date.now() });113    await peer.receive(); // SYNC_ERROR114    await initPromise;115  });116117  it("skips manifest exchange and returns completed when versions are equal", async () => {118    const [session, peer] = inMemoryPair();119    const vault = makeVaultAdapter({ localVersion: 10 });120121    const initPromise = runSyncInitiator(session, vault);122123    await peer.receive(); // SYNC_HELLO124    await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 10, deviceTime: Date.now() });125126    // Coordinator should send SYNC_DONE without any manifest exchange.127    const done = await peer.receive();128    expect(done.type).toBe("SYNC_DONE");129130    // No SYNC_MANIFEST was sent — peer's queue only has the DONE.131    await peer.send({ type: "SYNC_DONE" });132133    const outcome = await initPromise;134    expect(outcome.status).toBe("completed");135    expect(outcome.entriesReceived).toBe(0);136    expect(outcome.entriesSent).toBe(0);137  });138139  it("logs a clock-skew warning but does not abort when clocks differ by > 120s", async () => {140    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);141    const [session, peer] = inMemoryPair();142    const vault = makeVaultAdapter({ localVersion: 5 });143144    const initPromise = runSyncInitiator(session, vault);145146    await peer.receive(); // SYNC_HELLO147    // Peer reports a deviceTime 200 s in the past — beyond the 120 s threshold.148    await peer.send({149      type: "SYNC_HELLO",150      vaultId: "vault-abc",151      deviceId: "p",152      version: 3, // different version so exchange proceeds153      deviceTime: Date.now() - 200_000,154    });155    // Yield so the coordinator's async continuation can run the clock-skew check.156    await new Promise((r) => setTimeout(r, 0));157    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("clock skew"));158159    // Drive the sync to completion so the test doesn't leak.160    await peer.receive(); // SYNC_MANIFEST161    await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });162    await peer.receive(); // SYNC_REQUEST163    await peer.send({ type: "SYNC_REQUEST", entryIds: [] });164    await peer.receive(); // SYNC_ENTRIES165    await peer.receive(); // SYNC_TOMBSTONES166    await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });167    await peer.receive(); // SYNC_DONE168    await peer.send({ type: "SYNC_DONE" });169170    const outcome = await initPromise;171    expect(outcome.status).toBe("completed");172    warnSpy.mockRestore();173  });174});175176describe("runSyncInitiator", () => {177  it("reports the privacy-safe protocol stage when a device stops answering", async () => {178    vi.useFakeTimers();179    try {180      const [session, peer] = inMemoryPair();181      const vault = makeVaultAdapter({});182      const initPromise = runSyncInitiator(session, vault);183184      const hello = await peer.receive();185      expect(hello.type).toBe("SYNC_HELLO");186187      await vi.advanceTimersByTimeAsync(30_000);188      await expect(initPromise).resolves.toMatchObject({189        status: "timeout",190        error: "sync_timeout:waiting_for_device_hello",191      });192    } finally {193      vi.useRealTimers();194    }195  });196197  it("completes a full sync with bi-directional updates", async () => {198    const [session, peer] = inMemoryPair();199    const vault = makeVaultAdapter({200      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50201      tombstones: { x: 70 },202      ciphertext: { a: "ct-a", b: "ct-b" },203    });204205    const initPromise = runSyncInitiator(session, vault);206207    const peerProtocol = (async () => {208      const hello = await peer.receive();209      expect(hello.type).toBe("SYNC_HELLO");210      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now() });211212      const ourManifest = await peer.receive();213      expect(ourManifest.type).toBe("SYNC_MANIFEST");214      await peer.send({215        type: "SYNC_MANIFEST",216        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'217        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'218      });219220      const ourRequest = await peer.receive();221      expect(ourRequest.type).toBe("SYNC_REQUEST");222      // We should request the entries the peer reported newer223      if (ourRequest.type === "SYNC_REQUEST") {224        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);225      }226      // Peer asks us for entries the peer claimed it had older versions of227      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });228229      const ourEntries = await peer.receive();230      expect(ourEntries.type).toBe("SYNC_ENTRIES");231      if (ourEntries.type === "SYNC_ENTRIES") {232        expect(ourEntries.entries).toEqual({ a: "ct-a" });233      }234      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });235236      const ourTombs = await peer.receive();237      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");238      if (ourTombs.type === "SYNC_TOMBSTONES") {239        // We had x@70, peer had x@60, so we ship x@70240        expect(ourTombs.tombstones).toEqual({ x: 70 });241      }242      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });243244      const ourDone = await peer.receive();245      expect(ourDone.type).toBe("SYNC_DONE");246      await peer.send({ type: "SYNC_DONE" });247    })();248249    const outcome = await initPromise;250    await peerProtocol;251252    expect(outcome.status).toBe("completed");253    expect(outcome.entriesSent).toBe(1);254    expect(outcome.entriesReceived).toBe(2);255    expect(outcome.tombstonesSent).toBe(1);256    expect(outcome.tombstonesReceived).toBe(1);257    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);258    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);259  });260261  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {262    const [session, peer] = inMemoryPair();263    const vault = makeVaultAdapter({ vaultId: "vault-A" });264265    const initPromise = runSyncInitiator(session, vault);266    const peerProtocol = (async () => {267      const hello = await peer.receive();268      expect(hello.type).toBe("SYNC_HELLO");269      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now() });270      // Peer should receive a SYNC_ERROR and we should stop.271      const err = await peer.receive();272      expect(err.type).toBe("SYNC_ERROR");273    })();274275    const outcome = await initPromise;276    await peerProtocol;277    expect(outcome.status).toBe("vault_mismatch");278  });279280  it("exchanges device manifests when both adapters opt in", async () => {281    const [session, peer] = inMemoryPair();282    const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us");283    const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them");284285    const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey);286    const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey);287    const applied: SignedDeviceManifest[] = [];288289    const base = makeVaultAdapter({});290    const vault: SyncVaultAdapter = {291      ...base,292      async buildOutgoingDeviceManifest() { return outgoing; },293      async applyIncomingDeviceManifest(m, authenticatedKey) {294        expect(authenticatedKey).toEqual(session.remoteStaticPublicKey);295        applied.push(m);296      },297    };298299    const initPromise = runSyncInitiator(session, vault);300    const peerProtocol = (async () => {301      await peer.receive(); // hello302      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });303      await peer.receive(); // manifest304      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });305      await peer.receive(); // request306      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });307      await peer.receive(); // entries308      await peer.receive(); // tombstones309      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });310      const ourManifest = await peer.receive();311      expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST");312      if (ourManifest.type === "SYNC_DEVICE_MANIFEST") {313        expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us");314      }315      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer });316      await peer.receive(); // done317      await peer.send({ type: "SYNC_DONE" });318    })();319320    const outcome = await initPromise;321    await peerProtocol;322    expect(outcome.status).toBe("completed");323    expect(applied).toHaveLength(1);324    expect(applied[0].body.issuerDeviceId).toBe("them");325  });326327  it("uses null to represent an absent device manifest", async () => {328    const [session, peer] = inMemoryPair();329    const applyIncoming = vi.fn();330    const vault: SyncVaultAdapter = {331      ...makeVaultAdapter({}),332      async buildOutgoingDeviceManifest() { return null; },333      applyIncomingDeviceManifest: applyIncoming,334    };335336    const initPromise = runSyncInitiator(session, vault);337    const peerProtocol = (async () => {338      await peer.receive();339      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });340      await peer.receive();341      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });342      await peer.receive();343      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });344      await peer.receive();345      await peer.receive();346      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });347      const manifest = await peer.receive();348      expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null });349      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null });350      await peer.receive();351      await peer.send({ type: "SYNC_DONE" });352    })();353354    const outcome = await initPromise;355    await peerProtocol;356    expect(outcome.status).toBe("completed");357    expect(applyIncoming).not.toHaveBeenCalled();358  });359360  it("returns errored when peer sends a wrong frame", async () => {361    const [session, peer] = inMemoryPair();362    const vault = makeVaultAdapter({ vaultId: "v" });363364    const initPromise = runSyncInitiator(session, vault);365    void (async () => {366      await peer.receive(); // hello367      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now() });368      await peer.receive(); // our manifest369      await peer.send({ type: "SYNC_DONE" }); // wrong frame370    })();371372    const outcome = await initPromise;373    expect(outcome.status).toBe("errored");374    expect(outcome.error).toMatch(/SYNC_MANIFEST/);375  });376377  it("reports completed when the peer closes the socket while we wait for its closing SYNC_DONE", async () => {378    // Bug: by the time we send our own SYNC_DONE, every entry and tombstone379    // has already been exchanged and applied on both sides. The peer's reply380    // is a courtesy, not a commit. Android closes the socket the moment it381    // has our DONE; previously the initiator's parked receive() rejected with382    // "WebSocket closed" and the run was reported as `errored` → run_failed383    // → red, even though the data had demonstrably landed.384    const [session, peer] = inMemoryPair();385    const vault = makeVaultAdapter({ localVersion: 5 });386387    const initPromise = runSyncInitiator(session, vault);388    void (async () => {389      await peer.receive(); // hello390      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 5, deviceTime: Date.now() });391      await peer.receive(); // our done392      // No SYNC_DONE echo — Android just hangs up.393      await peer.closeInitiatorReceive();394    })();395396    const outcome = await initPromise;397    expect(outcome.status).toBe("completed");398  });399});