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/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/core/peer-lifecycle.ts b/src/core/peer-lifecycle.ts index 6463d4d0..3bb2d3d2 100644 --- a/src/core/peer-lifecycle.ts +++ b/src/core/peer-lifecycle.ts @@ -126,4 +126,29 @@ 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; + + // 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] = 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( + { id: successor.id }, + { method: "become_coordinator", peerList: handoffList }, + ); + } } 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); + }); +}); 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. 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", 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();