diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4f18821..bc9682b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -40,6 +40,7 @@ jobs: test: name: Test runs-on: ubuntu-latest + timeout-minutes: 10 steps: - uses: actions/checkout@v7 - uses: pnpm/action-setup@v6 diff --git a/package.json b/package.json index 69cf3fc..57a8efc 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/peer-id-verification.integration.test.js dist/test/become-coordinator-actual-port.integration.test.js dist/test/state-sync-convergence.test.js dist/test/downtime-replay.test.js dist/test/downtime-replay.integration.test.js dist/test/filestore.test.js dist/test/handshake.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/tls-transport.integration.test.js dist/test/peer-id-verification.integration.test.js dist/test/become-coordinator-actual-port.integration.test.js dist/test/state-sync-convergence.test.js dist/test/downtime-replay.test.js dist/test/downtime-replay.integration.test.js dist/test/filestore.test.js dist/test/handshake.test.js dist/test/approval.integration.test.js dist/test/listener-policy.integration.test.js dist/test/mesh-smoke.integration.test.js", "test:visibility": "node --test dist/test/visibility.integration.test.js", "test:delivery": "node dist/test/delivery-receipt.runner.js", "test:federation": "node dist/test/federation.integration.test.js", diff --git a/src/core/comms-store.ts b/src/core/comms-store.ts index 130629c..19a7466 100644 --- a/src/core/comms-store.ts +++ b/src/core/comms-store.ts @@ -1,26 +1,23 @@ /** * CommsStore — abstract interface for the agent communication store. * - * Two implementations: - * FileStore — filesystem-backed, no server process (fallback) - * MeshStore — TCP peer mesh, in-memory, real-time push (preferred) + * Two implementations: FileStore — filesystem-backed, no server process (fallback) MeshStore — TCP peer mesh, in-memory, real-time push (preferred) * * Bridges depend on this interface, not on a specific implementation. + * + * Deliberately excludes listener management, federation, and connection approval: those are transport concerns MeshStore alone can support -- FileStore has no network transport to manage listeners on, federate through, or approve inbound connections for. Widening this interface to cover them (as it once did, via always-throwing FileStore stubs) is what forced server.ts and the bridge controller to reach past CommsStore into the concrete MeshStore anyway; CommsTool, the one consumer that genuinely needs to expose these when a MeshStore backs it, takes them as an optional extension (see MeshOnlyFeatures in tool.ts) rather than the shared interface pretending every implementation supports them. */ import type { AgentIdentity, DeliveryEvent, DmMessage, - NetworkInterface, Room, RoomMessage, RoomType, StreamingBehavior, Visibility, } from "./types.js"; -import type { ListenerInfo } from "./transport.js"; -import type { FedLink } from "./federation.js"; export interface CommsStore { // -- Identity -- @@ -95,43 +92,6 @@ export interface CommsStore { deliver(agentId: string, event: DeliveryEvent): Promise; drainDelivery(agentId: string): Promise; - // -- Listener management (coordinator only) -- - addListener(host: string, port: number, policy: string): Promise; - removeListener(id: string): Promise; - listListeners(): ListenerInfo[]; - getNetworkInterfaces(): NetworkInterface[]; - - // -- Federation (coordinator-to-coordinator) -- - fedConnect(host: string, port: number, name?: string): Promise; - fedDisconnect(linkId: string): Promise; - fedLinks(): FedLink[]; - /** This instance's own federation TLS fingerprint, to hand to an operator on the other side to pin. */ - getFederationFingerprint(): string; - /** Pin a remote mesh's certificate fingerprint as trusted for federation, inbound or outbound. */ - fedTrust(fingerprint: string): Promise; - /** Remove a previously pinned federation fingerprint. */ - fedUntrust(fingerprint: string): Promise; - /** List currently trusted federation fingerprints. */ - fedTrustedFingerprints(): string[]; - /** Start accepting inbound federation links on host:port. Rejects any connection whose certificate isn't pinned via fedTrust(). */ - fedListen(host: string, port: number): Promise; - /** Stop accepting new inbound federation connections. Existing links are unaffected. */ - fedStopListening(): Promise; - // -- Connection approval -- - acceptConnection(connectionId: string): Promise; - rejectConnection(connectionId: string, reason: string): Promise; - listPendingConnections(): { - connectionId: string; - peerId: string; - dataPort: number; - name: string; - fingerprint: string; - }[]; - connectToRemote(host: string, port: number): Promise; - - /** Start only the data server without connecting to a coordinator. */ - startDataServerOnly(): Promise; - // -- Lifecycle -- init(): Promise; } diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 24af96f..382eeb7 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -7,14 +7,16 @@ * Delivery events are pushed directly over the transport — no polling, * no filesystem. * - * Transport is injected via the constructor. TcpTransport for localhost - * TCP, TlsTransport for encrypted connections, etc. + * Transport is set via setTransport() (e.g. TlsTransport for encrypted + * connections) before init() or any other transport-using method is called + * -- there is no default, since every real bridge builds its own transport + * from this store's own events getter, which needs the store to already + * exist. */ import * as os from "node:os"; import { nanoid } from "./nanoid.js"; import { CommsError } from "./store.js"; -import { TcpTransport } from "./tcp-transport.js"; import { dmKey, normaliseWireState } from "./wire-protocol.js"; import type { SerialisedState } from "./wire-protocol.js"; import { DiscoveryManager } from "./discovery.js"; @@ -99,7 +101,7 @@ export class MeshStore implements CommsStore { private deliveryQueues = new Map(); private identityCache = new Map(); - private transport: MeshTransport; + private transport: MeshTransport | undefined; private peerInfo = new Map(); private staleCheckTimer: ReturnType | undefined; private isShutDown = false; @@ -109,7 +111,8 @@ export class MeshStore implements CommsStore { /** Whether the mesh has a live coordinator connection. */ get connected(): boolean { return ( - this.transport.isCoordinator || this.transport.hasCoordinatorConnection + this.requireTransport().isCoordinator || + this.requireTransport().hasCoordinatorConnection ); } @@ -126,6 +129,11 @@ export class MeshStore implements CommsStore { /** Fires for every state patch — both locally generated and remote. */ onPatch: ((patch: MeshStatePatch) => void | Promise) | undefined; + /** + * Fires for a transport-level error a caller may want to observe (a connection rejected for presenting a certificate that doesn't match its claimed peer ID, this store's own inability to join or create a mesh, etc). Previously wired to `this.events.onError` inside this class's own methods, but the `events` getter never actually implemented `onError` on the object it returns, so every one of those calls was silently a no-op -- this store had no way to observe its own transport failures at all. Left undefined by default (matching onDelivery/onPatch): a caller that wants visibility sets it, exactly like those two. + */ + onError: ((error: Error) => void) | undefined; + /** Serialise the full mesh state for state_sync messages. */ serialise(): SerialisedState { return { @@ -147,9 +155,6 @@ export class MeshStore implements CommsStore { this.peerId = nanoid(8); this.startedAt = new Date().toISOString(); this.coordinatorPort = coordinatorPort; - // Wire up transport with this store's event handlers. - // TcpTransport is the default; callers can replace via setTransport(). - this.transport = new TcpTransport(this.events); // Discovery manager — registers available backends this.discovery = new DiscoveryManager(); @@ -176,11 +181,21 @@ export class MeshStore implements CommsStore { ); } - /** Replace the transport (e.g. with TlsTransport for encrypted connections). */ + /** Sets the transport (e.g. TlsTransport for encrypted connections). Must be called before init() or any other transport-using method. */ setTransport(transport: MeshTransport): void { this.transport = transport; } + /** The set transport, or throws if setTransport() hasn't been called yet -- the single place every transport-using method reads through, so the "must call setTransport() first" contract is enforced at one boundary rather than checked ad hoc at each call site. */ + private requireTransport(): MeshTransport { + if (this.transport === undefined) { + throw new Error( + "MeshStore: no transport set; call setTransport() before using the store", + ); + } + return this.transport; + } + // ----------------------------------------------------------------------- // Mesh lifecycle // ----------------------------------------------------------------------- @@ -188,12 +203,12 @@ export class MeshStore implements CommsStore { async init(): Promise { if (this.initialised) return; this.initialised = true; - await this.transport.startDataServer(); + await this.requireTransport().startDataServer(); // Register our own peer info this.peerInfo.set(this.peerId, { id: this.peerId, - port: this.transport.dataPort, + port: this.requireTransport().dataPort, startedAt: this.startedAt, }); @@ -208,16 +223,16 @@ export class MeshStore implements CommsStore { let connected = false; try { - await this.transport.connectToCoordinator( + await this.requireTransport().connectToCoordinator( COORDINATOR_HOST, this.coordinatorPort, this.peerId, - this.transport.dataPort, + this.requireTransport().dataPort, ); connected = true; } catch { try { - await this.transport.becomeCoordinator( + await this.requireTransport().becomeCoordinator( COORDINATOR_HOST, this.coordinatorPort, ); @@ -243,7 +258,7 @@ export class MeshStore implements CommsStore { return; } - this.transport.unref(); + this.requireTransport().unref(); } // ----------------------------------------------------------------------- @@ -253,13 +268,16 @@ export class MeshStore implements CommsStore { private handlePeerList(peers: PeerInfo[]): void { for (const peer of peers) { this.peerInfo.set(peer.id, peer); - void this.transport.connectToPeer(peer, this.peerId); + // The list always includes this store's own entry — dialling yourself is a wasted connection attempt (and, on some platforms, an immediate self-inflicted ECONNRESET) that never needs to happen. + if (peer.id === this.peerId) continue; + void this.requireTransport().connectToPeer(peer, this.peerId); } } private handlePeerJoined(peer: PeerInfo): void { this.peerInfo.set(peer.id, peer); - void this.transport.connectToPeer(peer, this.peerId); + if (peer.id === this.peerId) return; + void this.requireTransport().connectToPeer(peer, this.peerId); } private async handleIntroduction( @@ -278,14 +296,14 @@ export class MeshStore implements CommsStore { method: "peer_list", peers: [...this.peerInfo.values()], }; - await this.transport.send(handle, peerList); + await this.requireTransport().send(handle, peerList); // Broadcast arrival to all existing peers const joined: MeshMessage = { method: "peer_joined", peer: newPeer }; - await this.transport.broadcast(joined); + await this.requireTransport().broadcast(joined); // Connect to the new peer's data server - void this.transport.connectToPeer(newPeer, this.peerId); + void this.requireTransport().connectToPeer(newPeer, this.peerId); } private async handlePeerConnected( @@ -295,7 +313,7 @@ export class MeshStore implements CommsStore { // If we have state and the peer doesn't, send state sync if (this.agents.size > 0) { const state: SerialisedState = this.serialise(); - await this.transport.send(handle, { + await this.requireTransport().send(handle, { method: "state_sync", state, }); @@ -402,14 +420,14 @@ export class MeshStore implements CommsStore { private async handleBecomeCoordinator(peerList: PeerInfo[]): Promise { // Take over as coordinator using the data server we already have - await this.transport.becomeCoordinator( + await this.requireTransport().becomeCoordinator( COORDINATOR_HOST, this.coordinatorPort, ); this.peerInfo.clear(); for (const peer of peerList) { this.peerInfo.set(peer.id, peer); - void this.transport.connectToPeer(peer, this.peerId); + void this.requireTransport().connectToPeer(peer, this.peerId); } this.startStaleCheck(); } @@ -462,7 +480,7 @@ export class MeshStore implements CommsStore { } this.pendingInboundConnections.delete(connectionId); const handle: ConnectionHandle = { id: connectionId }; - await this.transport.acceptConnection(handle); + await this.requireTransport().acceptConnection(handle); } /** Reject a pending inbound connection. */ @@ -473,7 +491,7 @@ export class MeshStore implements CommsStore { } this.pendingInboundConnections.delete(connectionId); const handle: ConnectionHandle = { id: connectionId }; - await this.transport.rejectConnection(handle, reason); + await this.requireTransport().rejectConnection(handle, reason); } /** List all pending inbound connections awaiting approval. */ @@ -498,12 +516,12 @@ export class MeshStore implements CommsStore { // Fire-and-forget: don't await the full approval handshake. // The coordinator will either accept (triggering normal introduction flow) // or reject (closing the socket). Handle rejection to avoid unhandled rejection. - this.transport + this.requireTransport() .connectToRemote( host, port, this.peerId, - this.transport.dataPort, + this.requireTransport().dataPort, agent?.name ?? "", "", ) @@ -518,13 +536,13 @@ export class MeshStore implements CommsStore { /** Start only the data server without connecting to a coordinator. * Used for testing scenarios where the peer connects via connectToRemote. */ async startDataServerOnly(): Promise { - await this.transport.startDataServer(); + await this.requireTransport().startDataServer(); this.peerInfo.set(this.peerId, { id: this.peerId, - port: this.transport.dataPort, + port: this.requireTransport().dataPort, startedAt: this.startedAt, }); - this.transport.unref(); + this.requireTransport().unref(); } // ----------------------------------------------------------------------- @@ -558,6 +576,9 @@ export class MeshStore implements CommsStore { onConnectionRequest: (handle, request) => { this.handleConnectionRequest(handle, request); }, + onError: (error) => { + this.onError?.(error); + }, }; } @@ -766,7 +787,7 @@ export class MeshStore implements CommsStore { // ----------------------------------------------------------------------- private async broadcastPatch(patch: MeshStatePatch): Promise { - await this.transport.broadcast({ method: "state_update", patch }); + await this.requireTransport().broadcast({ method: "state_update", patch }); if (this.onPatch) { await this.onPatch(patch); } @@ -1601,15 +1622,15 @@ export class MeshStore implements CommsStore { if (!isListenerPolicy(policy)) { throw new CommsError(`Invalid policy "${policy}"`, "INVALID_POLICY"); } - return this.transport.addListener(host, port, policy); + return this.requireTransport().addListener(host, port, policy); } async removeListener(id: string): Promise { - return this.transport.removeListener(id); + return this.requireTransport().removeListener(id); } listListeners(): ListenerInfo[] { - return this.transport.listListeners(); + return this.requireTransport().listListeners(); } getNetworkInterfaces(): NetworkInterface[] { @@ -1822,6 +1843,6 @@ export class MeshStore implements CommsStore { this.stopStaleCheck(); await this.federation.shutdown(); - await this.transport.shutdown(); + await this.requireTransport().shutdown(); } } diff --git a/src/core/store.ts b/src/core/store.ts index 8fc78e7..15bff04 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -26,7 +26,6 @@ import type { AgentIdentity, DeliveryEvent, DmMessage, - NetworkInterface, Room, RoomMessage, RoomType, @@ -34,8 +33,6 @@ import type { Visibility, } from "./types.js"; import type { CommsStore } from "./comms-store.js"; -import type { ListenerInfo } from "./transport.js"; -import type { FedLink } from "./federation.js"; // --------------------------------------------------------------------------- // CommsError @@ -657,125 +654,4 @@ export class FileStore implements CommsStore { await fs.mkdir(dir, { recursive: true }); } } - - // ----------------------------------------------------------------------- - // Listener management — not supported by FileStore - // ----------------------------------------------------------------------- - - addListener(): Promise { - throw new CommsError( - "FileStore does not support listener management", - "NOT_SUPPORTED", - ); - } - - removeListener(): Promise { - throw new CommsError( - "FileStore does not support listener management", - "NOT_SUPPORTED", - ); - } - - listListeners(): ListenerInfo[] { - return []; - } - - getNetworkInterfaces(): NetworkInterface[] { - return []; - } - - // ----------------------------------------------------------------------- - // Federation — not supported by FileStore - // ----------------------------------------------------------------------- - - fedConnect(): Promise { - throw new CommsError( - "FileStore does not support federation", - "NOT_SUPPORTED", - ); - } - - fedDisconnect(): Promise { - throw new CommsError( - "FileStore does not support federation", - "NOT_SUPPORTED", - ); - } - - fedLinks(): FedLink[] { - return []; - } - - getFederationFingerprint(): string { - return ""; - } - - fedTrust(): Promise { - throw new CommsError( - "FileStore does not support federation", - "NOT_SUPPORTED", - ); - } - - fedUntrust(): Promise { - throw new CommsError( - "FileStore does not support federation", - "NOT_SUPPORTED", - ); - } - - fedTrustedFingerprints(): string[] { - return []; - } - - fedListen(): Promise { - throw new CommsError( - "FileStore does not support federation", - "NOT_SUPPORTED", - ); - } - - fedStopListening(): Promise { - return Promise.resolve(); - } - // Connection approval — not supported by FileStore - // ----------------------------------------------------------------------- - - acceptConnection(): Promise { - throw new CommsError( - "FileStore does not support connection approval", - "NOT_SUPPORTED", - ); - } - - rejectConnection(): Promise { - throw new CommsError( - "FileStore does not support connection approval", - "NOT_SUPPORTED", - ); - } - - listPendingConnections(): { - connectionId: string; - peerId: string; - dataPort: number; - name: string; - fingerprint: string; - }[] { - return []; - } - - connectToRemote(): Promise { - throw new CommsError( - "FileStore does not support remote connections", - "NOT_SUPPORTED", - ); - } - - startDataServerOnly(): Promise { - throw new CommsError( - "FileStore does not support network operations", - "NOT_SUPPORTED", - ); - } } diff --git a/src/core/tcp-transport.ts b/src/core/tcp-transport.ts deleted file mode 100644 index bcadee6..0000000 --- a/src/core/tcp-transport.ts +++ /dev/null @@ -1,846 +0,0 @@ -/** - * TcpTransport — TCP localhost transport for the peer mesh. - * - * Owns all net.Server and net.Socket instances. Handles connection - * establishment, framing, and message routing. Fires events via - * TransportEvents for MeshStore to handle state management. - * - * Lifecycle mirrors the MeshTransport interface: - * 1. startDataServer() — listen for incoming peer data connections - * 2. connectToCoordinator() — join an existing mesh - * OR becomeCoordinator() — start a new mesh as coordinator - * 3. connectToPeer() — establish data connection to a discovered peer - * 4. send() / broadcast() — send wire messages - * 5. shutdown() — close all connections and servers - */ - -import * as net from "node:net"; -import { encode, isMeshMessage, MessageBuffer } from "./wire-protocol.js"; -import type { MeshMessage, PeerInfo } from "./wire-protocol.js"; -import { attachSocketHandshake } from "./handshake.js"; -import type { - ConnectionHandle, - ListenerInfo, - ListenerPolicy, - MeshTransport, - TransportEvents, -} from "./transport.js"; -import { nanoid } from "./nanoid.js"; - -// --------------------------------------------------------------------------- -// Constants -// --------------------------------------------------------------------------- - -/** Host all mesh servers bind to. */ -const COORDINATOR_HOST = "127.0.0.1"; - -/** Timeout for connecting to the coordinator before giving up. */ -const CONNECT_TIMEOUT_MS = 1000; - -// --------------------------------------------------------------------------- -// Async socket write helper (not exported) -// --------------------------------------------------------------------------- - -/** - * Safety valve so a dial that never completes cannot grow its outbound queue - * unbounded; oldest entries are dropped first (#23). - */ -const MAX_PENDING_PER_PEER = 100; - -function writeAsync(socket: net.Socket, data: string): Promise { - return new Promise((resolve, reject) => { - if (socket.destroyed) { - resolve(); - return; - } - socket.write(data, "utf-8", (err) => { - if (err) reject(err); - else resolve(); - }); - }); -} - -// --------------------------------------------------------------------------- -// TcpTransport -// --------------------------------------------------------------------------- - -export class TcpTransport implements MeshTransport { - // -- Data server (accepts incoming peer data connections) -- - private dataServer: net.Server | undefined; - private _dataPort = 0; - - // -- Coordinator listeners (multiple adapters) -- - private coordinatorListeners = new Map< - string, - { - server: net.Server; - policy: ListenerPolicy; - host: string; - port: number; - isDefault: boolean; - } - >(); - /** The default localhost listener ID (set once during becomeCoordinator). */ - private defaultListenerId: string | undefined; - private _isCoordinator = false; - - // -- Coordinator client socket (connection to the coordinator) -- - private coordinatorSocket: net.Socket | undefined; - - // -- Coordinator introduction handshake (resolved on the peer list) -- - private resolveCoordinatorHandshake: (() => void) | undefined; - private coordinatorHandshakeTimer: ReturnType | undefined; - - // -- Peer data connections (peer ID → socket + buffer) -- - private peerConnections = new Map< - string, - { socket: net.Socket; buffer: MessageBuffer } - >(); - - // -- Messages queued for peers whose dial is still in flight (#23) -- - private pendingOutbound = new Map(); - - // -- All sockets accepted by the data server (for shutdown cleanup) -- - private dataServerSockets = new Set(); - - // -- All sockets accepted by coordinator servers (for shutdown cleanup) -- - private coordinatorServerSockets = new Set(); - - // -- Coordinator introduction connections (handle ID → socket) -- - // Maps the introducing peer's ID to the coordinator server socket, so - // MeshStore can send the peer_list response via transport.send(). - private introConnections = new Map(); - - // -- Pending connections awaiting approval (handle ID → socket + request info) -- - private pendingConnections = new Map< - string, - { - socket: net.Socket; - peerId: string; - dataPort: number; - name: string; - fingerprint: string; - policy: ListenerPolicy; - } - >(); - - // -- Shutdown sentinel — prevents callbacks after shutdown() -- - private shutDown = false; - - // -- This peer's ID (set during connectToCoordinator or becomeCoordinator) -- - private _peerId = ""; - - private readonly events: TransportEvents; - - constructor(events: TransportEvents) { - this.events = events; - } - - // -- Public getters for interface properties -- - - get dataPort(): number { - return this._dataPort; - } - - get isCoordinator(): boolean { - return this._isCoordinator; - } - - get hasCoordinatorConnection(): boolean { - return ( - this.coordinatorSocket !== undefined && !this.coordinatorSocket.destroyed - ); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Data server - // ----------------------------------------------------------------------- - - async startDataServer(): Promise { - await new Promise((resolve, reject) => { - this.dataServer = net.createServer((socket) => { - this.handleIncomingDataConnection(socket); - }); - this.dataServer.listen(0, COORDINATOR_HOST, () => { - const addr = this.dataServer?.address(); - if (typeof addr === "object" && addr !== null) { - this._dataPort = addr.port; - } - resolve(); - }); - this.dataServer.on("error", reject); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Coordinator connection (client side) - // ----------------------------------------------------------------------- - - async connectToCoordinator( - host: string, - port: number, - peerId: string, - localDataPort: number, - ): Promise { - this._peerId = peerId; - await new Promise((resolve, reject) => { - const socket = net.createConnection({ port, host }, () => { - this.coordinatorSocket = socket; - - // Wire up the protocol handshake first (client role: sends its frame immediately as the connection's very first bytes) so the introduction below lands after it on the wire, not before. - const buffer = new MessageBuffer(); - attachSocketHandshake( - socket, - "client", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (isMeshMessage(item)) { - this.dispatchCoordinatorClientMessage(item); - } - } - }, - (error) => this.events.onError?.(error), - ); - - // Send introduction - const intro: MeshMessage = { - method: "introduce", - peerId, - dataPort: localDataPort, - }; - socket.write(encode(intro)); - socket.on("error", () => { - /* ignore late errors on coordinator connection */ - }); - - clearTimeout(timer); - // Resolve on the coordinator's peer list rather than on sending the - // introduction: MeshStore.init() then returns only after the post-join - // peer dials have started, so the first broadcasts are queued for the - // dialling peers instead of dropped (#23). A timeout keeps today's - // degraded behaviour for a coordinator that never answers. - this.resolveCoordinatorHandshake = resolve; - this.coordinatorHandshakeTimer = setTimeout(() => { - this.resolveCoordinatorHandshake = undefined; - resolve(); - }, CONNECT_TIMEOUT_MS); - }); - - const timer = setTimeout(() => { - socket.destroy(); - reject(new Error("Coordinator connection timeout")); - }, CONNECT_TIMEOUT_MS); - - socket.on("error", (err) => { - clearTimeout(timer); - socket.destroy(); - reject(err); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Connect to remote coordinator with approval - // ----------------------------------------------------------------------- - - async connectToRemote( - host: string, - port: number, - peerId: string, - localDataPort: number, - name: string, - fingerprint: string, - ): Promise { - this._peerId = peerId; - await new Promise((resolve, reject) => { - const socket = net.createConnection({ port, host }, () => { - this.coordinatorSocket = socket; - - // Wire up the protocol handshake first (client role) so connect_request below lands after it on the wire, pending: wait for connect_accepted/rejected, then normal dispatch - const buffer = new MessageBuffer(); - let approved = false; - attachSocketHandshake( - socket, - "client", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (!isMeshMessage(item)) continue; - - if (!approved) { - if (item.method === "connect_accepted") { - approved = true; - resolve(); - } else if (item.method === "connect_rejected") { - socket.destroy(); - reject(new Error(`Connection rejected: ${item.reason}`)); - return; - } - } else { - this.dispatchCoordinatorClientMessage(item); - } - } - }, - (error) => this.events.onError?.(error), - ); - - // Send connect_request instead of introduce - const req: MeshMessage = { - method: "connect_request", - peerId, - dataPort: localDataPort, - name, - fingerprint, - }; - socket.write(encode(req)); - socket.on("error", () => { - /* ignore late errors on coordinator connection */ - }); - }); - - const timer = setTimeout(() => { - socket.destroy(); - reject(new Error("Remote connection timeout")); - }, CONNECT_TIMEOUT_MS); - - socket.on("error", (err) => { - clearTimeout(timer); - socket.destroy(); - reject(err); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Become coordinator (server side) - // ----------------------------------------------------------------------- - - async becomeCoordinator(host: string, port: number): Promise { - const id = nanoid(8); - await new Promise((resolve, reject) => { - const server = net.createServer((socket) => { - this.handleCoordinatorServerConnection(socket, "full"); - }); - - server.listen(port, host, () => { - const addr = server.address(); - const actualPort = - typeof addr === "object" && addr !== null ? addr.port : port; - this._isCoordinator = true; - this.coordinatorListeners.set(id, { - server, - policy: "full", - host, - port: actualPort, - isDefault: true, - }); - this.defaultListenerId = id; - resolve(); - }); - - server.on("error", (err: unknown) => { - reject(err instanceof Error ? err : new Error(String(err))); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Multi-listener management - // ----------------------------------------------------------------------- - - async addListener( - host: string, - port: number, - policy: ListenerPolicy, - ): Promise { - if (!this._isCoordinator) { - throw new Error("Only the coordinator can add listeners"); - } - - const id = nanoid(8); - await new Promise((resolve, reject) => { - const server = net.createServer((socket) => { - this.handleCoordinatorServerConnection(socket, policy); - }); - - server.listen(port, host, () => { - const addr = server.address(); - const actualPort = - typeof addr === "object" && addr !== null ? addr.port : port; - this.coordinatorListeners.set(id, { - server, - policy, - host, - port: actualPort, - isDefault: false, - }); - resolve(); - }); - - server.on("error", (err: unknown) => { - reject(err instanceof Error ? err : new Error(String(err))); - }); - }); - return id; - } - - removeListener(id: string): Promise { - const listener = this.coordinatorListeners.get(id); - if (!listener) { - throw new Error(`Listener ${id} not found`); - } - if (listener.isDefault) { - throw new Error("Cannot remove the default localhost listener"); - } - - listener.server.unref(); - listener.server.close(); - this.coordinatorListeners.delete(id); - - return Promise.resolve(); - } - - listListeners(): ListenerInfo[] { - const result: ListenerInfo[] = []; - for (const [id, listener] of this.coordinatorListeners) { - result.push({ - id, - host: listener.host, - port: listener.port, - policy: listener.policy, - isDefault: listener.isDefault, - }); - } - return result; - } - - // ----------------------------------------------------------------------- - // MeshTransport — Connect to a peer's data server - // ----------------------------------------------------------------------- - - async connectToPeer(peer: PeerInfo, ownPeerId: string): Promise { - if (this.peerConnections.has(peer.id)) return; - - // Queue broadcasts until the connection registers: messages sent in the - // dial window previously had nowhere to go and were silently dropped. - this.pendingOutbound.set(peer.id, this.pendingOutbound.get(peer.id) ?? []); - - await new Promise((resolve) => { - const socket = net.createConnection( - { port: peer.port, host: COORDINATOR_HOST }, - () => { - const buffer = new MessageBuffer(); - this.peerConnections.set(peer.id, { socket, buffer }); - - // Wire up the protocol handshake first (client role) so everything below lands after it on the wire, not before. - attachSocketHandshake( - socket, - "client", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (isMeshMessage(item)) { - const handle: ConnectionHandle = { id: peer.id }; - this.dispatchDataMessage(handle, item); - } - } - }, - (error) => this.events.onError?.(error), - ); - - // Identify ourselves - const pong: MeshMessage = { method: "pong", peerId: ownPeerId }; - socket.write(encode(pong)); - - void this.flushPending(peer.id, socket); - - resolve(); - }, - ); - - const handle: ConnectionHandle = { id: peer.id }; - let disconnected = false; - - const onDisconnect = (): void => { - if (disconnected) return; - disconnected = true; - const wasConnected = this.peerConnections.has(peer.id); - this.peerConnections.delete(peer.id); - if (wasConnected && !this.shutDown) { - this.events.onPeerDisconnected(handle); - } - }; - - socket.on("close", onDisconnect); - socket.on("error", () => { - this.pendingOutbound.delete(peer.id); - onDisconnect(); - socket.destroy(); - resolve(); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Send / broadcast - // ----------------------------------------------------------------------- - - async send(handle: ConnectionHandle, message: MeshMessage): Promise { - // Check data connections first - const peerConn = this.peerConnections.get(handle.id); - if (peerConn) { - await writeAsync(peerConn.socket, encode(message)); - return; - } - - // Check coordinator introduction connections - const introSocket = this.introConnections.get(handle.id); - if (introSocket) { - await writeAsync(introSocket, encode(message)); - return; - } - - throw new Error(`No connection for handle ${handle.id}`); - } - - async acceptConnection(handle: ConnectionHandle): Promise { - const pending = this.pendingConnections.get(handle.id); - if (!pending) { - throw new Error(`No pending connection for handle ${handle.id}`); - } - this.pendingConnections.delete(handle.id); - - const { socket, peerId, dataPort, policy } = pending; - - // Move to introConnections so send() can reach this peer - this.introConnections.set(peerId, socket); - - // Send acceptance to the connecting peer - const accepted: MeshMessage = { - method: "connect_accepted", - peerId: this._peerId, - dataPort: this._dataPort, - }; - await writeAsync(socket, encode(accepted)); - - // Fire onIntroduction so MeshStore processes the new peer normally - const connHandle: ConnectionHandle = { id: peerId, policy }; - this.events.onIntroduction(connHandle, { peerId, dataPort }); - } - - async rejectConnection( - handle: ConnectionHandle, - reason: string, - ): Promise { - const pending = this.pendingConnections.get(handle.id); - if (!pending) { - throw new Error(`No pending connection for handle ${handle.id}`); - } - this.pendingConnections.delete(handle.id); - - const { socket } = pending; - - const rejected: MeshMessage = { - method: "connect_rejected", - peerId: handle.id, - reason, - }; - await writeAsync(socket, encode(rejected)); - socket.destroy(); - } - - async broadcast(message: MeshMessage): Promise { - const data = encode(message); - const writes: Promise[] = []; - for (const [, peer] of this.peerConnections) { - writes.push( - writeAsync(peer.socket, data).catch(() => { - /* broken connection — cleanup handled by close/error listeners */ - }), - ); - } - for (const queue of this.pendingOutbound.values()) { - queue.push(message); - if (queue.length > MAX_PENDING_PER_PEER) queue.shift(); - } - await Promise.all(writes); - } - - /** Send messages queued while the peer's connection was being dialled. */ - private async flushPending( - peerId: string, - socket: net.Socket, - ): Promise { - const queue = this.pendingOutbound.get(peerId); - this.pendingOutbound.delete(peerId); - if (queue === undefined) return; - for (const message of queue) { - const sent = await writeAsync(socket, encode(message)).then( - () => true, - () => false, - ); - if (!sent) return; // connection is dying; close/error listeners clean up - } - } - - // ----------------------------------------------------------------------- - // MeshTransport — Shutdown / unref - // ----------------------------------------------------------------------- - - shutdown(): Promise { - this.shutDown = true; - if (this.coordinatorHandshakeTimer !== undefined) { - clearTimeout(this.coordinatorHandshakeTimer); - this.coordinatorHandshakeTimer = undefined; - } - this.resolveCoordinatorHandshake = undefined; - - // Destroy the coordinator client socket - this.coordinatorSocket?.unref(); - this.coordinatorSocket?.destroy(); - this.coordinatorSocket = undefined; - - // Destroy all identified peer connections - for (const [, peer] of this.peerConnections) { - peer.socket.unref(); - peer.socket.destroy(); - } - this.peerConnections.clear(); - - // Destroy all data server accepted sockets (including unidentified) - for (const socket of this.dataServerSockets) { - socket.unref(); - socket.destroy(); - } - this.dataServerSockets.clear(); - - // Destroy all coordinator server accepted sockets - for (const socket of this.coordinatorServerSockets) { - socket.unref(); - socket.destroy(); - } - this.coordinatorServerSockets.clear(); - - // Clear introduction connection tracking - this.introConnections.clear(); - - // Clear pending connections - for (const [, pending] of this.pendingConnections) { - pending.socket.unref(); - pending.socket.destroy(); - } - this.pendingConnections.clear(); - - // Close data server - this.dataServer?.unref(); - this.dataServer?.close(); - this.dataServer = undefined; - - // Close coordinator listener servers — stop accepting new connections - for (const [, listener] of this.coordinatorListeners) { - listener.server.unref(); - listener.server.close(); - } - this.coordinatorListeners.clear(); - this.defaultListenerId = undefined; - - this._isCoordinator = false; - - return Promise.resolve(); - } - - unref(): void { - this.dataServer?.unref(); - for (const [, listener] of this.coordinatorListeners) { - listener.server.unref(); - } - this.coordinatorSocket?.unref(); - } - - // ----------------------------------------------------------------------- - // Internal — Coordinator client message dispatch - // ----------------------------------------------------------------------- - - /** - * Handles messages received on the coordinator client socket (the - * connection from this peer TO the coordinator). Fires the appropriate - * TransportEvents for peer_list, peer_joined, and become_coordinator. - */ - private dispatchCoordinatorClientMessage(msg: MeshMessage): void { - if (this.shutDown) return; - - if (msg.method === "peer_list") { - // Fire onPeerList first: it starts the post-join dials (and their - // broadcast queues) before init() resolves. - this.events.onPeerList(msg.peers); - this.completeCoordinatorHandshake(); - } else if (msg.method === "peer_joined") { - this.events.onPeerJoined(msg.peer); - } else if (msg.method === "become_coordinator") { - this.events.onBecomeCoordinator(msg.peerList); - } - } - - /** Complete connectToCoordinator's handshake after the peer list arrives. */ - private completeCoordinatorHandshake(): void { - if (this.coordinatorHandshakeTimer !== undefined) { - clearTimeout(this.coordinatorHandshakeTimer); - this.coordinatorHandshakeTimer = undefined; - } - this.resolveCoordinatorHandshake?.(); - this.resolveCoordinatorHandshake = undefined; - } - - // ----------------------------------------------------------------------- - // Internal — Data message dispatch - // ----------------------------------------------------------------------- - - /** - * Routes a message received on a peer data connection to the - * appropriate event. become_coordinator has its own event; all - * other messages fire onMessage for MeshStore to handle. - */ - private dispatchDataMessage( - handle: ConnectionHandle, - msg: MeshMessage, - ): void { - if (this.shutDown) return; - - if (msg.method === "become_coordinator") { - this.events.onBecomeCoordinator(msg.peerList); - } else { - this.events.onMessage(handle, msg); - } - } - - // ----------------------------------------------------------------------- - // Internal — Coordinator server connection handling - // ----------------------------------------------------------------------- - - /** - * Accepts a new connection on a coordinator server. Reads - * introduce messages and fires onIntroduction so MeshStore can - * respond with the peer list and broadcast the arrival. - * The connection is tagged with the listener's policy. - */ - private handleCoordinatorServerConnection( - socket: net.Socket, - policy: ListenerPolicy, - ): void { - if (this.shutDown) { - socket.destroy(); - return; - } - - this.coordinatorServerSockets.add(socket); - socket.on("close", () => this.coordinatorServerSockets.delete(socket)); - socket.on("error", () => this.coordinatorServerSockets.delete(socket)); - - const buffer = new MessageBuffer(); - attachSocketHandshake( - socket, - "server", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (!isMeshMessage(item)) continue; - - if (item.method === "introduce") { - const handle: ConnectionHandle = { id: item.peerId, policy }; - this.introConnections.set(handle.id, socket); - this.events.onIntroduction(handle, { - peerId: item.peerId, - dataPort: item.dataPort, - }); - } else if (item.method === "connect_request") { - const handle: ConnectionHandle = { id: item.peerId, policy }; - this.pendingConnections.set(handle.id, { - socket, - peerId: item.peerId, - dataPort: item.dataPort, - name: item.name, - fingerprint: item.fingerprint, - policy, - }); - this.events.onConnectionRequest(handle, { - peerId: item.peerId, - dataPort: item.dataPort, - name: item.name, - fingerprint: item.fingerprint, - }); - } - } - }, - (error) => this.events.onError?.(error), - ); - } - - // ----------------------------------------------------------------------- - // Internal — Incoming data connection handling - // ----------------------------------------------------------------------- - - /** - * Accepts a new connection on the data server. Reads the initial - * pong to identify the peer, then fires onMessage for subsequent - * messages. Tracks the connection for cleanup on close/error. - */ - private handleIncomingDataConnection(socket: net.Socket): void { - if (this.shutDown) { - socket.destroy(); - return; - } - - this.dataServerSockets.add(socket); - socket.on("close", () => this.dataServerSockets.delete(socket)); - - const buffer = new MessageBuffer(); - let remotePeerId: string | undefined; - let disconnected = false; - - attachSocketHandshake( - socket, - "server", - (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (isMeshMessage(item)) { - if (item.method === "pong") { - const peerId = item.peerId; - remotePeerId = peerId; - if (!this.peerConnections.has(peerId)) { - this.peerConnections.set(peerId, { socket, buffer }); - } - void this.flushPending(peerId, socket); - const handle: ConnectionHandle = { id: peerId }; - const info: PeerInfo = { - id: peerId, - port: 0, - startedAt: new Date().toISOString(), - }; - this.events.onPeerConnected(handle, info); - } else if (remotePeerId !== undefined) { - const handle: ConnectionHandle = { id: remotePeerId }; - this.dispatchDataMessage(handle, item); - } - } - } - }, - (error) => this.events.onError?.(error), - ); - - const onDisconnect = (): void => { - if (disconnected) return; - disconnected = true; - if (remotePeerId !== undefined) { - this.peerConnections.delete(remotePeerId); - if (!this.shutDown) { - this.events.onPeerDisconnected({ id: remotePeerId }); - } - } - }; - - socket.on("close", onDisconnect); - socket.on("error", onDisconnect); - } -} diff --git a/src/core/tls-transport.ts b/src/core/tls-transport.ts index 6a05615..acb221b 100644 --- a/src/core/tls-transport.ts +++ b/src/core/tls-transport.ts @@ -1,20 +1,21 @@ /** - * TlsTransport — TLS-encrypted transport for the peer mesh. + * TlsTransport — TLS-encrypted transport for the peer mesh, the one + * production MeshTransport implementation. * - * Same wire protocol as TcpTransport, but all connections are wrapped in TLS - * with certificate fingerprint authentication. Each peer generates an ECDSA - * P-256 keypair and self-signed X.509 certificate on startup. The peer ID - * is the certificate fingerprint — verifying a peer's identity is simply - * checking that the presented certificate's fingerprint matches the known ID. + * All connections are wrapped in TLS with certificate fingerprint + * authentication. Each peer generates an ECDSA P-256 keypair and self-signed + * X.509 certificate on startup. The peer ID is the certificate fingerprint — + * verifying a peer's identity is simply checking that the presented + * certificate's fingerprint matches the known ID. * * This is the Syncthing trust model: no CA, no PKI, just certificate pinning. * Overlay networks (Tailscale, WireGuard) are still recommended for NAT * traversal but are not required for security — TLS handles encryption and * authentication at the protocol level. * - * The transport is a drop-in replacement for TcpTransport. Same interface, - * same wire protocol, same event callbacks. MeshStore cannot tell the - * difference. + * Implements the same MeshTransport interface every transport does. Same + * wire protocol, same event callbacks — MeshStore only ever sees the + * interface, never a concrete transport. */ import * as net from "node:net"; @@ -90,6 +91,19 @@ function writeAsync( }); } +/** + * writeAsync(), but a failed write (the peer disconnected, or its own shutdown destroyed the socket mid-write) resolves instead of rejecting -- cleanup is already handled by the socket's own close/error listeners wherever it's registered, exactly the same ordinary-and-expected outcome broadcast() already treats this way. A caller that genuinely needs to know whether the write landed gets that back as a boolean rather than an exception, since a write racing a legitimate connection teardown is not itself a program error and must never surface as an unhandled rejection. + */ +async function writeBestEffort( + socket: net.Socket | tls.TLSSocket, + data: string, +): Promise { + return writeAsync(socket, data).then( + () => true, + () => false, + ); +} + // --------------------------------------------------------------------------- // TlsTransport // --------------------------------------------------------------------------- @@ -152,6 +166,12 @@ export class TlsTransport { } >(); + // -- Outbound connectToPeer sockets still mid-dial, not yet in peerConnections or already failed — tracked so shutdown() can destroy an in-flight dial rather than leaving it running in the background indefinitely, long after this transport is gone -- + private pendingConnectSockets = new Set(); + + // -- Peers this side has dialled out to (in flight or established) — deliberately separate from peerConnections, which also holds sockets THIS side merely accepted an inbound dial from and identified via pong. A peer can legitimately be dialling us at the same time we need to dial them (mesh formation establishes one connection in each direction so each side's acceptor can push its own state), so connectToPeer's own "don't dial twice" guard must not be satisfied by an inbound connection that happens to share the same peer ID. -- + private outboundDials = new Set(); + // -- Shutdown sentinel — prevents callbacks after shutdown() -- private shutDown = false; @@ -513,18 +533,26 @@ export class TlsTransport { // ----------------------------------------------------------------------- async connectToPeer(peer: PeerInfo, ownPeerId: string): Promise { - if (this.peerConnections.has(peer.id)) return; + if (this.shutDown || this.outboundDials.has(peer.id)) return; + this.outboundDials.add(peer.id); - // Queue broadcasts until the connection registers: messages sent in the - // dial window previously had nowhere to go and were silently dropped. + // Queue broadcasts until the connection registers: messages sent in the dial window previously had nowhere to go and were silently dropped. this.pendingOutbound.set(peer.id, this.pendingOutbound.get(peer.id) ?? []); await new Promise((resolve) => { const socket = tls.connect( { ...this.connectOptions, host: COORDINATOR_HOST, port: peer.port }, () => { + this.pendingConnectSockets.delete(socket); + if (this.shutDown) { + this.pendingOutbound.delete(peer.id); + socket.destroy(); + resolve(); + return; + } if (!this.verifyClaimedPeerId(socket, peer.id)) { this.pendingOutbound.delete(peer.id); + this.outboundDials.delete(peer.id); resolve(); return; } @@ -557,6 +585,8 @@ export class TlsTransport { resolve(); }, ); + // Tracked from the moment tls.connect() returns (before the connect callback above ever fires) so shutdown() can destroy a dial that's still mid-handshake — otherwise it keeps running in the background with nothing to cancel it, eventually erroring out (or not) long after this transport, and the test or process that started it, are gone. + this.pendingConnectSockets.add(socket); const handle: ConnectionHandle = { id: peer.id }; let disconnected = false; @@ -564,15 +594,26 @@ export class TlsTransport { const onDisconnect = (): void => { if (disconnected) return; disconnected = true; - const wasConnected = this.peerConnections.has(peer.id); - this.peerConnections.delete(peer.id); + this.pendingConnectSockets.delete(socket); + this.outboundDials.delete(peer.id); + // Only remove peerConnections' entry if it still points at THIS socket -- the peer may also have dialled us independently (a second, inbound connection for the same peer ID), and that connection's own close must never delete a live entry this one already replaced, or vice versa. + const current = this.peerConnections.get(peer.id); + const wasConnected = current?.socket === socket; + if (wasConnected) this.peerConnections.delete(peer.id); if (wasConnected && !this.shutDown) { this.events.onPeerDisconnected(handle); } }; socket.on("close", onDisconnect); - socket.on("error", () => { + socket.on("error", (err) => { + if (!this.shutDown) { + this.events.onError?.( + new Error( + `connectToPeer(${peer.id}, port ${String(peer.port)}) socket error: ${err.message}`, + ), + ); + } this.pendingOutbound.delete(peer.id); onDisconnect(); socket.destroy(); @@ -589,14 +630,14 @@ export class TlsTransport { // Check data connections first const peerConn = this.peerConnections.get(handle.id); if (peerConn) { - await writeAsync(peerConn.socket, encode(message)); + await writeBestEffort(peerConn.socket, encode(message)); return; } // Check coordinator introduction connections const introSocket = this.introConnections.get(handle.id); if (introSocket) { - await writeAsync(introSocket, encode(message)); + await writeBestEffort(introSocket, encode(message)); return; } @@ -615,13 +656,13 @@ export class TlsTransport { // Move to introConnections so send() can reach this peer this.introConnections.set(peerId, socket); - // Send acceptance to the connecting peer + // Send acceptance to the connecting peer -- best-effort: the peer having already vanished (or this side's own concurrent shutdown destroying the socket mid-write) is an ordinary disconnect, not a program error, and must not surface as an unhandled rejection once onIntroduction's own downstream state-sync work below is already in flight. const accepted: MeshMessage = { method: "connect_accepted", peerId: this._peerId, dataPort: this._dataPort, }; - await writeAsync(socket, encode(accepted)); + await writeBestEffort(socket, encode(accepted)); // Fire onIntroduction so MeshStore processes the new peer normally const connHandle: ConnectionHandle = { id: peerId, policy }; @@ -645,7 +686,7 @@ export class TlsTransport { peerId: handle.id, reason, }; - await writeAsync(socket, encode(rejected)); + await writeBestEffort(socket, encode(rejected)); socket.destroy(); } @@ -653,11 +694,7 @@ export class TlsTransport { const data = encode(message); const writes: Promise[] = []; for (const [, peer] of this.peerConnections) { - writes.push( - writeAsync(peer.socket, data).catch(() => { - /* broken connection — cleanup handled by close/error listeners */ - }), - ); + writes.push(writeBestEffort(peer.socket, data).then(() => undefined)); } for (const queue of this.pendingOutbound.values()) { queue.push(message); @@ -675,10 +712,7 @@ export class TlsTransport { this.pendingOutbound.delete(peerId); if (queue === undefined) return; for (const message of queue) { - const sent = await writeAsync(socket, encode(message)).then( - () => true, - () => false, - ); + const sent = await writeBestEffort(socket, encode(message)); if (!sent) return; // connection is dying; close/error listeners clean up } } @@ -731,6 +765,13 @@ export class TlsTransport { } this.pendingConnections.clear(); + // Destroy any outbound connectToPeer dials still mid-flight — without this they keep retrying/erroring out in the background indefinitely, long after this transport (and whatever test or process started it) is gone. + for (const socket of this.pendingConnectSockets) { + socket.unref(); + socket.destroy(); + } + this.pendingConnectSockets.clear(); + // Close data server this.dataServer?.unref(); this.dataServer?.close(); @@ -919,7 +960,11 @@ export class TlsTransport { if (disconnected) return; disconnected = true; if (remotePeerId !== undefined) { - this.peerConnections.delete(remotePeerId); + // Only remove peerConnections' entry if it still points at THIS socket -- an outbound dial to the same peer ID can register its own, separate connection there, and this accepted socket closing must never delete that live entry. + const current = this.peerConnections.get(remotePeerId); + if (current?.socket === socket) { + this.peerConnections.delete(remotePeerId); + } if (!this.shutDown) { this.events.onPeerDisconnected({ id: remotePeerId }); } diff --git a/src/core/tool.ts b/src/core/tool.ts index e7ff167..8fdb6ac 100644 --- a/src/core/tool.ts +++ b/src/core/tool.ts @@ -19,6 +19,7 @@ import type { import type { ListenerInfo } from "./transport.js"; import type { CommsStore } from "./comms-store.js"; import type { DiscoveryManager } from "./discovery.js"; +import type { FedLink } from "./federation.js"; import { CommsError } from "./store.js"; export interface CommsContext { @@ -34,12 +35,48 @@ export interface CommsResult { isError: boolean; } +/** + * Listener management, federation, and connection approval: transport concerns CommsStore deliberately excludes (see comms-store.ts's own header) since only MeshStore, never FileStore, can support them. Every method here is optional for exactly that reason -- a CommsTool backed by a FileStore simply doesn't have them, and each call site below reports that as an ordinary CommsResult error rather than assuming they exist. + */ +export interface MeshOnlyFeatures { + addListener?(host: string, port: number, policy: string): Promise; + removeListener?(id: string): Promise; + listListeners?(): ListenerInfo[]; + getNetworkInterfaces?(): NetworkInterface[]; + fedConnect?(host: string, port: number, name?: string): Promise; + fedDisconnect?(linkId: string): Promise; + fedLinks?(): FedLink[]; + getFederationFingerprint?(): string; + fedTrust?(fingerprint: string): Promise; + fedUntrust?(fingerprint: string): Promise; + fedTrustedFingerprints?(): string[]; + fedListen?(host: string, port: number): Promise; + fedStopListening?(): Promise; + acceptConnection?(connectionId: string): Promise; + rejectConnection?(connectionId: string, reason: string): Promise; + listPendingConnections?(): { + connectionId: string; + peerId: string; + dataPort: number; + name: string; + fingerprint: string; + }[]; + connectToRemote?(host: string, port: number): Promise; + setVisibility?(level: MeshVisibility, adapter?: string): Promise; + getVisibility?(adapter?: string): MeshVisibility; +} + +/** Uniform "this bridge isn't backed by a mesh transport" result for a MeshOnlyFeatures method that isn't present on the current store. */ +function notMeshBacked(action: string): CommsResult { + return { + isError: true, + content: `${action} requires a mesh-backed store (this session is running on FileStore)`, + }; +} + export class CommsTool { constructor( - private readonly store: CommsStore & { - setVisibility?(level: MeshVisibility, adapter?: string): Promise; - getVisibility?(adapter?: string): MeshVisibility; - }, + private readonly store: CommsStore & MeshOnlyFeatures, private readonly discovery?: DiscoveryManager, ) {} @@ -410,6 +447,8 @@ export class CommsTool { } private meshInterfaces(): CommsResult { + if (!this.store.getNetworkInterfaces) + return notMeshBacked("mesh_interfaces"); const interfaces = this.store.getNetworkInterfaces(); if (interfaces.length === 0) return { content: "No network interfaces found.", isError: false }; @@ -449,6 +488,7 @@ export class CommsTool { isError: true, }; } + if (!this.store.addListener) return notMeshBacked("mesh_listen"); try { const id = await this.store.addListener( action.host, @@ -471,6 +511,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_unlisten" }, ): Promise { + if (!this.store.removeListener) return notMeshBacked("mesh_unlisten"); try { await this.store.removeListener(action.id); return { content: `Listener ${action.id} removed.`, isError: false }; @@ -483,6 +524,7 @@ export class CommsTool { } private meshListeners(_ctx: CommsContext): CommsResult { + if (!this.store.listListeners) return notMeshBacked("mesh_listeners"); const listeners = this.store.listListeners(); if (listeners.length === 0) return { content: "No listeners (not coordinator).", isError: false }; @@ -534,6 +576,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_fed_connect" }, ): Promise { + if (!this.store.fedConnect) return notMeshBacked("mesh_fed_connect"); try { const linkId = await this.store.fedConnect( action.host, @@ -556,6 +599,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_connect" }, ): Promise { + if (!this.store.connectToRemote) return notMeshBacked("mesh_connect"); try { await this.store.connectToRemote(action.host, action.port); return { @@ -574,6 +618,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_fed_disconnect" }, ): Promise { + if (!this.store.fedDisconnect) return notMeshBacked("mesh_fed_disconnect"); try { await this.store.fedDisconnect(action.linkId); return { @@ -592,6 +637,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_accept" }, ): Promise { + if (!this.store.acceptConnection) return notMeshBacked("mesh_accept"); try { await this.store.acceptConnection(action.connectionId); return { @@ -607,6 +653,7 @@ export class CommsTool { } private meshFedLinks(_ctx: CommsContext): CommsResult { + if (!this.store.fedLinks) return notMeshBacked("mesh_fed_links"); const links = this.store.fedLinks(); if (links.length === 0) return { content: "No federation links.", isError: false }; @@ -622,6 +669,8 @@ export class CommsTool { } private meshFedFingerprint(_ctx: CommsContext): CommsResult { + if (!this.store.getFederationFingerprint) + return notMeshBacked("mesh_fed_fingerprint"); const fingerprint = this.store.getFederationFingerprint(); return { content: `This mesh's federation fingerprint: ${fingerprint}\nHand this to the operator on the other side so they can run mesh_fed_trust with it — and do the same in reverse before either side connects.`, @@ -633,6 +682,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_fed_trust" }, ): Promise { + if (!this.store.fedTrust) return notMeshBacked("mesh_fed_trust"); try { await this.store.fedTrust(action.fingerprint); return { @@ -651,6 +701,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_fed_untrust" }, ): Promise { + if (!this.store.fedUntrust) return notMeshBacked("mesh_fed_untrust"); try { await this.store.fedUntrust(action.fingerprint); return { @@ -666,6 +717,8 @@ export class CommsTool { } private meshFedTrusted(_ctx: CommsContext): CommsResult { + if (!this.store.fedTrustedFingerprints) + return notMeshBacked("mesh_fed_trusted"); const fingerprints = this.store.fedTrustedFingerprints(); if (fingerprints.length === 0) return { content: "No trusted federation fingerprints.", isError: false }; @@ -679,6 +732,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_fed_listen" }, ): Promise { + if (!this.store.fedListen) return notMeshBacked("mesh_fed_listen"); try { await this.store.fedListen(action.host, action.port); return { @@ -694,6 +748,8 @@ export class CommsTool { } private async meshFedStopListening(_ctx: CommsContext): Promise { + if (!this.store.fedStopListening) + return notMeshBacked("mesh_fed_stop_listening"); try { await this.store.fedStopListening(); return { @@ -712,6 +768,7 @@ export class CommsTool { _ctx: CommsContext, action: CommsAction & { action: "mesh_reject" }, ): Promise { + if (!this.store.rejectConnection) return notMeshBacked("mesh_reject"); try { await this.store.rejectConnection(action.connectionId, action.reason); return { @@ -727,6 +784,8 @@ export class CommsTool { } private meshPending(_ctx: CommsContext): CommsResult { + if (!this.store.listPendingConnections) + return notMeshBacked("mesh_pending"); const pending = this.store.listPendingConnections(); if (pending.length === 0) return { content: "No pending connections.", isError: false }; diff --git a/src/core/wire-protocol.ts b/src/core/wire-protocol.ts index c2bfa08..dd1c7cc 100644 --- a/src/core/wire-protocol.ts +++ b/src/core/wire-protocol.ts @@ -2,8 +2,9 @@ * Wire protocol — framing, encoding, and message types for the TCP mesh. * * Transport-agnostic: carries the protocol contract between peers without - * depending on net.Socket or any specific transport implementation. - * TcpTransport, TlsTransport, and WebSocketTransport all use these types. + * depending on net.Socket or any specific transport implementation. Any + * MeshTransport implementation (TlsTransport is the one production + * transport today) uses these types. */ import type { diff --git a/src/core/ws-transport.ts b/src/core/ws-transport.ts deleted file mode 100644 index d0817f7..0000000 --- a/src/core/ws-transport.ts +++ /dev/null @@ -1,782 +0,0 @@ -/** - * WebSocketTransport — WebSocket transport for the peer mesh. - * - * Carries the same wire protocol as TcpTransport but over WebSocket frames. - * Each WS frame is a complete JSON message — no newline delimiter or - * MessageBuffer needed. This enables browser-based peers (PWAs) to - * participate in the mesh. - * - * The transport is a drop-in replacement for TcpTransport. Same interface, - * same wire protocol, same event callbacks. MeshStore cannot tell the - * difference. - */ - -import { WebSocket, WebSocketServer } from "ws"; -import { isMeshMessage } from "./wire-protocol.js"; -import { attachWsHandshake } from "./handshake.js"; -import type { MeshMessage, PeerInfo } from "./wire-protocol.js"; -import type { - ConnectionHandle, - ListenerInfo, - MeshTransport, - TransportEvents, -} from "./transport.js"; - -// --------------------------------------------------------------------------- -// Constants -// --------------------------------------------------------------------------- - -/** Timeout for connecting to the coordinator before giving up. */ -const CONNECT_TIMEOUT_MS = 2000; - -// --------------------------------------------------------------------------- -// Async WS send helper (not exported) -// --------------------------------------------------------------------------- - -/** - * Safety valve so a dial that never completes cannot grow its outbound queue - * unbounded; oldest entries are dropped first (#23). - */ -const MAX_PENDING_PER_PEER = 100; - -function sendAsync(ws: WebSocket, data: string): Promise { - return new Promise((resolve, reject) => { - if (ws.readyState !== WebSocket.OPEN) { - resolve(); - return; - } - ws.send(data, (err) => { - if (err) reject(err); - else resolve(); - }); - }); -} - -// --------------------------------------------------------------------------- -// WebSocketTransport -// --------------------------------------------------------------------------- - -export class WebSocketTransport implements MeshTransport { - // -- Data server (accepts incoming peer data connections) -- - private dataServer: WebSocketServer | undefined; - private _dataPort = 0; - - // -- Coordinator server (accepts introductions from new peers) -- - private coordinatorServer: WebSocketServer | undefined; - private _isCoordinator = false; - - // -- Coordinator client socket (connection to the coordinator) -- - private coordinatorWs: WebSocket | undefined; - - // -- Coordinator introduction handshake (resolved on the peer list) -- - private resolveCoordinatorHandshake: (() => void) | undefined; - private coordinatorHandshakeTimer: ReturnType | undefined; - - // -- Peer data connections (peer ID → WebSocket) -- - private peerConnections = new Map(); - - // -- Messages queued for peers whose dial is still in flight (#23) -- - private pendingOutbound = new Map(); - - // -- All WS connections accepted by the data server (for shutdown cleanup) -- - private dataServerSockets = new Set(); - - // -- All WS connections accepted by the coordinator server (for shutdown cleanup) -- - private coordinatorServerSockets = new Set(); - - // -- Coordinator introduction connections (handle ID → WebSocket) -- - // Maps the introducing peer's ID to the coordinator server WS, so - // MeshStore can send the peer_list response via transport.send(). - private introConnections = new Map(); - - // -- Pending connections awaiting approval (handle ID → WS + request info) -- - private pendingConnections = new Map< - string, - { - ws: WebSocket; - peerId: string; - dataPort: number; - name: string; - fingerprint: string; - } - >(); - - // -- Shutdown sentinel — prevents callbacks after shutdown() -- - private shutDown = false; - - // -- This peer's ID (set during connectToCoordinator or becomeCoordinator) -- - private _peerId = ""; - - private readonly events: TransportEvents; - - constructor(events: TransportEvents) { - this.events = events; - } - - // -- Public getters for interface properties -- - - get dataPort(): number { - return this._dataPort; - } - - get isCoordinator(): boolean { - return this._isCoordinator; - } - - get hasCoordinatorConnection(): boolean { - return this.coordinatorWs?.readyState === WebSocket.OPEN; - } - - // ----------------------------------------------------------------------- - // MeshTransport — Data server - // ----------------------------------------------------------------------- - - async startDataServer(): Promise { - await new Promise((resolve, reject) => { - // WebSocketServer needs an underlying HTTP server to get an - // OS-assigned port via port 0. - this.dataServer = new WebSocketServer({ port: 0, host: "0.0.0.0" }); - - this.dataServer.on("error", reject); - - this.dataServer.on("listening", () => { - const addr = this.dataServer?.address(); - if (typeof addr === "object" && addr !== null) { - this._dataPort = addr.port; - } - resolve(); - }); - - this.dataServer.on("connection", (ws) => { - this.handleIncomingDataConnection(ws); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Coordinator connection (client side) - // ----------------------------------------------------------------------- - - async connectToCoordinator( - host: string, - port: number, - peerId: string, - localDataPort: number, - ): Promise { - this._peerId = peerId; - await new Promise((resolve, reject) => { - const url = `ws://${host}:${String(port)}`; - const ws = new WebSocket(url); - - const timer = setTimeout(() => { - ws.terminate(); - reject(new Error("Coordinator connection timeout")); - }, CONNECT_TIMEOUT_MS); - - ws.on("unexpected-response", () => { - clearTimeout(timer); - ws.terminate(); - reject(new Error("Coordinator connection rejected")); - }); - - ws.on("open", () => { - this.coordinatorWs = ws; - - // Wire up the protocol handshake first (client role: attachWsHandshake sends its own binary frame immediately) so the introduction below lands after it on the wire, not before, and not duplicated. - attachWsHandshake( - ws, - "client", - (raw) => { - const msg = parseMessage(raw); - if (msg !== undefined) { - this.dispatchCoordinatorClientMessage(msg); - } - }, - (error) => this.events.onError?.(error), - ); - - // Send introduction - const intro: MeshMessage = { - method: "introduce", - peerId, - dataPort: localDataPort, - }; - ws.send(JSON.stringify(intro)); - ws.on("error", () => { - /* ignore late errors on coordinator connection */ - }); - - clearTimeout(timer); - // Resolve on the coordinator's peer list rather than on sending the - // introduction: MeshStore.init() then returns only after the post-join - // peer dials have started, so the first broadcasts are queued for the - // dialling peers instead of dropped (#23). A timeout keeps today's - // degraded behaviour for a coordinator that never answers. - this.resolveCoordinatorHandshake = resolve; - this.coordinatorHandshakeTimer = setTimeout(() => { - this.resolveCoordinatorHandshake = undefined; - resolve(); - }, CONNECT_TIMEOUT_MS); - }); - - ws.on("error", (err) => { - clearTimeout(timer); - ws.terminate(); - reject(err); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Connect to remote coordinator with approval - // ----------------------------------------------------------------------- - - async connectToRemote( - host: string, - port: number, - peerId: string, - _localDataPort: number, - name: string, - fingerprint: string, - ): Promise { - this._peerId = peerId; - await new Promise((resolve, reject) => { - const url = `ws://${host}:${String(port)}`; - const ws = new WebSocket(url); - - const timer = setTimeout(() => { - ws.terminate(); - reject(new Error("Remote connection timeout")); - }, CONNECT_TIMEOUT_MS); - - ws.on("unexpected-response", () => { - clearTimeout(timer); - ws.terminate(); - reject(new Error("Remote connection rejected")); - }); - - ws.on("open", () => { - this.coordinatorWs = ws; - - // Wire up the protocol handshake first (client role) so connect_request below lands after it on the wire, not duplicated. Wait for connect_accepted or connect_rejected. - let approved = false; - attachWsHandshake( - ws, - "client", - (raw) => { - const msg = parseMessage(raw); - if (msg === undefined) return; - - if (!approved) { - if (msg.method === "connect_accepted") { - approved = true; - clearTimeout(timer); - resolve(); - } else if (msg.method === "connect_rejected") { - clearTimeout(timer); - ws.terminate(); - reject(new Error(`Connection rejected: ${msg.reason}`)); - return; - } - } else { - this.dispatchCoordinatorClientMessage(msg); - } - }, - (error) => this.events.onError?.(error), - ); - - // Send connect_request instead of introduce - const req: MeshMessage = { - method: "connect_request", - peerId, - dataPort: _localDataPort, - name, - fingerprint, - }; - ws.send(JSON.stringify(req)); - ws.on("error", () => { - /* ignore late errors on coordinator connection */ - }); - }); - - ws.on("error", (err) => { - clearTimeout(timer); - ws.terminate(); - reject(err); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Become coordinator (server side) - // ----------------------------------------------------------------------- - - async becomeCoordinator(host: string, port: number): Promise { - await new Promise((resolve, reject) => { - this.coordinatorServer = new WebSocketServer({ host, port }); - - this.coordinatorServer.on("error", (err: unknown) => { - reject(err instanceof Error ? err : new Error(String(err))); - }); - - this.coordinatorServer.on("listening", () => { - this._isCoordinator = true; - resolve(); - }); - - this.coordinatorServer.on("connection", (ws) => { - this.handleCoordinatorServerConnection(ws); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Connect to a peer's data server - // ----------------------------------------------------------------------- - - async connectToPeer(peer: PeerInfo, ownPeerId: string): Promise { - if (this.peerConnections.has(peer.id)) return; - - // Queue broadcasts until the connection registers: messages sent in the - // dial window previously had nowhere to go and were silently dropped. - this.pendingOutbound.set(peer.id, this.pendingOutbound.get(peer.id) ?? []); - - await new Promise((resolve) => { - const url = `ws://127.0.0.1:${String(peer.port)}`; - const ws = new WebSocket(url); - - ws.on("open", () => { - this.peerConnections.set(peer.id, ws); - - // Wire up the protocol handshake first (client role) so everything below lands after it on the wire, not duplicated. - attachWsHandshake( - ws, - "client", - (raw) => { - const msg = parseMessage(raw); - if (msg !== undefined) { - const handle: ConnectionHandle = { id: peer.id }; - this.dispatchDataMessage(handle, msg); - } - }, - (error) => this.events.onError?.(error), - ); - - // Identify ourselves - const pong: MeshMessage = { method: "pong", peerId: ownPeerId }; - ws.send(JSON.stringify(pong)); - - void this.flushPending(peer.id, ws); - - resolve(); - }); - - const handle: ConnectionHandle = { id: peer.id }; - let disconnected = false; - - const onDisconnect = (): void => { - if (disconnected) return; - disconnected = true; - const wasConnected = this.peerConnections.has(peer.id); - this.peerConnections.delete(peer.id); - if (wasConnected && !this.shutDown) { - this.events.onPeerDisconnected(handle); - } - }; - - ws.on("close", onDisconnect); - ws.on("error", () => { - this.pendingOutbound.delete(peer.id); - onDisconnect(); - ws.terminate(); - resolve(); - }); - }); - } - - // ----------------------------------------------------------------------- - // MeshTransport — Send / broadcast - // ----------------------------------------------------------------------- - - async send(handle: ConnectionHandle, message: MeshMessage): Promise { - const data = JSON.stringify(message); - - // Check data connections first - const peerWs = this.peerConnections.get(handle.id); - if (peerWs) { - await sendAsync(peerWs, data); - return; - } - - // Check coordinator introduction connections - const introWs = this.introConnections.get(handle.id); - if (introWs) { - await sendAsync(introWs, data); - return; - } - - throw new Error(`No connection for handle ${handle.id}`); - } - - async acceptConnection(handle: ConnectionHandle): Promise { - const pending = this.pendingConnections.get(handle.id); - if (!pending) { - throw new Error(`No pending connection for handle ${handle.id}`); - } - this.pendingConnections.delete(handle.id); - - const { ws, peerId, dataPort } = pending; - - // Move to introConnections so send() can reach this peer - this.introConnections.set(peerId, ws); - - // Send acceptance to the connecting peer - const accepted: MeshMessage = { - method: "connect_accepted", - peerId: this._peerId, - dataPort: this._dataPort, - }; - await sendAsync(ws, JSON.stringify(accepted)); - - // Fire onIntroduction so MeshStore processes the new peer normally - const connHandle: ConnectionHandle = { id: peerId }; - this.events.onIntroduction(connHandle, { peerId, dataPort }); - } - - async rejectConnection( - handle: ConnectionHandle, - reason: string, - ): Promise { - const pending = this.pendingConnections.get(handle.id); - if (!pending) { - throw new Error(`No pending connection for handle ${handle.id}`); - } - this.pendingConnections.delete(handle.id); - - const { ws } = pending; - - const rejected: MeshMessage = { - method: "connect_rejected", - peerId: handle.id, - reason, - }; - await sendAsync(ws, JSON.stringify(rejected)); - ws.terminate(); - } - - async broadcast(message: MeshMessage): Promise { - const data = JSON.stringify(message); - const writes: Promise[] = []; - for (const [, ws] of this.peerConnections) { - writes.push( - sendAsync(ws, data).catch(() => { - /* broken connection — cleanup handled by close/error listeners */ - }), - ); - } - for (const queue of this.pendingOutbound.values()) { - queue.push(message); - if (queue.length > MAX_PENDING_PER_PEER) queue.shift(); - } - await Promise.all(writes); - } - - /** Send messages queued while the peer's connection was being dialled. */ - private async flushPending(peerId: string, ws: WebSocket): Promise { - const queue = this.pendingOutbound.get(peerId); - this.pendingOutbound.delete(peerId); - if (queue === undefined) return; - for (const message of queue) { - const sent = await sendAsync(ws, JSON.stringify(message)).then( - () => true, - () => false, - ); - if (!sent) return; // connection is dying; close listeners clean up - } - } - - // ----------------------------------------------------------------------- - // MeshTransport — Listener management (not supported over WebSocket) - // ----------------------------------------------------------------------- - - addListener(): Promise { - throw new Error("WebSocketTransport does not support listener management"); - } - - removeListener(): Promise { - throw new Error("WebSocketTransport does not support listener management"); - } - - listListeners(): ListenerInfo[] { - return []; - } - - // ----------------------------------------------------------------------- - // MeshTransport — Shutdown / unref - // ----------------------------------------------------------------------- - - shutdown(): Promise { - this.shutDown = true; - if (this.coordinatorHandshakeTimer !== undefined) { - clearTimeout(this.coordinatorHandshakeTimer); - this.coordinatorHandshakeTimer = undefined; - } - this.resolveCoordinatorHandshake = undefined; - - // Destroy the coordinator client socket - this.coordinatorWs?.terminate(); - this.coordinatorWs = undefined; - - // Destroy all identified peer connections - for (const [, ws] of this.peerConnections) { - ws.terminate(); - } - this.peerConnections.clear(); - - // Destroy all data server accepted sockets (including unidentified) - for (const ws of this.dataServerSockets) { - ws.terminate(); - } - this.dataServerSockets.clear(); - - // Destroy all coordinator server accepted sockets - for (const ws of this.coordinatorServerSockets) { - ws.terminate(); - } - this.coordinatorServerSockets.clear(); - - // Clear introduction connection tracking - this.introConnections.clear(); - - // Clear pending connections - for (const [, pending] of this.pendingConnections) { - pending.ws.terminate(); - } - this.pendingConnections.clear(); - - // Close servers — stop accepting new connections - this.dataServer?.close(); - this.dataServer = undefined; - this.coordinatorServer?.close(); - this.coordinatorServer = undefined; - - this._isCoordinator = false; - - return Promise.resolve(); - } - - unref(): void { - // WebSocketServer doesn't have an unref() method the way net.Server - // does. The servers will be closed on shutdown(). Peer WS connections - // are cleaned up on shutdown() as well. - } - - // ----------------------------------------------------------------------- - // Internal — Coordinator client message dispatch - // ----------------------------------------------------------------------- - - /** - * Handles messages received on the coordinator client socket (the - * connection from this peer TO the coordinator). Fires the appropriate - * TransportEvents for peer_list, peer_joined, and become_coordinator. - */ - private dispatchCoordinatorClientMessage(msg: MeshMessage): void { - if (this.shutDown) return; - - if (msg.method === "peer_list") { - // Fire onPeerList first: it starts the post-join dials (and their - // broadcast queues) before init() resolves. - this.events.onPeerList(msg.peers); - this.completeCoordinatorHandshake(); - } else if (msg.method === "peer_joined") { - this.events.onPeerJoined(msg.peer); - } else if (msg.method === "become_coordinator") { - this.events.onBecomeCoordinator(msg.peerList); - } - } - - /** Complete connectToCoordinator's handshake after the peer list arrives. */ - private completeCoordinatorHandshake(): void { - if (this.coordinatorHandshakeTimer !== undefined) { - clearTimeout(this.coordinatorHandshakeTimer); - this.coordinatorHandshakeTimer = undefined; - } - this.resolveCoordinatorHandshake?.(); - this.resolveCoordinatorHandshake = undefined; - } - - // ----------------------------------------------------------------------- - // Internal — Data message dispatch - // ----------------------------------------------------------------------- - - /** - * Routes a message received on a peer data connection to the - * appropriate event. become_coordinator has its own event; all - * other messages fire onMessage for MeshStore to handle. - */ - private dispatchDataMessage( - handle: ConnectionHandle, - msg: MeshMessage, - ): void { - if (this.shutDown) return; - - if (msg.method === "become_coordinator") { - this.events.onBecomeCoordinator(msg.peerList); - } else { - this.events.onMessage(handle, msg); - } - } - - // ----------------------------------------------------------------------- - // Internal — Coordinator server connection handling - // ----------------------------------------------------------------------- - - /** - * Accepts a new connection on the coordinator server. Reads - * introduce messages and fires onIntroduction so MeshStore can - * respond with the peer list and broadcast the arrival. - */ - private handleCoordinatorServerConnection(ws: WebSocket): void { - if (this.shutDown) { - ws.terminate(); - return; - } - - this.coordinatorServerSockets.add(ws); - ws.on("close", () => this.coordinatorServerSockets.delete(ws)); - ws.on("error", () => this.coordinatorServerSockets.delete(ws)); - - attachWsHandshake( - ws, - "server", - (raw) => { - const msg = parseMessage(raw); - if (msg === undefined) return; - - if (msg.method === "introduce") { - const handle: ConnectionHandle = { id: msg.peerId }; - this.introConnections.set(handle.id, ws); - this.events.onIntroduction(handle, { - peerId: msg.peerId, - dataPort: msg.dataPort, - }); - } else if (msg.method === "connect_request") { - const handle: ConnectionHandle = { id: msg.peerId }; - this.pendingConnections.set(handle.id, { - ws, - peerId: msg.peerId, - dataPort: msg.dataPort, - name: msg.name, - fingerprint: msg.fingerprint, - }); - this.events.onConnectionRequest(handle, { - peerId: msg.peerId, - dataPort: msg.dataPort, - name: msg.name, - fingerprint: msg.fingerprint, - }); - } - }, - (error) => this.events.onError?.(error), - ); - } - - // ----------------------------------------------------------------------- - // Internal — Incoming data connection handling - // ----------------------------------------------------------------------- - - /** - * Accepts a new connection on the data server. Reads the initial - * pong to identify the peer, then fires onMessage for subsequent - * messages. Tracks the connection for cleanup on close/error. - */ - private handleIncomingDataConnection(ws: WebSocket): void { - if (this.shutDown) { - ws.terminate(); - return; - } - - this.dataServerSockets.add(ws); - ws.on("close", () => this.dataServerSockets.delete(ws)); - - let remotePeerId: string | undefined; - let disconnected = false; - - attachWsHandshake( - ws, - "server", - (raw) => { - const msg = parseMessage(raw); - if (msg === undefined) return; - - if (msg.method === "pong") { - const peerId = msg.peerId; - remotePeerId = peerId; - if (!this.peerConnections.has(peerId)) { - this.peerConnections.set(peerId, ws); - } - void this.flushPending(peerId, ws); - const handle: ConnectionHandle = { id: peerId }; - const info: PeerInfo = { - id: peerId, - port: 0, - startedAt: new Date().toISOString(), - }; - this.events.onPeerConnected(handle, info); - } else if (remotePeerId !== undefined) { - const handle: ConnectionHandle = { id: remotePeerId }; - this.dispatchDataMessage(handle, msg); - } - }, - (error) => this.events.onError?.(error), - ); - - const onDisconnect = (): void => { - if (disconnected) return; - disconnected = true; - if (remotePeerId !== undefined) { - this.peerConnections.delete(remotePeerId); - if (!this.shutDown) { - this.events.onPeerDisconnected({ id: remotePeerId }); - } - } - }; - - ws.on("close", onDisconnect); - ws.on("error", onDisconnect); - } -} - -// --------------------------------------------------------------------------- -// Message parsing (not exported) -// --------------------------------------------------------------------------- - -/** - * Parse a raw WS message into a MeshMessage, returning undefined for - * malformed data. Unlike TCP, each WS frame is a complete message — - * no newline delimiter or buffer splitting required. - */ -function parseMessage(raw: unknown): MeshMessage | undefined { - if (raw === undefined) return undefined; - - // WS data arrives as Buffer in Node, string in browsers - let text: string; - if (typeof raw === "string") { - text = raw; - } else if (raw instanceof Buffer || ArrayBuffer.isView(raw)) { - text = new TextDecoder().decode( - raw instanceof Buffer ? raw : new Uint8Array(raw.buffer), - ); - } else if (raw instanceof ArrayBuffer) { - text = new TextDecoder().decode(raw); - } else { - return undefined; - } - - try { - const parsed: unknown = JSON.parse(text); - return isMeshMessage(parsed) ? parsed : undefined; - } catch { - return undefined; - } -} diff --git a/src/test/approval.integration.test.ts b/src/test/approval.integration.test.ts index 283579d..e2f1243 100644 --- a/src/test/approval.integration.test.ts +++ b/src/test/approval.integration.test.ts @@ -13,6 +13,7 @@ import { MeshStore } from "../core/mesh-store.js"; import { CommsTool } from "../core/tool.js"; import { buildAction } from "../core/bridge.js"; import type { DeliveryEvent } from "../core/types.js"; +import { waitFor, wireTestTransport } from "./test-transport.js"; /** Find a free port on localhost by binding to port 0. */ function findFreePort(): Promise { @@ -48,67 +49,74 @@ describe("connection approval", () => { // Set up coordinator (store A) const storeA = new MeshStore(portA); - const receivedRequests: Extract< - DeliveryEvent, - { type: "connection_request" } - >[] = []; - storeA.onDelivery = (_id, event) => { - if (event.type === "connection_request") { - receivedRequests.push(event); - } - }; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - await sleep(100); - - // Set up connecting peer (store B) that uses connectToRemote + wireTestTransport(storeA); const storeB = new MeshStore(portB); - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); + wireTestTransport(storeB); + try { + const receivedRequests: Extract< + DeliveryEvent, + { type: "connection_request" } + >[] = []; + storeA.onDelivery = (_id, event) => { + if (event.type === "connection_request") { + receivedRequests.push(event); + } + }; + await storeA.init(); + await storeA.registerAgent({ + name: "coordinator", + harness: "test", + cwd: "/test/a", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + await sleep(100); - // Initiate connection request - storeB.connectToRemote("127.0.0.1", portA); - - // Wait for the coordinator to receive the request - await sleep(300); - - // Coordinator should have received a connection_request event - assert.equal( - receivedRequests.length, - 1, - "Coordinator should receive exactly one connection request", - ); - const request = receivedRequests[0]; - assert.equal(request?.type, "connection_request"); - assert.ok(request?.connectionId, "Request should have a connectionId"); - assert.equal( - request?.peerId, - storeB.peerId, - "Request should contain connector's peer ID", - ); - assert.equal( - request?.name, - "connector", - "Request should contain connector's name", - ); - - await storeB.shutdown(); - await storeA.shutdown(); + // Set up connecting peer (store B) that uses connectToRemote + await storeB.startDataServerOnly(); + await storeB.registerAgent({ + name: "connector", + harness: "test", + cwd: "/test/b", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // Initiate connection request + storeB.connectToRemote("127.0.0.1", portA); + + // Wait for the coordinator to receive the request + await waitFor( + () => receivedRequests.length === 1, + "coordinator receives the connection request", + ); + + // Coordinator should have received a connection_request event + assert.equal( + receivedRequests.length, + 1, + "Coordinator should receive exactly one connection request", + ); + const request = receivedRequests[0]; + assert.equal(request?.type, "connection_request"); + assert.ok(request?.connectionId, "Request should have a connectionId"); + assert.equal( + request?.peerId, + storeB.peerId, + "Request should contain connector's peer ID", + ); + assert.equal( + request?.name, + "connector", + "Request should contain connector's name", + ); + } finally { + await storeB.shutdown(); + await storeA.shutdown(); + } }); void test("accept establishes the peer connection", async () => { @@ -116,71 +124,78 @@ describe("connection approval", () => { const portB = portA + 100; const storeA = new MeshStore(portA); - const receivedRequests: Extract< - DeliveryEvent, - { type: "connection_request" } - >[] = []; - storeA.onDelivery = (_id, event) => { - if (event.type === "connection_request") { - receivedRequests.push(event); - } - }; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - await sleep(100); - + wireTestTransport(storeA); const storeB = new MeshStore(portB); - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - storeB.connectToRemote("127.0.0.1", portA); - - await sleep(300); - - assert.equal( - receivedRequests.length, - 1, - "Coordinator should receive a connection request", - ); - const request = receivedRequests[0]; - assert.ok(request?.connectionId); - - // Accept the connection - await storeA.acceptConnection(request.connectionId); - - // Wait for state sync (peer_list → connectToPeer → state_sync → handlePeerConnected) - await sleep(500); - - // Verify both stores see each other's agents - const agentsA = await storeA.listAgents(storeA.peerId); - const agentsB = await storeB.listAgents(storeB.peerId); + wireTestTransport(storeB); + try { + const receivedRequests: Extract< + DeliveryEvent, + { type: "connection_request" } + >[] = []; + storeA.onDelivery = (_id, event) => { + if (event.type === "connection_request") { + receivedRequests.push(event); + } + }; + await storeA.init(); + await storeA.registerAgent({ + name: "coordinator", + harness: "test", + cwd: "/test/a", + pid: process.pid, + visibility: "visible", + tags: [], + }); - assert.ok( - agentsA.some((a) => a.id === storeB.peerId), - "Coordinator should see the connector agent", - ); - assert.ok( - agentsB.some((a) => a.id === storeA.peerId), - "Connector should see the coordinator agent", - ); + await sleep(100); - await storeB.shutdown(); - await storeA.shutdown(); + await storeB.startDataServerOnly(); + await storeB.registerAgent({ + name: "connector", + harness: "test", + cwd: "/test/b", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + storeB.connectToRemote("127.0.0.1", portA); + + await waitFor( + () => receivedRequests.length === 1, + "coordinator receives the connection request", + ); + const request = receivedRequests[0]; + assert.ok(request?.connectionId); + + // Accept the connection + await storeA.acceptConnection(request.connectionId); + + // State sync (peer_list -> connectToPeer -> state_sync -> handlePeerConnected) is a real TLS round trip, not instantaneous -- poll rather than assume any fixed delay is enough under a loaded CI runner. + await waitFor( + () => storeA.serialise().agents[storeB.peerId] !== undefined, + "coordinator sees the connector agent", + ); + await waitFor( + () => storeB.serialise().agents[storeA.peerId] !== undefined, + "connector sees the coordinator agent", + ); + + const agentsA = await storeA.listAgents(storeA.peerId); + const agentsB = await storeB.listAgents(storeB.peerId); + + assert.ok( + agentsA.some((a) => a.id === storeB.peerId), + "Coordinator should see the connector agent", + ); + assert.ok( + agentsB.some((a) => a.id === storeA.peerId), + "Connector should see the coordinator agent", + ); + } finally { + await storeB.shutdown(); + await storeA.shutdown(); + } }); void test("reject closes with reason", async () => { @@ -188,65 +203,72 @@ describe("connection approval", () => { const portB = portA + 100; const storeA = new MeshStore(portA); - const receivedRequests: Extract< - DeliveryEvent, - { type: "connection_request" } - >[] = []; - storeA.onDelivery = (_id, event) => { - if (event.type === "connection_request") { - receivedRequests.push(event); - } - }; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - await sleep(100); - + wireTestTransport(storeA); const storeB = new MeshStore(portB); - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - storeB.connectToRemote("127.0.0.1", portA); - - await sleep(300); - - assert.equal( - receivedRequests.length, - 1, - "Coordinator should receive a connection request", - ); - const request = receivedRequests[0]; - assert.ok(request?.connectionId); - - // Reject the connection - await storeA.rejectConnection(request.connectionId, "unauthorised"); - - // Give the rejection time to propagate - await sleep(200); + wireTestTransport(storeB); + try { + const receivedRequests: Extract< + DeliveryEvent, + { type: "connection_request" } + >[] = []; + storeA.onDelivery = (_id, event) => { + if (event.type === "connection_request") { + receivedRequests.push(event); + } + }; + await storeA.init(); + await storeA.registerAgent({ + name: "coordinator", + harness: "test", + cwd: "/test/a", + pid: process.pid, + visibility: "visible", + tags: [], + }); - // Verify the coordinator no longer sees the connector agent - const agentsA = await storeA.listAgents(storeA.peerId); - assert.ok( - !agentsA.some((a) => a.id === storeB.peerId), - "Rejected peer should not appear in agent list", - ); + await sleep(100); - await storeB.shutdown(); - await storeA.shutdown(); + await storeB.startDataServerOnly(); + await storeB.registerAgent({ + name: "connector", + harness: "test", + cwd: "/test/b", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + storeB.connectToRemote("127.0.0.1", portA); + + await waitFor( + () => receivedRequests.length === 1, + "coordinator receives the connection request", + ); + + assert.equal( + receivedRequests.length, + 1, + "Coordinator should receive a connection request", + ); + const request = receivedRequests[0]; + assert.ok(request?.connectionId); + + // Reject the connection + await storeA.rejectConnection(request.connectionId, "unauthorised"); + + // Give the rejection time to propagate -- there is no positive event to poll for here (the assertion below proves an absence), so a fixed wait is the right shape, unlike the accept-flow tests above. + await sleep(200); + + // Verify the coordinator no longer sees the connector agent + const agentsA = await storeA.listAgents(storeA.peerId); + assert.ok( + !agentsA.some((a) => a.id === storeB.peerId), + "Rejected peer should not appear in agent list", + ); + } finally { + await storeB.shutdown(); + await storeA.shutdown(); + } }); void test("mesh_pending lists pending connections", async () => { @@ -254,75 +276,77 @@ describe("connection approval", () => { const portB = portA + 100; const storeA = new MeshStore(portA); - storeA.onDelivery = () => {}; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - await sleep(100); - + wireTestTransport(storeA); const storeB = new MeshStore(portB); - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); + wireTestTransport(storeB); + try { + storeA.onDelivery = () => {}; + await storeA.init(); + await storeA.registerAgent({ + name: "coordinator", + harness: "test", + cwd: "/test/a", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + await sleep(100); - // No pending connections initially - let pending = storeA.listPendingConnections(); - assert.equal( - pending.length, - 0, - "Should have no pending connections initially", - ); - - // Initiate connection request - storeB.connectToRemote("127.0.0.1", portA); - - await sleep(300); - - // Should now have one pending connection - pending = storeA.listPendingConnections(); - assert.equal(pending.length, 1, "Should have one pending connection"); - assert.equal( - pending[0]?.name, - "connector", - "Pending connection should show connector name", - ); - assert.equal( - pending[0]?.peerId, - storeB.peerId, - "Pending connection should show connector peer ID", - ); - - // Accept to clean up - const connectionId = pending[0]?.connectionId; - assert.ok(connectionId); - await storeA.acceptConnection(connectionId); - - // Wait for state sync - await sleep(300); - - // No more pending connections after acceptance - pending = storeA.listPendingConnections(); - assert.equal( - pending.length, - 0, - "Should have no pending connections after accept", - ); - - await storeB.shutdown(); - await storeA.shutdown(); + await storeB.startDataServerOnly(); + await storeB.registerAgent({ + name: "connector", + harness: "test", + cwd: "/test/b", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // No pending connections initially + let pending = storeA.listPendingConnections(); + assert.equal( + pending.length, + 0, + "Should have no pending connections initially", + ); + + // Initiate connection request + storeB.connectToRemote("127.0.0.1", portA); + + await waitFor( + () => storeA.listPendingConnections().length === 1, + "coordinator sees the pending connection", + ); + + // Should now have one pending connection + pending = storeA.listPendingConnections(); + assert.equal(pending.length, 1, "Should have one pending connection"); + assert.equal( + pending[0]?.name, + "connector", + "Pending connection should show connector name", + ); + assert.equal( + pending[0]?.peerId, + storeB.peerId, + "Pending connection should show connector peer ID", + ); + + // Accept to clean up + const connectionId = pending[0]?.connectionId; + assert.ok(connectionId); + await storeA.acceptConnection(connectionId); + + // No more pending connections after acceptance + await waitFor( + () => storeA.listPendingConnections().length === 0, + "pending connection is cleared after accept", + ); + } finally { + await storeB.shutdown(); + await storeA.shutdown(); + } }); void test("tool handles mesh_connect/mesh_accept/mesh_reject/mesh_pending actions", async () => { @@ -330,132 +354,151 @@ describe("connection approval", () => { const portB = portA + 100; const storeA = new MeshStore(portA); - storeA.onDelivery = () => {}; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - const toolA = new CommsTool(storeA); - - await sleep(100); - + wireTestTransport(storeA); const storeB = new MeshStore(portB); - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - const toolB = new CommsTool(storeB); - - // Use buildAction to construct the mesh_connect action - const connectAction = buildAction({ - action: "mesh_connect", - host: "127.0.0.1", - port: portA, - }); - - // Give coordinator time to settle - await sleep(200); - - // Verify coordinator is listening - const listeners = storeA.listListeners(); - assert.ok( - listeners.length > 0, - `Coordinator should have listeners, got ${listeners.length}`, - ); - assert.equal( - listeners[0]?.port, - portA, - `Coordinator should be on port ${portA}`, - ); - - // B initiates the connection via the tool - const connectResult = await toolB.handle( - { - agentId: storeB.peerId, - harness: "test", - cwd: "/test/b", - pid: process.pid, - }, - connectAction, - ); - assert.ok( - !connectResult.isError, - `mesh_connect should succeed: ${connectResult.content}`, - ); - - await sleep(300); - - // A checks pending connections via the tool - const pendingAction = buildAction({ action: "mesh_pending" }); - const pendingResult = await toolA.handle( - { - agentId: storeA.peerId, + wireTestTransport(storeB); + try { + storeA.onDelivery = () => {}; + await storeA.init(); + await storeA.registerAgent({ + name: "coordinator", harness: "test", cwd: "/test/a", pid: process.pid, - }, - pendingAction, - ); - assert.ok( - !pendingResult.isError, - `mesh_pending should succeed: ${pendingResult.content}`, - ); - assert.ok( - pendingResult.content.includes("connector"), - "Pending list should show connector name", - ); - - // Extract connectionId from the pending connections list - const pendingConns = storeA.listPendingConnections(); - assert.equal(pendingConns.length, 1, "Should have one pending connection"); - const connectionId = pendingConns[0]?.connectionId; - assert.ok(connectionId); - - // A accepts the connection via the tool - const acceptAction = buildAction({ - action: "mesh_accept", - connectionId, - }); - const acceptResult = await toolA.handle( - { - agentId: storeA.peerId, + visibility: "visible", + tags: [], + }); + const toolA = new CommsTool(storeA); + + await sleep(100); + + await storeB.startDataServerOnly(); + await storeB.registerAgent({ + name: "connector", harness: "test", - cwd: "/test/a", + cwd: "/test/b", pid: process.pid, - }, - acceptAction, - ); - assert.ok( - !acceptResult.isError, - `mesh_accept should succeed: ${acceptResult.content}`, - ); - - await sleep(200); - - // Verify both see each other - const agentsA = await storeA.listAgents(storeA.peerId); - const agentsB = await storeB.listAgents(storeB.peerId); - assert.ok( - agentsA.some((a) => a.id === storeB.peerId), - "A should see B after accept", - ); - assert.ok( - agentsB.some((a) => a.id === storeA.peerId), - "B should see A after accept", - ); - - await storeB.shutdown(); - await storeA.shutdown(); + visibility: "visible", + tags: [], + }); + const toolB = new CommsTool(storeB); + + // Use buildAction to construct the mesh_connect action + const connectAction = buildAction({ + action: "mesh_connect", + host: "127.0.0.1", + port: portA, + }); + + // Give coordinator time to settle + await sleep(200); + + // Verify coordinator is listening + const listeners = storeA.listListeners(); + assert.ok( + listeners.length > 0, + `Coordinator should have listeners, got ${listeners.length}`, + ); + assert.equal( + listeners[0]?.port, + portA, + `Coordinator should be on port ${portA}`, + ); + + // B initiates the connection via the tool + const connectResult = await toolB.handle( + { + agentId: storeB.peerId, + harness: "test", + cwd: "/test/b", + pid: process.pid, + }, + connectAction, + ); + assert.ok( + !connectResult.isError, + `mesh_connect should succeed: ${connectResult.content}`, + ); + + await waitFor( + () => storeA.listPendingConnections().length === 1, + "coordinator sees the pending connection", + ); + + // A checks pending connections via the tool + const pendingAction = buildAction({ action: "mesh_pending" }); + const pendingResult = await toolA.handle( + { + agentId: storeA.peerId, + harness: "test", + cwd: "/test/a", + pid: process.pid, + }, + pendingAction, + ); + assert.ok( + !pendingResult.isError, + `mesh_pending should succeed: ${pendingResult.content}`, + ); + assert.ok( + pendingResult.content.includes("connector"), + "Pending list should show connector name", + ); + + // Extract connectionId from the pending connections list + const pendingConns = storeA.listPendingConnections(); + assert.equal( + pendingConns.length, + 1, + "Should have one pending connection", + ); + const connectionId = pendingConns[0]?.connectionId; + assert.ok(connectionId); + + // A accepts the connection via the tool + const acceptAction = buildAction({ + action: "mesh_accept", + connectionId, + }); + const acceptResult = await toolA.handle( + { + agentId: storeA.peerId, + harness: "test", + cwd: "/test/a", + pid: process.pid, + }, + acceptAction, + ); + assert.ok( + !acceptResult.isError, + `mesh_accept should succeed: ${acceptResult.content}`, + ); + + // State sync is a real TLS round trip, not instantaneous -- poll rather than assume any fixed delay is enough under a loaded CI runner. + await waitFor( + () => storeA.serialise().agents[storeB.peerId] !== undefined, + "A sees B after accept", + ); + await waitFor( + () => storeB.serialise().agents[storeA.peerId] !== undefined, + "B sees A after accept", + ); + + // Verify both see each other + const agentsA = await storeA.listAgents(storeA.peerId); + const agentsB = await storeB.listAgents(storeB.peerId); + assert.ok( + agentsA.some((a) => a.id === storeB.peerId), + "A should see B after accept", + ); + assert.ok( + agentsB.some((a) => a.id === storeA.peerId), + "B should see A after accept", + ); + } finally { + await storeB.shutdown(); + await storeA.shutdown(); + } }); void test("tool mesh_reject returns error message", async () => { @@ -463,69 +506,76 @@ describe("connection approval", () => { const portB = portA + 100; const storeA = new MeshStore(portA); - storeA.onDelivery = () => {}; - await storeA.init(); - await storeA.registerAgent({ - name: "coordinator", - harness: "test", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - const toolA = new CommsTool(storeA); - - await sleep(100); - + wireTestTransport(storeA); const storeB = new MeshStore(portB); - await storeB.startDataServerOnly(); - await storeB.registerAgent({ - name: "connector", - harness: "test", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - // B connects - storeB.connectToRemote("127.0.0.1", portA); - - await sleep(300); + wireTestTransport(storeB); + try { + storeA.onDelivery = () => {}; + await storeA.init(); + await storeA.registerAgent({ + name: "coordinator", + harness: "test", + cwd: "/test/a", + pid: process.pid, + visibility: "visible", + tags: [], + }); + const toolA = new CommsTool(storeA); - const pendingConns = storeA.listPendingConnections(); - const connectionId = pendingConns[0]?.connectionId; - assert.ok(connectionId); + await sleep(100); - // A rejects via the tool - const rejectAction = buildAction({ - action: "mesh_reject", - connectionId, - reason: "not allowed", - }); - const rejectResult = await toolA.handle( - { - agentId: storeA.peerId, + await storeB.startDataServerOnly(); + await storeB.registerAgent({ + name: "connector", harness: "test", - cwd: "/test/a", + cwd: "/test/b", pid: process.pid, - }, - rejectAction, - ); - assert.ok( - !rejectResult.isError, - `mesh_reject should succeed: ${rejectResult.content}`, - ); - assert.ok( - rejectResult.content.includes("not allowed"), - "Result should include the reason", - ); - - // Verify rejection took effect - const pending = storeA.listPendingConnections(); - assert.equal(pending.length, 0, "No pending connections after reject"); - - await storeB.shutdown(); - await storeA.shutdown(); + visibility: "visible", + tags: [], + }); + + // B connects + storeB.connectToRemote("127.0.0.1", portA); + + await waitFor( + () => storeA.listPendingConnections().length === 1, + "coordinator sees the pending connection", + ); + + const pendingConns = storeA.listPendingConnections(); + const connectionId = pendingConns[0]?.connectionId; + assert.ok(connectionId); + + // A rejects via the tool + const rejectAction = buildAction({ + action: "mesh_reject", + connectionId, + reason: "not allowed", + }); + const rejectResult = await toolA.handle( + { + agentId: storeA.peerId, + harness: "test", + cwd: "/test/a", + pid: process.pid, + }, + rejectAction, + ); + assert.ok( + !rejectResult.isError, + `mesh_reject should succeed: ${rejectResult.content}`, + ); + assert.ok( + rejectResult.content.includes("not allowed"), + "Result should include the reason", + ); + + // Verify rejection took effect + const pending = storeA.listPendingConnections(); + assert.equal(pending.length, 0, "No pending connections after reject"); + } finally { + await storeB.shutdown(); + await storeA.shutdown(); + } }); }); diff --git a/src/test/become-coordinator-actual-port.integration.test.ts b/src/test/become-coordinator-actual-port.integration.test.ts index 734524c..be240a7 100644 --- a/src/test/become-coordinator-actual-port.integration.test.ts +++ b/src/test/become-coordinator-actual-port.integration.test.ts @@ -4,11 +4,9 @@ * Run: node dist/test/become-coordinator-actual-port.integration.test.js [test-name] With no argument, every scenario runs in order. */ -import * as net from "node:net"; import * as tls from "node:tls"; import * as assert from "node:assert/strict"; import { TlsTransport } from "../core/tls-transport.js"; -import { TcpTransport } from "../core/tcp-transport.js"; import { generateIdentity } from "../core/identity.js"; import type { TransportEvents } from "../core/transport.js"; @@ -51,31 +49,6 @@ async function testTlsReportsActualPort(): Promise { await transport.shutdown(); } -async function testTcpReportsActualPort(): Promise { - const transport = new TcpTransport(noopEvents()); - - await transport.becomeCoordinator("127.0.0.1", 0); - const [listener] = transport.listListeners(); - assert.ok(listener); - assert.notStrictEqual(listener.port, 0); - - await new Promise((resolve, reject) => { - const socket = net.connect( - { host: "127.0.0.1", port: listener.port }, - () => { - socket.destroy(); - resolve(); - }, - ); - socket.once("error", reject); - }); - console.log( - ` ✓ TcpTransport reported and bound the same port (${listener.port})`, - ); - - await transport.shutdown(); -} - // --------------------------------------------------------------------------- // Runner // --------------------------------------------------------------------------- @@ -84,7 +57,6 @@ const testName = process.argv[2]; const tests: Record Promise> = { "tls-reports-actual-port": testTlsReportsActualPort, - "tcp-reports-actual-port": testTcpReportsActualPort, }; const selected = diff --git a/src/test/coordinator-socket-error.integration.test.ts b/src/test/coordinator-socket-error.integration.test.ts index 83dcdb0..791315a 100644 --- a/src/test/coordinator-socket-error.integration.test.ts +++ b/src/test/coordinator-socket-error.integration.test.ts @@ -9,6 +9,7 @@ import * as net from "node:net"; import { MeshStore } from "../core/mesh-store.js"; import * as assert from "node:assert/strict"; import { test } from "node:test"; +import { wireTestTransport } from "./test-transport.js"; const TEST_PORT = 19879; @@ -22,6 +23,7 @@ const TEST_PORT = 19879; */ void test("coordinator survives ECONNRESET on accepted socket", async () => { const store = new MeshStore(TEST_PORT); + wireTestTransport(store); await store.init(); // Connect a raw socket to the coordinator port diff --git a/src/test/delivery-receipt.helper.ts b/src/test/delivery-receipt.helper.ts index 200d878..d1c33ff 100644 --- a/src/test/delivery-receipt.helper.ts +++ b/src/test/delivery-receipt.helper.ts @@ -17,6 +17,7 @@ import { MeshStore } from "../core/mesh-store.js"; import type { DeliveryEvent } from "../core/types.js"; import * as net from "node:net"; import assert from "node:assert/strict"; +import { wireTestTransport } from "./test-transport.js"; const testName = process.argv[2]; if (testName === undefined) { @@ -62,6 +63,7 @@ async function cleanup(...stores: MeshStore[]): Promise { async function testPushRoom(): Promise { const port = await allocFreePort(); const a = new MeshStore(port); + wireTestTransport(a); const deliveriesA: unknown[] = []; a.onDelivery = () => { deliveriesA.push(1); @@ -78,6 +80,7 @@ async function testPushRoom(): Promise { await sleep(100); const b = new MeshStore(port); + wireTestTransport(b); const deliveriesB: DeliveryEvent[] = []; b.onDelivery = (_id: string, ev: DeliveryEvent) => { deliveriesB.push(ev); @@ -125,6 +128,7 @@ async function testPushRoom(): Promise { async function testPushDm(): Promise { const port = await allocFreePort(); const a = new MeshStore(port); + wireTestTransport(a); const deliveriesA: unknown[] = []; a.onDelivery = () => { deliveriesA.push(1); @@ -141,6 +145,7 @@ async function testPushDm(): Promise { await sleep(100); const b = new MeshStore(port); + wireTestTransport(b); const deliveriesB: DeliveryEvent[] = []; b.onDelivery = (_id: string, ev: DeliveryEvent) => { deliveriesB.push(ev); @@ -175,6 +180,7 @@ async function testPushDm(): Promise { async function testDrainRoom(): Promise { const port = await allocFreePort(); const a = new MeshStore(port); + wireTestTransport(a); await a.init(); await a.registerAgent({ name: "a", @@ -188,6 +194,7 @@ async function testDrainRoom(): Promise { // B has NO onDelivery — events queue for drain const b = new MeshStore(port); + wireTestTransport(b); await b.init(); await b.registerAgent({ name: "b", @@ -237,6 +244,7 @@ async function testDrainRoom(): Promise { async function testDrainDm(): Promise { const port = await allocFreePort(); const a = new MeshStore(port); + wireTestTransport(a); await a.init(); await a.registerAgent({ name: "a", @@ -249,6 +257,7 @@ async function testDrainDm(): Promise { await sleep(100); const b = new MeshStore(port); + wireTestTransport(b); await b.init(); await b.registerAgent({ name: "b", @@ -279,6 +288,7 @@ async function testDrainDm(): Promise { async function testReadReceiptPush(): Promise { const port = await allocFreePort(); const a = new MeshStore(port); + wireTestTransport(a); const deliveriesA: DeliveryEvent[] = []; a.onDelivery = (_id: string, ev: DeliveryEvent) => { deliveriesA.push(ev); @@ -295,6 +305,7 @@ async function testReadReceiptPush(): Promise { await sleep(100); const b = new MeshStore(port); + wireTestTransport(b); b.onDelivery = () => { /* intentionally empty — dummy handler for drain delivery */ }; @@ -336,6 +347,7 @@ async function testReadReceiptPush(): Promise { async function testReadReceiptDrain(): Promise { const port = await allocFreePort(); const a = new MeshStore(port); + wireTestTransport(a); const deliveriesA: DeliveryEvent[] = []; a.onDelivery = (_id: string, ev: DeliveryEvent) => { deliveriesA.push(ev); @@ -353,6 +365,7 @@ async function testReadReceiptDrain(): Promise { // B has NO onDelivery — drain triggers markRead const b = new MeshStore(port); + wireTestTransport(b); await b.init(); await b.registerAgent({ name: "b", @@ -396,6 +409,7 @@ async function testReadReceiptDrain(): Promise { async function testReadbyArray(): Promise { const port = await allocFreePort(); const a = new MeshStore(port); + wireTestTransport(a); a.onDelivery = () => { /* intentionally empty — dummy handler for drain delivery */ }; @@ -411,6 +425,7 @@ async function testReadbyArray(): Promise { await sleep(100); const b = new MeshStore(port); + wireTestTransport(b); b.onDelivery = () => { /* intentionally empty — dummy handler for drain delivery */ }; diff --git a/src/test/downtime-replay.test.ts b/src/test/downtime-replay.test.ts index 939135f..fd3afc9 100644 --- a/src/test/downtime-replay.test.ts +++ b/src/test/downtime-replay.test.ts @@ -7,15 +7,18 @@ 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"; +import { wireTestTransport } from "./test-transport.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. */ +/** A local-only store: transport is set (registerAgent's own broadcastPatch needs one) but never started, so no ports and no flake -- the stores in these tests never actually connect. */ function makeStore(): MeshStore { - return new MeshStore(); + const store = new MeshStore(); + wireTestTransport(store); + return store; } void test("events queued while the target was down replay on its first snapshot", async () => { diff --git a/src/test/federation.integration.test.ts b/src/test/federation.integration.test.ts index 2fb5e64..ee94e68 100644 --- a/src/test/federation.integration.test.ts +++ b/src/test/federation.integration.test.ts @@ -17,6 +17,7 @@ import { MeshStore } from "../core/mesh-store.js"; import type { DeliveryEvent, RoomMessage } from "../core/types.js"; import * as assert from "node:assert/strict"; import * as net from "node:net"; +import { wireTestTransport } from "./test-transport.js"; // Use high ports to avoid collisions with real meshes const MESH_A_PORT = 28876; @@ -38,6 +39,7 @@ async function createMesh( deliveries: DeliveryEvent[]; }> { const store = new MeshStore(coordinatorPort); + wireTestTransport(store); const deliveries: DeliveryEvent[] = []; store.onDelivery = (_agentId: string, event: DeliveryEvent) => { deliveries.push(event); diff --git a/src/test/listener-policy.integration.test.ts b/src/test/listener-policy.integration.test.ts index 66ccda3..b7534bd 100644 --- a/src/test/listener-policy.integration.test.ts +++ b/src/test/listener-policy.integration.test.ts @@ -7,259 +7,289 @@ */ import * as net from "node:net"; +import * as tls from "node:tls"; import { MeshStore } from "../core/mesh-store.js"; import { CommsTool } from "../core/tool.js"; import { buildAction } from "../core/bridge.js"; +import { generateIdentity } from "../core/identity.js"; import type { ConnectionHandle, TransportEvents } from "../core/transport.js"; import * as assert from "node:assert/strict"; import { test, describe } from "node:test"; +import { wireTestTransport } from "./test-transport.js"; const TEST_PORT = 19880; describe("listener policy", () => { void test("coordinator starts with a single default localhost listener", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const listeners = store.listListeners(); - assert.equal(listeners.length, 1, "Should have exactly one listener"); - assert.equal( - listeners[0]?.policy, - "full", - "Default listener should have full policy", - ); - assert.equal( - listeners[0]?.isDefault, - true, - "Default listener should be marked as default", - ); - assert.equal( - listeners[0]?.host, - "127.0.0.1", - "Default listener should be on localhost", - ); - assert.equal( - listeners[0]?.port, - TEST_PORT, - "Default listener should be on the coordinator port", - ); - - await store.shutdown(); + wireTestTransport(store); + try { + await store.init(); + + const listeners = store.listListeners(); + assert.equal(listeners.length, 1, "Should have exactly one listener"); + assert.equal( + listeners[0]?.policy, + "full", + "Default listener should have full policy", + ); + assert.equal( + listeners[0]?.isDefault, + true, + "Default listener should be marked as default", + ); + assert.equal( + listeners[0]?.host, + "127.0.0.1", + "Default listener should be on localhost", + ); + assert.equal( + listeners[0]?.port, + TEST_PORT, + "Default listener should be on the coordinator port", + ); + } finally { + await store.shutdown(); + } }); void test("addListener creates an additional listener", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const id = await store.addListener("127.0.0.1", 0, "observe"); - assert.ok(id, "Should return a listener ID"); - - const listeners = store.listListeners(); - assert.equal(listeners.length, 2, "Should have two listeners"); - - const newListener = listeners.find((l) => l.id === id); - assert.ok(newListener, "New listener should be listed"); - assert.equal( - newListener.policy, - "observe", - "New listener should have observe policy", - ); - assert.equal( - newListener.isDefault, - false, - "New listener should not be default", - ); - assert.ok(newListener.port > 0, "Should have an assigned port"); - - await store.shutdown(); + wireTestTransport(store); + try { + await store.init(); + + const id = await store.addListener("127.0.0.1", 0, "observe"); + assert.ok(id, "Should return a listener ID"); + + const listeners = store.listListeners(); + assert.equal(listeners.length, 2, "Should have two listeners"); + + const newListener = listeners.find((l) => l.id === id); + assert.ok(newListener, "New listener should be listed"); + assert.equal( + newListener.policy, + "observe", + "New listener should have observe policy", + ); + assert.equal( + newListener.isDefault, + false, + "New listener should not be default", + ); + assert.ok(newListener.port > 0, "Should have an assigned port"); + } finally { + await store.shutdown(); + } }); void test("removeListener removes a non-default listener", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const id = await store.addListener("127.0.0.1", 0, "observe"); - assert.equal(store.listListeners().length, 2); - - await store.removeListener(id); - const listeners = store.listListeners(); - assert.equal(listeners.length, 1, "Should be back to one listener"); - assert.equal( - listeners[0]?.isDefault, - true, - "Remaining listener should be the default", - ); - - await store.shutdown(); + wireTestTransport(store); + try { + await store.init(); + + const id = await store.addListener("127.0.0.1", 0, "observe"); + assert.equal(store.listListeners().length, 2); + + await store.removeListener(id); + const listeners = store.listListeners(); + assert.equal(listeners.length, 1, "Should be back to one listener"); + assert.equal( + listeners[0]?.isDefault, + true, + "Remaining listener should be the default", + ); + } finally { + await store.shutdown(); + } }); void test("removeListener rejects removing the default listener", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const listeners = store.listListeners(); - const defaultId = listeners[0]?.id; - assert.ok(defaultId, "Should have a default listener"); - - await assert.rejects( - () => store.removeListener(defaultId), - /Cannot remove the default/, - "Should reject removing default listener", - ); - - await store.shutdown(); + wireTestTransport(store); + try { + await store.init(); + + const listeners = store.listListeners(); + const defaultId = listeners[0]?.id; + assert.ok(defaultId, "Should have a default listener"); + + await assert.rejects( + () => store.removeListener(defaultId), + /Cannot remove the default/, + "Should reject removing default listener", + ); + } finally { + await store.shutdown(); + } }); void test("observe listener accepts connections but enforces policy", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const id = await store.addListener("127.0.0.1", 0, "observe"); - const listeners = store.listListeners(); - const observeListener = listeners.find((l) => l.id === id); - assert.ok(observeListener, "Observe listener should exist"); - - // Verify a peer can connect to the observe listener's port - const canConnect = await new Promise((resolve) => { - const socket = net.createConnection({ - port: observeListener.port, - host: "127.0.0.1", + wireTestTransport(store); + try { + await store.init(); + + const id = await store.addListener("127.0.0.1", 0, "observe"); + const listeners = store.listListeners(); + const observeListener = listeners.find((l) => l.id === id); + assert.ok(observeListener, "Observe listener should exist"); + + // Verify a peer can connect to the observe listener's port + const canConnect = await new Promise((resolve) => { + const socket = net.createConnection({ + port: observeListener.port, + host: "127.0.0.1", + }); + socket.on("connect", () => { + socket.destroy(); + resolve(true); + }); + socket.on("error", () => { + socket.destroy(); + resolve(false); + }); }); - socket.on("connect", () => { - socket.destroy(); - resolve(true); - }); - socket.on("error", () => { - socket.destroy(); - resolve(false); - }); - }); - assert.ok(canConnect, "Should be able to connect to observe listener"); - - await store.shutdown(); + assert.ok(canConnect, "Should be able to connect to observe listener"); + } finally { + await store.shutdown(); + } }); void test("mesh_listeners action returns all listeners via CommsTool", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const agent = await store.registerAgent({ - name: "test-agent", - harness: "test", - cwd: "/test", - pid: process.pid, - visibility: "visible", - tags: [], - }); + wireTestTransport(store); + try { + await store.init(); + + const agent = await store.registerAgent({ + name: "test-agent", + harness: "test", + cwd: "/test", + pid: process.pid, + visibility: "visible", + tags: [], + }); - await store.addListener("127.0.0.1", 0, "rooms-only"); - - const tool = new CommsTool(store); - const result = await tool.handle( - { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, - { action: "mesh_listeners" }, - ); - - assert.ok(!result.isError, "Should not be an error"); - assert.ok( - result.content.includes("full"), - "Should list the default full listener", - ); - assert.ok( - result.content.includes("rooms-only"), - "Should list the rooms-only listener", - ); - - await store.shutdown(); + await store.addListener("127.0.0.1", 0, "rooms-only"); + + const tool = new CommsTool(store); + const result = await tool.handle( + { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, + { action: "mesh_listeners" }, + ); + + assert.ok(!result.isError, "Should not be an error"); + assert.ok( + result.content.includes("full"), + "Should list the default full listener", + ); + assert.ok( + result.content.includes("rooms-only"), + "Should list the rooms-only listener", + ); + } finally { + await store.shutdown(); + } }); void test("mesh_interfaces action returns available network adapters", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const agent = await store.registerAgent({ - name: "test-agent", - harness: "test", - cwd: "/test", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - const tool = new CommsTool(store); - const result = await tool.handle( - { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, - { action: "mesh_interfaces" }, - ); - - assert.ok(!result.isError, "Should not be an error"); - assert.ok( - result.content.includes("lo") || result.content.includes("IPv4"), - "Should list network interfaces", - ); + wireTestTransport(store); + try { + await store.init(); + + const agent = await store.registerAgent({ + name: "test-agent", + harness: "test", + cwd: "/test", + pid: process.pid, + visibility: "visible", + tags: [], + }); - await store.shutdown(); + const tool = new CommsTool(store); + const result = await tool.handle( + { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, + { action: "mesh_interfaces" }, + ); + + assert.ok(!result.isError, "Should not be an error"); + assert.ok( + result.content.includes("lo") || result.content.includes("IPv4"), + "Should list network interfaces", + ); + } finally { + await store.shutdown(); + } }); void test("mesh_unlisten removes listener via CommsTool", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const agent = await store.registerAgent({ - name: "test-agent", - harness: "test", - cwd: "/test", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - const listenerId = await store.addListener("127.0.0.1", 0, "observe"); + wireTestTransport(store); + try { + await store.init(); + + const agent = await store.registerAgent({ + name: "test-agent", + harness: "test", + cwd: "/test", + pid: process.pid, + visibility: "visible", + tags: [], + }); - const tool = new CommsTool(store); - const result = await tool.handle( - { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, - { action: "mesh_unlisten", id: listenerId }, - ); + const listenerId = await store.addListener("127.0.0.1", 0, "observe"); - assert.ok(!result.isError, "Should not be an error"); - assert.ok(result.content.includes("removed"), "Should confirm removal"); + const tool = new CommsTool(store); + const result = await tool.handle( + { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, + { action: "mesh_unlisten", id: listenerId }, + ); - const listeners = store.listListeners(); - assert.equal(listeners.length, 1, "Should be back to one listener"); + assert.ok(!result.isError, "Should not be an error"); + assert.ok(result.content.includes("removed"), "Should confirm removal"); - await store.shutdown(); + const listeners = store.listListeners(); + assert.equal(listeners.length, 1, "Should be back to one listener"); + } finally { + await store.shutdown(); + } }); void test("mesh_listen adds listener via CommsTool", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - const agent = await store.registerAgent({ - name: "test-agent", - harness: "test", - cwd: "/test", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - const tool = new CommsTool(store); - const result = await tool.handle( - { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, - { action: "mesh_listen", host: "127.0.0.1", policy: "observe" }, - ); - - assert.ok(!result.isError, "Should not be an error"); - assert.ok( - result.content.includes("Listener added"), - "Should confirm addition", - ); - - const listeners = store.listListeners(); - assert.equal(listeners.length, 2, "Should have two listeners"); + wireTestTransport(store); + try { + await store.init(); + + const agent = await store.registerAgent({ + name: "test-agent", + harness: "test", + cwd: "/test", + pid: process.pid, + visibility: "visible", + tags: [], + }); - await store.shutdown(); + const tool = new CommsTool(store); + const result = await tool.handle( + { agentId: agent.id, harness: "test", cwd: "/test", pid: process.pid }, + { action: "mesh_listen", host: "127.0.0.1", policy: "observe" }, + ); + + assert.ok(!result.isError, "Should not be an error"); + assert.ok( + result.content.includes("Listener added"), + "Should confirm addition", + ); + + const listeners = store.listListeners(); + assert.equal(listeners.length, 2, "Should have two listeners"); + } finally { + await store.shutdown(); + } }); void test("buildAction parses mesh_listen with host and policy", () => { @@ -300,68 +330,73 @@ describe("listener policy", () => { void test("connections via non-default listener carry policy in handle", async () => { const store = new MeshStore(TEST_PORT); - await store.init(); - - // Add an observe listener - const listenerId = await store.addListener("127.0.0.1", 0, "observe"); - const listeners = store.listListeners(); - const observeListener = listeners.find((l) => l.id === listenerId); - assert.ok(observeListener, "Should find the observe listener"); - - // Connect to the observe listener and send an introduce message - // The transport should tag the connection handle with policy="observe" - // - // We intercept at the transport.events level because store.events - // is a getter that creates a fresh object each call. - const receivedHandle = await new Promise<{ - policy: string | undefined; - } | null>((resolve) => { - const timeout = setTimeout(() => resolve(null), 3000); - - const transport = ( - store as unknown as { transport: { events: TransportEvents } } - ).transport; - const originalOnIntroduction = ( - handle: ConnectionHandle, - msg: { peerId: string; dataPort: number }, - ) => transport.events.onIntroduction(handle, msg); - transport.events.onIntroduction = (handle, msg) => { - // Capture the handle's policy - resolve({ policy: handle.policy }); - clearTimeout(timeout); - // Call the original handler - originalOnIntroduction(handle, msg); - }; - - const socket = net.createConnection({ - port: observeListener.port, - host: "127.0.0.1", - }); - - socket.on("connect", () => { - socket.write( - JSON.stringify({ - method: "introduce", - peerId: "test-observe-peer", - dataPort: 19999, - }) + "\n", + wireTestTransport(store); + try { + await store.init(); + + // Add an observe listener + const listenerId = await store.addListener("127.0.0.1", 0, "observe"); + const listeners = store.listListeners(); + const observeListener = listeners.find((l) => l.id === listenerId); + assert.ok(observeListener, "Should find the observe listener"); + + // Connect to the observe listener and send an introduce message over a real TLS client connection -- the listener is a TlsTransport server, which requires an actual TLS handshake before any application bytes are readable, and separately verifies the introduce message's claimed peerId against the client certificate's own fingerprint, so the probe needs a real generated identity, not an arbitrary string. The transport should tag the connection handle with policy="observe" + // + // We intercept at the transport.events level because store.events is a getter that creates a fresh object each call. + const probeIdentity = generateIdentity(); + const receivedHandle = await new Promise<{ + policy: string | undefined; + } | null>((resolve) => { + const timeout = setTimeout(() => resolve(null), 3000); + + const transport = ( + store as unknown as { transport: { events: TransportEvents } } + ).transport; + // Captured by value (bound, so eslint doesn't flag an unsafely-detached method reference), not by a closure that re-reads transport.events.onIntroduction at call time -- a lazy re-lookup would resolve to the wrapper itself once the assignment below replaces it, recursing forever on the very first introduction. + const originalOnIntroduction = transport.events.onIntroduction.bind( + transport.events, ); + transport.events.onIntroduction = (handle, msg) => { + // Capture the handle's policy + resolve({ policy: handle.policy }); + clearTimeout(timeout); + // Call the original handler + originalOnIntroduction(handle, msg); + }; + + const socket = tls.connect({ + port: observeListener.port, + host: "127.0.0.1", + key: probeIdentity.privateKey, + cert: probeIdentity.certificate, + rejectUnauthorized: false, + }); + + socket.on("secureConnect", () => { + socket.write( + JSON.stringify({ + method: "introduce", + peerId: probeIdentity.fingerprint, + dataPort: 19999, + }) + "\n", + ); + }); + + socket.on("error", () => { + socket.destroy(); + clearTimeout(timeout); + resolve(null); + }); }); - socket.on("error", () => { - socket.destroy(); - clearTimeout(timeout); - resolve(null); - }); - }); - - assert.ok(receivedHandle, "Should receive an introduction"); - assert.equal( - receivedHandle.policy, - "observe", - "Handle should carry observe policy", - ); - - await store.shutdown(); + assert.ok(receivedHandle, "Should receive an introduction"); + assert.equal( + receivedHandle.policy, + "observe", + "Handle should carry observe policy", + ); + } finally { + await store.shutdown(); + } }); }); diff --git a/src/test/mesh-e2e.integration.test.ts b/src/test/mesh-e2e.integration.test.ts index f5a8786..ce82ff8 100644 --- a/src/test/mesh-e2e.integration.test.ts +++ b/src/test/mesh-e2e.integration.test.ts @@ -10,6 +10,7 @@ import { CommsTool } from "../core/tool.js"; import { buildAction } from "../core/bridge.js"; import type { DeliveryEvent } from "../core/types.js"; import * as assert from "node:assert/strict"; +import { wireTestTransport } from "./test-transport.js"; const E2E_PORT = 19878; @@ -18,6 +19,7 @@ async function createStore( harness: string, ): Promise<{ store: MeshStore; tool: CommsTool; deliveries: DeliveryEvent[] }> { const store = new MeshStore(E2E_PORT); + wireTestTransport(store); const deliveries: DeliveryEvent[] = []; store.onDelivery = (_agentId: string, event: DeliveryEvent) => { diff --git a/src/test/mesh-smoke.integration.test.ts b/src/test/mesh-smoke.integration.test.ts index 39b9d8a..9caf51c 100644 --- a/src/test/mesh-smoke.integration.test.ts +++ b/src/test/mesh-smoke.integration.test.ts @@ -1,9 +1,7 @@ /** - * Multi-process smoke test for MeshStore TCP mesh. + * Multi-process smoke test for the MeshStore mesh. * - * Spawns two separate Node.js processes running MeshStore instances, - * verifies they discover each other via the coordinator, exchange - * messages, and receive push delivery. + * Spawns two separate Node.js processes running MeshStore instances over TlsTransport, verifies they discover each other via the coordinator, exchange messages, and receive push delivery. */ import * as assert from "node:assert/strict"; @@ -23,6 +21,9 @@ interface PeerHandle { exit: Promise; } +// A hung child (a mesh operation that never resolves) must not hang this process indefinitely -- the child is killed and the promise rejects with a clear reason instead of Promise.all([...]) waiting forever. +const PEER_EXIT_TIMEOUT_MS = 15_000; + // ----------------------------------------------------------------------- // Child process peer // ----------------------------------------------------------------------- @@ -53,11 +54,30 @@ function spawnPeer(name: string, actions: string): PeerHandle { const exit = new Promise((resolve, reject) => { let stderr = ""; + let settled = false; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + child.kill(); + reject( + new Error( + `${name} did not exit within ${String(PEER_EXIT_TIMEOUT_MS)}ms:\n${stderr}`, + ), + ); + }, PEER_EXIT_TIMEOUT_MS); child.stderr.on("data", (data: Buffer) => { stderr += data.toString(); }); - child.on("error", reject); + child.on("error", (err) => { + if (settled) return; + settled = true; + clearTimeout(timer); + reject(err); + }); child.on("exit", (code) => { + if (settled) return; + settled = true; + clearTimeout(timer); if (buffer.trim()) { try { const parsed: unknown = JSON.parse(buffer); @@ -87,9 +107,14 @@ function buildScript(name: string, actions: string): string { return [ `const { MeshStore } = require("./dist/core/mesh-store.js");`, `const { CommsTool } = require("./dist/core/tool.js");`, + `const { TlsTransport } = require("./dist/core/tls-transport.js");`, + `const { generateIdentity } = require("./dist/core/identity.js");`, `function log(msg) { process.stdout.write(JSON.stringify(msg) + "\\n"); }`, `(async () => {`, ` const store = new MeshStore(${String(SMOKE_PORT)});`, + ` const identity = generateIdentity();`, + ` store.peerId = identity.fingerprint;`, + ` store.setTransport(new TlsTransport(store.events, identity));`, ` const tool = new CommsTool(store);`, ` const deliveries = [];`, ` store.onDelivery = (_id, event) => {`, diff --git a/src/test/state-sync-convergence.test.ts b/src/test/state-sync-convergence.test.ts index 3e9ffc1..f2c418a 100644 --- a/src/test/state-sync-convergence.test.ts +++ b/src/test/state-sync-convergence.test.ts @@ -8,10 +8,13 @@ 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 { wireTestTransport } from "./test-transport.js"; -/** A local-only store: no transport start, so no ports and no flake. */ +/** A local-only store: transport is set (registerAgent's own broadcastPatch needs one) but never started, so no ports and no flake -- the stores in these tests never actually connect. */ function makeStore(): MeshStore { - return new MeshStore(); + const store = new MeshStore(); + wireTestTransport(store); + return store; } function snapshotOf(store: MeshStore): SerialisedState { diff --git a/src/test/test-transport.ts b/src/test/test-transport.ts new file mode 100644 index 0000000..599035e --- /dev/null +++ b/src/test/test-transport.ts @@ -0,0 +1,41 @@ +/** Wires a real TlsTransport with a freshly generated identity onto a freshly constructed MeshStore -- the same setTransport() call every production bridge makes immediately after construction. MeshStore has no default transport, so every test that constructs one needs this (or an equivalent explicit setTransport() call) before init() or any other transport-using method runs. */ + +import { TlsTransport } from "../core/tls-transport.js"; +import { generateIdentity } from "../core/identity.js"; +import type { MeshStore } from "../core/mesh-store.js"; + +export function wireTestTransport(store: MeshStore): void { + const identity = generateIdentity(); + // Every real bridge sets peerId to the identity's own certificate fingerprint before wiring the transport -- TlsTransport's cert-pinning trust model means a peer's advertised ID and the fingerprint the other side actually authenticates the connection against must be the same value, or introduction/state-sync never recognises the peer as itself. + store.peerId = identity.fingerprint; + store.setTransport(new TlsTransport(store.events, identity)); + // Surface transport-level errors instead of leaving them silent — a genuine socket failure during a test run is signal worth seeing even when the test's own assertions still pass, since it can point at a real race the assertions don't happen to catch. + store.onError = (e) => { + console.error(`[transport error, peerId=${store.peerId}]`, e.message); + }; +} + +// Generous on purpose: waitFor returns the instant its condition holds, so a long ceiling costs nothing on the happy path (a local run settles in well under a second) and only matters for the worst case -- a loaded CI runner working through a real, sequential chain of TLS handshakes (each one genuine X.509 certificate work, not instant) for the accept-flow's second connection direction, confirmed to need meaningfully more than 5s on at least one real CI run. +const DEFAULT_WAIT_FOR_TIMEOUT_MS = 20_000; +const WAIT_FOR_POLL_INTERVAL_MS = 20; + +/** + * Polls condition() until it returns true or timeoutMs elapses, rather than a fixed sleep() before a single check. A real TLS handshake plus the peer_list -> connectToPeer -> state_sync -> handlePeerConnected round trip genuinely takes variable, load-dependent wall-clock time -- comfortably inside a fixed sleep on a fast local machine, not reliably so under a throttled CI runner. Throws with a descriptive message on timeout rather than letting the caller's own assertion fail with a less specific one. + */ +export async function waitFor( + condition: () => boolean, + description: string, + timeoutMs = DEFAULT_WAIT_FOR_TIMEOUT_MS, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!condition()) { + if (Date.now() >= deadline) { + throw new Error( + `waitFor timed out after ${String(timeoutMs)}ms: ${description}`, + ); + } + await new Promise((resolve) => + setTimeout(resolve, WAIT_FOR_POLL_INTERVAL_MS), + ); + } +} diff --git a/src/test/visibility.integration.test.ts b/src/test/visibility.integration.test.ts index e9e3aeb..5d02e9e 100644 --- a/src/test/visibility.integration.test.ts +++ b/src/test/visibility.integration.test.ts @@ -16,6 +16,7 @@ import { CommsTool } from "../core/tool.js"; import { buildAction } from "../core/bridge.js"; import { DiscoveryManager } from "../core/discovery.js"; import type { MeshVisibility } from "../core/types.js"; +import { wireTestTransport } from "./test-transport.js"; const TEST_PORT = 19881; @@ -73,6 +74,7 @@ describe("DiscoveryManager visibility", () => { describe("MeshStore visibility delegation", () => { void test("setVisibility delegates to discovery manager", async () => { const store = new MeshStore(TEST_PORT); + wireTestTransport(store); await store.init(); assert.equal(store.getVisibility(), "discoverable"); @@ -91,6 +93,7 @@ describe("MeshStore visibility delegation", () => { void test("setVisibility with adapter delegates per-adapter", async () => { const store = new MeshStore(TEST_PORT); + wireTestTransport(store); await store.init(); await store.setVisibility("quiet", "mdns"); @@ -108,6 +111,7 @@ describe("MeshStore visibility delegation", () => { describe("CommsTool visibility actions", () => { void test("mesh_set_visibility action sets visibility", async () => { const store = new MeshStore(TEST_PORT); + wireTestTransport(store); await store.init(); const agent = await store.registerAgent({ @@ -134,6 +138,7 @@ describe("CommsTool visibility actions", () => { void test("mesh_set_visibility with adapter sets per-adapter", async () => { const store = new MeshStore(TEST_PORT); + wireTestTransport(store); await store.init(); const agent = await store.registerAgent({ @@ -160,6 +165,7 @@ describe("CommsTool visibility actions", () => { void test("mesh_get_visibility returns current visibility", async () => { const store = new MeshStore(TEST_PORT); + wireTestTransport(store); await store.init(); const agent = await store.registerAgent({ diff --git a/src/test/ws-broadcast-window.integration.test.ts b/src/test/ws-broadcast-window.integration.test.ts deleted file mode 100644 index 9f2166d..0000000 --- a/src/test/ws-broadcast-window.integration.test.ts +++ /dev/null @@ -1,110 +0,0 @@ -/** - * Integration test for the WebSocket transport's broadcast queue: state patches broadcast before the WS data connections are established must be queued and flushed on registration, not dropped (#23). - * - * Mirrors broadcast-window.integration.test.ts over TlsTransport; the queue logic is implemented per transport, so each needs its own coverage. - */ - -import * as assert from "node:assert/strict"; -import { MeshStore } from "../core/mesh-store.js"; -import { WebSocketTransport } from "../core/ws-transport.js"; - -const TEST_PORT = 19892; -const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); - -/** A peer wired like a WS-based participant (browser relay worker). */ -function makePeer(): MeshStore { - const store = new MeshStore(TEST_PORT); - store.setTransport(new WebSocketTransport(store.events)); - return store; -} - -/** 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 { - // A is the coordinator and stays up throughout. - const a = makePeer(); - await a.init(); - await a.registerAgent({ - name: "peer-a", - harness: "user", - cwd: "/test/a", - pid: process.pid, - visibility: "visible", - tags: [], - }); - - // B joins and registers IMMEDIATELY after init() — no settle delay. This is exactly the pattern that used to race the dials and lose the upsert. - const b = makePeer(); - await b.init(); - await b.registerAgent({ - name: "peer-b", - harness: "user", - cwd: "/test/b", - pid: process.pid, - visibility: "visible", - tags: [], - }); - const bId = b.peerId; - - await waitFor( - "the coordinator to see the immediately-registered agent", - async () => { - const agents = await a.listAgents(a.peerId); - const seen = agents.find((agent) => agent.id === bId); - return seen?.status === "active"; - }, - ); - - await waitFor("the joining peer to see the coordinator's agent", async () => { - const agents = await b.listAgents(b.peerId); - return agents.some((agent) => agent.id === a.peerId); - }); - - // A room message from the coordinator must push to the joiner over WS. - await a.createRoom({ - name: "ws-window", - type: "public", - owner: a.peerId, - description: "ws broadcast window", - }); - await waitFor("the room to reach the joiner", async () => { - const rooms = await b.listRooms(b.peerId); - return rooms.some((room) => room.id === "ws-window"); - }); - await b.joinRoom("ws-window", b.peerId); - // Wait for the membership to reach the sender before sending: delivery is - // computed from the sender's local room state. - await waitFor("the coordinator to see the joiner in the room", async () => { - const room = await a.getRoom("ws-window"); - return room?.members.includes(bId) === true; - }); - const deliveries: string[] = []; - b.onDelivery = (_id, ev) => { - if (ev.type === "room_message") deliveries.push(ev.message.content); - }; - await a.sendRoomMessage("ws-window", a.peerId, "hello over ws"); - await waitFor("the WS room message push", async () => - deliveries.includes("hello over ws"), - ); - - await b.shutdown(); - await a.shutdown(); - console.log("✓ immediate registration and delivery work over WebSocket"); -} - -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); -});