From 0b4cf90bf1175c0c3e813724a5a95c02e445fe8f Mon Sep 17 00:00:00 2001 From: Anso Date: Sat, 16 May 2026 20:11:31 -0400 Subject: [PATCH] fix(mesh): peer-initiated callback bridge installs reverse dialer (#1071) * fix(mesh): peer-initiated callback bridge installs reverse dialer The R1-A2 symmetric-dial work added a peer-side dialer that opens a callback WS to central at `/api/mesh/proxy-tunnel-from-peer`. The WS upgrade, JWT validation, and `mesh_centrals` persistence all worked, but the peer-side handler put the wrong object on its end of the pipe: a `PilotTunnelBridge` (the central-role object) rather than a `TcpStreamSwitchboard` with a registered `reverseDialer`. The design doc states: "Central retains PilotTunnelBridge ownership; peer retains TcpStreamSwitchboard + reverseDialer ownership." The central-initiated handler at `meshProxyTunnel.ts:115-163` honors this. The peer-initiated handler at `PeerToCentralMeshSessionDialer.attachBridge` did not. It created a `PilotTunnelBridge(0, ws)` and stored it in a private `currentSession` field. Result: `MeshService.reverseDialer` stayed null, `dialMeshTcpStream` fell through to `PilotTunnelManager.ensureBridge(target.nodeId)`, and `NodeRegistry. getProxyTarget` on the peer returned null because proxy-mode peers do not enroll their central. Cross-fleet dispatch from a peer container failed with `proxy-tunnel.open.fail nodeId= reason=no_target` even though the callback bridge was up and healthy (`bridgeOpen: true`, `last_used_at` populated). Replace the peer-side `PilotTunnelBridge` with the same wiring the central-initiated handler uses: - `attachTcpStreamSwitchboard` with `resolveByComposeLabels` (allows inbound `tcp_open` from central if it ever uses the callback bridge for a reverse-direction dispatch; matches the central- initiated pipe's resolver). - A `SwitchboardReverseDialer` that delegates `openMeshTcpStream` to `switchboard.openReverseStream`. - `MeshService.setReverseDialer(localDialer, null)` with the same CAS guard the central-initiated handler uses, so a concurrent central-initiated tunnel does not get silently overwritten. - `ws.on('message', onMessage)` dispatches JSON / binary frames to the switchboard. - `ws.once('close'/'error', teardown)` clears the switchboard, the reverseDialer, and the `currentSession` reference idempotently. The `currentSession` field type changes from `PilotTunnelBridge | null` to `TcpStreamSwitchboard | null`. Callers in `MeshService.openCrossNode` only use the return value for truthiness; the cast is harmless. The public `hasSession()` signature is unchanged. A new `currentWs` field holds the WS reference so `resetForTest` can close the connection on teardown (the switchboard's own `cleanup` does not close its WS). Test update mirrors the production wiring: the existing "marks the row used on successful WS open" case now asserts the return value is a `TcpStreamSwitchboard`, that `MeshService.reverseDialer` is non-null after `ensureSession`, and explicitly closes the WS in the test's finally so `wss.close` can resolve. Pre-existing test logic that asserted `bridge instanceof PilotTunnelBridge` on the same path is removed in favor of the new switchboard assertion. Discovered during the v0.81.1 live verification: the peer-side `route.dispatch cross-node` event always paired with `proxy-tunnel.open.fail no_target` and `tunnel.fail`, even though `centralCallback.bridgeOpen: true` and `mesh_centrals.last_used_at` were populated. The callback path opened cleanly in isolation but the dispatch path never consumed it. This fix makes the dispatch path consume it. Tests: - cd backend && npx tsc --noEmit clean - cd backend && npx vitest run mesh 170/170 green - cd backend && npx vitest run 2259/2268 green; only failures are the pre-existing Windows EBUSY flake in filesystem-backup.test.ts (documented in prior handoff) * fix(mesh): drop unused imports and align reverse-dialer interface name CI lint failures on the previous push were two @typescript-eslint/no-unused-vars errors in the test file: 24:76 'vi' is defined but never used 31:5 'TcpStreamSwitchboardCtor' is assigned a value but only used as a type Both were collateral from a local-debugging simplification: vi.waitFor was swapped for a synchronous assertion when the test's wss.close hang was traced to an unclosed client WS; the toBeInstanceOf assertion that used the runtime binding was removed at the same time. Restore the instanceof assertion next to the existing not.toBeNull check so the runtime binding is in use and the test guards against future regressions where the wrong end-of-pipe object type is returned on the callback bridge (the bug this PR fixes). Drop the now-unused `vi` symbol from the vitest import. Also fold in a /simplify convergent finding: rename the local CallbackReverseDialer interface to SwitchboardReverseDialer to match the sibling declaration in meshProxyTunnel.ts:72. Same shape, same name; readers of either handler now see the same mental model. Cross-file extraction of the interface to tcpStreamSwitchboard.ts is the cleaner long-term shape but touches a file outside this PR's scope; filing as a follow-up. Considered-and-deferred findings from the /simplify pass (file as separate PRs to keep one-branch-one-concern): - extract shared attachSwitchboard helper (~40 duplicated lines between meshProxyTunnel.ts and PeerToCentralMeshSessionDialer.ts) - public TcpStreamSwitchboard.getWs accessor + drop currentWs field - public MeshService.hasReverseDialer for the test bracket-cast - shared close-reason constants (drift risk across the two handlers) Tests: - cd backend && npx tsc --noEmit clean - cd backend && npx eslint clean (lint_exit=0) - cd backend && npx vitest run mesh 170/170, dialer 8/8 green --- ...eer-to-central-mesh-session-dialer.test.ts | 49 +++++-- .../PeerToCentralMeshSessionDialer.ts | 126 +++++++++++++++--- 2 files changed, 146 insertions(+), 29 deletions(-) diff --git a/backend/src/__tests__/peer-to-central-mesh-session-dialer.test.ts b/backend/src/__tests__/peer-to-central-mesh-session-dialer.test.ts index d5f55d58..364abd05 100644 --- a/backend/src/__tests__/peer-to-central-mesh-session-dialer.test.ts +++ b/backend/src/__tests__/peer-to-central-mesh-session-dialer.test.ts @@ -20,15 +20,16 @@ */ import http from 'http'; import type { AddressInfo } from 'net'; -import { WebSocketServer } from 'ws'; -import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; +import { WebSocketServer, type WebSocket as WsClient } from 'ws'; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest'; import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; let tmpDir: string; let PeerToCentralMeshSessionDialer: typeof import('../services/PeerToCentralMeshSessionDialer').PeerToCentralMeshSessionDialer; let MeshCentralRegistry: typeof import('../services/MeshCentralRegistry').MeshCentralRegistry; let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; -let PilotTunnelBridge: typeof import('../services/PilotTunnelBridge').PilotTunnelBridge; +let TcpStreamSwitchboardCtor: typeof import('../mesh/tcpStreamSwitchboard').TcpStreamSwitchboard; +let MeshService: typeof import('../services/MeshService').MeshService; interface RejectingServer { server: http.Server; @@ -79,7 +80,8 @@ beforeAll(async () => { ({ PeerToCentralMeshSessionDialer } = await import('../services/PeerToCentralMeshSessionDialer')); ({ MeshCentralRegistry } = await import('../services/MeshCentralRegistry')); ({ DatabaseService } = await import('../services/DatabaseService')); - ({ PilotTunnelBridge } = await import('../services/PilotTunnelBridge')); + ({ TcpStreamSwitchboard: TcpStreamSwitchboardCtor } = await import('../mesh/tcpStreamSwitchboard')); + ({ MeshService } = await import('../services/MeshService')); }); afterAll(() => { @@ -207,7 +209,7 @@ describe('PeerToCentralMeshSessionDialer', () => { } }); - it('marks the row used on successful WS open', async () => { + it('marks the row used on successful WS open and installs a reverseDialer', async () => { const wss = new WebSocketServer({ noServer: true }); const srv = http.createServer(); srv.on('upgrade', (req, socket, head) => { @@ -220,7 +222,10 @@ describe('PeerToCentralMeshSessionDialer', () => { await new Promise((resolve) => srv.listen(0, '127.0.0.1', () => resolve())); const port = (srv.address() as AddressInfo).port; const url = `http://127.0.0.1:${port}`; - let bridge: InstanceType | null = null; + // Start with a clean reverseDialer slot on the local MeshService so + // setReverseDialer's CAS swap succeeds inside attachSwitchboard. + MeshService.getInstance().setReverseDialer(null); + let switchboard: InstanceType | null = null; try { MeshCentralRegistry.getInstance().upsert({ centralInstanceId: 'inst-success', @@ -229,14 +234,34 @@ describe('PeerToCentralMeshSessionDialer', () => { jwtIssuedAt: 1, jwtExpiresAt: 9999999999, }); - bridge = await PeerToCentralMeshSessionDialer.getInstance().ensureSession(); - expect(bridge).toBeInstanceOf(PilotTunnelBridge); - await vi.waitFor(() => { - expect(MeshCentralRegistry.getInstance().getActive()?.lastUsedAt ?? 0).toBeGreaterThan(0); - }); + switchboard = await PeerToCentralMeshSessionDialer.getInstance().ensureSession(); + expect(switchboard).not.toBeNull(); + // The R1-A2 design puts a TcpStreamSwitchboard on the peer end of + // this WS, not a PilotTunnelBridge (which is central's role). A + // future regression that returns the wrong shape would surface + // here before the more abstract reverseDialer-installed check. + expect(switchboard).toBeInstanceOf(TcpStreamSwitchboardCtor); expect(PeerToCentralMeshSessionDialer.getInstance().hasSession()).toBe(true); + // markUsed is called synchronously inside attachSwitchboard before + // ensureSession resolves, so the DB row reflects it immediately. + expect(MeshCentralRegistry.getInstance().getActive()?.lastUsedAt ?? 0).toBeGreaterThan(0); + // The R1-A2 wiring under test: MeshService.reverseDialer must be + // populated so MeshService.dialMeshTcpStream routes peer-side + // cross-fleet traffic through this callback bridge instead of + // falling through to PilotTunnelManager.ensureBridge(centralId), + // which has no record for central on a proxy peer. + const meshSvc = MeshService.getInstance() as unknown as { reverseDialer: unknown }; + expect(meshSvc.reverseDialer).not.toBeNull(); } finally { - try { bridge?.close(1000, 'test done'); } catch { /* ignore */ } + try { switchboard?.cleanup('test done'); } catch { /* ignore */ } + // The dialer owns the client-side WS and the switchboard doesn't + // close it on cleanup; tear it down explicitly so wss.close can + // resolve (it waits for all client connections to disconnect). + try { + const inst = PeerToCentralMeshSessionDialer.getInstance() as unknown as { currentWs: WsClient | null }; + inst.currentWs?.close(1000, 'test done'); + } catch { /* ignore */ } + MeshService.getInstance().setReverseDialer(null); await new Promise((resolve) => wss.close(() => resolve())); await new Promise((resolve) => srv.close(() => resolve())); } diff --git a/backend/src/services/PeerToCentralMeshSessionDialer.ts b/backend/src/services/PeerToCentralMeshSessionDialer.ts index 8e25cb55..5a489456 100644 --- a/backend/src/services/PeerToCentralMeshSessionDialer.ts +++ b/backend/src/services/PeerToCentralMeshSessionDialer.ts @@ -21,13 +21,28 @@ */ import { EventEmitter } from 'events'; import WebSocket from 'ws'; -import { MAX_FRAME_SIZE_BYTES } from '../pilot/protocol'; +import { + MAX_FRAME_SIZE_BYTES, + decodeBinaryFrame, + decodeJsonFrame, + wsDataToBuffer, + wsDataToString, +} from '../pilot/protocol'; import { MeshCentralRegistry } from './MeshCentralRegistry'; -import { PilotTunnelBridge } from './PilotTunnelBridge'; +import { + attachTcpStreamSwitchboard, + resolveByComposeLabels, + type TcpStreamSwitchboard, + type ReverseTcpStreamHandle, +} from '../mesh/tcpStreamSwitchboard'; import { PilotMetrics } from './PilotMetrics'; import { httpUrlToWs } from '../utils/wsUrl'; import { sanitizeForLog } from '../utils/safeLog'; +interface SwitchboardReverseDialer { + openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null; +} + const HANDSHAKE_TIMEOUT_MS = 15_000; const RATE_LIMIT_WINDOW_MS = 60_000; const RATE_LIMIT_MAX = 5; @@ -59,8 +74,9 @@ interface DialError extends Error { export class PeerToCentralMeshSessionDialer extends EventEmitter { private static instance: PeerToCentralMeshSessionDialer | null = null; - private currentSession: PilotTunnelBridge | null = null; - private inflight: Promise | null = null; + private currentSession: TcpStreamSwitchboard | null = null; + private currentWs: WebSocket | null = null; + private inflight: Promise | null = null; private recentDials: number[] = []; private endpointUnavailableUntil = 0; @@ -73,7 +89,8 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter { public static resetForTest(): void { if (this.instance) { - try { this.instance.currentSession?.close(1000, 'test reset'); } catch { /* ignore */ } + try { this.instance.currentSession?.cleanup('test reset'); } catch { /* ignore */ } + try { this.instance.currentWs?.close(1000, 'test reset'); } catch { /* ignore */ } } this.instance = null; } @@ -82,7 +99,7 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter { return this.currentSession !== null; } - public async ensureSession(): Promise { + public async ensureSession(): Promise { if (this.currentSession) return this.currentSession; if (Date.now() < this.endpointUnavailableUntil) return null; if (this.isRateLimited()) return null; @@ -97,7 +114,7 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter { return this.recentDials.length >= RATE_LIMIT_MAX; } - private async dial(): Promise { + private async dial(): Promise { const material = MeshCentralRegistry.getInstance().getActive(); if (!material) return null; this.recentDials.push(Date.now()); @@ -115,25 +132,100 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter { this.handleDialFailure(err, material.centralInstanceId); return null; } - return this.attachBridge(ws, material.centralInstanceId); + return this.attachSwitchboard(ws, material.centralInstanceId); } - private async attachBridge(ws: WebSocket, instanceId: string): Promise { - const bridge = new PilotTunnelBridge(0, ws); + /** + * Wire the peer-initiated callback WS into the local MeshService. The + * R1-A2 design puts the peer end of the bridge in TcpStreamSwitchboard + * mode (peer multiplexes streams; central side runs PilotTunnelBridge). + * Without this wiring the WS opens cleanly but MeshService.reverseDialer + * stays null, so MeshService.dialMeshTcpStream falls through to + * PilotTunnelManager.ensureBridge(centralNodeId), which has no record + * for central on a proxy-mode peer (peers do not enroll central) and + * fails with proxy-tunnel.open.fail reason=no_target. End state matches + * the v0.78.1 reverse-direction failure even though the callback bridge + * is alive. + * + * Wiring symmetric to the central-initiated handler at + * `meshProxyTunnel.ts:115-163`: + * - attachTcpStreamSwitchboard with the same compose-label resolver + * - SwitchboardReverseDialer that delegates to switchboard.openReverseStream + * - setReverseDialer(localDialer, null) with CAS so a concurrent + * central-initiated tunnel does not get silently overwritten + * - ws.on('message') dispatches JSON/binary frames to the switchboard + * - ws.on('close'/'error') tears down switchboard + clears reverseDialer + */ + private async attachSwitchboard(ws: WebSocket, instanceId: string): Promise { + let switchboard: TcpStreamSwitchboard; try { - await bridge.start(); - } catch { - try { bridge.close(1011, 'bridge start failed'); } catch { /* ignore */ } + switchboard = attachTcpStreamSwitchboard({ + ws, + resolveTarget: resolveByComposeLabels, + logLabel: 'MeshCallback', + }); + } catch (err) { + try { ws.close(1011, 'switchboard attach failed'); } catch { /* ignore */ } + PilotMetrics.increment('mesh_callback_dials_failed_total'); + console.warn(`[PeerToCentralMeshSessionDialer] attach failed: ${sanitizeForLog((err as Error).message)}`); + return null; + } + + const { MeshService } = await import('./MeshService'); + const meshService = MeshService.getInstance(); + const localDialer: SwitchboardReverseDialer = { + openMeshTcpStream(target) { + return switchboard.openReverseStream(target); + }, + }; + const installed = meshService.setReverseDialer(localDialer, null); + if (!installed) { + console.warn('[PeerToCentralMeshSessionDialer] reverse dialer already installed; rejecting concurrent callback bridge'); + switchboard.cleanup('reverse dialer already installed'); + try { ws.close(1013, 'reverse dialer already installed'); } catch { /* ignore */ } PilotMetrics.increment('mesh_callback_dials_failed_total'); return null; } - bridge.once('closed', () => { - if (this.currentSession === bridge) this.currentSession = null; + + const onMessage = (data: unknown, isBinary: boolean): void => { + try { + if (isBinary) { + const buf = wsDataToBuffer(data); + if (!buf) return; + switchboard.handleBinaryFrame(decodeBinaryFrame(buf)); + return; + } + const text = wsDataToString(data); + if (text == null) return; + switchboard.handleJsonFrame(decodeJsonFrame(text)); + } catch (err) { + console.warn(`[PeerToCentralMeshSessionDialer] malformed frame: ${sanitizeForLog((err as Error).message)}`); + } + }; + + let tornDown = false; + const teardown = (): void => { + if (tornDown) return; + tornDown = true; + ws.off('message', onMessage); + try { switchboard.cleanup('mesh callback bridge closed'); } catch { /* ignore */ } + meshService.setReverseDialer(null, localDialer); + if (this.currentSession === switchboard) this.currentSession = null; + if (this.currentWs === ws) this.currentWs = null; + }; + + ws.on('message', onMessage); + ws.once('close', teardown); + ws.once('error', (err) => { + console.warn(`[PeerToCentralMeshSessionDialer] ws error: ${sanitizeForLog(err.message)}`); + teardown(); }); - this.currentSession = bridge; + + this.currentSession = switchboard; + this.currentWs = ws; PilotMetrics.increment('mesh_central_bootstraps_total'); MeshCentralRegistry.getInstance().markUsed(instanceId); - return bridge; + return switchboard; } private awaitOpen(ws: WebSocket): Promise {