sync-coordinator.test.ts

1import { describe, expect, it, vi } from "vitest";2import {3  runSyncInitiator,4  runSyncResponder,5  type SyncVaultAdapter,6} from "./sync-coordinator";7import type { SyncMessage, SignedDeviceManifest } from "../shared/sync-types";8import type { SyncSession } from "../shared/sync-transport";9import { buildSignedDeviceManifest } from "../shared/device-manifest";10import { deriveSyncSigningKeypair } from "../shared/sync-identity";1112function inMemoryPair(): [SyncSession, FakePeer] {13  const a: Uint8Array[] = [];14  const b: Uint8Array[] = [];15  const aR: Array<(v: Uint8Array) => void> = [];16  const bR: Array<(v: Uint8Array) => void> = [];17  const aRej: Array<(e: Error) => void> = [];18  let closed = false;1920  const ourSession: SyncSession = {21    send(bytes) {22      const r = bR.shift();23      if (r) r(bytes); else b.push(bytes);24    },25    receive(): Promise<Uint8Array> {26      const v = a.shift();27      if (v) return Promise.resolve(v);28      if (closed) return Promise.reject(new Error("WebSocket closed"));29      return new Promise<Uint8Array>((res, rej) => { aR.push(res); aRej.push(rej); });30    },31    async close() { /* no-op */ },32    remoteStaticPublicKey: new Uint8Array(32),33  };3435  const peer: FakePeer = {36    async send(msg: SyncMessage) {37      const bytes = new TextEncoder().encode(JSON.stringify(msg));38      const r = aR.shift();39      if (r) r(bytes); else a.push(bytes);40    },41    async receive(): Promise<SyncMessage> {42      const v = b.shift();43      const bytes = v ?? await new Promise<Uint8Array>((res) => bR.push(res));44      return JSON.parse(new TextDecoder().decode(bytes)) as SyncMessage;45    },46    /**47     * Android's listener closes the socket the moment it has our final48     * SYNC_DONE — there is no courtesy echo. Surface that as a "WebSocket49     * closed" rejection on the initiator's parked receive().50     */51    async closeInitiatorReceive() {52      if (closed) return;53      closed = true;54      while (aRej.length > 0) aRej.shift()!(new Error("WebSocket closed"));55    },56  };57  return [ourSession, peer];58}5960function symmetricSessionPair(): [SyncSession, SyncSession] {61  const queues: [Uint8Array[], Uint8Array[]] = [[], []];62  const readers: [63    Array<(value: Uint8Array) => void>,64    Array<(value: Uint8Array) => void>,65  ] = [[], []];66  const make = (side: 0 | 1): SyncSession => ({67    send(bytes) {68      const other = side === 0 ? 1 : 0;69      const reader = readers[other].shift();70      if (reader) reader(bytes);71      else queues[other].push(bytes);72    },73    receive() {74      const queued = queues[side].shift();75      if (queued) return Promise.resolve(queued);76      return new Promise<Uint8Array>((resolve) => readers[side].push(resolve));77    },78    async close() {},79    remoteStaticPublicKey: new Uint8Array(32).fill(side === 0 ? 0x22 : 0x11),80  });81  return [make(0), make(1)];82}8384interface FakePeer {85  send(msg: SyncMessage): Promise<void>;86  receive(): Promise<SyncMessage>;87  /** Hang up on the initiator's parked receive(), as Android does at DONE. */88  closeInitiatorReceive(): Promise<void>;89}9091function makeVaultAdapter(opts: {92  vaultId?: string;93  deviceId?: string;94  localVersion?: number;95  entries?: Record<string, number>;96  tombstones?: Record<string, number>;97  ciphertext?: Record<string, string>;98}): SyncVaultAdapter & {99  appliedEntries: Record<string, string>[];100  appliedTombstones: Record<string, number>[];101} {102  const appliedEntries: Record<string, string>[] = [];103  const appliedTombstones: Record<string, number>[] = [];104  return {105    async computeVaultId() { return opts.vaultId ?? "vault-abc"; },106    async getOurDeviceId() { return opts.deviceId ?? "device-init"; },107    // Use a non-zero default so the equal-version short circuit doesn't fire in108    // tests that aren't explicitly testing that path.109    async getLocalVersion() { return opts.localVersion ?? 5; },110    async buildEntryManifest() { return { ...(opts.entries ?? {}) }; },111    async buildTombstoneManifest() { return { ...(opts.tombstones ?? {}) }; },112    async getEncryptedEntries(ids) {113      const out: Record<string, string> = {};114      const cipher = opts.ciphertext ?? {};115      for (const id of ids) if (cipher[id] !== undefined) out[id] = cipher[id];116      return out;117    },118    async applyEntries(entries) { appliedEntries.push(entries); },119    async applyTombstones(t) { appliedTombstones.push(t); },120    appliedEntries,121    appliedTombstones,122  };123}124125describe("runSyncInitiator — HELLO fields and version optimisation", () => {126  it("sends HELLO with version and deviceTime", async () => {127    const [session, peer] = inMemoryPair();128    const vault = makeVaultAdapter({ localVersion: 7 });129130    const initPromise = runSyncInitiator(session, vault);131132    const sentHello = await peer.receive();133    expect(sentHello.type).toBe("SYNC_HELLO");134    if (sentHello.type === "SYNC_HELLO") {135      expect(sentHello.version).toBe(7);136      expect(typeof sentHello.deviceTime).toBe("number");137      expect(sentHello.deviceTime).toBeGreaterThan(0);138    }139    // Send a mismatched vaultId to abort early so the test doesn't hang.140    await peer.send({ type: "SYNC_HELLO", vaultId: "different-vault", deviceId: "p", version: 7, deviceTime: Date.now() });141    await peer.receive(); // SYNC_ERROR142    await initPromise;143  });144145  it("skips manifest exchange and returns completed when versions are equal", async () => {146    const [session, peer] = inMemoryPair();147    const vault = makeVaultAdapter({ localVersion: 10 });148149    const initPromise = runSyncInitiator(session, vault);150151    await peer.receive(); // SYNC_HELLO152    await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 10, deviceTime: Date.now() });153154    // Coordinator should send SYNC_DONE without any manifest exchange.155    const done = await peer.receive();156    expect(done.type).toBe("SYNC_DONE");157158    // No SYNC_MANIFEST was sent — peer's queue only has the DONE.159    await peer.send({ type: "SYNC_DONE" });160161    const outcome = await initPromise;162    expect(outcome.status).toBe("completed");163    expect(outcome.entriesReceived).toBe(0);164    expect(outcome.entriesSent).toBe(0);165  });166167  it("logs a clock-skew warning but does not abort when clocks differ by > 120s", async () => {168    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);169    const [session, peer] = inMemoryPair();170    const vault = makeVaultAdapter({ localVersion: 5 });171172    const initPromise = runSyncInitiator(session, vault);173174    await peer.receive(); // SYNC_HELLO175    // Peer reports a deviceTime 200 s in the past — beyond the 120 s threshold.176    await peer.send({177      type: "SYNC_HELLO",178      vaultId: "vault-abc",179      deviceId: "p",180      version: 3, // different version so exchange proceeds181      deviceTime: Date.now() - 200_000,182    });183    // Yield so the coordinator's async continuation can run the clock-skew check.184    await new Promise((r) => setTimeout(r, 0));185    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("clock skew"));186187    // Drive the sync to completion so the test doesn't leak.188    await peer.receive(); // SYNC_MANIFEST189    await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });190    await peer.receive(); // SYNC_REQUEST191    await peer.send({ type: "SYNC_REQUEST", entryIds: [] });192    await peer.receive(); // SYNC_ENTRIES193    await peer.receive(); // SYNC_TOMBSTONES194    await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });195    await peer.receive(); // SYNC_DONE196    await peer.send({ type: "SYNC_DONE" });197198    const outcome = await initPromise;199    expect(outcome.status).toBe("completed");200    warnSpy.mockRestore();201  });202});203204describe("runSyncInitiator", () => {205  it("syncs two mesh peers directly with their introducer absent", async () => {206    const [dialerSession, waiterSession] = symmetricSessionPair();207    const dialer = makeVaultAdapter({208      deviceId: "laptop-a",209      localVersion: 10,210      entries: { a: 10 },211      ciphertext: { a: "cipher-a" },212    });213    const waiter = makeVaultAdapter({214      deviceId: "laptop-c",215      localVersion: 11,216      entries: { c: 11 },217      ciphertext: { c: "cipher-c" },218    });219220    const [dialerOutcome, waiterOutcome] = await Promise.all([221      runSyncInitiator(dialerSession, dialer),222      runSyncResponder(waiterSession, waiter),223    ]);224225    expect(dialerOutcome.status).toBe("completed");226    expect(waiterOutcome.status).toBe("completed");227    expect(dialer.appliedEntries).toEqual([{ c: "cipher-c" }]);228    expect(waiter.appliedEntries).toEqual([{ a: "cipher-a" }]);229  });230231  it("reports the privacy-safe protocol stage when a device stops answering", async () => {232    vi.useFakeTimers();233    try {234      const [session, peer] = inMemoryPair();235      const vault = makeVaultAdapter({});236      const initPromise = runSyncInitiator(session, vault);237238      const hello = await peer.receive();239      expect(hello.type).toBe("SYNC_HELLO");240241      await vi.advanceTimersByTimeAsync(30_000);242      await expect(initPromise).resolves.toMatchObject({243        status: "timeout",244        error: "sync_timeout:waiting_for_device_hello",245      });246    } finally {247      vi.useRealTimers();248    }249  });250251  it("completes a full sync with bi-directional updates", async () => {252    const [session, peer] = inMemoryPair();253    const vault = makeVaultAdapter({254      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50255      tombstones: { x: 70 },256      ciphertext: { a: "ct-a", b: "ct-b" },257    });258259    const initPromise = runSyncInitiator(session, vault);260261    const peerProtocol = (async () => {262      const hello = await peer.receive();263      expect(hello.type).toBe("SYNC_HELLO");264      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now() });265266      const ourManifest = await peer.receive();267      expect(ourManifest.type).toBe("SYNC_MANIFEST");268      await peer.send({269        type: "SYNC_MANIFEST",270        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'271        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'272      });273274      const ourRequest = await peer.receive();275      expect(ourRequest.type).toBe("SYNC_REQUEST");276      // We should request the entries the peer reported newer277      if (ourRequest.type === "SYNC_REQUEST") {278        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);279      }280      // Peer asks us for entries the peer claimed it had older versions of281      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });282283      const ourEntries = await peer.receive();284      expect(ourEntries.type).toBe("SYNC_ENTRIES");285      if (ourEntries.type === "SYNC_ENTRIES") {286        expect(ourEntries.entries).toEqual({ a: "ct-a" });287      }288      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });289290      const ourTombs = await peer.receive();291      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");292      if (ourTombs.type === "SYNC_TOMBSTONES") {293        // We had x@70, peer had x@60, so we ship x@70294        expect(ourTombs.tombstones).toEqual({ x: 70 });295      }296      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });297298      const ourDone = await peer.receive();299      expect(ourDone.type).toBe("SYNC_DONE");300      await peer.send({ type: "SYNC_DONE" });301    })();302303    const outcome = await initPromise;304    await peerProtocol;305306    expect(outcome.status).toBe("completed");307    expect(outcome.entriesSent).toBe(1);308    expect(outcome.entriesReceived).toBe(2);309    expect(outcome.tombstonesSent).toBe(1);310    expect(outcome.tombstonesReceived).toBe(1);311    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);312    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);313  });314315  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {316    const [session, peer] = inMemoryPair();317    const vault = makeVaultAdapter({ vaultId: "vault-A" });318319    const initPromise = runSyncInitiator(session, vault);320    const peerProtocol = (async () => {321      const hello = await peer.receive();322      expect(hello.type).toBe("SYNC_HELLO");323      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now() });324      // Peer should receive a SYNC_ERROR and we should stop.325      const err = await peer.receive();326      expect(err.type).toBe("SYNC_ERROR");327    })();328329    const outcome = await initPromise;330    await peerProtocol;331    expect(outcome.status).toBe("vault_mismatch");332  });333334  it("exchanges device manifests when both adapters opt in", async () => {335    const [session, peer] = inMemoryPair();336    const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us");337    const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them");338339    const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey);340    const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey);341    const applied: SignedDeviceManifest[] = [];342343    const base = makeVaultAdapter({});344    const vault: SyncVaultAdapter = {345      ...base,346      async buildOutgoingDeviceManifest() { return outgoing; },347      async applyIncomingDeviceManifest(m, authenticatedKey) {348        expect(authenticatedKey).toEqual(session.remoteStaticPublicKey);349        applied.push(m);350      },351    };352353    const initPromise = runSyncInitiator(session, vault);354    const peerProtocol = (async () => {355      await peer.receive(); // hello356      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });357      await peer.receive(); // manifest358      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });359      await peer.receive(); // request360      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });361      await peer.receive(); // entries362      await peer.receive(); // tombstones363      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });364      const ourManifest = await peer.receive();365      expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST");366      if (ourManifest.type === "SYNC_DEVICE_MANIFEST") {367        expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us");368      }369      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer });370      await peer.receive(); // done371      await peer.send({ type: "SYNC_DONE" });372    })();373374    const outcome = await initPromise;375    await peerProtocol;376    expect(outcome.status).toBe("completed");377    expect(applied).toHaveLength(1);378    expect(applied[0].body.issuerDeviceId).toBe("them");379  });380381  it("uses null to represent an absent device manifest", async () => {382    const [session, peer] = inMemoryPair();383    const applyIncoming = vi.fn();384    const vault: SyncVaultAdapter = {385      ...makeVaultAdapter({}),386      async buildOutgoingDeviceManifest() { return null; },387      applyIncomingDeviceManifest: applyIncoming,388    };389390    const initPromise = runSyncInitiator(session, vault);391    const peerProtocol = (async () => {392      await peer.receive();393      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });394      await peer.receive();395      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });396      await peer.receive();397      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });398      await peer.receive();399      await peer.receive();400      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });401      const manifest = await peer.receive();402      expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null });403      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null });404      await peer.receive();405      await peer.send({ type: "SYNC_DONE" });406    })();407408    const outcome = await initPromise;409    await peerProtocol;410    expect(outcome.status).toBe("completed");411    expect(applyIncoming).not.toHaveBeenCalled();412  });413414  it("returns errored when peer sends a wrong frame", async () => {415    const [session, peer] = inMemoryPair();416    const vault = makeVaultAdapter({ vaultId: "v" });417418    const initPromise = runSyncInitiator(session, vault);419    void (async () => {420      await peer.receive(); // hello421      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now() });422      await peer.receive(); // our manifest423      await peer.send({ type: "SYNC_DONE" }); // wrong frame424    })();425426    const outcome = await initPromise;427    expect(outcome.status).toBe("errored");428    expect(outcome.error).toMatch(/SYNC_MANIFEST/);429  });430431  it("reports completed when the peer closes the socket while we wait for its closing SYNC_DONE", async () => {432    // Bug: by the time we send our own SYNC_DONE, every entry and tombstone433    // has already been exchanged and applied on both sides. The peer's reply434    // is a courtesy, not a commit. Android closes the socket the moment it435    // has our DONE; previously the initiator's parked receive() rejected with436    // "WebSocket closed" and the run was reported as `errored` → run_failed437    // → red, even though the data had demonstrably landed.438    const [session, peer] = inMemoryPair();439    const vault = makeVaultAdapter({ localVersion: 5 });440441    const initPromise = runSyncInitiator(session, vault);442    void (async () => {443      await peer.receive(); // hello444      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 5, deviceTime: Date.now() });445      await peer.receive(); // our done446      // No SYNC_DONE echo — Android just hangs up.447      await peer.closeInitiatorReceive();448    })();449450    const outcome = await initPromise;451    expect(outcome.status).toBe("completed");452  });453});