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("reports the privacy-safe protocol stage when a device stops answering", async () => {163    vi.useFakeTimers();164    try {165      const [session, peer] = inMemoryPair();166      const vault = makeVaultAdapter({});167      const initPromise = runSyncInitiator(session, vault);168169      const hello = await peer.receive();170      expect(hello.type).toBe("SYNC_HELLO");171172      await vi.advanceTimersByTimeAsync(30_000);173      await expect(initPromise).resolves.toMatchObject({174        status: "timeout",175        error: "sync_timeout:waiting_for_device_hello",176      });177    } finally {178      vi.useRealTimers();179    }180  });181182  it("completes a full sync with bi-directional updates", async () => {183    const [session, peer] = inMemoryPair();184    const vault = makeVaultAdapter({185      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50186      tombstones: { x: 70 },187      ciphertext: { a: "ct-a", b: "ct-b" },188    });189190    const initPromise = runSyncInitiator(session, vault);191192    const peerProtocol = (async () => {193      const hello = await peer.receive();194      expect(hello.type).toBe("SYNC_HELLO");195      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now() });196197      const ourManifest = await peer.receive();198      expect(ourManifest.type).toBe("SYNC_MANIFEST");199      await peer.send({200        type: "SYNC_MANIFEST",201        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'202        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'203      });204205      const ourRequest = await peer.receive();206      expect(ourRequest.type).toBe("SYNC_REQUEST");207      // We should request the entries the peer reported newer208      if (ourRequest.type === "SYNC_REQUEST") {209        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);210      }211      // Peer asks us for entries the peer claimed it had older versions of212      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });213214      const ourEntries = await peer.receive();215      expect(ourEntries.type).toBe("SYNC_ENTRIES");216      if (ourEntries.type === "SYNC_ENTRIES") {217        expect(ourEntries.entries).toEqual({ a: "ct-a" });218      }219      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });220221      const ourTombs = await peer.receive();222      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");223      if (ourTombs.type === "SYNC_TOMBSTONES") {224        // We had x@70, peer had x@60, so we ship x@70225        expect(ourTombs.tombstones).toEqual({ x: 70 });226      }227      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });228229      const ourDone = await peer.receive();230      expect(ourDone.type).toBe("SYNC_DONE");231      await peer.send({ type: "SYNC_DONE" });232    })();233234    const outcome = await initPromise;235    await peerProtocol;236237    expect(outcome.status).toBe("completed");238    expect(outcome.entriesSent).toBe(1);239    expect(outcome.entriesReceived).toBe(2);240    expect(outcome.tombstonesSent).toBe(1);241    expect(outcome.tombstonesReceived).toBe(1);242    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);243    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);244  });245246  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {247    const [session, peer] = inMemoryPair();248    const vault = makeVaultAdapter({ vaultId: "vault-A" });249250    const initPromise = runSyncInitiator(session, vault);251    const peerProtocol = (async () => {252      const hello = await peer.receive();253      expect(hello.type).toBe("SYNC_HELLO");254      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now() });255      // Peer should receive a SYNC_ERROR and we should stop.256      const err = await peer.receive();257      expect(err.type).toBe("SYNC_ERROR");258    })();259260    const outcome = await initPromise;261    await peerProtocol;262    expect(outcome.status).toBe("vault_mismatch");263  });264265  it("exchanges device manifests when both adapters opt in", async () => {266    const [session, peer] = inMemoryPair();267    const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us");268    const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them");269270    const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey);271    const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey);272    const applied: SignedDeviceManifest[] = [];273274    const base = makeVaultAdapter({});275    const vault: SyncVaultAdapter = {276      ...base,277      async buildOutgoingDeviceManifest() { return outgoing; },278      async applyIncomingDeviceManifest(m, authenticatedKey) {279        expect(authenticatedKey).toEqual(session.remoteStaticPublicKey);280        applied.push(m);281      },282    };283284    const initPromise = runSyncInitiator(session, vault);285    const peerProtocol = (async () => {286      await peer.receive(); // hello287      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });288      await peer.receive(); // manifest289      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });290      await peer.receive(); // request291      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });292      await peer.receive(); // entries293      await peer.receive(); // tombstones294      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });295      const ourManifest = await peer.receive();296      expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST");297      if (ourManifest.type === "SYNC_DEVICE_MANIFEST") {298        expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us");299      }300      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer });301      await peer.receive(); // done302      await peer.send({ type: "SYNC_DONE" });303    })();304305    const outcome = await initPromise;306    await peerProtocol;307    expect(outcome.status).toBe("completed");308    expect(applied).toHaveLength(1);309    expect(applied[0].body.issuerDeviceId).toBe("them");310  });311312  it("uses null to represent an absent device manifest", async () => {313    const [session, peer] = inMemoryPair();314    const applyIncoming = vi.fn();315    const vault: SyncVaultAdapter = {316      ...makeVaultAdapter({}),317      async buildOutgoingDeviceManifest() { return null; },318      applyIncomingDeviceManifest: applyIncoming,319    };320321    const initPromise = runSyncInitiator(session, vault);322    const peerProtocol = (async () => {323      await peer.receive();324      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });325      await peer.receive();326      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });327      await peer.receive();328      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });329      await peer.receive();330      await peer.receive();331      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });332      const manifest = await peer.receive();333      expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null });334      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null });335      await peer.receive();336      await peer.send({ type: "SYNC_DONE" });337    })();338339    const outcome = await initPromise;340    await peerProtocol;341    expect(outcome.status).toBe("completed");342    expect(applyIncoming).not.toHaveBeenCalled();343  });344345  it("returns errored when peer sends a wrong frame", async () => {346    const [session, peer] = inMemoryPair();347    const vault = makeVaultAdapter({ vaultId: "v" });348349    const initPromise = runSyncInitiator(session, vault);350    void (async () => {351      await peer.receive(); // hello352      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now() });353      await peer.receive(); // our manifest354      await peer.send({ type: "SYNC_DONE" }); // wrong frame355    })();356357    const outcome = await initPromise;358    expect(outcome.status).toBe("errored");359    expect(outcome.error).toMatch(/SYNC_MANIFEST/);360  });361});