import { describe, expect, it, vi } from "vitest"; import { SYNC_LIMITS } from "../shared/sync-limits"; import { runSyncInitiator, runSyncResponder, type SyncVaultAdapter, } from "./sync-coordinator"; import { DEVICE_MANIFEST_VERSION, SYNC_PROTOCOL_VERSION, type SyncMessage, type SignedDeviceManifest, } from "../shared/sync-types"; import type { SyncSession } from "../shared/sync-transport"; import { buildSignedDeviceManifest } from "../shared/device-manifest"; import { deriveSyncSigningKeypair } from "../shared/legacy-identity-derivation.test-support"; function inMemoryPair(): [SyncSession, FakePeer] { const a: Uint8Array[] = []; const b: Uint8Array[] = []; const aR: Array<(v: Uint8Array) => void> = []; const bR: Array<(v: Uint8Array) => void> = []; const aRej: Array<(e: Error) => void> = []; let closed = false; const ourSession: SyncSession = { send(bytes) { const r = bR.shift(); if (r) r(bytes); else b.push(bytes); }, receive(): Promise { const v = a.shift(); if (v) return Promise.resolve(v); if (closed) return Promise.reject(new Error("WebSocket closed")); return new Promise((res, rej) => { aR.push(res); aRej.push(rej); }); }, async close() { /* no-op */ }, remoteStaticPublicKey: new Uint8Array(32), }; const peer: FakePeer = { async send(msg: SyncMessage) { const bytes = new TextEncoder().encode(JSON.stringify(msg)); const r = aR.shift(); if (r) r(bytes); else a.push(bytes); }, async receive(): Promise { const v = b.shift(); const bytes = v ?? await new Promise((res) => bR.push(res)); return JSON.parse(new TextDecoder().decode(bytes)) as SyncMessage; }, /** * Android's listener closes the socket the moment it has our final * SYNC_DONE — there is no courtesy echo. Surface that as a "WebSocket * closed" rejection on the initiator's parked receive(). */ async closeInitiatorReceive() { if (closed) return; closed = true; while (aRej.length > 0) aRej.shift()!(new Error("WebSocket closed")); }, }; return [ourSession, peer]; } function symmetricSessionPair(): [SyncSession, SyncSession] { const queues: [Uint8Array[], Uint8Array[]] = [[], []]; const readers: [ Array<(value: Uint8Array) => void>, Array<(value: Uint8Array) => void>, ] = [[], []]; const make = (side: 0 | 1): SyncSession => ({ send(bytes) { const other = side === 0 ? 1 : 0; const reader = readers[other].shift(); if (reader) reader(bytes); else queues[other].push(bytes); }, receive() { const queued = queues[side].shift(); if (queued) return Promise.resolve(queued); return new Promise((resolve) => readers[side].push(resolve)); }, async close() {}, remoteStaticPublicKey: new Uint8Array(32).fill(side === 0 ? 0x22 : 0x11), }); return [make(0), make(1)]; } interface FakePeer { send(msg: SyncMessage): Promise; receive(): Promise; /** Hang up on the initiator's parked receive(), as Android does at DONE. */ closeInitiatorReceive(): Promise; } function makeVaultAdapter(opts: { vaultId?: string; deviceId?: string; localVersion?: number; entries?: Record; tombstones?: Record; ciphertext?: Record; }): SyncVaultAdapter & { appliedEntries: Record[]; appliedTombstones: Record[]; } { const appliedEntries: Record[] = []; const appliedTombstones: Record[] = []; return { async computeVaultId() { return opts.vaultId ?? "vault-abc"; }, async getOurDeviceId() { return opts.deviceId ?? "device-init"; }, // Use a non-zero default so the equal-version short circuit doesn't fire in // tests that aren't explicitly testing that path. async getLocalVersion() { return opts.localVersion ?? 5; }, async buildEntryManifest() { return { ...(opts.entries ?? {}) }; }, async buildTombstoneManifest() { return { ...(opts.tombstones ?? {}) }; }, async getEncryptedEntries(ids) { const out: Record = {}; const cipher = opts.ciphertext ?? {}; for (const id of ids) if (cipher[id] !== undefined) out[id] = cipher[id]; return out; }, async applyEntries(entries) { appliedEntries.push(entries); }, async applyTombstones(t) { appliedTombstones.push(t); }, appliedEntries, appliedTombstones, }; } // PB-26-111: SYNC_REQUEST.entryIds was bounded only by the 24 MiB frame size — // roughly a million UUIDs — while each requested id costs an HKDF derivation // plus an AES-GCM encryption of the entry envelope. An authenticated peer could // burn ~30s of service-worker CPU per dial, repeatably, with no data effect that // would look like an attack. describe("runSyncInitiator — bounded SYNC_REQUEST (PB-26-111)", () => { it("rejects an oversized id list before any per-entry work", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ vaultId: "vault-abc", entries: { "id-1": 1 } }); let encryptedFor: string[] | null = null; const spied = Object.assign({}, vault, { async getEncryptedEntries(ids: string[]) { encryptedFor = ids; return {}; }, }); const initPromise = runSyncInitiator(session, spied); void (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 0, deviceTime: Date.now(), protocolVersion: 3, }); await peer.receive(); // our manifest await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} }); await peer.receive(); // our request // Ask for more ids than the protocol allows. await peer.send({ type: "SYNC_REQUEST", entryIds: Array.from( { length: SYNC_LIMITS.MAX_REQUEST_ENTRY_COUNT + 1 }, (_, i) => `id-${i}`, ), }); })(); const outcome = await initPromise; expect(outcome.status).toBe("errored"); expect(outcome.error).toMatch(/entry count exceeds limit/); // The whole point: the bound fires before the expensive per-id work. expect(encryptedFor).toBeNull(); }); it("still serves a request at the limit", async () => { const [session, peer] = inMemoryPair(); const ids = Array.from({ length: 3 }, (_, i) => `id-${i}`); const vault = makeVaultAdapter({ vaultId: "vault-abc", entries: { "id-0": 1, "id-1": 1, "id-2": 1 }, ciphertext: { "id-0": "AAAA", "id-1": "BBBB", "id-2": "CCCC" }, }); const initPromise = runSyncInitiator(session, vault); void (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 0, deviceTime: Date.now(), protocolVersion: 3, }); await peer.receive(); // our manifest await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} }); await peer.receive(); // our request await peer.send({ type: "SYNC_REQUEST", entryIds: ids }); await peer.receive(); // SYNC_ENTRIES await peer.receive(); // SYNC_TOMBSTONES await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} }); await peer.receive(); // SYNC_DONE await peer.send({ type: "SYNC_DONE" }); })(); const outcome = await initPromise; expect(outcome.status).toBe("completed"); expect(outcome.entriesSent).toBe(3); }); }); describe("runSyncInitiator — SYNC_MANIFEST limits (PB-26-312)", () => { it("rejects oversized entry and tombstone manifests before diffing", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ vaultId: "vault-abc" }); const initPromise = runSyncInitiator(session, vault); void (async () => { await peer.receive(); await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 0, deviceTime: Date.now(), protocolVersion: 3, }); await peer.receive(); const entries = Object.fromEntries( Array.from({ length: SYNC_LIMITS.MAX_ENTRY_COUNT + 1 }, (_, i) => [`id-${i}`, 1]), ); await peer.send({ type: "SYNC_MANIFEST", entries, tombstones: {} }); })(); await expect(initPromise).resolves.toMatchObject({ status: "errored", error: expect.stringMatching(/manifest entry count exceeds limit/), }); const [tombstoneSession, tombstonePeer] = inMemoryPair(); const tombstonePromise = runSyncInitiator(tombstoneSession, vault); void (async () => { await tombstonePeer.receive(); await tombstonePeer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 0, deviceTime: Date.now(), protocolVersion: 3, }); await tombstonePeer.receive(); const tombstones = Object.fromEntries( Array.from({ length: SYNC_LIMITS.MAX_TOMBSTONE_COUNT + 1 }, (_, i) => [`id-${i}`, 1]), ); await tombstonePeer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones }); })(); await expect(tombstonePromise).resolves.toMatchObject({ status: "errored", error: expect.stringMatching(/manifest tombstone count exceeds limit/), }); }); }); describe("runSyncInitiator — HELLO fields and version optimisation", () => { it("sends HELLO with version and deviceTime", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ localVersion: 7 }); const initPromise = runSyncInitiator(session, vault); const sentHello = await peer.receive(); expect(sentHello.type).toBe("SYNC_HELLO"); if (sentHello.type === "SYNC_HELLO") { expect(sentHello.version).toBe(7); expect(typeof sentHello.deviceTime).toBe("number"); expect(sentHello.deviceTime).toBeGreaterThan(0); } // Send a mismatched vaultId to abort early so the test doesn't hang. await peer.send({ type: "SYNC_HELLO", vaultId: "different-vault", deviceId: "p", version: 7, deviceTime: Date.now(), protocolVersion: 3 }); await peer.receive(); // SYNC_ERROR await initPromise; }); it("exchanges entry manifests when vault versions are equal", async () => { const [dialerSession, waiterSession] = symmetricSessionPair(); const dialer = makeVaultAdapter({ deviceId: "dialer", localVersion: 10, entries: { a: 10 }, ciphertext: { a: "cipher-a" }, }); const waiter = makeVaultAdapter({ deviceId: "waiter", localVersion: 10, entries: { b: 11 }, ciphertext: { b: "cipher-b" }, }); const [dialerOutcome, waiterOutcome] = await Promise.all([ runSyncInitiator(dialerSession, dialer), runSyncResponder(waiterSession, waiter), ]); expect(dialerOutcome.status).toBe("completed"); expect(waiterOutcome.status).toBe("completed"); expect(dialer.appliedEntries).toEqual([{ b: "cipher-b" }]); expect(waiter.appliedEntries).toEqual([{ a: "cipher-a" }]); }); it("exchanges device graphs at equal entry versions when both peers advertise support", async () => { const [dialerSession, waiterSession] = symmetricSessionPair(); const dialerManifest = { body: { manifestVersion: 5, issuerDeviceId: "dialer", issuedAt: 1, devices: [], revoked: [], }, signatureHex: "11".repeat(64), } satisfies SignedDeviceManifest; const waiterManifest = { body: { manifestVersion: 5, issuerDeviceId: "waiter", issuedAt: 1, devices: [], revoked: [], }, signatureHex: "22".repeat(64), } satisfies SignedDeviceManifest; const dialerApplied: SignedDeviceManifest[] = []; const waiterApplied: SignedDeviceManifest[] = []; const dialer: SyncVaultAdapter = { ...makeVaultAdapter({ deviceId: "dialer", localVersion: 7 }), async buildOutgoingDeviceManifest() { return dialerManifest; }, async applyIncomingDeviceManifest(manifest) { dialerApplied.push(manifest); }, }; const waiter: SyncVaultAdapter = { ...makeVaultAdapter({ deviceId: "waiter", localVersion: 7 }), async buildOutgoingDeviceManifest() { return waiterManifest; }, async applyIncomingDeviceManifest(manifest) { waiterApplied.push(manifest); }, }; const [dialerOutcome, waiterOutcome] = await Promise.all([ runSyncInitiator(dialerSession, dialer), runSyncResponder(waiterSession, waiter), ]); expect(dialerOutcome.status).toBe("completed"); expect(waiterOutcome.status).toBe("completed"); expect(dialerApplied).toEqual([waiterManifest]); expect(waiterApplied).toEqual([dialerManifest]); }); it("negotiates the device-manifest step from both HELLOs in every role and version case", async () => { for (const initiatorSupports of [false, true]) { for (const responderSupports of [false, true]) { for (const equalVersions of [false, true]) { const [initiatorSession, responderSession] = symmetricSessionPair(); const appliedByInitiator: SignedDeviceManifest[] = []; const appliedByResponder: SignedDeviceManifest[] = []; const manifest = (issuerDeviceId: string): SignedDeviceManifest => ({ body: { manifestVersion: DEVICE_MANIFEST_VERSION, issuerDeviceId, issuedAt: 1, devices: [], revoked: [], }, signatureHex: "11".repeat(64), }); const adapter = ( deviceId: string, localVersion: number, supports: boolean, applied: SignedDeviceManifest[], ): SyncVaultAdapter => ({ ...makeVaultAdapter({ deviceId, localVersion }), ...(supports ? { async buildOutgoingDeviceManifest() { return manifest(deviceId); }, async applyIncomingDeviceManifest(incoming: SignedDeviceManifest) { applied.push(incoming); }, } : {}), }); const initiator = adapter("initiator", 7, initiatorSupports, appliedByInitiator); const responder = adapter( "responder", equalVersions ? 7 : 9, responderSupports, appliedByResponder, ); const [initiatorOutcome, responderOutcome] = await Promise.all([ runSyncInitiator(initiatorSession, initiator), runSyncResponder(responderSession, responder), ]); const bothSupport = initiatorSupports && responderSupports; expect(initiatorOutcome.status).toBe("completed"); expect(responderOutcome.status).toBe("completed"); expect(appliedByInitiator).toHaveLength(bothSupport ? 1 : 0); expect(appliedByResponder).toHaveLength(bothSupport ? 1 : 0); } } } }); it("logs a clock-skew warning but does not abort when clocks differ by > 120s", async () => { const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ localVersion: 5 }); const initPromise = runSyncInitiator(session, vault); await peer.receive(); // SYNC_HELLO // Peer reports a deviceTime 200 s in the past — beyond the 120 s threshold. await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, // different version so exchange proceeds deviceTime: Date.now() - 200_000, protocolVersion: 3, }); // Yield so the coordinator's async continuation can run the clock-skew check. await new Promise((r) => setTimeout(r, 0)); expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("clock skew")); // Drive the sync to completion so the test doesn't leak. await peer.receive(); // SYNC_MANIFEST await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} }); await peer.receive(); // SYNC_REQUEST await peer.send({ type: "SYNC_REQUEST", entryIds: [] }); await peer.receive(); // SYNC_ENTRIES await peer.receive(); // SYNC_TOMBSTONES await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} }); await peer.receive(); // SYNC_DONE await peer.send({ type: "SYNC_DONE" }); const outcome = await initPromise; expect(outcome.status).toBe("completed"); warnSpy.mockRestore(); }); }); describe("runSyncInitiator", () => { it("syncs two mesh peers directly with their introducer absent", async () => { const [dialerSession, waiterSession] = symmetricSessionPair(); const dialer = makeVaultAdapter({ deviceId: "laptop-a", localVersion: 10, entries: { a: 10 }, ciphertext: { a: "cipher-a" }, }); const waiter = makeVaultAdapter({ deviceId: "laptop-c", localVersion: 11, entries: { c: 11 }, ciphertext: { c: "cipher-c" }, }); const [dialerOutcome, waiterOutcome] = await Promise.all([ runSyncInitiator(dialerSession, dialer), runSyncResponder(waiterSession, waiter), ]); expect(dialerOutcome.status).toBe("completed"); expect(waiterOutcome.status).toBe("completed"); expect(dialer.appliedEntries).toEqual([{ c: "cipher-c" }]); expect(waiter.appliedEntries).toEqual([{ a: "cipher-a" }]); }); it("reports the privacy-safe protocol stage when a device stops answering", async () => { vi.useFakeTimers(); try { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({}); const initPromise = runSyncInitiator(session, vault); const hello = await peer.receive(); expect(hello.type).toBe("SYNC_HELLO"); await vi.advanceTimersByTimeAsync(30_000); await expect(initPromise).resolves.toMatchObject({ status: "timeout", error: "sync_timeout:waiting_for_device_hello", }); } finally { vi.useRealTimers(); } }); it("completes a full sync with bi-directional updates", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=50 tombstones: { x: 70 }, ciphertext: { a: "ct-a", b: "ct-b" }, }); const initPromise = runSyncInitiator(session, vault); const peerProtocol = (async () => { const hello = await peer.receive(); expect(hello.type).toBe("SYNC_HELLO"); await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer", version: 3, deviceTime: Date.now(), protocolVersion: 3 }); const ourManifest = await peer.receive(); expect(ourManifest.type).toBe("SYNC_MANIFEST"); await peer.send({ type: "SYNC_MANIFEST", entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c' tombstones: { x: 60, y: 99 }, // peer has newer tombstone for 'y' }); const ourRequest = await peer.receive(); expect(ourRequest.type).toBe("SYNC_REQUEST"); // We should request the entries the peer reported newer if (ourRequest.type === "SYNC_REQUEST") { expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]); } // Peer asks us for entries the peer claimed it had older versions of await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] }); const ourEntries = await peer.receive(); expect(ourEntries.type).toBe("SYNC_ENTRIES"); if (ourEntries.type === "SYNC_ENTRIES") { expect(ourEntries.entries).toEqual({ a: "ct-a" }); } await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } }); const ourTombs = await peer.receive(); expect(ourTombs.type).toBe("SYNC_TOMBSTONES"); if (ourTombs.type === "SYNC_TOMBSTONES") { // We had x@70, peer had x@60, so we ship x@70 expect(ourTombs.tombstones).toEqual({ x: 70 }); } await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } }); const ourDone = await peer.receive(); expect(ourDone.type).toBe("SYNC_DONE"); await peer.send({ type: "SYNC_DONE" }); })(); const outcome = await initPromise; await peerProtocol; expect(outcome.status).toBe("completed"); expect(outcome.entriesSent).toBe(1); expect(outcome.entriesReceived).toBe(2); expect(outcome.tombstonesSent).toBe(1); expect(outcome.tombstonesReceived).toBe(1); expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]); expect(vault.appliedTombstones).toEqual([{ y: 99 }]); }); it("aborts with vault_mismatch on a peer with a different vault_id", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ vaultId: "vault-A" }); const initPromise = runSyncInitiator(session, vault); const peerProtocol = (async () => { const hello = await peer.receive(); expect(hello.type).toBe("SYNC_HELLO"); await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil", version: 3, deviceTime: Date.now(), protocolVersion: 3 }); // Peer should receive a SYNC_ERROR and we should stop. const err = await peer.receive(); expect(err.type).toBe("SYNC_ERROR"); })(); const outcome = await initPromise; await peerProtocol; expect(outcome.status).toBe("vault_mismatch"); }); it("rejects a stale peer with PEER_PROTOCOL_OUTDATED", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({}); const initPromise = runSyncInitiator(session, vault); const peerProtocol = (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), }); const err = await peer.receive(); expect(err.type).toBe("SYNC_ERROR"); })(); const outcome = await initPromise; await peerProtocol; expect(outcome.status).toBe("errored"); expect(outcome.error).toBe("PEER_PROTOCOL_OUTDATED"); }); it("rejects a lower explicit protocol version as repair-required", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({}); const initPromise = runSyncInitiator(session, vault); const peerProtocol = (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: SYNC_PROTOCOL_VERSION - 1, }); const err = await peer.receive(); expect(err).toEqual({ type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" }); })(); const outcome = await initPromise; await peerProtocol; expect(outcome.status).toBe("errored"); expect(outcome.error).toBe("PEER_PROTOCOL_OUTDATED"); }); it("rejects a future protocol version rather than failing open", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({}); const initPromise = runSyncInitiator(session, vault); const peerProtocol = (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: SYNC_PROTOCOL_VERSION + 1, }); const err = await peer.receive(); expect(err).toEqual({ type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" }); })(); const outcome = await initPromise; await peerProtocol; expect(outcome.status).toBe("errored"); expect(outcome.error).toBe("PEER_PROTOCOL_OUTDATED"); }); it("responder rejects an incompatible peer before sending any manifest", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({}); const respondPromise = runSyncResponder(session, vault); await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: SYNC_PROTOCOL_VERSION + 1, }); const reply = await peer.receive(); expect(reply).toEqual({ type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" }); const outcome = await respondPromise; expect(outcome.status).toBe("errored"); expect(outcome.error).toBe("PEER_PROTOCOL_OUTDATED"); }); it("initiator surfaces repair-required when the responder rejects first", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({}); const initPromise = runSyncInitiator(session, vault); await peer.receive(); // hello await peer.send({ type: "SYNC_ERROR", error: "PEER_PROTOCOL_OUTDATED" }); const outcome = await initPromise; expect(outcome.status).toBe("errored"); expect(outcome.error).toBe("PEER_PROTOCOL_OUTDATED"); }); it("exchanges device manifests when both adapters opt in", async () => { const [session, peer] = inMemoryPair(); const ourKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x11), "us"); const peerKp = await deriveSyncSigningKeypair(new Uint8Array(32).fill(0x22), "them"); const outgoing = await buildSignedDeviceManifest("us", [], ourKp.privateKey); const incomingFromPeer = await buildSignedDeviceManifest("them", [], peerKp.privateKey); const applied: SignedDeviceManifest[] = []; const base = makeVaultAdapter({}); const vault: SyncVaultAdapter = { ...base, async buildOutgoingDeviceManifest() { return outgoing; }, async applyIncomingDeviceManifest(m, authenticatedKey) { expect(authenticatedKey).toEqual(session.remoteStaticPublicKey); applied.push(m); }, }; const initPromise = runSyncInitiator(session, vault); const peerProtocol = (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: 3, deviceManifestVersion: 5 }); await peer.receive(); // manifest await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} }); await peer.receive(); // request await peer.send({ type: "SYNC_REQUEST", entryIds: [] }); await peer.receive(); // entries await peer.receive(); // tombstones await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} }); const ourManifest = await peer.receive(); expect(ourManifest.type).toBe("SYNC_DEVICE_MANIFEST"); if (ourManifest.type === "SYNC_DEVICE_MANIFEST") { expect(ourManifest.manifest?.body.issuerDeviceId).toBe("us"); } await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: incomingFromPeer }); await peer.receive(); // done await peer.send({ type: "SYNC_DONE" }); })(); const outcome = await initPromise; await peerProtocol; expect(outcome.status).toBe("completed"); expect(applied).toHaveLength(1); expect(applied[0].body.issuerDeviceId).toBe("them"); }); it("uses null to represent an absent device manifest", async () => { const [session, peer] = inMemoryPair(); const applyIncoming = vi.fn(); const vault: SyncVaultAdapter = { ...makeVaultAdapter({}), async buildOutgoingDeviceManifest() { return null; }, applyIncomingDeviceManifest: applyIncoming, }; const initPromise = runSyncInitiator(session, vault); const peerProtocol = (async () => { await peer.receive(); await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: 3, deviceManifestVersion: 5 }); await peer.receive(); await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} }); await peer.receive(); await peer.send({ type: "SYNC_REQUEST", entryIds: [] }); await peer.receive(); await peer.receive(); await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} }); const manifest = await peer.receive(); expect(manifest).toEqual({ type: "SYNC_DEVICE_MANIFEST", manifest: null }); await peer.send({ type: "SYNC_DEVICE_MANIFEST", manifest: null }); await peer.receive(); await peer.send({ type: "SYNC_DONE" }); })(); const outcome = await initPromise; await peerProtocol; expect(outcome.status).toBe("completed"); expect(applyIncoming).not.toHaveBeenCalled(); }); it("returns errored when peer sends a wrong frame", async () => { const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ vaultId: "v" }); const initPromise = runSyncInitiator(session, vault); void (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p", version: 3, deviceTime: Date.now(), protocolVersion: 3 }); await peer.receive(); // our manifest await peer.send({ type: "SYNC_DONE" }); // wrong frame })(); const outcome = await initPromise; expect(outcome.status).toBe("errored"); expect(outcome.error).toMatch(/SYNC_MANIFEST/); }); it("reports completed when the peer closes the socket while we wait for its closing SYNC_DONE", async () => { // Bug: by the time we send our own SYNC_DONE, every entry and tombstone // has already been exchanged and applied on both sides. The peer's reply // is a courtesy, not a commit. Android closes the socket the moment it // has our DONE; previously the initiator's parked receive() rejected with // "WebSocket closed" and the run was reported as `errored` → run_failed // → red, even though the data had demonstrably landed. const [session, peer] = inMemoryPair(); const vault = makeVaultAdapter({ localVersion: 5 }); const initPromise = runSyncInitiator(session, vault); void (async () => { await peer.receive(); // hello await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "p", version: 5, deviceTime: Date.now(), protocolVersion: 3 }); await peer.receive(); // manifest await peer.send({ type: "SYNC_MANIFEST", entries: {}, tombstones: {} }); await peer.receive(); // request await peer.send({ type: "SYNC_REQUEST", entryIds: [] }); await peer.receive(); // unconditional entries await peer.receive(); // tombstones await peer.send({ type: "SYNC_TOMBSTONES", tombstones: {} }); await peer.receive(); // done // No SYNC_DONE echo — Android just hangs up. await peer.closeInitiatorReceive(); })(); const outcome = await initPromise; expect(outcome.status).toBe("completed"); }); });