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> = [];1314  const ourSession: SyncSession = {15    send(bytes) {16      const r = bR.shift();17      if (r) r(bytes); else b.push(bytes);18    },19    receive(): Promise<Uint8Array> {20      const v = a.shift();21      if (v) return Promise.resolve(v);22      return new Promise<Uint8Array>((res) => aR.push(res));23    },24    async close() { /* no-op */ },25    remoteStaticPublicKey: new Uint8Array(32),26  };2728  const peer: FakePeer = {29    async send(msg: SyncMessage) {30      const bytes = new TextEncoder().encode(JSON.stringify(msg));31      const r = aR.shift();32      if (r) r(bytes); else a.push(bytes);33    },34    async receive(): Promise<SyncMessage> {35      const v = b.shift();36      const bytes = v ?? await new Promise<Uint8Array>((res) => bR.push(res));37      return JSON.parse(new TextDecoder().decode(bytes)) as SyncMessage;38    },39  };40  return [ourSession, peer];41}4243interface FakePeer {44  send(msg: SyncMessage): Promise<void>;45  receive(): Promise<SyncMessage>;46}4748function makeVaultAdapter(opts: {49  vaultId?: string;50  deviceId?: string;51  localVersion?: number;52  entries?: Record<string, number>;53  tombstones?: Record<string, number>;54  ciphertext?: Record<string, string>;55}): SyncVaultAdapter & {56  appliedEntries: Record<string, string>[];57  appliedTombstones: Record<string, number>[];58} {59  const appliedEntries: Record<string, string>[] = [];60  const appliedTombstones: Record<string, number>[] = [];61  return {62    async computeVaultId() { return opts.vaultId ?? "vault-abc"; },63    async getOurDeviceId() { return opts.deviceId ?? "device-init"; },64    // Use a non-zero default so the equal-version short circuit doesn't fire in65    // tests that aren't explicitly testing that path.66    async getLocalVersion() { return opts.localVersion ?? 5; },67    async buildEntryManifest() { return { ...(opts.entries ?? {}) }; },68    async buildTombstoneManifest() { return { ...(opts.tombstones ?? {}) }; },69    async getEncryptedEntries(ids) {70      const out: Record<string, string> = {};71      const cipher = opts.ciphertext ?? {};72      for (const id of ids) if (cipher[id] !== undefined) out[id] = cipher[id];73      return out;74    },75    async applyEntries(entries) { appliedEntries.push(entries); },76    async applyTombstones(t) { appliedTombstones.push(t); },77    appliedEntries,78    appliedTombstones,79  };80}8182describe("runSyncInitiator — HELLO fields and version optimisation", () => {83  it("sends HELLO with version and deviceTime", async () => {84    const [session, peer] = inMemoryPair();85    const vault = makeVaultAdapter({ localVersion: 7 });8687    const initPromise = runSyncInitiator(session, vault);8889    const sentHello = await peer.receive();90    expect(sentHello.type).toBe("SYNC_HELLO");91    if (sentHello.type === "SYNC_HELLO") {92      expect(sentHello.version).toBe(7);93      expect(typeof sentHello.deviceTime).toBe("number");94      expect(sentHello.deviceTime).toBeGreaterThan(0);95    }96    // Send a mismatched vaultId to abort early so the test doesn't hang.97    await peer.send({ type: "SYNC_HELLO", vaultId: "different-vault", deviceId: "p", version: 7, deviceTime: Date.now() });98    await peer.receive(); // SYNC_ERROR99    await initPromise;100  });101102  it("skips manifest exchange and returns completed when versions are equal", async () => {103    const [session, peer] = inMemoryPair();104    const vault = makeVaultAdapter({ localVersion: 10 });105106    const initPromise = runSyncInitiator(session, vault);107108    await peer.receive(); // SYNC_HELLO109    await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 10, deviceTime: Date.now() });110111    // Coordinator should send SYNC_DONE without any manifest exchange.112    const done = await peer.receive();113    expect(done.type).toBe("SYNC_DONE");114115    // No SYNC_MANIFEST was sent — peer's queue only has the DONE.116    await peer.send({ type: "SYNC_DONE" });117118    const outcome = await initPromise;119    expect(outcome.status).toBe("completed");120    expect(outcome.entriesReceived).toBe(0);121    expect(outcome.entriesSent).toBe(0);122  });123124  it("logs a clock-skew warning but does not abort when clocks differ by > 120s", async () => {125    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);126    const [session, peer] = inMemoryPair();127    const vault = makeVaultAdapter({ localVersion: 5 });128129    const initPromise = runSyncInitiator(session, vault);130131    await peer.receive(); // SYNC_HELLO132    // Peer reports a deviceTime 200 s in the past — beyond the 120 s threshold.133    await peer.send({134      type: "SYNC_HELLO",135      vaultId: "vault-abc",136      deviceId: "p",137      version: 3, // different version so exchange proceeds138      deviceTime: Date.now() - 200_000,139    });140    // Yield so the coordinator's async continuation can run the clock-skew check.141    await new Promise((r) => setTimeout(r, 0));142    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("clock skew"));143144    // Drive the sync to completion so the test doesn't leak.145    await peer.receive(); // SYNC_MANIFEST146    await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });147    await peer.receive(); // SYNC_REQUEST148    await peer.send({ type: "SYNC_REQUEST", entryIds: [] });149    await peer.receive(); // SYNC_ENTRIES150    await peer.receive(); // SYNC_TOMBSTONES151    await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });152    await peer.receive(); // SYNC_DONE153    await peer.send({ type: "SYNC_DONE" });154155    const outcome = await initPromise;156    expect(outcome.status).toBe("completed");157    warnSpy.mockRestore();158  });159});160161describe("runSyncInitiator", () => {162  it("completes a full sync with bi-directional updates", async () => {163    const [session, peer] = inMemoryPair();164    const vault = makeVaultAdapter({165      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50166      tombstones: { x: 70 },167      ciphertext: { a: "ct-a", b: "ct-b" },168    });169170    const initPromise = runSyncInitiator(session, vault);171172    const peerProtocol = (async () => {173      const hello = await peer.receive();174      expect(hello.type).toBe("SYNC_HELLO");175      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now() });176177      const ourManifest = await peer.receive();178      expect(ourManifest.type).toBe("SYNC_MANIFEST");179      await peer.send({180        type: "SYNC_MANIFEST",181        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'182        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'183      });184185      const ourRequest = await peer.receive();186      expect(ourRequest.type).toBe("SYNC_REQUEST");187      // We should request the entries the peer reported newer188      if (ourRequest.type === "SYNC_REQUEST") {189        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);190      }191      // Peer asks us for entries the peer claimed it had older versions of192      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });193194      const ourEntries = await peer.receive();195      expect(ourEntries.type).toBe("SYNC_ENTRIES");196      if (ourEntries.type === "SYNC_ENTRIES") {197        expect(ourEntries.entries).toEqual({ a: "ct-a" });198      }199      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });200201      const ourTombs = await peer.receive();202      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");203      if (ourTombs.type === "SYNC_TOMBSTONES") {204        // We had x@70, peer had x@60, so we ship x@70205        expect(ourTombs.tombstones).toEqual({ x: 70 });206      }207      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });208209      const ourDone = await peer.receive();210      expect(ourDone.type).toBe("SYNC_DONE");211      await peer.send({ type: "SYNC_DONE" });212    })();213214    const outcome = await initPromise;215    await peerProtocol;216217    expect(outcome.status).toBe("completed");218    expect(outcome.entriesSent).toBe(1);219    expect(outcome.entriesReceived).toBe(2);220    expect(outcome.tombstonesSent).toBe(1);221    expect(outcome.tombstonesReceived).toBe(1);222    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);223    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);224  });225226  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {227    const [session, peer] = inMemoryPair();228    const vault = makeVaultAdapter({ vaultId: "vault-A" });229230    const initPromise = runSyncInitiator(session, vault);231    const peerProtocol = (async () => {232      const hello = await peer.receive();233      expect(hello.type).toBe("SYNC_HELLO");234      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now() });235      // Peer should receive a SYNC_ERROR and we should stop.236      const err = await peer.receive();237      expect(err.type).toBe("SYNC_ERROR");238    })();239240    const outcome = await initPromise;241    await peerProtocol;242    expect(outcome.status).toBe("vault_mismatch");243  });244245  it("exchanges device manifests when both adapters opt in", async () => {246    const [session, peer] = inMemoryPair();247    const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us");248    const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them");249250    const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey);251    const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey);252    const applied: SignedDeviceManifest[] = [];253254    const base = makeVaultAdapter({});255    const vault: SyncVaultAdapter = {256      ...base,257      async buildOutgoingDeviceManifest() { return outgoing; },258      async applyIncomingDeviceManifest(m, authenticatedKey) {259        expect(authenticatedKey).toEqual(session.remoteStaticPublicKey);260        applied.push(m);261      },262    };263264    const initPromise = runSyncInitiator(session, vault);265    const peerProtocol = (async () => {266      await peer.receive(); // hello267      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });268      await peer.receive(); // manifest269      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });270      await peer.receive(); // request271      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });272      await peer.receive(); // entries273      await peer.receive(); // tombstones274      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });275      const ourManifest = await peer.receive();276      expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST");277      if (ourManifest.type === "SYNC_DEVICE_MANIFEST") {278        expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us");279      }280      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer });281      await peer.receive(); // done282      await peer.send({ type: "SYNC_DONE" });283    })();284285    const outcome = await initPromise;286    await peerProtocol;287    expect(outcome.status).toBe("completed");288    expect(applied).toHaveLength(1);289    expect(applied[0].body.issuerDeviceId).toBe("them");290  });291292  it("uses null to represent an absent device manifest", async () => {293    const [session, peer] = inMemoryPair();294    const applyIncoming = vi.fn();295    const vault: SyncVaultAdapter = {296      ...makeVaultAdapter({}),297      async buildOutgoingDeviceManifest() { return null; },298      applyIncomingDeviceManifest: applyIncoming,299    };300301    const initPromise = runSyncInitiator(session, vault);302    const peerProtocol = (async () => {303      await peer.receive();304      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });305      await peer.receive();306      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });307      await peer.receive();308      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });309      await peer.receive();310      await peer.receive();311      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });312      const manifest = await peer.receive();313      expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null });314      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null });315      await peer.receive();316      await peer.send({ type: "SYNC_DONE" });317    })();318319    const outcome = await initPromise;320    await peerProtocol;321    expect(outcome.status).toBe("completed");322    expect(applyIncoming).not.toHaveBeenCalled();323  });324325  it("returns errored when peer sends a wrong frame", async () => {326    const [session, peer] = inMemoryPair();327    const vault = makeVaultAdapter({ vaultId: "v" });328329    const initPromise = runSyncInitiator(session, vault);330    void (async () => {331      await peer.receive(); // hello332      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now() });333      await peer.receive(); // our manifest334      await peer.send({ type: "SYNC_DONE" }); // wrong frame335    })();336337    const outcome = await initPromise;338    expect(outcome.status).toBe("errored");339    expect(outcome.error).toMatch(/SYNC_MANIFEST/);340  });341});