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("exchanges device graphs at equal entry versions when both peers advertise support", async () => {168    const [dialerSession, waiterSession] = symmetricSessionPair();169    const dialerManifest = {170      body: {171        manifestVersion: 3,172        issuerDeviceId: "dialer",173        issuedAt: 1,174        devices: [],175        revoked: [],176      },177      signatureHex: "11".repeat(64),178    } satisfies SignedDeviceManifest;179    const waiterManifest = {180      body: {181        manifestVersion: 3,182        issuerDeviceId: "waiter",183        issuedAt: 1,184        devices: [],185        revoked: [],186      },187      signatureHex: "22".repeat(64),188    } satisfies SignedDeviceManifest;189    const dialerApplied: SignedDeviceManifest[] = [];190    const waiterApplied: SignedDeviceManifest[] = [];191    const dialer: SyncVaultAdapter = {192      ...makeVaultAdapter({ deviceId: "dialer", localVersion: 7 }),193      async buildOutgoingDeviceManifest() { return dialerManifest; },194      async applyIncomingDeviceManifest(manifest) { dialerApplied.push(manifest); },195    };196    const waiter: SyncVaultAdapter = {197      ...makeVaultAdapter({ deviceId: "waiter", localVersion: 7 }),198      async buildOutgoingDeviceManifest() { return waiterManifest; },199      async applyIncomingDeviceManifest(manifest) { waiterApplied.push(manifest); },200    };201202    const [dialerOutcome, waiterOutcome] = await Promise.all([203      runSyncInitiator(dialerSession, dialer),204      runSyncResponder(waiterSession, waiter),205    ]);206207    expect(dialerOutcome.status).toBe("completed");208    expect(waiterOutcome.status).toBe("completed");209    expect(dialerApplied).toEqual([waiterManifest]);210    expect(waiterApplied).toEqual([dialerManifest]);211  });212213  it("logs a clock-skew warning but does not abort when clocks differ by > 120s", async () => {214    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);215    const [session, peer] = inMemoryPair();216    const vault = makeVaultAdapter({ localVersion: 5 });217218    const initPromise = runSyncInitiator(session, vault);219220    await peer.receive(); // SYNC_HELLO221    // Peer reports a deviceTime 200 s in the past — beyond the 120 s threshold.222    await peer.send({223      type: "SYNC_HELLO",224      vaultId: "vault-abc",225      deviceId: "p",226      version: 3, // different version so exchange proceeds227      deviceTime: Date.now() - 200_000,228    });229    // Yield so the coordinator's async continuation can run the clock-skew check.230    await new Promise((r) => setTimeout(r, 0));231    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("clock skew"));232233    // Drive the sync to completion so the test doesn't leak.234    await peer.receive(); // SYNC_MANIFEST235    await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });236    await peer.receive(); // SYNC_REQUEST237    await peer.send({ type: "SYNC_REQUEST", entryIds: [] });238    await peer.receive(); // SYNC_ENTRIES239    await peer.receive(); // SYNC_TOMBSTONES240    await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });241    await peer.receive(); // SYNC_DONE242    await peer.send({ type: "SYNC_DONE" });243244    const outcome = await initPromise;245    expect(outcome.status).toBe("completed");246    warnSpy.mockRestore();247  });248});249250describe("runSyncInitiator", () => {251  it("syncs two mesh peers directly with their introducer absent", async () => {252    const [dialerSession, waiterSession] = symmetricSessionPair();253    const dialer = makeVaultAdapter({254      deviceId: "laptop-a",255      localVersion: 10,256      entries: { a: 10 },257      ciphertext: { a: "cipher-a" },258    });259    const waiter = makeVaultAdapter({260      deviceId: "laptop-c",261      localVersion: 11,262      entries: { c: 11 },263      ciphertext: { c: "cipher-c" },264    });265266    const [dialerOutcome, waiterOutcome] = await Promise.all([267      runSyncInitiator(dialerSession, dialer),268      runSyncResponder(waiterSession, waiter),269    ]);270271    expect(dialerOutcome.status).toBe("completed");272    expect(waiterOutcome.status).toBe("completed");273    expect(dialer.appliedEntries).toEqual([{ c: "cipher-c" }]);274    expect(waiter.appliedEntries).toEqual([{ a: "cipher-a" }]);275  });276277  it("reports the privacy-safe protocol stage when a device stops answering", async () => {278    vi.useFakeTimers();279    try {280      const [session, peer] = inMemoryPair();281      const vault = makeVaultAdapter({});282      const initPromise = runSyncInitiator(session, vault);283284      const hello = await peer.receive();285      expect(hello.type).toBe("SYNC_HELLO");286287      await vi.advanceTimersByTimeAsync(30_000);288      await expect(initPromise).resolves.toMatchObject({289        status: "timeout",290        error: "sync_timeout:waiting_for_device_hello",291      });292    } finally {293      vi.useRealTimers();294    }295  });296297  it("completes a full sync with bi-directional updates", async () => {298    const [session, peer] = inMemoryPair();299    const vault = makeVaultAdapter({300      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50301      tombstones: { x: 70 },302      ciphertext: { a: "ct-a", b: "ct-b" },303    });304305    const initPromise = runSyncInitiator(session, vault);306307    const peerProtocol = (async () => {308      const hello = await peer.receive();309      expect(hello.type).toBe("SYNC_HELLO");310      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now() });311312      const ourManifest = await peer.receive();313      expect(ourManifest.type).toBe("SYNC_MANIFEST");314      await peer.send({315        type: "SYNC_MANIFEST",316        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'317        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'318      });319320      const ourRequest = await peer.receive();321      expect(ourRequest.type).toBe("SYNC_REQUEST");322      // We should request the entries the peer reported newer323      if (ourRequest.type === "SYNC_REQUEST") {324        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);325      }326      // Peer asks us for entries the peer claimed it had older versions of327      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });328329      const ourEntries = await peer.receive();330      expect(ourEntries.type).toBe("SYNC_ENTRIES");331      if (ourEntries.type === "SYNC_ENTRIES") {332        expect(ourEntries.entries).toEqual({ a: "ct-a" });333      }334      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });335336      const ourTombs = await peer.receive();337      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");338      if (ourTombs.type === "SYNC_TOMBSTONES") {339        // We had x@70, peer had x@60, so we ship x@70340        expect(ourTombs.tombstones).toEqual({ x: 70 });341      }342      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });343344      const ourDone = await peer.receive();345      expect(ourDone.type).toBe("SYNC_DONE");346      await peer.send({ type: "SYNC_DONE" });347    })();348349    const outcome = await initPromise;350    await peerProtocol;351352    expect(outcome.status).toBe("completed");353    expect(outcome.entriesSent).toBe(1);354    expect(outcome.entriesReceived).toBe(2);355    expect(outcome.tombstonesSent).toBe(1);356    expect(outcome.tombstonesReceived).toBe(1);357    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);358    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);359  });360361  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {362    const [session, peer] = inMemoryPair();363    const vault = makeVaultAdapter({ vaultId: "vault-A" });364365    const initPromise = runSyncInitiator(session, vault);366    const peerProtocol = (async () => {367      const hello = await peer.receive();368      expect(hello.type).toBe("SYNC_HELLO");369      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now() });370      // Peer should receive a SYNC_ERROR and we should stop.371      const err = await peer.receive();372      expect(err.type).toBe("SYNC_ERROR");373    })();374375    const outcome = await initPromise;376    await peerProtocol;377    expect(outcome.status).toBe("vault_mismatch");378  });379380  it("exchanges device manifests when both adapters opt in", async () => {381    const [session, peer] = inMemoryPair();382    const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us");383    const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them");384385    const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey);386    const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey);387    const applied: SignedDeviceManifest[] = [];388389    const base = makeVaultAdapter({});390    const vault: SyncVaultAdapter = {391      ...base,392      async buildOutgoingDeviceManifest() { return outgoing; },393      async applyIncomingDeviceManifest(m, authenticatedKey) {394        expect(authenticatedKey).toEqual(session.remoteStaticPublicKey);395        applied.push(m);396      },397    };398399    const initPromise = runSyncInitiator(session, vault);400    const peerProtocol = (async () => {401      await peer.receive(); // hello402      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });403      await peer.receive(); // manifest404      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });405      await peer.receive(); // request406      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });407      await peer.receive(); // entries408      await peer.receive(); // tombstones409      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });410      const ourManifest = await peer.receive();411      expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST");412      if (ourManifest.type === "SYNC_DEVICE_MANIFEST") {413        expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us");414      }415      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer });416      await peer.receive(); // done417      await peer.send({ type: "SYNC_DONE" });418    })();419420    const outcome = await initPromise;421    await peerProtocol;422    expect(outcome.status).toBe("completed");423    expect(applied).toHaveLength(1);424    expect(applied[0].body.issuerDeviceId).toBe("them");425  });426427  it("uses null to represent an absent device manifest", async () => {428    const [session, peer] = inMemoryPair();429    const applyIncoming = vi.fn();430    const vault: SyncVaultAdapter = {431      ...makeVaultAdapter({}),432      async buildOutgoingDeviceManifest() { return null; },433      applyIncomingDeviceManifest: applyIncoming,434    };435436    const initPromise = runSyncInitiator(session, vault);437    const peerProtocol = (async () => {438      await peer.receive();439      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now() });440      await peer.receive();441      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });442      await peer.receive();443      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });444      await peer.receive();445      await peer.receive();446      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });447      const manifest = await peer.receive();448      expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null });449      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null });450      await peer.receive();451      await peer.send({ type: "SYNC_DONE" });452    })();453454    const outcome = await initPromise;455    await peerProtocol;456    expect(outcome.status).toBe("completed");457    expect(applyIncoming).not.toHaveBeenCalled();458  });459460  it("returns errored when peer sends a wrong frame", async () => {461    const [session, peer] = inMemoryPair();462    const vault = makeVaultAdapter({ vaultId: "v" });463464    const initPromise = runSyncInitiator(session, vault);465    void (async () => {466      await peer.receive(); // hello467      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now() });468      await peer.receive(); // our manifest469      await peer.send({ type: "SYNC_DONE" }); // wrong frame470    })();471472    const outcome = await initPromise;473    expect(outcome.status).toBe("errored");474    expect(outcome.error).toMatch(/SYNC_MANIFEST/);475  });476477  it("reports completed when the peer closes the socket while we wait for its closing SYNC_DONE", async () => {478    // Bug: by the time we send our own SYNC_DONE, every entry and tombstone479    // has already been exchanged and applied on both sides. The peer's reply480    // is a courtesy, not a commit. Android closes the socket the moment it481    // has our DONE; previously the initiator's parked receive() rejected with482    // "WebSocket closed" and the run was reported as `errored` → run_failed483    // → red, even though the data had demonstrably landed.484    const [session, peer] = inMemoryPair();485    const vault = makeVaultAdapter({ localVersion: 5 });486487    const initPromise = runSyncInitiator(session, vault);488    void (async () => {489      await peer.receive(); // hello490      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 5, deviceTime: Date.now() });491      await peer.receive(); // our done492      // No SYNC_DONE echo — Android just hangs up.493      await peer.closeInitiatorReceive();494    })();495496    const outcome = await initPromise;497    expect(outcome.status).toBe("completed");498  });499});