From 567e52445099cbd8e13b8386ed94c14e37cb1fc9 Mon Sep 17 00:00:00 2001 From: Anso Date: Fri, 8 May 2026 16:21:34 -0400 Subject: [PATCH] feat(mesh): bidirectional routing via tcp_open_reverse and central relay (#1003) * feat(mesh): bidirectional routing via tcp_open_reverse and central relay Phase B of the mesh redesign. Adds the reverse-direction protocol so a pilot's MeshForwarder can route cross-node traffic to central or to another pilot. Central relays pilot-to-pilot streams transparently; pilots keep their existing single outbound WS to central. Protocol additions (backend/src/pilot/protocol.ts): - TcpOpenReverseFrame { s, targetNodeId, stack, service, port } sent agent to primary. Reuses existing tcp_open_ack, tcp_close, and TcpData binary frames for the response and byte plane. - AGENT_REVERSE_ID_BASE = 0x40000001 splits the 32-bit id space so agent-allocated reverse stream ids never collide with primary- allocated forward ids on the same tunnel. - StreamIdAllocator now wraps to its configured start (not always 1) so an allocator parameterized with the agent base stays in the agent half across the wrap. Agent (backend/src/pilot/agent.ts): - New ReverseTcpStreamHandle exported class, EventEmitter facade matching PilotTunnelBridge.TcpStream's surface (write, end, destroy plus open, data, error, close events). - Public openMeshTcpStream(target) allocates a reverse id, sends tcp_open_reverse, returns the handle. - onTcpOpenAckReverse handles inbound ack, dispatches open or error+close to the matching handle. - TcpData and tcp_close binary/JSON paths route ids in the reverse range to reverseTcpStreams; existing primary-allocated paths unchanged. - streamCount, cleanupAfterDisconnect, onStreamIdle include reverse streams. The allocator is reset to a fresh base on disconnect so long-lived agents that reconnect many times do not drift up the id space. - startPilotAgent registers the agent as MeshService's reverse dialer via lazy import. Bridge (backend/src/services/PilotTunnelBridge.ts): - Two new StreamState kinds: reverse_local (target = central, dial Dockerode container IP) and reverse_relay (target = another pilot, open a forward TcpStream on the target pilot's bridge). - handleTcpOpenReverse validates the agent-id range and stream cap, then dispatches to acceptReverseLocal or acceptReverseRelay via lazy imports of MeshService, NodeRegistry, and PilotTunnelManager (avoids module cycles). - acceptReverseLocal calls MeshService.resolveContainerIp (now public), opens net.createConnection, splices bytes through the tunnel. Pre-connect error handler is removed inside connect to avoid double-firing with the mid-stream error path. - acceptReverseRelay opens a forward TcpStream on the target bridge, splices bytes between the two tunnels. - Existing tcp_close and TcpData handlers extended for the new state kinds. teardownStream extended. MeshService (backend/src/services/MeshService.ts): - resolveContainerIp made public so the bridge's local-target path can dial the same shape. - New reverseDialer field plus setReverseDialer setter. Pilot mode registers the agent at boot; central mode leaves it null. - New private dialMeshTcpStream dispatcher: pilot side uses the reverse dialer, central side uses PilotTunnelManager.getBridge. - openCrossNode refactored to call the dispatcher and use the active-stream record's id for activity logging. - New exported MeshTcpStreamLike interface that both PilotTunnelBridge.TcpStream and ReverseTcpStreamHandle satisfy structurally; ReverseMeshDialer is the contract for the setter. Tests (3 files, 16 new cases, 172 total pass): - pilot-protocol-tcp.test.ts: tcp_open_reverse round-trip; agent base invariant. - pilot-agent-reverse-stream.test.ts: openMeshTcpStream allocation, ws-not-open, frame shape, write encoding, end emits tcp_close, ack-success, ack-failure, low-id ack ignored, inbound TcpData routing. - pilot-bridge-reverse.test.ts: id-range validation, no-target failure, successful local dial against a real upstream server. Together with PR #1000 (Phase A) and PR #1001 (deps), Phase B completes the central-pilot-pilot mesh routing matrix. Phase C (mesh over proxy-mode remotes) is the planned follow-up. * test(pilot): drop unused encodeJsonFrame import Lint failed on pilot-agent-reverse-stream.test.ts after the test changed from constructing tcp_open_reverse frames inline to driving the agent's frame-dispatch path with synthetic objects. The import is no longer referenced; ESLint's no-unused-vars rule rejected it. --- .../pilot-agent-reverse-stream.test.ts | 198 ++++++++++++++++++ .../__tests__/pilot-bridge-reverse.test.ts | 161 ++++++++++++++ .../src/__tests__/pilot-protocol-tcp.test.ts | 31 +++ backend/src/pilot/agent.ts | 184 +++++++++++++++- backend/src/pilot/protocol.ts | 59 +++++- backend/src/services/MeshService.ts | 88 ++++++-- backend/src/services/PilotTunnelBridge.ts | 198 +++++++++++++++++- 7 files changed, 884 insertions(+), 35 deletions(-) create mode 100644 backend/src/__tests__/pilot-agent-reverse-stream.test.ts create mode 100644 backend/src/__tests__/pilot-bridge-reverse.test.ts diff --git a/backend/src/__tests__/pilot-agent-reverse-stream.test.ts b/backend/src/__tests__/pilot-agent-reverse-stream.test.ts new file mode 100644 index 00000000..84afd321 --- /dev/null +++ b/backend/src/__tests__/pilot-agent-reverse-stream.test.ts @@ -0,0 +1,198 @@ +/** + * Phase B: PilotAgent's outbound reverse mesh stream. The agent's + * MeshForwarder hands off cross-node connections to PilotAgent via + * `openMeshTcpStream`, which sends a `tcp_open_reverse` frame and returns + * a stream handle. The test exercises the wire shape and the lifecycle + * dispatchers (open / data / close) without needing a real WebSocket or + * the full Sencho boot. + */ +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; +import { + AGENT_REVERSE_ID_BASE, + BinaryFrameType, + decodeBinaryFrame, + decodeJsonFrame, + encodeBinaryFrame, +} from '../pilot/protocol'; + +let tmpDir: string; +let PilotAgent: typeof import('../pilot/agent').PilotAgent; +let ReverseTcpStreamHandle: typeof import('../pilot/agent').ReverseTcpStreamHandle; + +interface CapturedSend { + raw: string | Buffer; + binary: boolean; +} + +function makeMockAgent(): { agent: import('../pilot/agent').PilotAgent; sent: CapturedSend[]; mockWs: { readyState: number; send: (data: unknown, opts?: { binary?: boolean }) => void } } { + const sent: CapturedSend[] = []; + const mockWs = { + readyState: 1, // WebSocket.OPEN + send(data: unknown, opts?: { binary?: boolean }) { + const isBinary = opts?.binary === true; + sent.push({ + raw: isBinary ? (Buffer.isBuffer(data) ? data : Buffer.from(data as Uint8Array)) : String(data), + binary: isBinary, + }); + }, + }; + const agent = new PilotAgent({ + primaryUrl: 'http://primary.invalid', + loopbackPort: 1, + initialToken: 'irrelevant', + enrolling: false, + }); + // Inject the mock ws into the private slot. The agent's + // openMeshTcpStream checks readyState and calls send(); the mock + // captures both for assertions without needing real connect/handshake. + (agent as unknown as { ws: typeof mockWs }).ws = mockWs; + return { agent, sent, mockWs }; +} + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ PilotAgent, ReverseTcpStreamHandle } = await import('../pilot/agent')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +describe('PilotAgent.openMeshTcpStream (Phase B)', () => { + it('allocates an id in the agent-reverse range and emits a tcp_open_reverse frame with the target', () => { + const { agent, sent } = makeMockAgent(); + const handle = agent.openMeshTcpStream({ + nodeId: 12, + stack: 'api', + service: 'db', + port: 5432, + }); + expect(handle).toBeInstanceOf(ReverseTcpStreamHandle); + expect(handle!.streamId).toBeGreaterThanOrEqual(AGENT_REVERSE_ID_BASE); + + const textFrames = sent.filter((s) => !s.binary); + expect(textFrames.length).toBe(1); + const decoded = decodeJsonFrame(textFrames[0].raw as string); + expect(decoded.t).toBe('tcp_open_reverse'); + if (decoded.t !== 'tcp_open_reverse') throw new Error('narrowing'); + expect(decoded.s).toBe(handle!.streamId); + expect(decoded.targetNodeId).toBe(12); + expect(decoded.stack).toBe('api'); + expect(decoded.service).toBe('db'); + expect(decoded.port).toBe(5432); + }); + + it('returns null when the tunnel is not OPEN', () => { + const { agent, mockWs } = makeMockAgent(); + mockWs.readyState = 0; // CONNECTING + const handle = agent.openMeshTcpStream({ nodeId: 1, stack: 'a', service: 'b', port: 1 }); + expect(handle).toBeNull(); + }); + + it('handle.write encodes a TcpData binary frame with the agent-allocated streamId', () => { + const { agent, sent } = makeMockAgent(); + const handle = agent.openMeshTcpStream({ nodeId: 2, stack: 's', service: 'svc', port: 80 }); + if (!handle) throw new Error('handle should exist'); + sent.length = 0; // clear the open frame + + const ok = handle.write(Buffer.from('hello')); + expect(ok).toBe(true); + expect(sent.length).toBe(1); + expect(sent[0].binary).toBe(true); + const decoded = decodeBinaryFrame(sent[0].raw as Buffer); + expect(decoded.type).toBe(BinaryFrameType.TcpData); + expect(decoded.streamId).toBe(handle.streamId); + expect(decoded.payload.toString()).toBe('hello'); + }); + + it('handle.end sends a tcp_close JSON frame and removes the stream from the agent map', () => { + const { agent, sent } = makeMockAgent(); + const handle = agent.openMeshTcpStream({ nodeId: 3, stack: 's', service: 'svc', port: 80 }); + if (!handle) throw new Error('handle should exist'); + sent.length = 0; + + handle.end(); + + const text = sent.find((s) => !s.binary); + expect(text).toBeDefined(); + const decoded = decodeJsonFrame(text!.raw as string); + expect(decoded.t).toBe('tcp_close'); + if (decoded.t !== 'tcp_close') throw new Error('narrowing'); + expect(decoded.s).toBe(handle.streamId); + + const internalMap = (agent as unknown as { reverseTcpStreams: Map }).reverseTcpStreams; + expect(internalMap.has(handle.streamId)).toBe(false); + }); + + it('inbound tcp_open_ack {ok: true} fires the open event on the matching handle', () => { + const { agent, mockWs } = makeMockAgent(); + const handle = agent.openMeshTcpStream({ nodeId: 4, stack: 's', service: 'svc', port: 80 }); + if (!handle) throw new Error('handle should exist'); + + let opened = false; + handle.on('open', () => { opened = true; }); + + // Simulate an inbound ack frame: invoke the agent's private json- + // dispatch path with a synthetic frame. + const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent); + onTcpOpenAckReverse({ t: 'tcp_open_ack', s: handle.streamId, ok: true }); + + expect(opened).toBe(true); + // Sanity: mockWs's readyState wasn't touched. + expect(mockWs.readyState).toBe(1); + }); + + it('inbound tcp_open_ack {ok: false} emits error and close, then drops the handle', () => { + const { agent } = makeMockAgent(); + const handle = agent.openMeshTcpStream({ nodeId: 5, stack: 's', service: 'svc', port: 80 }); + if (!handle) throw new Error('handle should exist'); + + let errMessage: string | undefined; + let closed = false; + handle.on('error', (err: Error) => { errMessage = err.message; }); + handle.on('close', () => { closed = true; }); + + const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent); + onTcpOpenAckReverse({ t: 'tcp_open_ack', s: handle.streamId, ok: false, err: 'unreachable' }); + + expect(errMessage).toBe('unreachable'); + expect(closed).toBe(true); + const internalMap = (agent as unknown as { reverseTcpStreams: Map }).reverseTcpStreams; + expect(internalMap.has(handle.streamId)).toBe(false); + }); + + it('forward-direction tcp_open_ack ids (low half) do not match reverse handles', () => { + const { agent } = makeMockAgent(); + const handle = agent.openMeshTcpStream({ nodeId: 6, stack: 's', service: 'svc', port: 80 }); + if (!handle) throw new Error('handle should exist'); + + let opened = false; + handle.on('open', () => { opened = true; }); + + // Primary-direction id (< AGENT_REVERSE_ID_BASE) must not bleed into + // the reverse map. + const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent); + onTcpOpenAckReverse({ t: 'tcp_open_ack', s: 5, ok: true }); + + expect(opened).toBe(false); + }); + + it('inbound TcpData binary frame (encoded against the wire) emits data on the handle', () => { + const { agent } = makeMockAgent(); + const handle = agent.openMeshTcpStream({ nodeId: 7, stack: 's', service: 'svc', port: 80 }); + if (!handle) throw new Error('handle should exist'); + + const received: Buffer[] = []; + handle.on('data', (chunk: Buffer) => received.push(chunk)); + + // Drive the agent's binary-frame handler with a wire-encoded TcpData + // frame to also exercise the routing branch added in Phase B. + const buf = encodeBinaryFrame(BinaryFrameType.TcpData, handle.streamId, Buffer.from('echo!')); + const decoded = decodeBinaryFrame(buf); + const handleBinaryFrame = (agent as unknown as { handleBinaryFrame: (frame: unknown) => void }).handleBinaryFrame.bind(agent); + handleBinaryFrame(decoded); + + expect(Buffer.concat(received).toString()).toBe('echo!'); + }); +}); diff --git a/backend/src/__tests__/pilot-bridge-reverse.test.ts b/backend/src/__tests__/pilot-bridge-reverse.test.ts new file mode 100644 index 00000000..f24e9d19 --- /dev/null +++ b/backend/src/__tests__/pilot-bridge-reverse.test.ts @@ -0,0 +1,161 @@ +/** + * Phase B: PilotTunnelBridge handles `tcp_open_reverse` from the agent. + * Two paths: + * - target is the central's local node: dial via Dockerode container IP + * and splice bytes between the resulting `net.Socket` and the tunnel. + * - target is another pilot: open a forward `TcpStream` on the target + * pilot's bridge and relay. + * + * The local-dial path is the common case for small fleets. This file + * exercises the validation gate (agent-id-range, stream cap) and the + * local-dial outcome by intercepting `MeshService.resolveContainerIp`. + */ +import { describe, it, expect, beforeAll, afterAll, vi } from 'vitest'; +import net from 'net'; +import { EventEmitter } from 'events'; +import { WebSocket } from 'ws'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; +import { + AGENT_REVERSE_ID_BASE, + decodeJsonFrame, + encodeJsonFrame, +} from '../pilot/protocol'; + +let tmpDir: string; +let PilotTunnelBridge: typeof import('../services/PilotTunnelBridge').PilotTunnelBridge; +let MeshService: typeof import('../services/MeshService').MeshService; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ PilotTunnelBridge } = await import('../services/PilotTunnelBridge')); + ({ MeshService } = await import('../services/MeshService')); +}); + +afterAll(() => { + vi.restoreAllMocks(); + cleanupTestDb(tmpDir); +}); + +function makeMockTunnelWs(): EventEmitter & { + sent: unknown[]; + readyState: number; + bufferedAmount: number; + send: (data: unknown) => void; + ping: () => void; + close: () => void; +} { + const ws = new EventEmitter() as EventEmitter & { + sent: unknown[]; readyState: number; bufferedAmount: number; + send: (data: unknown) => void; ping: () => void; close: () => void; + }; + ws.sent = []; + ws.readyState = WebSocket.OPEN; + ws.bufferedAmount = 0; + ws.send = (data: unknown) => { ws.sent.push(data); }; + ws.ping = () => { /* no-op */ }; + ws.close = () => { ws.readyState = WebSocket.CLOSED; ws.emit('close'); }; + return ws; +} + +function findAck(ws: { sent: unknown[] }, s: number): { ok: boolean; err?: string } | undefined { + for (const item of ws.sent) { + if (typeof item !== 'string') continue; + try { + const f = decodeJsonFrame(item); + if (f.t === 'tcp_open_ack' && f.s === s) return { ok: f.ok, err: f.err }; + } catch { /* ignore */ } + } + return undefined; +} + +async function waitFor(check: () => T | undefined): Promise { + const deadline = Date.now() + 1500; + while (Date.now() < deadline) { + const v = check(); + if (v !== undefined) return v; + await new Promise((r) => setTimeout(r, 10)); + } + throw new Error('timeout'); +} + +describe('PilotTunnelBridge handles tcp_open_reverse (Phase B)', () => { + it('rejects frames whose s is below the agent-reverse range', async () => { + const mockWs = makeMockTunnelWs(); + const bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket); + await bridge.start(); + + // Drive the tunnel-message handler with a low-range id; the + // validation gate must reject and emit a tcp_open_ack {ok: false}. + mockWs.emit('message', encodeJsonFrame({ + t: 'tcp_open_reverse', s: 5, + targetNodeId: 1, stack: 's', service: 'svc', port: 80, + }), false); + + const ack = await waitFor(() => findAck(mockWs, 5)); + expect(ack.ok).toBe(false); + expect(ack.err).toBe('agent_error'); + + bridge.close(); + }); + + it('local target with no Dockerode resolution returns ok:false err:no_target', async () => { + const mockWs = makeMockTunnelWs(); + const bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket); + await bridge.start(); + + // The default node id from the test DB. Force resolveContainerIp to + // return null to simulate a stack that does not exist locally. + const localNodeId = (await import('../services/NodeRegistry')).NodeRegistry.getInstance().getDefaultNodeId(); + const spy = vi.spyOn(MeshService.getInstance(), 'resolveContainerIp').mockResolvedValue(null); + + const s = AGENT_REVERSE_ID_BASE + 1; + mockWs.emit('message', encodeJsonFrame({ + t: 'tcp_open_reverse', s, + targetNodeId: localNodeId, stack: 'missing', service: 'svc', port: 80, + }), false); + + const ack = await waitFor(() => findAck(mockWs, s)); + expect(ack.ok).toBe(false); + expect(ack.err).toBe('no_target'); + expect(spy).toHaveBeenCalledTimes(1); + + bridge.close(); + vi.restoreAllMocks(); + }); + + it('local target with a working dial sends ok:true and registers a reverse stream', async () => { + const mockWs = makeMockTunnelWs(); + const bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket); + await bridge.start(); + + // Spin up a real local server the bridge will dial as the target. + const upstream = net.createServer((socket) => { + socket.write('hello-upstream'); + }); + await new Promise((resolve) => upstream.listen(0, '127.0.0.1', () => resolve())); + const addr = upstream.address(); + if (!addr || typeof addr === 'string') throw new Error('no address'); + const upstreamPort = addr.port; + + const localNodeId = (await import('../services/NodeRegistry')).NodeRegistry.getInstance().getDefaultNodeId(); + vi.spyOn(MeshService.getInstance(), 'resolveContainerIp').mockResolvedValue('127.0.0.1'); + + const s = AGENT_REVERSE_ID_BASE + 2; + mockWs.emit('message', encodeJsonFrame({ + t: 'tcp_open_reverse', s, + targetNodeId: localNodeId, stack: 'real', service: 'svc', port: upstreamPort, + }), false); + + const ack = await waitFor(() => findAck(mockWs, s)); + expect(ack.ok).toBe(true); + + // Stream is registered on the bridge. + const streams = (bridge as unknown as { streams: Map }).streams; + const state = streams.get(s); + expect(state?.kind).toBe('reverse_local'); + + upstream.close(); + bridge.close(); + vi.restoreAllMocks(); + }); +}); diff --git a/backend/src/__tests__/pilot-protocol-tcp.test.ts b/backend/src/__tests__/pilot-protocol-tcp.test.ts index 695c5992..f95dadf8 100644 --- a/backend/src/__tests__/pilot-protocol-tcp.test.ts +++ b/backend/src/__tests__/pilot-protocol-tcp.test.ts @@ -3,6 +3,7 @@ */ import { describe, it, expect } from 'vitest'; import { + AGENT_REVERSE_ID_BASE, BinaryFrameType, decodeBinaryFrame, decodeJsonFrame, @@ -104,3 +105,33 @@ describe('Mesh TcpData binary frames', () => { } }); }); + +describe('Phase B: tcp_open_reverse', () => { + it('roundtrips a tcp_open_reverse frame including targetNodeId', () => { + const raw = encodeJsonFrame({ + t: 'tcp_open_reverse', + s: AGENT_REVERSE_ID_BASE + 7, + targetNodeId: 12, + stack: 'api', + service: 'db', + port: 5432, + }); + const decoded = decodeJsonFrame(raw); + expect(decoded.t).toBe('tcp_open_reverse'); + if (decoded.t !== 'tcp_open_reverse') throw new Error('narrowing'); + expect(decoded.s).toBe(AGENT_REVERSE_ID_BASE + 7); + expect(decoded.targetNodeId).toBe(12); + expect(decoded.stack).toBe('api'); + expect(decoded.service).toBe('db'); + expect(decoded.port).toBe(5432); + }); + + it('AGENT_REVERSE_ID_BASE is in the upper half of the 32-bit space and clears primary allocator collision risk', () => { + // Primary allocator wraps at 0x7fffffff. Agent base must be above + // that range so primary's id sequence (1..0x7fffffff) and the + // agent's reverse sequence never collide on the same tunnel for + // the lifetime of the tunnel (well beyond MAX_STREAMS_PER_TUNNEL). + expect(AGENT_REVERSE_ID_BASE).toBeGreaterThan(0x40000000); + expect(AGENT_REVERSE_ID_BASE).toBeLessThan(0x80000000); + }); +}); diff --git a/backend/src/pilot/agent.ts b/backend/src/pilot/agent.ts index 6c9983dc..435befc1 100644 --- a/backend/src/pilot/agent.ts +++ b/backend/src/pilot/agent.ts @@ -2,18 +2,21 @@ import fs from 'fs'; import net from 'net'; import path from 'path'; import http from 'http'; +import { EventEmitter } from 'events'; import jwt from 'jsonwebtoken'; import WebSocket from 'ws'; import { getSenchoVersion } from '../services/CapabilityRegistry'; import { DatabaseService } from '../services/DatabaseService'; import { NodeRegistry } from '../services/NodeRegistry'; import { + AGENT_REVERSE_ID_BASE, BinaryFrameType, MAX_FRAME_SIZE_BYTES, MAX_STREAMS_PER_TUNNEL, MeshErrCode, PROTOCOL_VERSION, STREAM_IDLE_TIMEOUT_MS, + StreamIdAllocator, decodeBinaryFrame, decodeJsonFrame, encodeBinaryFrame, @@ -58,6 +61,16 @@ export function startPilotAgent(loopbackPort: number): void { initialToken: persistedToken || enrollToken!, enrolling: !persistedToken, }); + // Register the agent as MeshService's reverse dialer so outbound + // cross-node mesh traffic from this pilot's MeshForwarder routes via + // `tcp_open_reverse` over the existing pilot tunnel instead of trying + // to use the central-only `PilotTunnelManager.getBridge` path. Lazy + // import keeps `MeshService` outside the cold-boot critical path. + void import('../services/MeshService').then(({ MeshService }) => { + MeshService.getInstance().setReverseDialer(agent); + }).catch((err) => { + console.warn('[Pilot] reverse dialer registration failed:', sanitizeForLog((err as Error).message)); + }); agent.start(); } @@ -78,6 +91,10 @@ export class PilotAgent { private readonly httpStreams = new Map(); private readonly wsStreams = new Map(); private readonly tcpStreams = new Map(); + /** Reverse mesh streams the agent itself initiated via `tcp_open_reverse`. Keyed on the agent-allocated id. Disjoint from `tcpStreams` because those are primary-allocated and live in the lower id half. */ + private readonly reverseTcpStreams = new Map(); + /** Allocator is recreated on every disconnect (`cleanupAfterDisconnect`) so a long-lived agent that reconnects many times doesn't drift up the id range. */ + private reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE); private readonly idleTimers = new Map(); private shuttingDown = false; private readonly agentVersion: string; @@ -221,12 +238,23 @@ export class PilotAgent { try { stream.socket.destroy(); } catch { /* ignore */ } } this.tcpStreams.clear(); + for (const [, handle] of this.reverseTcpStreams) { + try { + handle._dispatchError(new Error('pilot tunnel closed')); + handle._dispatchClose(); + } catch { /* ignore */ } + } + this.reverseTcpStreams.clear(); + // Reset the reverse allocator so a long-lived agent that + // reconnects many times doesn't drift up the id range and + // approach the wrap point unnecessarily. + this.reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE); for (const [, timer] of this.idleTimers) clearTimeout(timer); this.idleTimers.clear(); } private streamCount(): number { - return this.httpStreams.size + this.wsStreams.size + this.tcpStreams.size; + return this.httpStreams.size + this.wsStreams.size + this.tcpStreams.size + this.reverseTcpStreams.size; } private refreshIdleTimer(streamId: number): void { @@ -272,6 +300,16 @@ export class PilotAgent { if (ws) { try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ } } + return; + } + const reverseEntry = this.reverseTcpStreams.get(streamId); + if (reverseEntry) { + this.reverseTcpStreams.delete(streamId); + reverseEntry._dispatchError(new Error('agent idle timeout')); + reverseEntry._dispatchClose(); + if (ws) { + try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ } + } } } @@ -337,6 +375,7 @@ export class PilotAgent { case 'ws_msg_text': this.onWsMsgText(frame.s, frame.data); break; case 'ws_close': this.onWsClose(frame.s, frame.code, frame.reason); break; case 'tcp_open': this.onTcpOpen(frame); break; + case 'tcp_open_ack': this.onTcpOpenAckReverse(frame); break; case 'tcp_close': this.onTcpClose(frame.s); break; default: // Other frame types are primary-bound only; agent ignores. @@ -361,6 +400,13 @@ export class PilotAgent { break; } case BinaryFrameType.TcpData: { + if (frame.streamId >= AGENT_REVERSE_ID_BASE) { + const reverse = this.reverseTcpStreams.get(frame.streamId); + if (!reverse) return; + reverse._dispatchData(frame.payload); + this.refreshIdleTimer(frame.streamId); + return; + } const stream = this.tcpStreams.get(frame.streamId); if (!stream) return; try { stream.socket.write(frame.payload); } catch { /* ignore */ } @@ -591,6 +637,17 @@ export class PilotAgent { } private onTcpClose(streamId: number): void { + // Reverse stream (agent-initiated): primary is closing the + // upstream half. Tear down the local socket the agent's + // MeshForwarder handed us via openMeshTcpStream. + if (streamId >= AGENT_REVERSE_ID_BASE) { + const handle = this.reverseTcpStreams.get(streamId); + if (!handle) return; + this.reverseTcpStreams.delete(streamId); + this.clearIdleTimer(streamId); + handle._dispatchClose(); + return; + } const entry = this.tcpStreams.get(streamId); if (!entry) return; this.tcpStreams.delete(streamId); @@ -598,6 +655,74 @@ export class PilotAgent { try { entry.socket.destroy(); } catch { /* ignore */ } } + /** + * Inbound `tcp_open_ack` for an agent-initiated reverse stream. The + * primary acknowledges (or rejects) the dial; emit 'open' or 'error' on + * the local handle so MeshService's splice setup can either start + * piping bytes or tear down the source socket. + */ + private onTcpOpenAckReverse(frame: Extract, { t: 'tcp_open_ack' }>): void { + if (frame.s < AGENT_REVERSE_ID_BASE) return; // forward-direction acks are primary-bound; ignore. + const handle = this.reverseTcpStreams.get(frame.s); + if (!handle) return; + if (frame.ok) { + handle._dispatchOpen(); + this.refreshIdleTimer(frame.s); + } else { + this.reverseTcpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); + handle._dispatchError(new Error(frame.err ?? 'tcp_open_reverse rejected')); + handle._dispatchClose(); + } + } + + /** + * Public entry point for the agent's mesh forwarder. Allocates a + * reverse stream id, sends `tcp_open_reverse`, and returns a handle + * MeshService can splice bytes through. Returns null if the tunnel is + * not currently open or the per-tunnel stream cap is reached. + */ + public openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null { + const ws = this.ws; + if (!ws || ws.readyState !== WebSocket.OPEN) return null; + if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) return null; + const streamId = this.reverseStreamIds.allocate(); + const handle = new ReverseTcpStreamHandle( + streamId, + (sid, payload) => { + if (this.ws?.readyState !== WebSocket.OPEN) return; + try { this.ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, sid, payload), { binary: true }); } catch { /* ignore */ } + this.refreshIdleTimer(sid); + }, + (sid) => { + if (!this.reverseTcpStreams.has(sid)) return; + this.reverseTcpStreams.delete(sid); + this.clearIdleTimer(sid); + if (this.ws?.readyState !== WebSocket.OPEN) return; + try { this.ws.send(encodeJsonFrame({ t: 'tcp_close', s: sid })); } catch { /* ignore */ } + }, + ); + this.reverseTcpStreams.set(streamId, handle); + this.refreshIdleTimer(streamId); + try { + ws.send(encodeJsonFrame({ + t: 'tcp_open_reverse', + s: streamId, + targetNodeId: target.nodeId, + stack: target.stack, + service: target.service, + port: target.port, + })); + } catch (err) { + this.reverseTcpStreams.delete(streamId); + this.clearIdleTimer(streamId); + handle._dispatchError(err as Error); + handle._dispatchClose(); + return null; + } + return handle; + } + /** * Resolves a mesh target by consulting the local mesh_stacks opt-in table * and Compose container labels. Refuses if the target stack is not opted @@ -645,6 +770,63 @@ interface MeshTcpStream { accepted: boolean; } +/** + * Handle returned by `PilotAgent.openMeshTcpStream` to MeshService. Mirrors + * the surface of `PilotTunnelBridge.TcpStream` (write/end/destroy + + * 'open'/'data'/'error'/'close' events) so MeshService.openCrossNode can + * splice bytes against it without caring whether it's running on central + * or on a pilot. The agent owns the per-stream WS plumbing through the + * `sendData` and `sendClose` callbacks; this class is a thin EventEmitter + * facade. + */ +export class ReverseTcpStreamHandle extends EventEmitter { + public readonly streamId: number; + private readonly sendData: (streamId: number, payload: Buffer) => void; + private readonly sendClose: (streamId: number) => void; + private closed = false; + + constructor( + streamId: number, + sendData: (streamId: number, payload: Buffer) => void, + sendClose: (streamId: number) => void, + ) { + super(); + this.streamId = streamId; + this.sendData = sendData; + this.sendClose = sendClose; + } + + public write(chunk: Buffer): boolean { + if (this.closed) return false; + this.sendData(this.streamId, chunk); + return true; + } + + public end(): void { + if (this.closed) return; + this.closed = true; + this.sendClose(this.streamId); + } + + public destroy(): void { this.end(); } + + /** @internal Called by PilotAgent on inbound `tcp_open_ack { ok: true }`. */ + public _dispatchOpen(): void { this.emit('open'); } + + /** @internal Called by PilotAgent on inbound `TcpData` for this stream. */ + public _dispatchData(chunk: Buffer): void { this.emit('data', chunk); } + + /** @internal Called by PilotAgent on tunnel-side error or rejection. */ + public _dispatchError(err: Error): void { this.emit('error', err); } + + /** @internal Called by PilotAgent on tunnel-side close. */ + public _dispatchClose(): void { + if (this.closed) return; + this.closed = true; + this.emit('close'); + } +} + type MeshResolveResult = | { ok: true; host: string; port: number } | { ok: false; err: MeshErrCode }; diff --git a/backend/src/pilot/protocol.ts b/backend/src/pilot/protocol.ts index 1ec79728..895d6251 100644 --- a/backend/src/pilot/protocol.ts +++ b/backend/src/pilot/protocol.ts @@ -75,6 +75,7 @@ export type JsonFrame = | WsCloseFrame | ControlFrame | TcpOpenFrame + | TcpOpenReverseFrame | TcpOpenAckFrame | TcpCloseFrame; @@ -157,9 +158,25 @@ export interface ControlFrame { } /** - * Sencho Mesh TCP frames. The primary asks the agent to open a TCP connection - * to a Compose service on the agent's local Docker host. Bytes flow as - * BinaryFrameType.TcpData. Mid-stream failures send tcp_close. + * Sencho Mesh TCP frames. Two directions: + * + * - `tcp_open` (primary -> agent): primary asks the agent to open a TCP + * connection to a Compose service on the agent's local Docker host. + * Stream id is allocated by the primary's `StreamIdAllocator` (low + * range, starts at 1). + * + * - `tcp_open_reverse` (agent -> primary): the agent's mesh forwarder + * accepted a connection destined for another node and asks the primary + * to dial that node. If `targetNodeId` matches the primary's own node, + * the primary dials a local container directly. Otherwise the primary + * relays via its bridge to the target pilot. Stream id is allocated by + * the agent and uses the upper half of the 32-bit space + * (>= 0x40000001) so it cannot collide with primary-allocated ids on + * the same tunnel; the wrap distance (~2^30 vs the 1024 stream cap) + * makes collisions statistically unreachable. + * + * In both directions, `tcp_open_ack` confirms acceptance, bytes flow as + * `BinaryFrameType.TcpData`, and `tcp_close` ends the stream. */ export interface TcpOpenFrame { t: 'tcp_open'; @@ -169,6 +186,15 @@ export interface TcpOpenFrame { port: number; } +export interface TcpOpenReverseFrame { + t: 'tcp_open_reverse'; + s: number; + targetNodeId: number; + stack: string; + service: string; + port: number; +} + export type MeshErrCode = 'mesh_not_enabled' | 'denied' | 'no_target' | 'unreachable' | 'agent_error'; export interface TcpOpenAckFrame { @@ -183,6 +209,14 @@ export interface TcpCloseFrame { s: number; } +/** + * Stream id space split for `tcp_open` (primary-allocated) vs + * `tcp_open_reverse` (agent-allocated). Used by the agent's reverse + * allocator and by the primary's `tcp_open_reverse` handler to verify + * incoming ids are in the agent half. + */ +export const AGENT_REVERSE_ID_BASE = 0x40000001; + // --- Serialize / parse --- export function encodeJsonFrame(frame: JsonFrame): string { @@ -271,24 +305,29 @@ export const PilotCloseCode = { // --- Stream id allocation --- /** - * Monotonic stream id generator. Primary is the sole allocator. + * Monotonic stream id generator. * - * Wraps at 2^31. With MAX_STREAMS_PER_TUNNEL = 1024 the allocator - * cannot collide with a still-live stream during a single tunnel - * lifetime: the wrap distance (~2.1 billion) is more than six orders - * of magnitude larger than the cap. A new tunnel restarts the - * sequence at 1, so cross-tunnel reuse is also harmless. + * Wraps at 2^31 back to the configured `start` value (not back to `1`), + * so allocators initialized with `AGENT_REVERSE_ID_BASE` stay in the + * agent half of the id space across the wrap. With + * MAX_STREAMS_PER_TUNNEL = 1024 the allocator cannot collide with a + * still-live stream during a single tunnel lifetime: the wrap distance + * (~2.1 billion) is more than six orders of magnitude larger than the + * cap. A new tunnel restarts the sequence at `start`, so cross-tunnel + * reuse is also harmless. */ export class StreamIdAllocator { + private readonly start: number; private next: number; constructor(start = 1) { + this.start = start; this.next = start; } allocate(): number { const id = this.next; - this.next = this.next >= 0x7fffffff ? 1 : this.next + 1; + this.next = this.next >= 0x7fffffff ? this.start : this.next + 1; return id; } } diff --git a/backend/src/services/MeshService.ts b/backend/src/services/MeshService.ts index c6a5b558..618d8914 100644 --- a/backend/src/services/MeshService.ts +++ b/backend/src/services/MeshService.ts @@ -590,8 +590,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { src.on('close', () => teardown()); } - /** Find the bridge-network IP of the first container of `/`. */ - private async resolveContainerIp(target: MeshTarget): Promise { + /** + * Find the bridge-network IP of the first container of + * `/`. Public so the central-side + * `PilotTunnelBridge` reverse-open handler (Phase B) can dial the same + * target shape when a pilot's mesh forwarder routes traffic back to + * central. + */ + public async resolveContainerIp(target: { stack: string; service: string }): Promise { try { const docker = DockerController.getInstance().getDocker(); // Compose default container name pattern; -1 is the first replica. @@ -644,28 +650,54 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { return info.NetworkSettings?.IPAddress || null; } - private openCrossNode(target: MeshTarget, src: net.Socket): void { + /** + * Pluggable reverse dialer. Set by `PilotAgent` on a pilot host; left + * null on a central host. When set, `openCrossNode` routes outbound + * mesh dials through the agent's `tcp_open_reverse` path; when unset, + * `openCrossNode` uses the central-side `PilotTunnelManager.getBridge` + * directly. Lets the same MeshService code work on both sides. + */ + private reverseDialer: ReverseMeshDialer | null = null; + + public setReverseDialer(dialer: ReverseMeshDialer | null): void { + this.reverseDialer = dialer; + } + + private dialMeshTcpStream(target: MeshTarget): MeshTcpStreamLike | null { + if (this.reverseDialer) { + return this.reverseDialer.openMeshTcpStream({ + nodeId: target.nodeId, + stack: target.stack, + service: target.service, + port: target.port, + }); + } const ptm = PilotTunnelManager.getInstance(); - if (!ptm.hasActiveTunnel(target.nodeId)) { + if (!ptm.hasActiveTunnel(target.nodeId)) return null; + const bridge = ptm.getBridge(target.nodeId); + if (!bridge) return null; + return bridge.openTcpStream({ stack: target.stack, service: target.service, port: target.port }); + } + + private openCrossNode(target: MeshTarget, src: net.Socket): void { + const tcpStream = this.dialMeshTcpStream(target); + if (!tcpStream) { this.logActivity({ source: 'pilot', level: 'error', type: 'tunnel.fail', nodeId: target.nodeId, alias: target.alias, - message: `no active pilot tunnel to node ${target.nodeId}`, + message: this.reverseDialer + ? `cannot open reverse mesh stream to node ${target.nodeId}` + : `no active pilot tunnel to node ${target.nodeId}`, }); try { src.destroy(); } catch { /* ignore */ } return; } - const bridge = ptm.getBridge(target.nodeId); - if (!bridge) { try { src.destroy(); } catch { /* ignore */ } return; } - const tcpStream = bridge.openTcpStream({ stack: target.stack, service: target.service, port: target.port }); - if (!tcpStream) { try { src.destroy(); } catch { /* ignore */ } return; } - const record = this.registerActiveStream(target.alias, tcpStream.streamId); const t0 = Date.now(); tcpStream.on('open', () => { this.logActivity({ source: 'mesh', level: 'info', type: 'route.resolve.ok', - nodeId: target.nodeId, alias: target.alias, streamId: tcpStream.streamId, + nodeId: target.nodeId, alias: target.alias, streamId: record.streamId, message: `cross-node connect to ${target.alias}`, }); this.routeLatencyMap.set(target.alias, Date.now() - t0); @@ -677,14 +709,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { tcpStream.on('error', (err: Error) => { this.logActivity({ source: 'pilot', level: 'error', type: 'tunnel.fail', - nodeId: target.nodeId, alias: target.alias, streamId: tcpStream.streamId, + nodeId: target.nodeId, alias: target.alias, streamId: record.streamId, message: err.message, }); - this.activeStreams.delete(tcpStream.streamId); + this.activeStreams.delete(record.streamId); try { src.destroy(); } catch { /* ignore */ } }); tcpStream.on('close', () => { - this.activeStreams.delete(tcpStream.streamId); + this.activeStreams.delete(record.streamId); try { src.end(); } catch { /* ignore */ } }); src.on('data', (chunk: Buffer) => { @@ -912,3 +944,31 @@ export class MeshError extends Error { this.code = code; } } + +/** + * Common surface of an outbound mesh TCP stream as MeshService consumes + * it. Both the central-side `PilotTunnelBridge.TcpStream` and the + * pilot-side `ReverseTcpStreamHandle` (from `pilot/agent.ts`) implement + * this shape structurally so MeshService.openCrossNode can splice bytes + * against either without caring which side initiated the stream. + */ +export interface MeshTcpStreamLike { + readonly streamId: number; + write(chunk: Buffer): boolean; + end(): void; + destroy(): void; + on(event: 'open', listener: () => void): this; + on(event: 'data', listener: (chunk: Buffer) => void): this; + on(event: 'error', listener: (err: Error) => void): this; + on(event: 'close', listener: () => void): this; +} + +/** + * Pilot-side reverse dialer. Set by `PilotAgent` when the agent boots + * (Phase B); leaves MeshService.openCrossNode able to route outbound + * cross-node mesh traffic over the agent's outbound `tcp_open_reverse` + * frame instead of central's `PilotTunnelManager.getBridge`. + */ +export interface ReverseMeshDialer { + openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): MeshTcpStreamLike | null; +} diff --git a/backend/src/services/PilotTunnelBridge.ts b/backend/src/services/PilotTunnelBridge.ts index 16a1d53a..8eb3fd5b 100644 --- a/backend/src/services/PilotTunnelBridge.ts +++ b/backend/src/services/PilotTunnelBridge.ts @@ -1,8 +1,9 @@ import http, { IncomingMessage, Server as HttpServer, ServerResponse } from 'http'; -import { Socket } from 'net'; +import net, { Socket } from 'net'; import { EventEmitter } from 'events'; import { WebSocket, WebSocketServer } from 'ws'; import { + AGENT_REVERSE_ID_BASE, BinaryFrameType, DecodedBinaryFrame, MAX_STREAMS_PER_TUNNEL, @@ -57,7 +58,34 @@ interface TcpStreamState extends StreamMeta { accepted: boolean; } -type StreamState = HttpStreamState | WsStreamState | TcpStreamState; +/** + * Pilot-initiated reverse stream where the agent's mesh forwarder asked the + * primary to dial a service ON the primary's local Docker host. The primary + * holds the resulting `net.Socket` and pumps bytes between the socket and + * the tunnel `TcpData` frames keyed on the agent-allocated `s`. + */ +interface ReverseLocalTcpStreamState extends StreamMeta { + kind: 'reverse_local'; + socket: Socket; + bytesIn: number; + bytesOut: number; +} + +/** + * Pilot-initiated reverse stream where the target is *another* pilot. The + * primary acts as a transparent relay: it allocates a forward `TcpStream` + * on the target pilot's bridge and splices bytes between this bridge's + * incoming `TcpData` frames (keyed on the source agent's `s`) and the + * target stream's `write` / `'data'`. + */ +interface ReverseRelayTcpStreamState extends StreamMeta { + kind: 'reverse_relay'; + target: TcpStream; + bytesIn: number; + bytesOut: number; +} + +type StreamState = HttpStreamState | WsStreamState | TcpStreamState | ReverseLocalTcpStreamState | ReverseRelayTcpStreamState; /** * Sencho Mesh TCP stream handle. EventEmitter-based duplex-like surface that @@ -513,11 +541,23 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle } break; } + case 'tcp_open_reverse': { + this.handleTcpOpenReverse(frame); + break; + } case 'tcp_close': { const s = this.streams.get(frame.s); - if (!s || s.kind !== 'tcp') return; - this.removeStream(frame.s); - s.handle.emit('close'); + if (!s) return; + if (s.kind === 'tcp') { + this.removeStream(frame.s); + s.handle.emit('close'); + } else if (s.kind === 'reverse_local') { + this.removeStream(frame.s); + try { s.socket.destroy(); } catch { /* ignore */ } + } else if (s.kind === 'reverse_relay') { + this.removeStream(frame.s); + try { s.target.destroy(); } catch { /* ignore */ } + } break; } default: @@ -551,10 +591,19 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle // Agent never originates request bodies; ignore for defense-in-depth. break; case BinaryFrameType.TcpData: { - if (s.kind !== 'tcp') return; - s.bytesIn += frame.payload.length; - this.refreshIdleTimer(frame.streamId, s); - s.handle.emit('data', frame.payload); + if (s.kind === 'tcp') { + s.bytesIn += frame.payload.length; + this.refreshIdleTimer(frame.streamId, s); + s.handle.emit('data', frame.payload); + } else if (s.kind === 'reverse_local') { + s.bytesIn += frame.payload.length; + this.refreshIdleTimer(frame.streamId, s); + try { s.socket.write(frame.payload); } catch { /* ignore */ } + } else if (s.kind === 'reverse_relay') { + s.bytesIn += frame.payload.length; + this.refreshIdleTimer(frame.streamId, s); + s.target.write(frame.payload); + } break; } default: @@ -673,11 +722,140 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle } else if (state.rawSocket) { try { state.rawSocket.destroy(); } catch { /* ignore */ } } - } else { + } else if (state.kind === 'tcp') { try { if (!state.accepted) state.handle.emit('error', new Error('tunnel closed before accept')); state.handle.emit('close'); } catch { /* ignore */ } + } else if (state.kind === 'reverse_local') { + try { state.socket.destroy(); } catch { /* ignore */ } + } else if (state.kind === 'reverse_relay') { + try { state.target.destroy(); } catch { /* ignore */ } } } + + // ---- Phase B: pilot-initiated reverse mesh streams ---- + + /** + * Handle an inbound `tcp_open_reverse` from the agent. The agent's + * `MeshForwarder` accepted a connection destined for a node other than + * the agent's own and is asking the primary to dial the target. Two + * cases: + * + * - target is the primary's own node: dial a local container directly + * and splice bytes between the resulting socket and the tunnel. + * - target is another pilot: open a forward `TcpStream` on the target + * pilot's bridge and relay bytes between the two tunnels (primary + * in the middle). + * + * Stream id is allocated by the agent; the primary stores state keyed + * on the agent's id. Agent ids are required to be in the upper half of + * the 32-bit space so they cannot collide with primary-allocated ids + * for forward `tcp_open` streams on the same tunnel. + */ + private handleTcpOpenReverse(frame: { s: number; targetNodeId: number; stack: string; service: string; port: number }): void { + const { s, targetNodeId, stack, service, port } = frame; + if (s < AGENT_REVERSE_ID_BASE) { + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'agent_error' }); + return; + } + if (this.streams.has(s) || this.streams.size >= MAX_STREAMS_PER_TUNNEL) { + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'agent_error' }); + return; + } + // Lazy-import services that import this module to avoid a cycle. + void (async () => { + const { NodeRegistry } = await import('./NodeRegistry'); + const localNodeId = NodeRegistry.getInstance().getDefaultNodeId(); + if (targetNodeId === localNodeId) { + await this.acceptReverseLocal(s, { stack, service, port }); + } else { + await this.acceptReverseRelay(s, targetNodeId, { stack, service, port }); + } + })().catch((err) => { + if (isDebugEnabled()) console.warn('[PilotBridge:diag] reverse-open dispatch failed:', sanitizeForLog((err as Error).message)); + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'agent_error' }); + }); + } + + private async acceptReverseLocal(s: number, target: { stack: string; service: string; port: number }): Promise { + const { MeshService } = await import('./MeshService'); + const ip = await MeshService.getInstance().resolveContainerIp({ stack: target.stack, service: target.service }); + if (!ip) { + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'no_target' }); + return; + } + const socket = net.createConnection({ host: ip, port: target.port }); + const state: ReverseLocalTcpStreamState = { kind: 'reverse_local', socket, bytesIn: 0, bytesOut: 0 }; + this.streams.set(s, state); + this.refreshIdleTimer(s, state); + + const teardown = (sendClose: boolean) => { + if (!this.streams.has(s)) return; + this.removeStream(s); + try { socket.destroy(); } catch { /* ignore */ } + if (sendClose) this.sendJson({ t: 'tcp_close', s }); + }; + + // Pre-connect failure: ack-fail and drop. The handler is removed in + // 'connect' below so post-connect errors fall through to the + // mid-stream teardown path instead of double-firing. + const onPreConnectError = () => { + if (!this.streams.has(s)) return; + this.streams.delete(s); + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' }); + }; + socket.once('error', onPreConnectError); + socket.once('connect', () => { + socket.off('error', onPreConnectError); + this.sendJson({ t: 'tcp_open_ack', s, ok: true }); + socket.on('data', (chunk: Buffer) => { + const cur = this.streams.get(s); + if (!cur || cur.kind !== 'reverse_local') return; + this.sendBinary(BinaryFrameType.TcpData, s, chunk); + cur.bytesOut += chunk.length; + this.refreshIdleTimer(s, cur); + }); + socket.on('close', () => teardown(true)); + socket.on('error', () => teardown(true)); + }); + } + + private async acceptReverseRelay(s: number, targetNodeId: number, target: { stack: string; service: string; port: number }): Promise { + const { PilotTunnelManager } = await import('./PilotTunnelManager'); + const targetBridge = PilotTunnelManager.getInstance().getBridge(targetNodeId); + if (!targetBridge) { + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' }); + return; + } + const targetStream = targetBridge.openTcpStream(target); + if (!targetStream) { + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' }); + return; + } + const state: ReverseRelayTcpStreamState = { kind: 'reverse_relay', target: targetStream, bytesIn: 0, bytesOut: 0 }; + this.streams.set(s, state); + this.refreshIdleTimer(s, state); + + targetStream.once('open', () => { + this.sendJson({ t: 'tcp_open_ack', s, ok: true }); + }); + targetStream.on('data', (chunk: Buffer) => { + const cur = this.streams.get(s); + if (!cur || cur.kind !== 'reverse_relay') return; + this.sendBinary(BinaryFrameType.TcpData, s, chunk); + cur.bytesOut += chunk.length; + this.refreshIdleTimer(s, cur); + }); + targetStream.once('error', () => { + if (!this.streams.has(s)) return; + this.streams.delete(s); + this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' }); + }); + targetStream.once('close', () => { + if (!this.streams.has(s)) return; + this.removeStream(s); + this.sendJson({ t: 'tcp_close', s }); + }); + } }