From 086ac6052e59ace0a3e53b62ea9995955647251f Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 9 Sep 2026 10:50:53 +0100 Subject: [PATCH] fix(mesh-store): converge state sync by per-entity version MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The state_sync receive path merged add-only, so an entity that changed while a peer was fully disconnected could never be corrected by a sync: the returning peer kept its stale copy, and a peer receiving that stale snapshot kept whatever it already had. Room member lists had the mirror problem — the union merge could add members but never remove a leaver. Give agents and rooms a monotonic revision the mutating store bumps and embeds in the record. Merges (both state_sync snapshots and incremental patches) are now tri-state: a strictly higher version replaces the record wholesale (which is what heals renames, edits, and leaves), an equal version unions memberships and subscriptions (so concurrent joins from the same base both survive), and a lower version is rejected (so a stale holder cannot regress a current one). Message and DM histories merge by message id, adding unseen entries and unioning read receipts. Owner-authority merging was rejected during design: a rejoining peer can carry an old copy of a third peer's entity and would clobber it. The merge itself is extracted to a public applyStateSync seam so the convergence contract is unit-testable without transports; the tests cover both directions (stale holder heals, stale snapshot rejected), room membership healing including leaves, and history/read-receipt merging. The wire format carries the version inside the entity records, so a mesh must be on a single build; the serial number fix and this change ship together. --- package.json | 2 +- src/core/mesh-store.ts | 157 ++++++++++++++++++----- src/core/store.ts | 7 ++ src/core/types.ts | 4 + src/test/state-sync-convergence.test.ts | 158 ++++++++++++++++++++++++ 5 files changed, 298 insertions(+), 30 deletions(-) create mode 100644 src/test/state-sync-convergence.test.ts diff --git a/package.json b/package.json index 36d3841..dc30a41 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", + "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: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 784ab5f..74403f2 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -52,6 +52,29 @@ import type { ListenerInfo, ListenerPolicy } from "./transport.js"; const DEFAULT_COORDINATOR_PORT = 19876; const COORDINATOR_HOST = "127.0.0.1"; +/** + * 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 + * it does. Local ordering is preserved; unseen entries are appended. + */ +function mergeMessageHistories( + local: T[], + incoming: T[], +): void { + const byId = new Map(local.map((m) => [m.id, m])); + for (const msg of incoming) { + const existing = byId.get(msg.id); + if (existing === undefined) { + local.push(msg); + byId.set(msg.id, msg); + continue; + } + for (const reader of msg.readBy) { + if (!existing.readBy.includes(reader)) existing.readBy.push(reader); + } + } +} + // --------------------------------------------------------------------------- // MeshStore // --------------------------------------------------------------------------- @@ -275,30 +298,78 @@ export class MeshStore implements CommsStore { } } - private async handleDataMessage( - handle: ConnectionHandle, - msg: MeshMessage, - ): Promise { - if (msg.method === "state_sync") { - // Merge — don't replace — so our own state isn't lost - const incoming = { - agents: new Map(Object.entries(msg.state.agents)), - rooms: new Map(Object.entries(msg.state.rooms)), - messages: new Map(Object.entries(msg.state.messages)), - dms: new Map(Object.entries(msg.state.dms)), - }; + /** + * Merge a peer's state snapshot into the local state. Agents and rooms + * accept the incoming copy when it carries a revision at least as high as + * the local one, so a peer holding stale entities converges when it + * receives a fresher snapshot, while its own stale copies are rejected by + * peers that stayed current (#27). Message and DM histories are + * append-only: add unseen entries and union read receipts. + */ + applyStateSync(state: SerialisedState): void { + const incoming = { + agents: new Map(Object.entries(state.agents)), + rooms: new Map(Object.entries(state.rooms)), + messages: new Map(Object.entries(state.messages)), + dms: new Map(Object.entries(state.dms)), + }; + { for (const [id, agent] of incoming.agents) { - if (!this.agents.has(id)) this.agents.set(id, agent); + const existingVersion = this.agents.get(id)?.version; + if (existingVersion !== undefined && agent.version < existingVersion) { + continue; + } + if (existingVersion === agent.version) { + const existing = this.agents.get(id); + if (existing) { + for (const r of existing.subscribedRooms) { + if (!agent.subscribedRooms.includes(r)) + agent.subscribedRooms.push(r); + } + } + } + this.agents.set(id, agent); } for (const [id, room] of incoming.rooms) { - if (!this.rooms.has(id)) this.rooms.set(id, room); + const existingVersion = this.rooms.get(id)?.version; + if (existingVersion !== undefined && room.version < existingVersion) { + continue; + } + if (existingVersion === room.version) { + const existing = this.rooms.get(id); + if (existing) { + for (const m of existing.members) { + if (!room.members.includes(m)) room.members.push(m); + } + } + } + this.rooms.set(id, room); } for (const [id, msgs] of incoming.messages) { - if (!this.messages.has(id)) this.messages.set(id, msgs); + const existing = this.messages.get(id); + if (existing === undefined) { + this.messages.set(id, msgs); + continue; + } + mergeMessageHistories(existing, msgs); } for (const [id, dmMsgs] of incoming.dms) { - if (!this.dms.has(id)) this.dms.set(id, dmMsgs); + const existing = this.dms.get(id); + if (existing === undefined) { + this.dms.set(id, dmMsgs); + continue; + } + mergeMessageHistories(existing, dmMsgs); } + } + } + + private async handleDataMessage( + handle: ConnectionHandle, + msg: MeshMessage, + ): Promise { + if (msg.method === "state_sync") { + this.applyStateSync(msg.state); } else if (msg.method === "state_update") { await this.applyPatch(msg.patch); } @@ -467,6 +538,12 @@ export class MeshStore implements CommsStore { }; } + /** Bump an entity's sync revision; call before broadcasting a local mutation. */ + private bump(entity: T): T { + entity.version += 1; + return entity; + } + // ----------------------------------------------------------------------- // State patch application // ----------------------------------------------------------------------- @@ -474,18 +551,24 @@ export class MeshStore implements CommsStore { private async applyPatch(patch: MeshStatePatch): Promise { switch (patch.type) { case "agent_upsert": { - // Merge subscribedRooms to avoid losing local room memberships. const existingAgent = this.agents.get(patch.agent.id); - if (existingAgent) { - const merged = patch.agent; + if ( + existingAgent !== undefined && + patch.agent.version < existingAgent.version + ) { + // Stale copy from a peer that missed updates (#27). + break; + } + const merged = patch.agent; + if (existingAgent?.version === patch.agent.version) { + // Concurrent mutations from the same base: keep subscriptions + // gained locally. A strictly higher version replaces the record. for (const r of existingAgent.subscribedRooms) { if (!merged.subscribedRooms.includes(r)) merged.subscribedRooms.push(r); } - this.agents.set(merged.id, merged); - } else { - this.agents.set(patch.agent.id, patch.agent); } + this.agents.set(merged.id, merged); break; } case "agent_offline": { @@ -497,18 +580,22 @@ export class MeshStore implements CommsStore { break; } case "room_upsert": { - // Merge members rather than overwriting — last-write-wins can lose - // members added locally when a remote patch arrives with a stale list. const existing = this.rooms.get(patch.room.id); - if (existing) { - const merged = patch.room; + if (existing !== undefined && patch.room.version < existing.version) { + // Stale copy from a peer that missed updates (#27). + break; + } + const merged = patch.room; + if (existing?.version === patch.room.version) { + // Concurrent mutations from the same base: union members so + // simultaneous joins both survive. A strictly higher version + // replaces the record wholesale, which is what heals a leave or + // edit a lagging peer missed. for (const m of existing.members) { if (!merged.members.includes(m)) merged.members.push(m); } - this.rooms.set(merged.id, merged); - } else { - this.rooms.set(patch.room.id, patch.room); } + this.rooms.set(merged.id, merged); break; } case "room_delete": @@ -819,6 +906,7 @@ export class MeshStore implements CommsStore { const id = this.peerId; const agent: AgentIdentity = { id, + version: 1, name: opts.name, harness: opts.harness, cwd: opts.cwd, @@ -858,6 +946,7 @@ export class MeshStore implements CommsStore { const oldStatus = agent.status; const oldName = agent.name; Object.assign(agent, patch); + this.bump(agent); this.agents.set(id, agent); await this.broadcastPatch({ type: "agent_upsert", agent }); @@ -891,6 +980,7 @@ export class MeshStore implements CommsStore { // Other stores learn about it via the agent_offline mesh patch. const isOwner = id === this.peerId; agent.status = "offline"; + this.bump(agent); this.agents.set(id, agent); if (isOwner) { @@ -917,6 +1007,7 @@ export class MeshStore implements CommsStore { const room: Room = { id, + version: 1, name: opts.name, type: opts.type, owner: opts.owner, @@ -968,11 +1059,13 @@ export class MeshStore implements CommsStore { if (!room.members.includes(agentId)) room.members.push(agentId); } + this.bump(room); this.rooms.set(roomId, room); const agent = this.agents.get(agentId); if (agent && !agent.subscribedRooms.includes(roomId)) { agent.subscribedRooms.push(roomId); + this.bump(agent); this.agents.set(agentId, agent); await this.broadcastPatch({ type: "agent_upsert", agent }); } @@ -1023,6 +1116,7 @@ export class MeshStore implements CommsStore { throw new CommsError(`Room ${roomId} not found`, "ROOM_NOT_FOUND"); room.members = room.members.filter((id) => id !== agentId); + this.bump(room); this.rooms.set(roomId, room); const agent = this.agents.get(agentId); @@ -1065,6 +1159,7 @@ export class MeshStore implements CommsStore { if (!room.invited.includes(targetId) && !room.members.includes(targetId)) { room.invited.push(targetId); } + this.bump(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); @@ -1095,6 +1190,7 @@ export class MeshStore implements CommsStore { ); room.invited = room.invited.filter((id) => id !== agentId); + this.bump(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); @@ -1121,6 +1217,7 @@ export class MeshStore implements CommsStore { room.members = room.members.filter((id) => id !== targetId); room.invited = room.invited.filter((id) => id !== targetId); + this.bump(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); } @@ -1476,6 +1573,7 @@ export class MeshStore implements CommsStore { if (!room.members.includes(remoteId)) { room.members.push(remoteId); + this.bump(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); } @@ -1501,6 +1599,7 @@ export class MeshStore implements CommsStore { const remoteId = `fed:${agentId}`; room.members = room.members.filter((m) => m !== remoteId); + this.bump(room); this.rooms.set(roomId, room); await this.broadcastPatch({ type: "room_upsert", room }); diff --git a/src/core/store.ts b/src/core/store.ts index 9d75cd5..f75ffa2 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -156,6 +156,7 @@ export class FileStore implements CommsStore { const id = nanoid(8); const agent: AgentIdentity = { id, + version: 1, name: opts.name, harness: opts.harness, cwd: opts.cwd, @@ -193,6 +194,7 @@ export class FileStore implements CommsStore { throw new CommsError(`Agent ${id} not found`, "AGENT_NOT_FOUND"); Object.assign(agent, patch); + agent.version += 1; await this.writeJsonFile(this.agentPath(id), agent); return agent; } @@ -220,6 +222,7 @@ export class FileStore implements CommsStore { const agent = await this.getAgent(id); if (agent) { agent.status = "offline"; + agent.version += 1; await this.writeJsonFile(this.agentPath(id), agent); } } @@ -241,6 +244,7 @@ export class FileStore implements CommsStore { const room: Room = { id, + version: 1, name: opts.name, type: opts.type, owner: opts.owner, @@ -306,6 +310,7 @@ export class FileStore implements CommsStore { } } + room.version += 1; await this.writeJsonFile(this.roomPath(roomId), room); const agent = await this.getAgent(agentId); @@ -329,6 +334,7 @@ export class FileStore implements CommsStore { throw new CommsError(`Room ${roomId} not found`, "ROOM_NOT_FOUND"); room.members = room.members.filter((id) => id !== agentId); + room.version += 1; await this.writeJsonFile(this.roomPath(roomId), room); const agent = await this.getAgent(agentId); @@ -418,6 +424,7 @@ export class FileStore implements CommsStore { room.members = room.members.filter((id) => id !== targetId); room.invited = room.invited.filter((id) => id !== targetId); + room.version += 1; await this.writeJsonFile(this.roomPath(roomId), room); } diff --git a/src/core/types.ts b/src/core/types.ts index 6ee580d..bfddf28 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -71,6 +71,8 @@ export type StreamingBehavior = z.infer; export const AgentIdentitySchema = defineSchema( z.object({ id: z.string(), + /** Monotonic revision, bumped by the mutating store; sync merges take the higher value. */ + version: z.number(), name: z.string(), harness: z.string(), cwd: z.string(), @@ -91,6 +93,8 @@ export type AgentIdentity = z.infer; export const RoomSchema = defineSchema( z.object({ id: z.string(), + /** Monotonic revision, bumped by the mutating store; sync merges take the higher value. */ + version: z.number(), name: z.string(), type: RoomType, owner: z.string(), diff --git a/src/test/state-sync-convergence.test.ts b/src/test/state-sync-convergence.test.ts new file mode 100644 index 0000000..fdfccee --- /dev/null +++ b/src/test/state-sync-convergence.test.ts @@ -0,0 +1,158 @@ +/** + * Unit tests for state_sync convergence (#27): entities that changed while a peer held a stale copy must converge on the fresher revision, in both directions — a stale holder heals when it receives a fresher snapshot, and a current holder rejects a stale snapshot instead of regressing. + * + * Also covers the append-only history merge: unseen messages are added and read receipts are unioned. These tests exercise the merge seam directly (no transport): the stores never connect, snapshots are exchanged through serialise() / applyStateSync(). + */ + +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"; + +/** A local-only store: no transport start, so no ports and no flake. */ +function makeStore(): MeshStore { + return new MeshStore(); +} + +function snapshotOf(store: MeshStore): SerialisedState { + return structuredClone(store.serialise()); +} + +void test("a stale holder converges when a fresher snapshot arrives", async () => { + const a = makeStore(); + const b = makeStore(); + const agent = await a.registerAgent({ + name: "old-name", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // B holds the pre-rename copy. + b.applyStateSync(snapshotOf(a)); + assert.equal((await b.getAgent(agent.id))?.name, "old-name"); + + // A renames while B is away; B heals on A's next snapshot. + await a.updateAgent(agent.id, { name: "new-name" }); + b.applyStateSync(snapshotOf(a)); + assert.equal((await b.getAgent(agent.id))?.name, "new-name"); +}); + +void test("a current holder rejects a stale snapshot instead of regressing", async () => { + const a = makeStore(); + const b = makeStore(); + const agent = await a.registerAgent({ + name: "old-name", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // B is current at the post-rename revision. + await a.updateAgent(agent.id, { name: "new-name" }); + b.applyStateSync(snapshotOf(a)); + assert.equal((await b.getAgent(agent.id))?.name, "new-name"); + + // A returning peer that still holds the pre-rename snapshot cannot regress B, where the old add-only merge kept whatever arrived first. + const stale = snapshotOf(a); + const staleAgent = stale.agents[agent.id]; + if (staleAgent === undefined) throw new Error("agent missing from snapshot"); + staleAgent.name = "old-name"; + staleAgent.version -= 1; + b.applyStateSync(stale); + assert.equal((await b.getAgent(agent.id))?.name, "new-name"); +}); + +void test("room membership changes converge and stale member lists are rejected", async () => { + const a = makeStore(); + const b = makeStore(); + const owner = await a.registerAgent({ + name: "owner", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const joiner = await b.registerAgent({ + name: "joiner", + harness: "claude-code", + cwd: "/tmp/j", + pid: process.pid, + visibility: "visible", + tags: [], + }); + await a.createRoom({ + name: "room", + type: "public", + owner: owner.id, + description: "x", + }); + + // A holds the room before the join; the joiner lives on B, so the join + // mutates B's copy and syncs back (one MeshStore is one peer identity, so + // a second agent cannot be registered on A). + a.applyStateSync(snapshotOf(b)); + b.applyStateSync(snapshotOf(a)); + assert.equal((await a.getRoom("room"))?.members.includes(joiner.id), false); + + await b.joinRoom("room", joiner.id); + a.applyStateSync(snapshotOf(b)); + assert.equal((await a.getRoom("room"))?.members.includes(joiner.id), true); + + // The joiner leaves; A (now current) must drop them — the union-only merge + // could never remove a leaver. + await b.leaveRoom("room", joiner.id); + a.applyStateSync(snapshotOf(b)); + assert.equal((await a.getRoom("room"))?.members.includes(joiner.id), false); + + // A stale member list that still contains them is rejected. + const stale = snapshotOf(b); + const staleRoom = stale.rooms.room; + if (staleRoom === undefined) throw new Error("room missing from snapshot"); + staleRoom.members.push(joiner.id); + staleRoom.version -= 1; + a.applyStateSync(stale); + assert.equal((await a.getRoom("room"))?.members.includes(joiner.id), false); +}); + +void test("history sync adds unseen messages and unions read receipts", async () => { + const a = makeStore(); + const b = makeStore(); + const agent = await a.registerAgent({ + name: "sender", + harness: "pi", + cwd: "/tmp/p", + pid: process.pid, + visibility: "visible", + tags: [], + }); + await a.createRoom({ + name: "room", + type: "public", + owner: agent.id, + description: "x", + }); + await a.sendRoomMessage("room", agent.id, "one"); + b.applyStateSync(snapshotOf(a)); + + // While B is away, a second message arrives and the first gains a reader elsewhere on the mesh; B's next sync takes both. + await a.sendRoomMessage("room", agent.id, "two"); + const fresh = snapshotOf(a); + const history = fresh.messages.room; + if (history?.[0] === undefined) + throw new Error("messages missing from snapshot"); + history[0].readBy.push("reader-elsewhere"); + b.applyStateSync(fresh); + + const merged = await b.readRoomMessages("room"); + assert.deepEqual( + merged.map((m) => m.content), + ["one", "two"], + ); + assert.equal(merged[0]?.readBy.includes("reader-elsewhere"), true); +});