sync-coordinator.test.ts

1import { describe, expect, it, vi } from "vitest";2import { SYNC_LIMITS } from "../shared/sync-limits";3import {4  runSyncInitiator,5  runSyncResponder,6  type SyncVaultAdapter,7} from "./sync-coordinator";8import {9  DEVICE_MANIFEST_VERSION,10  type SyncMessage,11  type SignedDeviceManifest,12} from "../shared/sync-types";13import type { SyncSession } from "../shared/sync-transport";14import { buildSignedDeviceManifest } from "../shared/device-manifest";15import { deriveSyncSigningKeypair } from "../shared/sync-identity";1617function inMemoryPair(): [SyncSession, FakePeer] {18  const a: Uint8Array[] = [];19  const b: Uint8Array[] = [];20  const aR: Array<(v: Uint8Array) => void> = [];21  const bR: Array<(v: Uint8Array) => void> = [];22  const aRej: Array<(e: Error) => void> = [];23  let closed = false;2425  const ourSession: SyncSession = {26    send(bytes) {27      const r = bR.shift();28      if (r) r(bytes); else b.push(bytes);29    },30    receive(): Promise<Uint8Array> {31      const v = a.shift();32      if (v) return Promise.resolve(v);33      if (closed) return Promise.reject(new Error("WebSocket closed"));34      return new Promise<Uint8Array>((res, rej) => { aR.push(res); aRej.push(rej); });35    },36    async close() { /* no-op */ },37    remoteStaticPublicKey: new Uint8Array(32),38  };3940  const peer: FakePeer = {41    async send(msg: SyncMessage) {42      const bytes = new TextEncoder().encode(JSON.stringify(msg));43      const r = aR.shift();44      if (r) r(bytes); else a.push(bytes);45    },46    async receive(): Promise<SyncMessage> {47      const v = b.shift();48      const bytes = v ?? await new Promise<Uint8Array>((res) => bR.push(res));49      return JSON.parse(new TextDecoder().decode(bytes)) as SyncMessage;50    },51    /**52     * Android's listener closes the socket the moment it has our final53     * SYNC_DONE — there is no courtesy echo. Surface that as a "WebSocket54     * closed" rejection on the initiator's parked receive().55     */56    async closeInitiatorReceive() {57      if (closed) return;58      closed = true;59      while (aRej.length > 0) aRej.shift()!(new Error("WebSocket closed"));60    },61  };62  return [ourSession, peer];63}6465function symmetricSessionPair(): [SyncSession, SyncSession] {66  const queues: [Uint8Array[], Uint8Array[]] = [[], []];67  const readers: [68    Array<(value: Uint8Array) => void>,69    Array<(value: Uint8Array) => void>,70  ] = [[], []];71  const make = (side: 0 | 1): SyncSession => ({72    send(bytes) {73      const other = side === 0 ? 1 : 0;74      const reader = readers[other].shift();75      if (reader) reader(bytes);76      else queues[other].push(bytes);77    },78    receive() {79      const queued = queues[side].shift();80      if (queued) return Promise.resolve(queued);81      return new Promise<Uint8Array>((resolve) => readers[side].push(resolve));82    },83    async close() {},84    remoteStaticPublicKey: new Uint8Array(32).fill(side === 0 ? 0x22 : 0x11),85  });86  return [make(0), make(1)];87}8889interface FakePeer {90  send(msg: SyncMessage): Promise<void>;91  receive(): Promise<SyncMessage>;92  /** Hang up on the initiator's parked receive(), as Android does at DONE. */93  closeInitiatorReceive(): Promise<void>;94}9596function makeVaultAdapter(opts: {97  vaultId?: string;98  deviceId?: string;99  localVersion?: number;100  entries?: Record<string, number>;101  tombstones?: Record<string, number>;102  ciphertext?: Record<string, string>;103}): SyncVaultAdapter & {104  appliedEntries: Record<string, string>[];105  appliedTombstones: Record<string, number>[];106} {107  const appliedEntries: Record<string, string>[] = [];108  const appliedTombstones: Record<string, number>[] = [];109  return {110    async computeVaultId() { return opts.vaultId ?? "vault-abc"; },111    async getOurDeviceId() { return opts.deviceId ?? "device-init"; },112    // Use a non-zero default so the equal-version short circuit doesn't fire in113    // tests that aren't explicitly testing that path.114    async getLocalVersion() { return opts.localVersion ?? 5; },115    async buildEntryManifest() { return { ...(opts.entries ?? {}) }; },116    async buildTombstoneManifest() { return { ...(opts.tombstones ?? {}) }; },117    async getEncryptedEntries(ids) {118      const out: Record<string, string> = {};119      const cipher = opts.ciphertext ?? {};120      for (const id of ids) if (cipher[id] !== undefined) out[id] = cipher[id];121      return out;122    },123    async applyEntries(entries) { appliedEntries.push(entries); },124    async applyTombstones(t) { appliedTombstones.push(t); },125    appliedEntries,126    appliedTombstones,127  };128}129130// PB-26-111: SYNC_REQUEST.entryIds was bounded only by the 24 MiB frame size —131// roughly a million UUIDs — while each requested id costs an HKDF derivation132// plus an AES-GCM encryption of the entry envelope. An authenticated peer could133// burn ~30s of service-worker CPU per dial, repeatably, with no data effect that134// would look like an attack.135describe("runSyncInitiator — bounded SYNC_REQUEST (PB-26-111)", () => {136  it("rejects an oversized id list before any per-entry work", async () => {137    const [session, peer] = inMemoryPair();138    const vault = makeVaultAdapter({ vaultId: "vault-abc", entries: { "id-1": 1 } });139    let encryptedFor: string[] | null = null;140    const spied = Object.assign({}, vault, {141      async getEncryptedEntries(ids: string[]) {142        encryptedFor = ids;143        return {};144      },145    });146147    const initPromise = runSyncInitiator(session, spied);148    void (async () => {149      await peer.receive(); // hello150      await peer.send({151        type: "SYNC_HELLO",152        vaultId: "vault-abc",153        deviceId: "p",154        version: 0,155        deviceTime: Date.now(),156        protocolVersion: 2,157      });158      await peer.receive(); // our manifest159      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });160      await peer.receive(); // our request161      // Ask for more ids than the protocol allows.162      await peer.send({163        type: "SYNC_REQUEST",164        entryIds: Array.from(165          { length: SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT + 1 },166          (_, i) => `id-${i}`,167        ),168      });169    })();170171    const outcome = await initPromise;172    expect(outcome.status).toBe("errored");173    expect(outcome.error).toMatch(/entry count exceeds limit/);174    // The whole point: the bound fires before the expensive per-id work.175    expect(encryptedFor).toBeNull();176  });177178  it("still serves a request at the limit", async () => {179    const [session, peer] = inMemoryPair();180    const ids = Array.from({ length: 3 }, (_, i) => `id-${i}`);181    const vault = makeVaultAdapter({182      vaultId: "vault-abc",183      entries: { "id-0": 1, "id-1": 1, "id-2": 1 },184      ciphertext: { "id-0": "AAAA", "id-1": "BBBB", "id-2": "CCCC" },185    });186187    const initPromise = runSyncInitiator(session, vault);188    void (async () => {189      await peer.receive(); // hello190      await peer.send({191        type: "SYNC_HELLO",192        vaultId: "vault-abc",193        deviceId: "p",194        version: 0,195        deviceTime: Date.now(),196        protocolVersion: 2,197      });198      await peer.receive(); // our manifest199      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });200      await peer.receive(); // our request201      await peer.send({ type: "SYNC_REQUEST", entryIds: ids });202      await peer.receive(); // SYNC_ENTRIES203      await peer.receive(); // SYNC_TOMBSTONES204      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });205      await peer.receive(); // SYNC_DONE206      await peer.send({ type: "SYNC_DONE" });207    })();208209    const outcome = await initPromise;210    expect(outcome.status).toBe("completed");211    expect(outcome.entriesSent).toBe(3);212  });213});214215describe("runSyncInitiator — SYNC_MANIFEST limits (PB-26-312)", () => {216  it("rejects oversized entry and tombstone manifests before diffing", async () => {217    const [session, peer] = inMemoryPair();218    const vault = makeVaultAdapter({ vaultId: "vault-abc" });219    const initPromise = runSyncInitiator(session, vault);220    void (async () => {221      await peer.receive();222      await peer.send({223        type: "SYNC_HELLO",224        vaultId: "vault-abc",225        deviceId: "p",226        version: 0,227        deviceTime: Date.now(),228        protocolVersion: 2,229      });230      await peer.receive();231      const entries = Object.fromEntries(232        Array.from({ length: SYNC_LIMITS.MAX_ENTRY_COUNT + 1 }, (_, i) => [`id-${i}`, 1]),233      );234      await peer.send({ type: "SYNC_MANIFEST", entries, tombstones: {} });235    })();236    await expect(initPromise).resolves.toMatchObject({237      status: "errored",238      error: expect.stringMatching(/manifest entry count exceeds limit/),239    });240241    const [tombstoneSession, tombstonePeer] = inMemoryPair();242    const tombstonePromise = runSyncInitiator(tombstoneSession, vault);243    void (async () => {244      await tombstonePeer.receive();245      await tombstonePeer.send({246        type: "SYNC_HELLO",247        vaultId: "vault-abc",248        deviceId: "p",249        version: 0,250        deviceTime: Date.now(),251        protocolVersion: 2,252      });253      await tombstonePeer.receive();254      const tombstones = Object.fromEntries(255        Array.from({ length: SYNC_LIMITS.MAX_TOMBSTONE_COUNT + 1 }, (_, i) => [`id-${i}`, 1]),256      );257      await tombstonePeer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones });258    })();259    await expect(tombstonePromise).resolves.toMatchObject({260      status: "errored",261      error: expect.stringMatching(/manifest tombstone count exceeds limit/),262    });263  });264});265266describe("runSyncInitiator — HELLO fields and version optimisation", () => {267  it("sends HELLO with version and deviceTime", async () => {268    const [session, peer] = inMemoryPair();269    const vault = makeVaultAdapter({ localVersion: 7 });270271    const initPromise = runSyncInitiator(session, vault);272273    const sentHello = await peer.receive();274    expect(sentHello.type).toBe("SYNC_HELLO");275    if (sentHello.type === "SYNC_HELLO") {276      expect(sentHello.version).toBe(7);277      expect(typeof sentHello.deviceTime).toBe("number");278      expect(sentHello.deviceTime).toBeGreaterThan(0);279    }280    // Send a mismatched vaultId to abort early so the test doesn't hang.281    await peer.send({ type: "SYNC_HELLO", vaultId: "different-vault", deviceId: "p", version: 7, deviceTime: Date.now(), protocolVersion: 2 });282    await peer.receive(); // SYNC_ERROR283    await initPromise;284  });285286  it("exchanges entry manifests when vault versions are equal", async () => {287    const [dialerSession, waiterSession] = symmetricSessionPair();288    const dialer = makeVaultAdapter({289      deviceId: "dialer",290      localVersion: 10,291      entries: { a: 10 },292      ciphertext: { a: "cipher-a" },293    });294    const waiter = makeVaultAdapter({295      deviceId: "waiter",296      localVersion: 10,297      entries: { b: 11 },298      ciphertext: { b: "cipher-b" },299    });300301    const [dialerOutcome, waiterOutcome] = await Promise.all([302      runSyncInitiator(dialerSession, dialer),303      runSyncResponder(waiterSession, waiter),304    ]);305306    expect(dialerOutcome.status).toBe("completed");307    expect(waiterOutcome.status).toBe("completed");308    expect(dialer.appliedEntries).toEqual([{ b: "cipher-b" }]);309    expect(waiter.appliedEntries).toEqual([{ a: "cipher-a" }]);310  });311312  it("exchanges device graphs at equal entry versions when both peers advertise support", async () => {313    const [dialerSession, waiterSession] = symmetricSessionPair();314    const dialerManifest = {315      body: {316        manifestVersion: 3,317        issuerDeviceId: "dialer",318        issuedAt: 1,319        devices: [],320        revoked: [],321      },322      signatureHex: "11".repeat(64),323    } satisfies SignedDeviceManifest;324    const waiterManifest = {325      body: {326        manifestVersion: 3,327        issuerDeviceId: "waiter",328        issuedAt: 1,329        devices: [],330        revoked: [],331      },332      signatureHex: "22".repeat(64),333    } satisfies SignedDeviceManifest;334    const dialerApplied: SignedDeviceManifest[] = [];335    const waiterApplied: SignedDeviceManifest[] = [];336    const dialer: SyncVaultAdapter = {337      ...makeVaultAdapter({ deviceId: "dialer", localVersion: 7 }),338      async buildOutgoingDeviceManifest() { return dialerManifest; },339      async applyIncomingDeviceManifest(manifest) { dialerApplied.push(manifest); },340    };341    const waiter: SyncVaultAdapter = {342      ...makeVaultAdapter({ deviceId: "waiter", localVersion: 7 }),343      async buildOutgoingDeviceManifest() { return waiterManifest; },344      async applyIncomingDeviceManifest(manifest) { waiterApplied.push(manifest); },345    };346347    const [dialerOutcome, waiterOutcome] = await Promise.all([348      runSyncInitiator(dialerSession, dialer),349      runSyncResponder(waiterSession, waiter),350    ]);351352    expect(dialerOutcome.status).toBe("completed");353    expect(waiterOutcome.status).toBe("completed");354    expect(dialerApplied).toEqual([waiterManifest]);355    expect(waiterApplied).toEqual([dialerManifest]);356  });357358  it("negotiates the device-manifest step from both HELLOs in every role and version case", async () => {359    for (const initiatorSupports of [false, true]) {360      for (const responderSupports of [false, true]) {361        for (const equalVersions of [false, true]) {362          const [initiatorSession, responderSession] = symmetricSessionPair();363          const appliedByInitiator: SignedDeviceManifest[] = [];364          const appliedByResponder: SignedDeviceManifest[] = [];365          const manifest = (issuerDeviceId: string): SignedDeviceManifest => ({366            body: {367              manifestVersion: DEVICE_MANIFEST_VERSION,368              issuerDeviceId,369              issuedAt: 1,370              devices: [],371              revoked: [],372            },373            signatureHex: "11".repeat(64),374          });375          const adapter = (376            deviceId: string,377            localVersion: number,378            supports: boolean,379            applied: SignedDeviceManifest[],380          ): SyncVaultAdapter => ({381            ...makeVaultAdapter({ deviceId, localVersion }),382            ...(supports ? {383              async buildOutgoingDeviceManifest() { return manifest(deviceId); },384              async applyIncomingDeviceManifest(incoming: SignedDeviceManifest) {385                applied.push(incoming);386              },387            } : {}),388          });389          const initiator = adapter("initiator", 7, initiatorSupports, appliedByInitiator);390          const responder = adapter(391            "responder",392            equalVersions ? 7 : 9,393            responderSupports,394            appliedByResponder,395          );396397          const [initiatorOutcome, responderOutcome] = await Promise.all([398            runSyncInitiator(initiatorSession, initiator),399            runSyncResponder(responderSession, responder),400          ]);401          const bothSupport = initiatorSupports && responderSupports;402403          expect(initiatorOutcome.status).toBe("completed");404          expect(responderOutcome.status).toBe("completed");405          expect(appliedByInitiator).toHaveLength(bothSupport ? 1 : 0);406          expect(appliedByResponder).toHaveLength(bothSupport ? 1 : 0);407        }408      }409    }410  });411412  it("logs a clock-skew warning but does not abort when clocks differ by > 120s", async () => {413    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);414    const [session, peer] = inMemoryPair();415    const vault = makeVaultAdapter({ localVersion: 5 });416417    const initPromise = runSyncInitiator(session, vault);418419    await peer.receive(); // SYNC_HELLO420    // Peer reports a deviceTime 200 s in the past — beyond the 120 s threshold.421    await peer.send({422      type: "SYNC_HELLO",423      vaultId: "vault-abc",424      deviceId: "p",425      version: 3, // different version so exchange proceeds426        deviceTime: Date.now() - 200_000,427        protocolVersion: 2,428      });429    // Yield so the coordinator's async continuation can run the clock-skew check.430    await new Promise((r) => setTimeout(r, 0));431    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("clock skew"));432433    // Drive the sync to completion so the test doesn't leak.434    await peer.receive(); // SYNC_MANIFEST435    await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });436    await peer.receive(); // SYNC_REQUEST437    await peer.send({ type: "SYNC_REQUEST", entryIds: [] });438    await peer.receive(); // SYNC_ENTRIES439    await peer.receive(); // SYNC_TOMBSTONES440    await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });441    await peer.receive(); // SYNC_DONE442    await peer.send({ type: "SYNC_DONE" });443444    const outcome = await initPromise;445    expect(outcome.status).toBe("completed");446    warnSpy.mockRestore();447  });448});449450describe("runSyncInitiator", () => {451  it("syncs two mesh peers directly with their introducer absent", async () => {452    const [dialerSession, waiterSession] = symmetricSessionPair();453    const dialer = makeVaultAdapter({454      deviceId: "laptop-a",455      localVersion: 10,456      entries: { a: 10 },457      ciphertext: { a: "cipher-a" },458    });459    const waiter = makeVaultAdapter({460      deviceId: "laptop-c",461      localVersion: 11,462      entries: { c: 11 },463      ciphertext: { c: "cipher-c" },464    });465466    const [dialerOutcome, waiterOutcome] = await Promise.all([467      runSyncInitiator(dialerSession, dialer),468      runSyncResponder(waiterSession, waiter),469    ]);470471    expect(dialerOutcome.status).toBe("completed");472    expect(waiterOutcome.status).toBe("completed");473    expect(dialer.appliedEntries).toEqual([{ c: "cipher-c" }]);474    expect(waiter.appliedEntries).toEqual([{ a: "cipher-a" }]);475  });476477  it("reports the privacy-safe protocol stage when a device stops answering", async () => {478    vi.useFakeTimers();479    try {480      const [session, peer] = inMemoryPair();481      const vault = makeVaultAdapter({});482      const initPromise = runSyncInitiator(session, vault);483484      const hello = await peer.receive();485      expect(hello.type).toBe("SYNC_HELLO");486487      await vi.advanceTimersByTimeAsync(30_000);488      await expect(initPromise).resolves.toMatchObject({489        status: "timeout",490        error: "sync_timeout:waiting_for_device_hello",491      });492    } finally {493      vi.useRealTimers();494    }495  });496497  it("completes a full sync with bi-directional updates", async () => {498    const [session, peer] = inMemoryPair();499    const vault = makeVaultAdapter({500      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50501      tombstones: { x: 70 },502      ciphertext: { a: "ct-a", b: "ct-b" },503    });504505    const initPromise = runSyncInitiator(session, vault);506507    const peerProtocol = (async () => {508      const hello = await peer.receive();509      expect(hello.type).toBe("SYNC_HELLO");510      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now(), protocolVersion: 2 });511512      const ourManifest = await peer.receive();513      expect(ourManifest.type).toBe("SYNC_MANIFEST");514      await peer.send({515        type: "SYNC_MANIFEST",516        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'517        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'518      });519520      const ourRequest = await peer.receive();521      expect(ourRequest.type).toBe("SYNC_REQUEST");522      // We should request the entries the peer reported newer523      if (ourRequest.type === "SYNC_REQUEST") {524        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);525      }526      // Peer asks us for entries the peer claimed it had older versions of527      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });528529      const ourEntries = await peer.receive();530      expect(ourEntries.type).toBe("SYNC_ENTRIES");531      if (ourEntries.type === "SYNC_ENTRIES") {532        expect(ourEntries.entries).toEqual({ a: "ct-a" });533      }534      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });535536      const ourTombs = await peer.receive();537      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");538      if (ourTombs.type === "SYNC_TOMBSTONES") {539        // We had x@70, peer had x@60, so we ship x@70540        expect(ourTombs.tombstones).toEqual({ x: 70 });541      }542      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });543544      const ourDone = await peer.receive();545      expect(ourDone.type).toBe("SYNC_DONE");546      await peer.send({ type: "SYNC_DONE" });547    })();548549    const outcome = await initPromise;550    await peerProtocol;551552    expect(outcome.status).toBe("completed");553    expect(outcome.entriesSent).toBe(1);554    expect(outcome.entriesReceived).toBe(2);555    expect(outcome.tombstonesSent).toBe(1);556    expect(outcome.tombstonesReceived).toBe(1);557    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);558    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);559  });560561  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {562    const [session, peer] = inMemoryPair();563    const vault = makeVaultAdapter({ vaultId: "vault-A" });564565    const initPromise = runSyncInitiator(session, vault);566    const peerProtocol = (async () => {567      const hello = await peer.receive();568      expect(hello.type).toBe("SYNC_HELLO");569      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now(), protocolVersion: 2 });570      // Peer should receive a SYNC_ERROR and we should stop.571      const err = await peer.receive();572      expect(err.type).toBe("SYNC_ERROR");573    })();574575    const outcome = await initPromise;576    await peerProtocol;577    expect(outcome.status).toBe("vault_mismatch");578  });579580  it("rejects a stale peer with PEER_PROTOCOL_OUTDATED", async () => {581    const [session, peer] = inMemoryPair();582    const vault = makeVaultAdapter({});583584    const initPromise = runSyncInitiator(session, vault);585    const peerProtocol = (async () => {586      await peer.receive(); // hello587      await peer.send({588        type: "SYNC_HELLO",589        vaultId: "vault-abc",590        deviceId: "p",591        version: 3,592        deviceTime: Date.now(),593      });594      const err = await peer.receive();595      expect(err.type).toBe("SYNC_ERROR");596    })();597598    const outcome = await initPromise;599    await peerProtocol;600    expect(outcome.status).toBe("errored");601    expect(outcome.error).toBe("PEER_PROTOCOL_OUTDATED");602  });603604  it("exchanges device manifests when both adapters opt in", async () => {605    const [session, peer] = inMemoryPair();606    const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us");607    const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them");608609    const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey);610    const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey);611    const applied: SignedDeviceManifest[] = [];612613    const base = makeVaultAdapter({});614    const vault: SyncVaultAdapter = {615      ...base,616      async buildOutgoingDeviceManifest() { return outgoing; },617      async applyIncomingDeviceManifest(m, authenticatedKey) {618        expect(authenticatedKey).toEqual(session.remoteStaticPublicKey);619        applied.push(m);620      },621    };622623    const initPromise = runSyncInitiator(session, vault);624    const peerProtocol = (async () => {625      await peer.receive(); // hello626      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: 2, deviceManifestVersion: 4 });627      await peer.receive(); // manifest628      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });629      await peer.receive(); // request630      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });631      await peer.receive(); // entries632      await peer.receive(); // tombstones633      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });634      const ourManifest = await peer.receive();635      expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST");636      if (ourManifest.type === "SYNC_DEVICE_MANIFEST") {637        expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us");638      }639      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer });640      await peer.receive(); // done641      await peer.send({ type: "SYNC_DONE" });642    })();643644    const outcome = await initPromise;645    await peerProtocol;646    expect(outcome.status).toBe("completed");647    expect(applied).toHaveLength(1);648    expect(applied[0].body.issuerDeviceId).toBe("them");649  });650651  it("uses null to represent an absent device manifest", async () => {652    const [session, peer] = inMemoryPair();653    const applyIncoming = vi.fn();654    const vault: SyncVaultAdapter = {655      ...makeVaultAdapter({}),656      async buildOutgoingDeviceManifest() { return null; },657      applyIncomingDeviceManifest: applyIncoming,658    };659660    const initPromise = runSyncInitiator(session, vault);661    const peerProtocol = (async () => {662      await peer.receive();663      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: 2, deviceManifestVersion: 4 });664      await peer.receive();665      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });666      await peer.receive();667      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });668      await peer.receive();669      await peer.receive();670      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });671      const manifest = await peer.receive();672      expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null });673      await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null });674      await peer.receive();675      await peer.send({ type: "SYNC_DONE" });676    })();677678    const outcome = await initPromise;679    await peerProtocol;680    expect(outcome.status).toBe("completed");681    expect(applyIncoming).not.toHaveBeenCalled();682  });683684  it("returns errored when peer sends a wrong frame", async () => {685    const [session, peer] = inMemoryPair();686    const vault = makeVaultAdapter({ vaultId: "v" });687688    const initPromise = runSyncInitiator(session, vault);689    void (async () => {690      await peer.receive(); // hello691      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: 2 });692      await peer.receive(); // our manifest693      await peer.send({ type: "SYNC_DONE" }); // wrong frame694    })();695696    const outcome = await initPromise;697    expect(outcome.status).toBe("errored");698    expect(outcome.error).toMatch(/SYNC_MANIFEST/);699  });700701  it("reports completed when the peer closes the socket while we wait for its closing SYNC_DONE", async () => {702    // Bug: by the time we send our own SYNC_DONE, every entry and tombstone703    // has already been exchanged and applied on both sides. The peer's reply704    // is a courtesy, not a commit. Android closes the socket the moment it705    // has our DONE; previously the initiator's parked receive() rejected with706    // "WebSocket closed" and the run was reported as `errored` → run_failed707    // → red, even though the data had demonstrably landed.708    const [session, peer] = inMemoryPair();709    const vault = makeVaultAdapter({ localVersion: 5 });710711    const initPromise = runSyncInitiator(session, vault);712    void (async () => {713      await peer.receive(); // hello714      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 5, deviceTime: Date.now(), protocolVersion: 2 });715      await peer.receive(); // manifest716      await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} });717      await peer.receive(); // request718      await peer.send({ type: "SYNC_REQUEST", entryIds: [] });719      await peer.receive(); // unconditional entries720      await peer.receive(); // tombstones721      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} });722      await peer.receive(); // done723      // No SYNC_DONE echo — Android just hangs up.724      await peer.closeInitiatorReceive();725    })();726727    const outcome = await initPromise;728    expect(outcome.status).toBe("completed");729  });730});