From a318e6b3c17a9aed734d8a8f5e309cc50b571ca3 Mon Sep 17 00:00:00 2001 From: Anso Date: Sun, 17 May 2026 14:01:39 -0400 Subject: [PATCH] fix(mesh): close data-plane race that dropped early TcpData frames (#1086) The tcp_open and tcp_open_reverse receivers registered their stream entry only after awaiting resolveTarget / resolveContainerIp. The peer, which is free to send TcpData immediately after the open frame, hit the lookup-miss path in handleBinaryFrame and the bytes were silently dropped. Probes passed because they only exercise the handshake; real HTTP hung with 0 bytes received. Reserve the stream entry synchronously, buffer early TcpData in a per-stream pendingData queue capped at 1 MiB, and flush onto the local socket inside the connect handler before sending tcp_open_ack. The payload is copied because decodeBinaryFrame returns a subarray view of the WS receive buffer; holding the view would pin the parent buffer past its lifecycle. Both sides of the data plane are patched: - tcpStreamSwitchboard.onTcpOpen (forward, agent + proxy-mode peer) - PilotTunnelBridge.handleTcpOpenReverse / acceptReverseLocal (reverse, primary acting as the local dial target) Adds unit coverage for the race window and for the overflow path. The cap constant lives in pilot/protocol.ts alongside the other per-stream limits. --- .../mesh-tcp-stream-switchboard.test.ts | 66 +++++++++++++ .../__tests__/pilot-bridge-reverse.test.ts | 54 +++++++++++ backend/src/mesh/tcpStreamSwitchboard.ts | 91 ++++++++++++++++-- backend/src/pilot/protocol.ts | 11 +++ backend/src/services/PilotTunnelBridge.ts | 96 +++++++++++++++++-- 5 files changed, 303 insertions(+), 15 deletions(-) diff --git a/backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts b/backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts index e860bf90..7dfa4e3f 100644 --- a/backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts +++ b/backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts @@ -188,6 +188,72 @@ describe('TcpStreamSwitchboard.onTcpOpen (forward path)', () => { expect(consumed).toBe(true); }); + it('buffers TcpData arriving during the resolveTarget await and flushes it on connect (F-9 race fix)', async () => { + const { port, server, firstConn } = await startEchoServer(); + try { + // Resolver delayed by 50 ms simulates a real Dockerode inspect. + // Without the fix, the TcpData frame sent inside this window + // hits a missing this.tcpStreams entry and is silently dropped. + const { switchboard, sent } = makeSwitchboard({ + resolve: () => new Promise((r) => setTimeout(() => r({ ok: true, host: '127.0.0.1', port }), 50)), + }); + + switchboard.handleJsonFrame({ t: 'tcp_open', s: 1, stack: 's', service: 'svc', port: 80 }); + // Immediately push the "HTTP request" bytes before the resolver + // finishes. Reservation must catch them in pendingData. + const reqFrame = encodeBinaryFrame(BinaryFrameType.TcpData, 1, Buffer.from('GET / HTTP/1.1\r\n\r\n')); + switchboard.handleBinaryFrame(decodeBinaryFrame(reqFrame)); + + const targetSocket = await firstConn; + const echoed = new Promise((r) => { + let acc = Buffer.alloc(0); + targetSocket.on('data', (chunk: Buffer) => { + acc = Buffer.concat([acc, chunk]); + if (acc.length >= 'GET / HTTP/1.1\r\n\r\n'.length) r(acc); + }); + }); + const received = await echoed; + expect(received.toString()).toBe('GET / HTTP/1.1\r\n\r\n'); + + // tcp_open_ack should have followed (not preceded) the flush. + const ack = sent.map((s) => s.binary ? null : decodeJsonFrame(s.raw as string)).find((d) => d?.t === 'tcp_open_ack'); + if (!ack || ack.t !== 'tcp_open_ack') throw new Error('no ack'); + expect(ack.ok).toBe(true); + + switchboard.cleanup('test'); + } finally { + server.close(); + } + }); + + it('caps pending-data buffer and sends tcp_close when a peer overflows it before resolve completes', async () => { + // Resolver that never resolves; reservation is held open and any + // queued bytes accumulate against the cap (1 MiB). + const { switchboard, sent } = makeSwitchboard({ + resolve: () => new Promise(() => { /* never */ }), + }); + try { + switchboard.handleJsonFrame({ t: 'tcp_open', s: 2, stack: 's', service: 'svc', port: 80 }); + // 16 chunks of 96 KiB = 1.5 MiB total, crosses the 1 MiB cap on + // the chunk that pushes pendingBytes past the limit. + const chunkSize = 96 * 1024; + const payload = Buffer.alloc(chunkSize, 0x41); + for (let i = 0; i < 16; i++) { + const buf = encodeBinaryFrame(BinaryFrameType.TcpData, 2, payload); + switchboard.handleBinaryFrame(decodeBinaryFrame(buf)); + } + await new Promise((r) => setImmediate(r)); + + const close = sent.map((s) => s.binary ? null : decodeJsonFrame(s.raw as string)).find((d) => d?.t === 'tcp_close' && d.s === 2); + expect(close).toBeDefined(); + expect(switchboard.tcpStreamCount()).toBe(0); + } finally { + // Idle-timer + resolver closures both pin entry state across + // tests when the resolver never resolves; cleanup detaches them. + switchboard.cleanup('test'); + } + }); + it('tcp_close on a forward stream destroys the socket and drops the entry', async () => { const { port, server, firstConn } = await startEchoServer(); try { diff --git a/backend/src/__tests__/pilot-bridge-reverse.test.ts b/backend/src/__tests__/pilot-bridge-reverse.test.ts index f24e9d19..2c1c110c 100644 --- a/backend/src/__tests__/pilot-bridge-reverse.test.ts +++ b/backend/src/__tests__/pilot-bridge-reverse.test.ts @@ -17,7 +17,9 @@ import { WebSocket } from 'ws'; import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; import { AGENT_REVERSE_ID_BASE, + BinaryFrameType, decodeJsonFrame, + encodeBinaryFrame, encodeJsonFrame, } from '../pilot/protocol'; @@ -123,6 +125,58 @@ describe('PilotTunnelBridge handles tcp_open_reverse (Phase B)', () => { vi.restoreAllMocks(); }); + it('buffers TcpData arriving during the resolveContainerIp await and flushes it on connect (F-9 reverse race fix)', async () => { + const mockWs = makeMockTunnelWs(); + const bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket); + await bridge.start(); + + // Upstream that captures received bytes so we can assert the peer's + // request body was delivered (not silently dropped during the + // resolveContainerIp window). + let resolveReceived!: (buf: Buffer) => void; + const received = new Promise((r) => { resolveReceived = r; }); + const upstream = net.createServer((socket) => { + let acc = Buffer.alloc(0); + socket.on('data', (chunk: Buffer) => { + acc = Buffer.concat([acc, chunk]); + if (acc.length >= 'POST /hook HTTP/1.1\r\n\r\n'.length) resolveReceived(acc); + }); + }); + 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(); + // Delay the resolve to widen the race window; without the fix the + // TcpData frame below lands while this.streams has no entry for s + // and the bytes are dropped at the lookup-miss path in + // handleBinaryFrame, so the upstream never sees the request. + vi.spyOn(MeshService.getInstance(), 'resolveContainerIp') + .mockImplementation(() => new Promise((r) => setTimeout(() => r('127.0.0.1'), 50))); + + const s = AGENT_REVERSE_ID_BASE + 3; + mockWs.emit('message', encodeJsonFrame({ + t: 'tcp_open_reverse', s, + targetNodeId: localNodeId, stack: 'real', service: 'svc', port: upstreamPort, + }), false); + // Immediately push the "HTTP request" bytes while resolve is in flight. + mockWs.emit('message', + encodeBinaryFrame(BinaryFrameType.TcpData, s, Buffer.from('POST /hook HTTP/1.1\r\n\r\n')), + true); + + const body = await received; + expect(body.toString()).toBe('POST /hook HTTP/1.1\r\n\r\n'); + + // ack:true should also have landed (after the flush). + const ack = await waitFor(() => findAck(mockWs, s)); + expect(ack.ok).toBe(true); + + upstream.close(); + 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); diff --git a/backend/src/mesh/tcpStreamSwitchboard.ts b/backend/src/mesh/tcpStreamSwitchboard.ts index cf46f780..64c52477 100644 --- a/backend/src/mesh/tcpStreamSwitchboard.ts +++ b/backend/src/mesh/tcpStreamSwitchboard.ts @@ -9,6 +9,7 @@ import { MAX_STREAMS_PER_TUNNEL, MeshErrCode, STREAM_IDLE_TIMEOUT_MS, + STREAM_PENDING_DATA_MAX_BYTES, StreamIdAllocator, encodeBinaryFrame, encodeJsonFrame, @@ -62,8 +63,21 @@ export interface SwitchboardCtx { } interface ForwardTcpStream { - socket: net.Socket; + /** + * Null between the synchronous reservation at the top of `onTcpOpen` + * and the moment `net.createConnection` returns. Once assigned the + * socket is filled in for the remainder of the stream's life. + */ + socket: net.Socket | null; accepted: boolean; + /** + * `TcpData` payloads received before `socket` exists. Drained onto the + * socket inside the `'connect'` handler so frames that arrive during + * the resolve / dial window are delivered in order rather than + * silently dropped by the lookup-miss path in `handleBinaryFrame`. + */ + pendingData: Buffer[]; + pendingBytes: number; } /** @@ -160,7 +174,12 @@ export class TcpStreamSwitchboard { const ws = this.ctx.ws; const fwd = this.tcpStreams.get(streamId); if (fwd) { - try { fwd.socket.destroy(); } catch { /* ignore */ } + // socket is null if the idle timer somehow fires while still + // inside resolveTarget; the in-flight onTcpOpen will see the + // missing reservation and abort the dial. + if (fwd.socket) { + try { fwd.socket.destroy(); } catch { /* ignore */ } + } this.tcpStreams.delete(streamId); try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ } return; @@ -211,7 +230,25 @@ export class TcpStreamSwitchboard { } const fwd = this.tcpStreams.get(frame.streamId); if (!fwd) return true; - try { fwd.socket.write(frame.payload); } catch { /* ignore */ } + if (fwd.socket) { + try { fwd.socket.write(frame.payload); } catch { /* ignore */ } + } else { + // Reservation exists but the local socket has not been created + // yet (we are still inside the resolveTarget await in onTcpOpen). + // Buffer up to STREAM_PENDING_DATA_MAX_BYTES; over the cap, drop + // the stream and signal close so peer tears down rather than us + // OOMing. Copy the payload because decodeBinaryFrame returns a + // subarray view of the WS frame; holding the view would pin the + // parent buffer (up to MAX_FRAME_SIZE_BYTES) past its lifecycle. + if (fwd.pendingBytes + frame.payload.length > STREAM_PENDING_DATA_MAX_BYTES) { + this.tcpStreams.delete(frame.streamId); + this.clearIdleTimer(frame.streamId); + try { this.ctx.ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.streamId })); } catch { /* ignore */ } + return true; + } + fwd.pendingData.push(Buffer.from(frame.payload)); + fwd.pendingBytes += frame.payload.length; + } this.refreshIdleTimer(frame.streamId); return true; } @@ -225,19 +262,38 @@ export class TcpStreamSwitchboard { return; } + // Reserve the slot synchronously BEFORE the resolveTarget await so + // any `TcpData` frame the peer sends back-to-back with `tcp_open` + // (the common HTTP-request case) lands on a known entry instead of + // being dropped by the lookup-miss path in handleBinaryFrame. + // pendingData captures bytes until the socket is created + connected. + const entry: ForwardTcpStream = { socket: null, accepted: false, pendingData: [], pendingBytes: 0 }; + this.tcpStreams.set(frame.s, entry); + this.refreshIdleTimer(frame.s); + const target = await this.ctx.resolveTarget(frame.stack, frame.service, frame.port); if (!target.ok) { + // Drop the reservation before sending the rejection ack so the + // stream cap stays honest and any over-cap-evicted pending frames + // do not strand on an empty entry. + if (this.tcpStreams.get(frame.s) === entry) { + this.tcpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); + } try { ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: target.err })); } catch { /* ignore */ } return; } + // The reservation may have been evicted under the pending-bytes cap + // (see handleBinaryFrame) while we were resolving. If so, do not + // dial: the peer has already been told via tcp_close to tear down. + if (this.tcpStreams.get(frame.s) !== entry) return; + const socket = net.createConnection({ host: target.host, port: target.port }); socket.setTimeout(MESH_CONNECT_TIMEOUT_MS); - const entry: ForwardTcpStream = { socket, accepted: false }; - this.tcpStreams.set(frame.s, entry); - this.refreshIdleTimer(frame.s); + entry.socket = socket; const sendAck = (ok: boolean, err?: MeshErrCode) => { try { ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok, err })); } catch { /* ignore */ } @@ -246,6 +302,16 @@ export class TcpStreamSwitchboard { socket.once('connect', () => { entry.accepted = true; socket.setTimeout(0); + // Ack only after buffered bytes are queued on the socket so + // peer sees a clean "ack + data" sequence rather than racing + // post-ack data ahead of the request body it carried in. + if (entry.pendingData.length > 0) { + for (const buf of entry.pendingData) { + try { socket.write(buf); } catch { /* ignore */ } + } + entry.pendingData = []; + entry.pendingBytes = 0; + } sendAck(true); this.refreshIdleTimer(frame.s); }); @@ -300,7 +366,12 @@ export class TcpStreamSwitchboard { if (!entry) return; this.tcpStreams.delete(streamId); this.clearIdleTimer(streamId); - try { entry.socket.destroy(); } catch { /* ignore */ } + // socket may be null if central sent tcp_close while we were still + // inside resolveTarget; the reservation goes away and the in-flight + // onTcpOpen sees the missing entry and aborts the dial. + if (entry.socket) { + try { entry.socket.destroy(); } catch { /* ignore */ } + } } private onTcpOpenAckReverse(frame: { s: number; ok: boolean; err?: MeshErrCode }): void { @@ -369,7 +440,11 @@ export class TcpStreamSwitchboard { */ public cleanup(reason = 'mesh tunnel closed'): void { for (const [, entry] of this.tcpStreams) { - try { entry.socket.destroy(); } catch { /* ignore */ } + // socket is null for reservations whose onTcpOpen is still + // inside resolveTarget; pending buffers go away with the map. + if (entry.socket) { + try { entry.socket.destroy(); } catch { /* ignore */ } + } } this.tcpStreams.clear(); for (const [, handle] of this.reverseTcpStreams) { diff --git a/backend/src/pilot/protocol.ts b/backend/src/pilot/protocol.ts index 895d6251..bb123f16 100644 --- a/backend/src/pilot/protocol.ts +++ b/backend/src/pilot/protocol.ts @@ -50,6 +50,17 @@ export const MAX_STREAMS_PER_TUNNEL = 1024; */ export const STREAM_IDLE_TIMEOUT_MS = 10 * 60 * 1000; +/** + * Per-stream cap on inbound `TcpData` bytes that a receiver buffers while + * waiting for its local socket to connect. `tcp_open` and `tcp_open_reverse` + * both trigger an async dial (resolveTarget + TCP handshake), and the + * protocol allows the peer to send data immediately. Anything received in + * that window is held in `pendingData` until the socket is ready; over + * this cap the stream is dropped and a `tcp_close` is sent back so a + * misbehaving (or compromised) peer cannot OOM the receiver. + */ +export const STREAM_PENDING_DATA_MAX_BYTES = 1024 * 1024; + // --- Binary frame types (first byte of a binary WS frame) --- export enum BinaryFrameType { diff --git a/backend/src/services/PilotTunnelBridge.ts b/backend/src/services/PilotTunnelBridge.ts index 4bfb2bfb..b45ff80c 100644 --- a/backend/src/services/PilotTunnelBridge.ts +++ b/backend/src/services/PilotTunnelBridge.ts @@ -8,6 +8,7 @@ import { DecodedBinaryFrame, MAX_STREAMS_PER_TUNNEL, STREAM_IDLE_TIMEOUT_MS, + STREAM_PENDING_DATA_MAX_BYTES, StreamIdAllocator, decodeBinaryFrame, decodeJsonFrame, @@ -66,9 +67,18 @@ interface TcpStreamState extends StreamMeta { */ interface ReverseLocalTcpStreamState extends StreamMeta { kind: 'reverse_local'; - socket: Socket; + /** + * Null between the synchronous reservation in `handleTcpOpenReverse` + * and `net.createConnection` inside `acceptReverseLocal`. Filled in + * once the dial begins. Any `TcpData` for `s` arriving in that window + * is held in `pendingData` so the lookup-miss path in handleBinaryFrame + * does not silently drop the peer's request bytes. + */ + socket: Socket | null; bytesIn: number; bytesOut: number; + pendingData: Buffer[]; + pendingBytes: number; } /** @@ -557,7 +567,13 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle s.handle.emit('close'); } else if (s.kind === 'reverse_local') { this.removeStream(frame.s); - try { s.socket.destroy(); } catch { /* ignore */ } + // socket may be null if the peer sends tcp_close while + // resolveContainerIp is still in flight; the in-flight + // acceptReverseLocal sees the missing reservation and + // aborts the dial. + if (s.socket) { + try { s.socket.destroy(); } catch { /* ignore */ } + } } else if (s.kind === 'reverse_relay') { this.removeStream(frame.s); try { s.target.destroy(); } catch { /* ignore */ } @@ -602,7 +618,24 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle } else if (s.kind === 'reverse_local') { s.bytesIn += frame.payload.length; this.refreshIdleTimer(frame.streamId, s); - try { s.socket.write(frame.payload); } catch { /* ignore */ } + if (s.socket) { + try { s.socket.write(frame.payload); } catch { /* ignore */ } + } else { + // Reservation exists but the local socket is not + // created yet (still inside resolveContainerIp / + // pre-connect). Buffer up to the cap; over the cap, + // drop the stream and tell the peer to tear down. + // Copy the payload because decodeBinaryFrame returns + // a subarray view of the WS frame; holding the view + // would pin the parent buffer past its lifecycle. + if (s.pendingBytes + frame.payload.length > STREAM_PENDING_DATA_MAX_BYTES) { + this.removeStream(frame.streamId); + this.sendJson({ t: 'tcp_close', s: frame.streamId }); + break; + } + s.pendingData.push(Buffer.from(frame.payload)); + s.pendingBytes += frame.payload.length; + } } else if (s.kind === 'reverse_relay') { s.bytesIn += frame.payload.length; this.refreshIdleTimer(frame.streamId, s); @@ -732,7 +765,11 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle state.handle.emit('close'); } catch { /* ignore */ } } else if (state.kind === 'reverse_local') { - try { state.socket.destroy(); } catch { /* ignore */ } + // socket may be null if teardown fires while the resolve is + // still in flight; the pending buffer goes away with the state. + if (state.socket) { + try { state.socket.destroy(); } catch { /* ignore */ } + } } else if (state.kind === 'reverse_relay') { try { state.target.destroy(); } catch { /* ignore */ } } @@ -767,6 +804,20 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'agent_error' }); return; } + // Reserve the slot synchronously here, BEFORE the IIFE awaits the + // NodeRegistry / MeshService dynamic imports plus resolveContainerIp. + // Without this, a TcpData frame the agent sends back-to-back with + // tcp_open_reverse lands while this.streams is empty for `s` and is + // dropped by the lookup-miss path in handleBinaryFrame. The + // reservation is reverse_local-shaped because that is the common + // case; the relay path discards it and sets up its own state when + // it discovers targetNodeId is remote. + const reservation: ReverseLocalTcpStreamState = { + kind: 'reverse_local', socket: null, + bytesIn: 0, bytesOut: 0, pendingData: [], pendingBytes: 0, + }; + this.streams.set(s, reservation); + this.refreshIdleTimer(s, reservation); // Lazy-import services that import this module to avoid a cycle. void (async () => { const { NodeRegistry } = await import('./NodeRegistry'); @@ -774,15 +825,28 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle if (targetNodeId === localNodeId) { await this.acceptReverseLocal(s, { stack, service, port }); } else { + // Relay path does not yet buffer early data (separate + // follow-up); drop the local-shaped reservation so the + // relay state machine starts from a clean slot. Any frames + // buffered up to this point are discarded with it. + if (this.streams.get(s) === reservation) this.removeStream(s); 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)); + if (this.streams.get(s) === reservation) this.removeStream(s); 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 { + // The reservation was placed in `this.streams` synchronously by + // handleTcpOpenReverse so any TcpData arriving in the resolve + // window is already buffered in state.pendingData by + // handleBinaryFrame. Recover it here and abort if it's gone + // (cleanup ran, idle-evicted, or a concurrent close removed it). + const state = this.streams.get(s); + if (!state || state.kind !== 'reverse_local') return; const { MeshService } = await import('./MeshService'); const meshSvc = MeshService.getInstance(); // Shared discriminator + target metadata so the Routing tab can @@ -798,6 +862,10 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle }; const ip = await meshSvc.resolveContainerIp({ stack: target.stack, service: target.service }); if (!ip) { + // Drop the reservation before acking the failure so the stream + // map stays honest and any over-cap-evicted pending frames do + // not strand on an empty entry. + if (this.streams.get(s) === state) this.removeStream(s); meshSvc.logActivity({ source: 'mesh', level: 'error', type: 'route.resolve.fail', nodeId: this.nodeId, @@ -807,10 +875,14 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'no_target' }); return; } + // The reservation may have been evicted under the pending-bytes cap + // (see handleBinaryFrame's reverse_local branch) while we were + // resolving. If so, do not dial: the peer has already been told via + // tcp_close to tear down. + if (this.streams.get(s) !== state) 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); + state.socket = socket; const teardown = (sendClose: boolean) => { if (!this.streams.has(s)) return; @@ -842,6 +914,16 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle message: `reverse dial ok: ${target.stack}/${target.service}:${target.port}`, details: baseDetails, }); + // Ack only after buffered bytes are queued on the socket so + // peer sees a clean "ack + data" sequence rather than racing + // post-ack data ahead of the request body it carried in. + if (state.pendingData.length > 0) { + for (const buf of state.pendingData) { + try { socket.write(buf); } catch { /* ignore */ } + } + state.pendingData = []; + state.pendingBytes = 0; + } this.sendJson({ t: 'tcp_open_ack', s, ok: true }); socket.on('data', (chunk: Buffer) => { const cur = this.streams.get(s);