From 84bf0f7fd7cde18fa7d9449dfc5ca023f1f233b2 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 9 Sep 2026 11:01:36 +0100 Subject: [PATCH] fix(mesh-store): replay pending deliveries to an agent returning from downtime Delivery queues were per-store in-memory maps excluded from serialised state, so a restarted bridge found downtime messages in synced history but was never woken for them: no push, no drain, until something new arrived. Every peer already replicates delivery events in its own queues, so the queues now travel in state snapshots, and applying a snapshot fires onDelivery for events targeting the receiving peer's own agent: a returning process replays exactly what accumulated while it was down. Replay is deduplicated two ways: room messages and DMs this agent has already read are skipped (read receipts mutate the event between snapshots, so a structural key alone would miss and re-fire), and a bounded structural key set covers the rest. An event fired locally is also dropped from the local pending queue, so steady-state snapshots stop carrying what a peer has already consumed; peers that never fired the event keep their copies, which is what a restart replays from. Queues are bounded per target agent (oldest dropped first) and are purged with their agent by the stale cleanup, which previously leaked queue entries for purged agents indefinitely. The wire format's SerialisedState gains the deliveryQueues field, so a mesh must be on a single build, as with the entity revisions. --- README.md | 2 +- package.json | 2 +- src/core/mesh-store.ts | 126 +++++++++++----- src/core/wire-protocol.ts | 7 + src/test/downtime-replay.integration.test.ts | 149 +++++++++++++++++++ src/test/downtime-replay.test.ts | 125 ++++++++++++++++ 6 files changed, 369 insertions(+), 42 deletions(-) create mode 100644 src/test/downtime-replay.integration.test.ts create mode 100644 src/test/downtime-replay.test.ts diff --git a/README.md b/README.md index 8c06621..048643e 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ graph LR B_Bridge -- "channel notification" --> B_LLM ``` -All state is held in memory and synchronised between peers. Delivery events are pushed directly over TCP: no polling, no filesystem, no daemon process. +All state is held in memory and synchronised between peers. Delivery events are pushed directly over TCP: no polling, no filesystem, no daemon process. Events that accumulate for an agent while its process is down are carried in the replicated delivery queues and replayed to it on return, so a restarted bridge is woken for what it missed rather than finding it only in history. ### Coordinator pattern diff --git a/package.json b/package.json index 4955a73..e35d375 100644 --- a/package.json +++ b/package.json @@ -52,7 +52,7 @@ "generate:server-json": "tsx scripts/sync-release-metadata.ts", "publish:mcp-registry": "tsx scripts/publish-mcp-registry.ts", "lint": "eslint .", - "test": "node --test --test-concurrency=1 dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js dist/test/identity-cert.test.js dist/test/broadcast-window.integration.test.js dist/test/identity-restart.integration.test.js dist/test/mesh-e2e.integration.test.js dist/test/ws-broadcast-window.integration.test.js dist/test/tls-transport.integration.test.js dist/test/state-sync-convergence.test.js", + "test": "node --test --test-concurrency=1 dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js dist/test/identity-cert.test.js dist/test/broadcast-window.integration.test.js dist/test/identity-restart.integration.test.js dist/test/mesh-e2e.integration.test.js dist/test/ws-broadcast-window.integration.test.js dist/test/tls-transport.integration.test.js dist/test/state-sync-convergence.test.js dist/test/downtime-replay.test.js dist/test/downtime-replay.integration.test.js", "test:visibility": "node --test dist/test/visibility.integration.test.js", "test:delivery": "node dist/test/delivery-receipt.runner.js", "test:all": "pnpm test && pnpm test:delivery", diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 74403f2..3137a77 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -52,6 +52,13 @@ import type { ListenerInfo, ListenerPolicy } from "./transport.js"; const DEFAULT_COORDINATOR_PORT = 19876; const COORDINATOR_HOST = "127.0.0.1"; +/** + * Bound on pending delivery events held per target agent. Events beyond the + * bound drop oldest-first: a long-offline agent's queue cannot grow without + * limit in memory or in synced snapshots (#28). + */ +const MAX_QUEUED_DELIVERIES_PER_AGENT = 100; + /** * Merge an incoming append-only message history into the local one: add * entries the local list does not have and union read receipts on the ones @@ -125,6 +132,7 @@ export class MeshStore implements CommsStore { rooms: Object.fromEntries(this.rooms), messages: Object.fromEntries(this.messages), dms: Object.fromEntries(this.dms), + deliveryQueues: Object.fromEntries(this.deliveryQueues), }; } @@ -285,12 +293,7 @@ export class MeshStore implements CommsStore { ): Promise { // If we have state and the peer doesn't, send state sync if (this.agents.size > 0) { - const state: SerialisedState = { - agents: Object.fromEntries(this.agents), - rooms: Object.fromEntries(this.rooms), - messages: Object.fromEntries(this.messages), - dms: Object.fromEntries(this.dms), - }; + const state: SerialisedState = this.serialise(); await this.transport.send(handle, { method: "state_sync", state, @@ -361,6 +364,20 @@ export class MeshStore implements CommsStore { } mergeMessageHistories(existing, dmMsgs); } + for (const [agentId, events] of Object.entries(state.deliveryQueues)) { + const seen = new Set( + (this.deliveryQueues.get(agentId) ?? []).map((e) => + JSON.stringify(e), + ), + ); + for (const event of events) { + if (seen.has(JSON.stringify(event))) continue; + this.queueDelivery(agentId, event); + // A returning peer replays its own pending queue: events pushed + // while its process was down fire onDelivery now (#28). + this.fireLocalDelivery(agentId, event); + } + } } } @@ -423,9 +440,7 @@ export class MeshStore implements CommsStore { name: request.name, fingerprint: request.fingerprint, }; - const arr = this.deliveryQueues.get(this.peerId) ?? []; - arr.push(event); - this.deliveryQueues.set(this.peerId, arr); + this.queueDelivery(this.peerId, event); if (this.onDelivery) { void this.onDelivery(this.peerId, event); } @@ -538,6 +553,16 @@ export class MeshStore implements CommsStore { }; } + /** Append to a target agent's delivery queue, bounded oldest-first (#28). */ + private queueDelivery(agentId: string, event: DeliveryEvent): void { + const arr = this.deliveryQueues.get(agentId) ?? []; + arr.push(event); + if (arr.length > MAX_QUEUED_DELIVERIES_PER_AGENT) { + arr.splice(0, arr.length - MAX_QUEUED_DELIVERIES_PER_AGENT); + } + this.deliveryQueues.set(agentId, arr); + } + /** Bump an entity's sync revision; call before broadcasting a local mutation. */ private bump(entity: T): T { entity.version += 1; @@ -614,9 +639,7 @@ export class MeshStore implements CommsStore { break; } case "delivery": { - const arr = this.deliveryQueues.get(patch.agentId) ?? []; - arr.push(patch.event); - this.deliveryQueues.set(patch.agentId, arr); + this.queueDelivery(patch.agentId, patch.event); if (patch.agentId === this.peerId && this.onDelivery) { // Deduplicate against local deliveries const eventKey = JSON.stringify(patch.event); @@ -672,9 +695,7 @@ export class MeshStore implements CommsStore { event: DeliveryEvent, ): Promise { // Local delivery - const arr = this.deliveryQueues.get(agentId) ?? []; - arr.push(event); - this.deliveryQueues.set(agentId, arr); + this.queueDelivery(agentId, event); // Auto-emit delivered status for messages if (event.type === "room_message") { @@ -688,37 +709,61 @@ export class MeshStore implements CommsStore { await this.emitDeliveryStatus(event.message.id, agentId, "delivered"); } - if (agentId === this.peerId && this.onDelivery) { - // Deduplicate: skip if this exact event was already delivered locally. - // The mesh can echo delivery patches through multiple peer paths, - // causing applyPatch to fire onDelivery for the same event. - const eventKey = JSON.stringify(event); - if (this.localDeliveryKeys.has(eventKey)) return; - this.localDeliveryKeys.add(eventKey); - // Prevent unbounded growth — evict oldest when cap reached - if (this.localDeliveryKeys.size > 50) { - const oldest = this.localDeliveryKeys.values().next().value; - if (oldest !== undefined) this.localDeliveryKeys.delete(oldest); - } - void this.onDelivery(agentId, event); - // Auto-mark read — scheduled as a macrotask to yield to the event - // loop (see matching comment in applyPatch for rationale). - const timer = setTimeout(() => { - if (this.isShutDown) return; - if (event.type === "room_message") { - void this.markRead(event.message.id, agentId, event.message.room); - } else if (event.type === "dm") { - void this.markRead(event.message.id, agentId); - } - }, 0); - if (!this.isShutDown) this.pendingMarkReadTimers.push(timer); - } + this.fireLocalDelivery(agentId, event); // Remote delivery const patch: MeshStatePatch = { type: "delivery", agentId, event }; await this.broadcastPatch(patch); } + /** + * Fire onDelivery for an event targeting this peer's own agent, deduped + * against events already delivered in this process. Used both for live + * deliveries and for replays of events that accumulated while this + * process was down (#28): the dedup set is per-process, so a replayed + * event this process never saw fires, and one it already handled does + * not. + */ + private fireLocalDelivery(agentId: string, event: DeliveryEvent): void { + if (agentId !== this.peerId || !this.onDelivery) return; + // A room message or DM this agent has already read was already pushed + // and consumed: read receipts mutate the event between snapshots, so a + // plain structural key would miss and re-fire on the next sync (#28). + if ( + (event.type === "room_message" || event.type === "dm") && + event.message.readBy.includes(agentId) + ) { + return; + } + const eventKey = JSON.stringify(event); + if (this.localDeliveryKeys.has(eventKey)) return; + this.localDeliveryKeys.add(eventKey); + // Prevent unbounded growth — evict oldest when cap reached + if (this.localDeliveryKeys.size > 50) { + const oldest = this.localDeliveryKeys.values().next().value; + if (oldest !== undefined) this.localDeliveryKeys.delete(oldest); + } + void this.onDelivery(agentId, event); + // Delivered to this process, so no longer pending for it. Peers that + // never fired the event keep their copies, which is what a restart + // replays from (#28). Key-based match: replayed events are JSON clones. + const queued = this.deliveryQueues.get(agentId); + if (queued !== undefined) { + const idx = queued.findIndex((e) => JSON.stringify(e) === eventKey); + if (idx !== -1) queued.splice(idx, 1); + } + // Auto-mark read — scheduled as a macrotask to yield to the event loop. + const timer = setTimeout(() => { + if (this.isShutDown) return; + if (event.type === "room_message") { + void this.markRead(event.message.id, agentId, event.message.room); + } else if (event.type === "dm") { + void this.markRead(event.message.id, agentId); + } + }, 0); + if (!this.isShutDown) this.pendingMarkReadTimers.push(timer); + } + private async deliverToRoom( roomId: string, event: DeliveryEvent, @@ -1423,6 +1468,7 @@ export class MeshStore implements CommsStore { this.agents.delete(id); this.peerInfo.delete(id); this.identityCache.delete(id); + this.deliveryQueues.delete(id); } } diff --git a/src/core/wire-protocol.ts b/src/core/wire-protocol.ts index 9d660af..d869b3b 100644 --- a/src/core/wire-protocol.ts +++ b/src/core/wire-protocol.ts @@ -33,6 +33,13 @@ export interface SerialisedState { rooms: Record; messages: Record; dms: Record; + /** + * Pending delivery events per target agent, replicated on every peer. A + * returning peer replays its own queue from the first snapshot it + * receives, so events pushed while its process was down still fire + * onDelivery (#28). + */ + deliveryQueues: Record; } // --------------------------------------------------------------------------- diff --git a/src/test/downtime-replay.integration.test.ts b/src/test/downtime-replay.integration.test.ts new file mode 100644 index 0000000..4675876 --- /dev/null +++ b/src/test/downtime-replay.integration.test.ts @@ -0,0 +1,149 @@ +/** + * Integration test for issue #28: a bridge restarted with a persisted identity must be push-delivered the events that accumulated while its process was down, not just find them in synced history. + * + * Peer B goes away; A sends a room message while B is down; B restarts in the same identity slot and must fire onDelivery for the missed message. + */ + +import * as assert from "node:assert/strict"; +import * as fs from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { MeshStore } from "../core/mesh-store.js"; +import { TlsTransport } from "../core/tls-transport.js"; +import { generateIdentity } from "../core/identity.js"; +import { + loadOrCreateIdentity, + releaseIdentityLock, + type IdentitySlot, +} from "../core/identity-store.js"; +import type { DeliveryEvent } from "../core/types.js"; +import type { PeerIdentity } from "../core/identity.js"; + +const TEST_PORT = 19896; +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +interface Peer { + store: MeshStore; + deliveries: DeliveryEvent[]; +} + +/** A peer wired like a real bridge: TLS transport, fingerprint peer ID. */ +function makePeer(identity: PeerIdentity): Peer { + const store = new MeshStore(TEST_PORT); + store.peerId = identity.fingerprint; + store.setTransport(new TlsTransport(store.events, identity)); + const deliveries: DeliveryEvent[] = []; + return { store, deliveries }; +} + +/** Poll until the predicate holds, or fail with the message. */ +async function waitFor( + what: string, + check: () => Promise, +): Promise { + for (let i = 0; i < 20; i++) { + if (await check()) return; + await sleep(100); + } + assert.ok(false, `timed out waiting for ${what}`); +} + +async function main(): Promise { + const dir = fs.mkdtempSync(path.join(tmpdir(), "agent-comms-downtime-")); + const slot: IdentitySlot = { harness: "pi", cwd: "/tmp/project", dir }; + + // A is a normal ephemeral bridge that stays up throughout. + const a = makePeer(generateIdentity()); + await a.store.init(); + await a.store.registerAgent({ + name: "peer-a", + harness: "claude-code", + cwd: "/tmp/a", + pid: process.pid, + visibility: "visible", + tags: [], + }); + await a.store.createRoom({ + name: "downtime", + type: "public", + owner: a.store.peerId, + description: "issue 28 acceptance", + }); + await sleep(200); + + // B joins with a persisted identity and becomes a room member. + const identityB = loadOrCreateIdentity(slot); + const b1 = makePeer(identityB); + await b1.store.init(); + // Registration must wait for the TLS data connections to establish (#23). + await sleep(300); + await b1.store.registerAgent({ + name: "peer-b", + harness: "pi", + cwd: "/tmp/project", + pid: process.pid, + visibility: "visible", + tags: [], + }); + await waitFor("the room to reach the joiner", async () => { + const rooms = await b1.store.listRooms(b1.store.peerId); + return rooms.some((room) => room.id === "downtime"); + }); + await b1.store.joinRoom("downtime", b1.store.peerId); + await waitFor("membership to reach the sender", async () => { + const room = await a.store.getRoom("downtime"); + return room?.members.includes(b1.store.peerId) === true; + }); + + // B goes down. + await b1.store.shutdown(); + await sleep(200); + + // A sends a room message while B is down. + await a.store.sendRoomMessage( + "downtime", + a.store.peerId, + "while you were down", + ); + await sleep(200); + + // B restarts in the same slot: same identity, same agent ID. + const identityB2 = loadOrCreateIdentity(slot); + assert.equal(identityB2.fingerprint, identityB.fingerprint); + const b2 = makePeer(identityB2); + b2.store.onDelivery = (_id, ev) => { + b2.deliveries.push(ev); + }; + await b2.store.init(); + await b2.store.registerAgent({ + name: "peer-b", + harness: "pi", + cwd: "/tmp/project", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // The message sent during downtime must push to the restarted bridge. + await waitFor( + "the downtime message to push to the restarted bridge", + async () => + b2.deliveries.some( + (ev) => + ev.type === "room_message" && + ev.message.content === "while you were down", + ), + ); + + await b2.store.shutdown(); + await a.store.shutdown(); + releaseIdentityLock(slot); + console.log("✓ downtime messages push-deliver on restart"); +} + +main().catch((err: unknown) => { + console.error("Test failed:", err); + process.exitCode = 1; + // The sequence above keeps mesh handles open when it fails partway; exit explicitly so a failure cannot hang the runner. + process.exit(1); +}); diff --git a/src/test/downtime-replay.test.ts b/src/test/downtime-replay.test.ts new file mode 100644 index 0000000..e29521f --- /dev/null +++ b/src/test/downtime-replay.test.ts @@ -0,0 +1,125 @@ +/** + * Unit tests for downtime delivery replay (#28): events that accumulated in peers' delivery queues while a target's process was down fire onDelivery when the target applies its first snapshot, without transport involvement. + */ + +import * as assert from "node:assert/strict"; +import { test } from "node:test"; +import { MeshStore } from "../core/mesh-store.js"; +import type { SerialisedState } from "../core/wire-protocol.js"; +import type { DeliveryEvent } from "../core/types.js"; + +/** A wire-accurate snapshot: production always applies parsed (cloned) state. */ +function snapshotOf(store: MeshStore): SerialisedState { + return structuredClone(store.serialise()); +} + +/** A local-only store: no transport start, so no ports and no flake. */ +function makeStore(): MeshStore { + return new MeshStore(); +} + +void test("events queued while the target was down replay on its first snapshot", async () => { + const sender = makeStore(); + const author = await sender.registerAgent({ + name: "author", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + // The target exists on the mesh (known to the sender) but its process is "down": modelled by only ever syncing snapshots into a future store. + const target = makeStore(); + const targetAgent = await target.registerAgent({ + name: "target", + harness: "claude-code", + cwd: "/tmp/t", + pid: process.pid, + visibility: "visible", + tags: [], + }); + // Make the sender aware of the target, then take the target's store away. + sender.applyStateSync(target.serialise()); + await sender.createRoom({ + name: "room", + type: "public", + owner: author.id, + description: "x", + }); + await sender.joinRoom("room", targetAgent.id); + + // While the target is down, a room message is sent to it. + await sender.sendRoomMessage("room", author.id, "while you were away"); + const pending = snapshotOf(sender).deliveryQueues[targetAgent.id]; + assert.ok(pending !== undefined && pending.length > 0); + + // The target returns (fresh process, same agent id) and receives the sender's snapshot: the pending event must fire onDelivery. + const returned = makeStore(); + const deliveries: DeliveryEvent[] = []; + returned.onDelivery = (_id, ev) => { + deliveries.push(ev); + }; + returned.peerId = targetAgent.id; + returned.applyStateSync(snapshotOf(sender)); + assert.equal( + deliveries.some( + (ev) => + ev.type === "room_message" && + ev.message.content === "while you were away", + ), + true, + ); + + // A second snapshot of the same state does not duplicate any push (the + // join notification replays too, exactly once). + returned.applyStateSync(snapshotOf(sender)); + assert.equal( + deliveries.filter( + (ev) => + ev.type === "room_message" && + ev.message.content === "while you were away", + ).length, + 1, + ); + assert.equal(deliveries.filter((ev) => ev.type === "room_members").length, 1); +}); + +void test("a queue is bounded oldest-first so downtime cannot grow it without limit", async () => { + const sender = makeStore(); + const author = await sender.registerAgent({ + name: "author", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const target = makeStore(); + const targetAgent = await target.registerAgent({ + name: "target", + harness: "claude-code", + cwd: "/tmp/t", + pid: process.pid, + visibility: "visible", + tags: [], + }); + sender.applyStateSync(target.serialise()); + await sender.createRoom({ + name: "room", + type: "public", + owner: author.id, + description: "x", + }); + await sender.joinRoom("room", targetAgent.id); + + for (let i = 0; i < 120; i++) { + await sender.sendRoomMessage("room", author.id, `msg-${String(i)}`); + } + const queued = snapshotOf(sender).deliveryQueues[targetAgent.id] ?? []; + assert.equal(queued.length, 100); + assert.equal( + queued[0]?.type === "room_message" && + queued[0].message.content === "msg-20", + true, + ); +});