From f0e7a94a5a93e8a901ccfa64231d1d36110160ed Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:19:15 +0100 Subject: [PATCH 1/5] feat(core): add PeerLifecycle.sendCoordinatorHandover Graceful coordinator shutdown never sent a become_coordinator message, so only the crash-race path (each surviving peer independently racing to rebind the coordinator port) ever produced a new coordinator, despite the README documenting handoff to the longest-running peer as live behaviour. sendCoordinatorHandover picks the remaining peer with the earliest recorded startedAt as successor and sends it become_coordinator carrying every other remaining peer, matching the peerList shape handleBecomeCoordinator already expects. A no-op when this side isn't the coordinator or no other peers remain. Not yet wired into MeshStore.shutdown(). --- src/core/peer-lifecycle.ts | 24 ++++++++++ src/test/peer-lifecycle.test.ts | 80 ++++++++++++++++++++++++++++++++- 2 files changed, 102 insertions(+), 2 deletions(-) diff --git a/src/core/peer-lifecycle.ts b/src/core/peer-lifecycle.ts index 6463d4d0..423c2ae8 100644 --- a/src/core/peer-lifecycle.ts +++ b/src/core/peer-lifecycle.ts @@ -126,4 +126,28 @@ export class PeerLifecycle { handlePeerDisconnected(handle: Readonly): void { this.deps.peerInfo.delete(handle.id); } + + /** Graceful coordinator handover (agent-comms#170): called by MeshStore.shutdown() while the transport is still up. A no-op unless this side currently holds the coordinator role and at least one other peer remains connected -- neither condition is this class's own business to log or report on, since a solo coordinator shutting down or a non-coordinator peer shutting down are both entirely ordinary. When both hold, picks the longest-running remaining peer (the one with the earliest recorded PeerInfo.startedAt, matching the README's documented policy) as the successor and sends it become_coordinator carrying every other remaining peer -- exactly the peerList shape handleBecomeCoordinator already expects (it dials each entry itself; the successor doesn't need to be told about itself). The crash-race path (each surviving peer independently racing to rebind the coordinator port) is a separate mechanism and untouched by this method. */ + async sendCoordinatorHandover(): Promise { + const transport = this.deps.requireTransport(); + if (!transport.isCoordinator) return; + + const selfId = this.deps.getPeerId(); + const remainingPeers = [...this.deps.peerInfo.values()].filter( + (peer) => peer.id !== selfId, + ); + if (remainingPeers.length === 0) return; + + const bySuccession = [...remainingPeers].sort( + (a, b) => Date.parse(a.startedAt) - Date.parse(b.startedAt), + ); + const [successor, ...handoffList] = bySuccession; + // remainingPeers.length > 0 (checked above) guarantees bySuccession has at least one entry -- this narrows the type for TypeScript rather than handling a real runtime case. + if (successor === undefined) return; + + await transport.send( + { id: successor.id }, + { method: "become_coordinator", peerList: handoffList }, + ); + } } diff --git a/src/test/peer-lifecycle.test.ts b/src/test/peer-lifecycle.test.ts index 7c7c2e72..f418f5b7 100644 --- a/src/test/peer-lifecycle.test.ts +++ b/src/test/peer-lifecycle.test.ts @@ -12,8 +12,12 @@ import type { PeerInfo, SerialisedState } from "../core/wire-protocol.js"; const OWNER_ID = "owner-device"; const COORDINATOR_PORT = 19876; -function peerInfo(id: string, port = 1): PeerInfo { - return { id, port, startedAt: "2026-01-01T00:00:00.000Z" }; +function peerInfo( + id: string, + port = 1, + startedAt = "2026-01-01T00:00:00.000Z", +): PeerInfo { + return { id, port, startedAt }; } function emptyState(): SerialisedState { @@ -24,6 +28,7 @@ interface Harness { deps: PeerLifecycleDeps; lifecycle: PeerLifecycle; transport: { + isCoordinator: boolean; connectToPeer: ReturnType; send: ReturnType; broadcast: ReturnType; @@ -39,6 +44,7 @@ interface Harness { function makeHarness(): Harness { const transport = { + isCoordinator: false, connectToPeer: vi.fn().mockResolvedValue(undefined), send: vi.fn().mockResolvedValue(undefined), broadcast: vi.fn().mockResolvedValue(undefined), @@ -231,6 +237,76 @@ describe("PeerLifecycle — handleBecomeCoordinator", () => { }); }); +describe("PeerLifecycle — sendCoordinatorHandover", () => { + it("does nothing when this side is not the coordinator", async () => { + const h = makeHarness(); + h.transport.isCoordinator = false; + h.deps.peerInfo.set(OWNER_ID, peerInfo(OWNER_ID)); + h.deps.peerInfo.set("other", peerInfo("other")); + + await h.lifecycle.sendCoordinatorHandover(); + + expect(h.transport.send).not.toHaveBeenCalled(); + expect(h.transport.broadcast).not.toHaveBeenCalled(); + }); + + it("does nothing when this side is the coordinator but no other peers remain", async () => { + const h = makeHarness(); + h.transport.isCoordinator = true; + h.deps.peerInfo.set(OWNER_ID, peerInfo(OWNER_ID)); + + await h.lifecycle.sendCoordinatorHandover(); + + expect(h.transport.send).not.toHaveBeenCalled(); + }); + + it("sends become_coordinator to the longest-running remaining peer, carrying every other remaining peer", async () => { + const h = makeHarness(); + h.transport.isCoordinator = true; + h.deps.peerInfo.set( + OWNER_ID, + peerInfo(OWNER_ID, 1, "2026-01-03T00:00:00.000Z"), + ); + const oldest = peerInfo("oldest", 1, "2026-01-01T00:00:00.000Z"); + const middle = peerInfo("middle", 1, "2026-01-02T00:00:00.000Z"); + const newest = peerInfo("newest", 1, "2026-01-04T00:00:00.000Z"); + h.deps.peerInfo.set(oldest.id, oldest); + h.deps.peerInfo.set(middle.id, middle); + h.deps.peerInfo.set(newest.id, newest); + + await h.lifecycle.sendCoordinatorHandover(); + + expect(h.transport.send).toHaveBeenCalledTimes(1); + expect(h.transport.send).toHaveBeenCalledWith( + { id: "oldest" }, + { + method: "become_coordinator", + peerList: expect.arrayContaining([middle, newest]) as PeerInfo[], + }, + ); + const [, message] = h.transport.send.mock.calls[0] as [ + unknown, + { peerList: PeerInfo[] }, + ]; + expect(message.peerList).toHaveLength(2); + }); + + it("picks the sole remaining peer as successor and sends an empty handoff list", async () => { + const h = makeHarness(); + h.transport.isCoordinator = true; + h.deps.peerInfo.set(OWNER_ID, peerInfo(OWNER_ID)); + const onlyPeer = peerInfo("only-peer"); + h.deps.peerInfo.set(onlyPeer.id, onlyPeer); + + await h.lifecycle.sendCoordinatorHandover(); + + expect(h.transport.send).toHaveBeenCalledWith( + { id: "only-peer" }, + { method: "become_coordinator", peerList: [] }, + ); + }); +}); + describe("PeerLifecycle — handlePeerDisconnected", () => { it("removes the disconnected peer's entry from peerInfo", () => { const h = makeHarness(); From de3c659e22315b5d01b0bd58649f6fd7ef4cd74b Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:27:15 +0100 Subject: [PATCH 2/5] feat(core): wire graceful coordinator handover into MeshStore.shutdown MeshStore.shutdown() stopped the stale-agent checker, dropped the hub gateway connection, and closed the transport, but never sent the become_coordinator handoff the README documents -- only the crash-race path (each surviving peer racing to rebind the coordinator port) ever actually produced a new coordinator. shutdown() now calls PeerLifecycle.sendCoordinatorHandover() after losing the gateway role but before the transport closes, since the handoff message rides the peer sessions shutdown() is about to tear down. The call is unconditional: sendCoordinatorHandover() is already a no-op when this side isn't the coordinator. --- src/core/mesh-store.ts | 2 + src/test/mesh-store-orchestration.test.ts | 64 ++++++++++++++++++++++- 2 files changed, 64 insertions(+), 2 deletions(-) diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index ff09532e..00a8c372 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -783,6 +783,8 @@ export class MeshStore implements CommsStore { this.staleAgentChecker.stop(); await this.coordinatorGateway.onLostCoordinator(); + // Graceful coordinator handover (agent-comms#170) -- a no-op unless this side currently holds the coordinator role, so this runs unconditionally rather than being gated behind an isCoordinator check duplicated here. Must run before the transport shuts down: the handoff message rides the very peer sessions shutdown() is about to close. + await this.peerLifecycle.sendCoordinatorHandover(); await this.onCoordinatorRoleChanged?.(false); await this.requireTransport().shutdown(); } diff --git a/src/test/mesh-store-orchestration.test.ts b/src/test/mesh-store-orchestration.test.ts index 3ad4669f..cfd634c7 100644 --- a/src/test/mesh-store-orchestration.test.ts +++ b/src/test/mesh-store-orchestration.test.ts @@ -14,10 +14,12 @@ const VALID_DEVICE_ID = "a".repeat(DEVICE_ID_HEX_LENGTH); /** Arbitrary, distinctive remote port for a connectToRemote test fixture -- no significance beyond "a real-looking port number". */ const REMOTE_PORT = 4242; -function fakeTransport(): MeshTransport { +function fakeTransport( + overrides: Partial> = {}, +): MeshTransport { return { dataPort: 4000, - isCoordinator: false, + isCoordinator: overrides.isCoordinator ?? false, hasCoordinatorConnection: false, startDataServer: vi.fn().mockResolvedValue(undefined), connectToCoordinator: vi.fn().mockResolvedValue(undefined), @@ -699,6 +701,64 @@ describe("MeshStore — shutdown()", () => { expect(onCoordinatorRoleChanged).toHaveBeenCalledWith(false); }); + it("hands the coordinator role to the longest-running remaining peer before shutting down the transport (agent-comms#170)", async () => { + const coordinatorStore = new MeshStore(); + const coordinatorTransport = fakeTransport({ isCoordinator: true }); + coordinatorStore.setTransport(coordinatorTransport); + const peerInfo = collaborator(coordinatorStore, "peerInfo") as Map< + string, + PeerInfo + >; + peerInfo.set(coordinatorStore.peerId, { + id: coordinatorStore.peerId, + port: 1, + startedAt: "2026-01-03T00:00:00.000Z", + }); + const oldest: PeerInfo = { + id: "oldest-peer", + port: 2, + startedAt: "2026-01-01T00:00:00.000Z", + }; + const newest: PeerInfo = { + id: "newest-peer", + port: 3, + startedAt: "2026-01-02T00:00:00.000Z", + }; + peerInfo.set(oldest.id, oldest); + peerInfo.set(newest.id, newest); + + await coordinatorStore.shutdown(); + + expect(coordinatorTransport.send).toHaveBeenCalledWith( + { id: oldest.id }, + { method: "become_coordinator", peerList: [newest] }, + ); + const sendOrder = (coordinatorTransport.send as ReturnType) + .mock.invocationCallOrder[0]; + const shutdownOrder = ( + coordinatorTransport.shutdown as ReturnType + ).mock.invocationCallOrder[0]; + expect(sendOrder).toBeLessThan(shutdownOrder as number); + }); + + it("sends no coordinator handoff when this side isn't the coordinator", async () => { + const peerInfo = collaborator(store, "peerInfo") as Map; + peerInfo.set(store.peerId, { + id: store.peerId, + port: 1, + startedAt: "2026-01-01T00:00:00.000Z", + }); + peerInfo.set("other-peer", { + id: "other-peer", + port: 2, + startedAt: "2026-01-02T00:00:00.000Z", + }); + + await store.shutdown(); + + expect(transport.send).not.toHaveBeenCalled(); + }); + it("broadcasts agent_offline for its own self agent when one is registered", async () => { const agent = await store.registerAgent({ name: "self", From 79c29fad97ed0857a6037cde5dc3dd06688f53db Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:31:23 +0100 Subject: [PATCH 3/5] refactor(core): avoid a redundant array copy in sendCoordinatorHandover Array.prototype.filter() already returns a fresh array, so sorting it in place needs no further defensive spread. --- src/core/peer-lifecycle.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/core/peer-lifecycle.ts b/src/core/peer-lifecycle.ts index 423c2ae8..3bb2d3d2 100644 --- a/src/core/peer-lifecycle.ts +++ b/src/core/peer-lifecycle.ts @@ -138,11 +138,12 @@ export class PeerLifecycle { ); if (remainingPeers.length === 0) return; - const bySuccession = [...remainingPeers].sort( + // filter() above already returned a fresh array, so sorting it in place is safe. + remainingPeers.sort( (a, b) => Date.parse(a.startedAt) - Date.parse(b.startedAt), ); - const [successor, ...handoffList] = bySuccession; - // remainingPeers.length > 0 (checked above) guarantees bySuccession has at least one entry -- this narrows the type for TypeScript rather than handling a real runtime case. + const [successor, ...handoffList] = remainingPeers; + // remainingPeers.length > 0 (checked above) guarantees at least one entry here -- this narrows the type for TypeScript rather than handling a real runtime case. if (successor === undefined) return; await transport.send( From 2aede9946d37946e50eff63bef445b4508e05a9d Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:53:49 +0100 Subject: [PATCH 4/5] fix(core): retry becomeCoordinator's bind after EADDRINUSE Sending become_coordinator during a graceful shutdown (agent-comms#170) races the outgoing coordinator's own listening-socket close against the successor's rebind of the identical port: onBecomeCoordinator's dispatch in mesh-store.ts is fire-and-forget, so a bind failure there becomes an unhandled promise rejection that crashes the process, and the real mesh-e2e integration test reproduced this deterministically on every run once the sender actually started firing. becomeCoordinator now retries its bind a bounded number of times with a short delay whenever wireTransport.listen() rejects with EADDRINUSE, extracted as retryOnAddrInUse in its own bind-retry.ts (kept wire-mesh-transport.ts under the repo's max-lines cap). Retrying a plain server bind carries none of the risk the existing comment on MeshStore.init() documents against retrying a TLS connectToCoordinator (a Node TLS session-cache bug) -- this is ordinary bind-after-close backoff, and benefits the pre-existing crash-race becomeCoordinator call in init() equally, not just the new graceful path. --- src/core/bind-retry.ts | 27 ++++++++++++++++ src/core/wire-mesh-transport.ts | 25 ++++++++------- src/test/bind-retry.test.ts | 57 +++++++++++++++++++++++++++++++++ 3 files changed, 98 insertions(+), 11 deletions(-) create mode 100644 src/core/bind-retry.ts create mode 100644 src/test/bind-retry.test.ts diff --git a/src/core/bind-retry.ts b/src/core/bind-retry.ts new file mode 100644 index 00000000..9ba9182b --- /dev/null +++ b/src/core/bind-retry.ts @@ -0,0 +1,27 @@ +/** + * retryOnAddrInUse — bind-retry backoff for WireMeshTransport.becomeCoordinator (agent-comms#170). Split out of wire-mesh-transport.ts to keep that file under the repo's max-lines cap, matching the same split-out-a-collaborator convention peer-lifecycle.ts and friends already use. + */ + +/** Number of extra attempts becomeCoordinator makes after an initial EADDRINUSE before giving up. Covers PeerLifecycle's own graceful coordinator handover: the outgoing coordinator's listening socket may not have finished releasing the coordinator port yet when the successor's own become_coordinator handler tries to rebind it moments later. Retrying a plain server bind carries none of the risk documented elsewhere against retrying a TLS connectToCoordinator (a Node TLS session-cache bug that can freeze the event loop) -- this is ordinary bind-after-close backoff, the same class of transient conflict any two processes racing to acquire the same port would need to tolerate. */ +export const BECOME_COORDINATOR_BIND_RETRIES = 5; +/** Delay between each becomeCoordinator bind retry -- five retries at this interval gives ~150ms of tolerance, comparable to the crash-race path's own documented "~100ms recovery". */ +export const BECOME_COORDINATOR_BIND_RETRY_DELAY_MS = 30; + +/** Retries `attempt` while it keeps rejecting with an EADDRINUSE-shaped error, up to `retries` further attempts, waiting `delayMs` between each. Any other error, or exhausting the retry budget, rethrows immediately -- never silently swallowed. A pure, transport-agnostic helper (no socket knowledge of its own) so becomeCoordinator's bind-retry race is fast and deterministic to test directly. */ +export async function retryOnAddrInUse( + attempt: () => Promise, + retries: number, + delayMs: number, +): Promise { + for (let tryNumber = 0; ; tryNumber++) { + try { + return await attempt(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (!message.includes("EADDRINUSE") || tryNumber >= retries) throw error; + await new Promise((resolve) => { + setTimeout(resolve, delayMs); + }); + } + } +} diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 63419912..d3a9d66a 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -9,6 +9,7 @@ */ import { createTlsTransport } from "wire-mesh-core/adapters/tls-transport"; +import * as bindRetry from "./bind-retry.js"; import { connectWsUrl } from "./ws-dial.js"; import { HubSession } from "./hub-session.js"; import { mergeKnownDevices } from "./gossip-directory.js"; @@ -653,17 +654,19 @@ export class WireMeshTransport implements MeshTransport { async becomeCoordinator(host: string, port: number): Promise { const id = nanoid(LISTENER_ID_LENGTH); - const listener = await this.wireTransport.listen( - `${host}:${String(port)}`, - (connection) => { - const tracked = this.coordinatorListeners.get(id); - void this.handleAcceptedConnection( - connection, - tracked?.policy, - false, - true, - ); - }, + const listener = await bindRetry.retryOnAddrInUse( + async () => + this.wireTransport.listen(`${host}:${String(port)}`, (connection) => { + const tracked = this.coordinatorListeners.get(id); + void this.handleAcceptedConnection( + connection, + tracked?.policy, + false, + true, + ); + }), + bindRetry.BECOME_COORDINATOR_BIND_RETRIES, + bindRetry.BECOME_COORDINATOR_BIND_RETRY_DELAY_MS, ); this._isCoordinator = true; this.coordinatorListeners.set(id, { diff --git a/src/test/bind-retry.test.ts b/src/test/bind-retry.test.ts new file mode 100644 index 00000000..dbbfb67d --- /dev/null +++ b/src/test/bind-retry.test.ts @@ -0,0 +1,57 @@ +/** + * Direct unit tests for bind-retry.ts's retryOnAddrInUse helper (agent-comms#170): WireMeshTransport.becomeCoordinator's own bind-retry safety net for the graceful coordinator handover race. When the outgoing coordinator sends become_coordinator and the successor's own handleBecomeCoordinator tries to rebind the coordinator port moments later, the outgoing side's listening socket may not have finished releasing that port yet -- an ordinary EADDRINUSE, not a design flaw, but one that would otherwise surface as an unhandled rejection crashing the process (onBecomeCoordinator's own dispatch in mesh-store.ts is fire-and-forget). retryOnAddrInUse is deliberately a pure, transport-agnostic helper (no real sockets) so this race's retry/backoff logic is fast and deterministic to test directly, independent of wire-mesh-transport's own real-TLS-socket test suite. + */ +import { describe, expect, it, vi } from "vitest"; +import { retryOnAddrInUse } from "../core/bind-retry.js"; + +const RETRIES = 3; +const DELAY_MS = 10; + +function addrInUseError(): Error { + return new Error("listen EADDRINUSE: address already in use 127.0.0.1:1"); +} + +describe("retryOnAddrInUse", () => { + it("returns the result immediately when the first attempt succeeds", async () => { + const attempt = vi.fn().mockResolvedValue("bound"); + + const result = await retryOnAddrInUse(attempt, RETRIES, DELAY_MS); + + expect(result).toBe("bound"); + expect(attempt).toHaveBeenCalledTimes(1); + }); + + it("retries on an EADDRINUSE-shaped rejection and returns once a later attempt succeeds", async () => { + const failuresBeforeSuccess = 2; + const attempt = vi + .fn() + .mockRejectedValueOnce(addrInUseError()) + .mockRejectedValueOnce(addrInUseError()) + .mockResolvedValue("bound"); + + const result = await retryOnAddrInUse(attempt, RETRIES, DELAY_MS); + + expect(result).toBe("bound"); + expect(attempt).toHaveBeenCalledTimes(failuresBeforeSuccess + 1); + }); + + it("rethrows immediately for a rejection that isn't EADDRINUSE, without retrying", async () => { + const otherError = new Error("listen EACCES: permission denied"); + const attempt = vi.fn().mockRejectedValue(otherError); + + await expect(retryOnAddrInUse(attempt, RETRIES, DELAY_MS)).rejects.toBe( + otherError, + ); + expect(attempt).toHaveBeenCalledTimes(1); + }); + + it("rethrows the last EADDRINUSE error once the retry budget is exhausted", async () => { + const attempt = vi.fn().mockRejectedValue(addrInUseError()); + + await expect(retryOnAddrInUse(attempt, RETRIES, DELAY_MS)).rejects.toThrow( + "EADDRINUSE", + ); + // The initial attempt plus exactly `retries` further attempts, never one more. + expect(attempt).toHaveBeenCalledTimes(RETRIES + 1); + }); +}); From bd03d5415d2283f0dde7c2d990e9e9f2083a6fb0 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:54:12 +0100 Subject: [PATCH 5/5] test(core): randomise mesh-e2e's coordinator port E2E_PORT was a fixed literal, so two concurrent vitest runs of this same file on the same machine (a real condition on a shared development host running several agent-comms worktrees at once) bind the identical port and one of them fails with EADDRINUSE. Picking a random port per process, in a band clear of every sibling integration test's own hardcoded port, avoids that collision. This is independent of, and layered on top of, becomeCoordinator's own bind-retry fix: that fix covers the real protocol-level race this test exists to exercise, this covers cross-process port contention on a shared machine. --- src/test/mesh-e2e.integration.test.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/test/mesh-e2e.integration.test.ts b/src/test/mesh-e2e.integration.test.ts index ab5e7cf6..43d0d8c9 100644 --- a/src/test/mesh-e2e.integration.test.ts +++ b/src/test/mesh-e2e.integration.test.ts @@ -5,6 +5,7 @@ * room creation, messaging, and delivery push. */ +import { randomInt } from "node:crypto"; import { test, expect } from "vitest"; import { MeshStore } from "../core/mesh-store.js"; import { CommsTool } from "../core/tool.js"; @@ -12,7 +13,12 @@ import { buildAction } from "../core/bridge.js"; import type { DeliveryEvent } from "../core/types.js"; import { waitFor, wireTestTransport } from "./test-transport.js"; -const E2E_PORT = 19878; +/** Start of this file's own reserved port band -- kept clear of the fixed literals every sibling integration test file hardcodes (19878-19897) so a random pick here can never collide with one of those. */ +const E2E_PORT_RANGE_START = 20_100; +/** Width of the reserved band -- wide enough that two concurrent runs of this exact file on the same machine picking the same port by chance is negligible. */ +const E2E_PORT_RANGE_WIDTH = 900; +/** Randomised per process rather than a fixed literal: a hardcoded port here deterministically collides (EADDRINUSE) with any other concurrent vitest run of this same file on the same machine -- confirmed repeatedly under real concurrent load, not hypothetical. */ +const E2E_PORT = E2E_PORT_RANGE_START + randomInt(E2E_PORT_RANGE_WIDTH); // --------------------------------------------------------------------------- // Timing constants — settle windows for asynchronous mesh propagation. There is no "operation complete" signal for these steps, so the test waits a fixed budget rather than polling.