From 8b5df9a0b76c741ee58e450aede2a4fb2926abb5 Mon Sep 17 00:00:00 2001 From: Juergen Kunz Date: Tue, 17 Mar 2026 15:36:23 +0000 Subject: [PATCH] update --- rust/crates/remoteingress-core/src/edge.rs | 1 + rust/crates/remoteingress-core/src/hub.rs | 1 + rust/crates/remoteingress-protocol/src/lib.rs | 76 +++++++++++++----- test/test.loadtest.node.ts | 77 +++++++++++-------- 4 files changed, 100 insertions(+), 55 deletions(-) diff --git a/rust/crates/remoteingress-core/src/edge.rs b/rust/crates/remoteingress-core/src/edge.rs index 4c30979..104bbd5 100644 --- a/rust/crates/remoteingress-core/src/edge.rs +++ b/rust/crates/remoteingress-core/src/edge.rs @@ -519,6 +519,7 @@ async fn connect_to_hub_and_run( // Single-owner I/O engine — no tokio::io::split, no mutex let mut tunnel_io = remoteingress_protocol::TunnelIo::new(tls_stream, Vec::new()); + let liveness_timeout_dur = Duration::from_secs(45); let mut last_activity = Instant::now(); let mut liveness_deadline = Box::pin(sleep_until(last_activity + liveness_timeout_dur)); diff --git a/rust/crates/remoteingress-core/src/hub.rs b/rust/crates/remoteingress-core/src/hub.rs index 7e9abd9..9aa9803 100644 --- a/rust/crates/remoteingress-core/src/hub.rs +++ b/rust/crates/remoteingress-core/src/hub.rs @@ -755,6 +755,7 @@ async fn handle_edge_connection( // Single-owner I/O engine — no tokio::io::split, no mutex let mut tunnel_io = remoteingress_protocol::TunnelIo::new(tls_stream, Vec::new()); + // Assigned in every break path of the hub_loop before use at the end. #[allow(unused_assignments)] let mut disconnect_reason = String::new(); diff --git a/rust/crates/remoteingress-protocol/src/lib.rs b/rust/crates/remoteingress-protocol/src/lib.rs index a5c8595..26f60a9 100644 --- a/rust/crates/remoteingress-protocol/src/lib.rs +++ b/rust/crates/remoteingress-protocol/src/lib.rs @@ -183,6 +183,11 @@ pub enum TunnelEvent { Cancelled, } +/// Maximum bytes written to TLS session buffer before requiring a flush. +/// Prevents unbounded buffer growth (OOM) while keeping the pipe saturated. +/// 256KB ≈ typical TCP send buffer — enough to fill the socket on each flush. +const MAX_UNFLUSHED_BYTES: usize = 256 * 1024; + /// Write state extracted into a sub-struct so the borrow checker can see /// disjoint field access between `self.write` and `self.stream`. struct WriteState { @@ -190,6 +195,7 @@ struct WriteState { data_queue: VecDeque>, // DATA, DATA_BACK — only when ctrl is empty offset: usize, // progress within current frame being written flush_needed: bool, + unflushed_bytes: usize, // bytes written to TLS since last successful flush } impl WriteState { @@ -231,6 +237,7 @@ impl TunnelIo { data_queue: VecDeque::new(), offset: 0, flush_needed: false, + unflushed_bytes: 0, }, } } @@ -312,12 +319,15 @@ impl TunnelIo { cancel_token: &tokio_util::sync::CancellationToken, ) -> Poll { // 1. WRITE: drain ctrl queue first, then data queue. - // Only write when flush is complete — otherwise the TLS session buffer - // grows without bound (poll_write always returns Ready, buffering plaintext - // in the TLS session even when TCP can't keep up). + // Allow up to MAX_UNFLUSHED_BYTES in the TLS session buffer before + // requiring a flush. This keeps the pipe saturated (unlike waiting for + // flush to complete) while preventing unbounded buffer growth (OOM). // Safe: `self.write` and `self.stream` are disjoint fields. let mut writes = 0; - while self.write.has_work() && writes < 16 && !self.write.flush_needed { + while self.write.has_work() && writes < 16 + && self.write.unflushed_bytes < MAX_UNFLUSHED_BYTES + && !self.write.flush_needed + { let from_ctrl = !self.write.ctrl_queue.is_empty(); let frame = if from_ctrl { self.write.ctrl_queue.front().unwrap() @@ -328,6 +338,8 @@ impl TunnelIo { match Pin::new(&mut self.stream).poll_write(cx, remaining) { Poll::Ready(Ok(0)) => { + log::error!("TunnelIo: poll_write returned 0 (write zero), ctrl_q={} data_q={} unflushed={}", + self.write.ctrl_queue.len(), self.write.data_queue.len(), self.write.unflushed_bytes); return Poll::Ready(TunnelEvent::WriteError( std::io::Error::new(std::io::ErrorKind::WriteZero, "write zero"), )); @@ -335,6 +347,7 @@ impl TunnelIo { Poll::Ready(Ok(n)) => { self.write.offset += n; self.write.flush_needed = true; + self.write.unflushed_bytes += n; if self.write.offset >= frame.len() { if from_ctrl { self.write.ctrl_queue.pop_front(); } else { self.write.data_queue.pop_front(); } @@ -342,7 +355,11 @@ impl TunnelIo { writes += 1; } } - Poll::Ready(Err(e)) => return Poll::Ready(TunnelEvent::WriteError(e)), + Poll::Ready(Err(e)) => { + log::error!("TunnelIo: poll_write error: {} (ctrl_q={} data_q={} unflushed={})", + e, self.write.ctrl_queue.len(), self.write.data_queue.len(), self.write.unflushed_bytes); + return Poll::Ready(TunnelEvent::WriteError(e)); + } Poll::Pending => break, } } @@ -350,8 +367,14 @@ impl TunnelIo { // 2. FLUSH: push encrypted data from TLS session to TCP. if self.write.flush_needed { match Pin::new(&mut self.stream).poll_flush(cx) { - Poll::Ready(Ok(())) => self.write.flush_needed = false, - Poll::Ready(Err(e)) => return Poll::Ready(TunnelEvent::WriteError(e)), + Poll::Ready(Ok(())) => { + self.write.flush_needed = false; + self.write.unflushed_bytes = 0; + } + Poll::Ready(Err(e)) => { + log::error!("TunnelIo: poll_flush error: {} (unflushed={})", e, self.write.unflushed_bytes); + return Poll::Ready(TunnelEvent::WriteError(e)); + } Poll::Pending => {} // TCP waker will notify us } } @@ -387,12 +410,19 @@ impl TunnelIo { // Partial data — loop to call poll_read again so the TCP // waker is re-registered when it finally returns Pending. } - Poll::Ready(Err(e)) => return Poll::Ready(TunnelEvent::ReadError(e)), + Poll::Ready(Err(e)) => { + log::error!("TunnelIo: poll_read error: {}", e); + return Poll::Ready(TunnelEvent::ReadError(e)); + } Poll::Pending => break, } } - // 4. CHANNELS: drain ctrl into ctrl_queue, data into data_queue. + // 4. CHANNELS: drain ctrl (always — priority), data (only if queue is small). + // Ctrl frames must never be delayed — always drain fully. + // Data frames are gated: keep data in the bounded channel for proper + // backpressure when TLS writes are slow. Without this gate, the internal + // data_queue (unbounded VecDeque) grows to hundreds of MB under throttle → OOM. let mut got_new = false; loop { match ctrl_rx.poll_recv(cx) { @@ -405,15 +435,17 @@ impl TunnelIo { Poll::Pending => break, } } - loop { - match data_rx.poll_recv(cx) { - Poll::Ready(Some(frame)) => { self.write.data_queue.push_back(frame); got_new = true; } - Poll::Ready(None) => { - return Poll::Ready(TunnelEvent::WriteError( - std::io::Error::new(std::io::ErrorKind::BrokenPipe, "data channel closed"), - )); + if self.write.data_queue.len() < 64 { + loop { + match data_rx.poll_recv(cx) { + Poll::Ready(Some(frame)) => { self.write.data_queue.push_back(frame); got_new = true; } + Poll::Ready(None) => { + return Poll::Ready(TunnelEvent::WriteError( + std::io::Error::new(std::io::ErrorKind::BrokenPipe, "data channel closed"), + )); + } + Poll::Pending => break, } - Poll::Pending => break, } } @@ -426,10 +458,12 @@ impl TunnelIo { } // 6. SELF-WAKE: only when flush is complete AND we have work. - // If flush is pending, the TCP write-readiness waker will notify us. - // CRITICAL: do NOT self-wake when flush_needed — this causes unbounded - // TLS session buffer growth (poll_write always accepts plaintext, but TCP - // can't drain it fast enough → OOM → process killed → ECONNRESET). + // When flush is Pending, the TCP write-readiness waker will notify us. + // CRITICAL: do NOT self-wake when flush_needed — poll_write always returns + // Ready (TLS buffers in-memory), so self-waking causes a tight spin loop + // that fills the TLS session buffer unboundedly → OOM → ECONNRESET. + // The write loop's MAX_UNFLUSHED_BYTES gate allows up to 64KB per poll_step + // even across flush boundaries, keeping the pipe saturated without spinning. if !self.write.flush_needed && (got_new || self.write.has_work()) { cx.waker().wake_by_ref(); } diff --git a/test/test.loadtest.node.ts b/test/test.loadtest.node.ts index 450f1d6..e10d3cd 100644 --- a/test/test.loadtest.node.ts +++ b/test/test.loadtest.node.ts @@ -142,7 +142,7 @@ class ThrottleTransform extends stream.Transform { this.bucket = 0; const delayMs = Math.min((deficit / this.bytesPerSec) * 1000, 1000); setTimeout(() => { - if (this.destroyed_) return; + if (this.destroyed_) { callback(); return; } this.lastRefill = Date.now(); this.bucket = 0; callback(null, chunk); @@ -179,7 +179,16 @@ async function startThrottleProxy( clientSock.pipe(throttleUp).pipe(upstream); upstream.pipe(throttleDown).pipe(clientSock); - const cleanup = () => { + let cleaned = false; + const cleanup = (source: string, err?: Error) => { + if (cleaned) return; + cleaned = true; + if (err) { + console.error(`[ThrottleProxy] cleanup triggered by ${source}: ${err.message}`); + } else { + console.error(`[ThrottleProxy] cleanup triggered by ${source} (no error)`); + } + console.error(`[ThrottleProxy] stack:`, new Error().stack); throttleUp.destroy(); throttleDown.destroy(); clientSock.destroy(); @@ -187,12 +196,12 @@ async function startThrottleProxy( connections.delete(clientSock); connections.delete(upstream); }; - clientSock.on('error', cleanup); - upstream.on('error', cleanup); - throttleUp.on('error', cleanup); - throttleDown.on('error', cleanup); - clientSock.on('close', cleanup); - upstream.on('close', cleanup); + clientSock.on('error', (e) => cleanup('clientSock.error', e)); + upstream.on('error', (e) => cleanup('upstream.error', e)); + throttleUp.on('error', (e) => cleanup('throttleUp.error', e)); + throttleDown.on('error', (e) => cleanup('throttleDown.error', e)); + clientSock.on('close', () => cleanup('clientSock.close')); + upstream.on('close', () => cleanup('upstream.close')); }); await new Promise((resolve) => server.listen(listenPort, '127.0.0.1', resolve)); @@ -222,13 +231,13 @@ let edgePort: number; // Tests // --------------------------------------------------------------------------- -tap.test('setup: start throttled tunnel (20 Mbit/s)', async () => { +tap.test('setup: start throttled tunnel (100 Mbit/s)', async () => { [hubPort, proxyPort, edgePort] = await findFreePorts(3); echoServer = await startEchoServer(edgePort, '127.0.0.2'); - // Throttle proxy: edge → proxy → hub at 20 Mbit/s (2.5 MB/s) - throttle = await startThrottleProxy(proxyPort, '127.0.0.1', hubPort, 2.5 * 1024 * 1024); + // Throttle proxy: edge → proxy → hub at 100 Mbit/s (12.5 MB/s) + throttle = await startThrottleProxy(proxyPort, '127.0.0.1', hubPort, 12.5 * 1024 * 1024); hub = new RemoteIngressHub(); edge = new RemoteIngressEdge(); @@ -246,7 +255,7 @@ tap.test('setup: start throttled tunnel (20 Mbit/s)', async () => { }); }); - // Edge connects to proxy, not hub directly + // Edge connects through throttle proxy await edge.start({ hubHost: '127.0.0.1', hubPort: proxyPort, @@ -262,12 +271,12 @@ tap.test('setup: start throttled tunnel (20 Mbit/s)', async () => { expect(status.connected).toBeTrue(); }); -tap.test('throttled: 10 streams x 50MB each through 10MB/s tunnel', async () => { - const streamCount = 10; - const payloadSize = 50 * 1024 * 1024; // 50MB per stream = 500MB total round-trip +tap.test('throttled: 5 streams x 20MB each through 100Mbit tunnel', async () => { + const streamCount = 5; + const payloadSize = 20 * 1024 * 1024; // 20MB per stream = 100MB total round-trip - const promises = Array.from({ length: streamCount }, () => { - const data = crypto.randomBytes(payloadSize); + const payloads = Array.from({ length: streamCount }, () => crypto.randomBytes(payloadSize)); + const promises = payloads.map((data) => { const hash = sha256(data); return sendAndReceive(edgePort, data, 300000).then((received) => ({ sent: hash, @@ -284,23 +293,23 @@ tap.test('throttled: 10 streams x 50MB each through 10MB/s tunnel', async () => expect(status.connected).toBeTrue(); }); -tap.test('throttled: slow consumer with 50MB does not kill other streams', async () => { - // Open a connection that creates massive download-direction backpressure: - // send 50MB but DON'T read the response — client TCP receive buffer fills +tap.test('throttled: slow consumer with 20MB does not kill other streams', async () => { + // Open a connection that creates download-direction backpressure: + // send 20MB but DON'T read the response — client TCP receive buffer fills const slowSock = net.createConnection({ host: '127.0.0.1', port: edgePort }); await new Promise((resolve) => slowSock.on('connect', resolve)); - const slowData = crypto.randomBytes(50 * 1024 * 1024); + const slowData = crypto.randomBytes(20 * 1024 * 1024); slowSock.write(slowData); slowSock.end(); // Don't read — backpressure builds on the download path // Wait for backpressure to develop - await new Promise((r) => setTimeout(r, 3000)); + await new Promise((r) => setTimeout(r, 2000)); - // Meanwhile, 10 normal echo streams with 50MB each must complete - const payload = crypto.randomBytes(50 * 1024 * 1024); + // Meanwhile, 5 normal echo streams with 20MB each must complete + const payload = crypto.randomBytes(20 * 1024 * 1024); const hash = sha256(payload); - const promises = Array.from({ length: 10 }, () => + const promises = Array.from({ length: 5 }, () => sendAndReceive(edgePort, payload, 300000).then((r) => ({ hash: sha256(r), sizeOk: r.length === payload.length, @@ -317,11 +326,11 @@ tap.test('throttled: slow consumer with 50MB does not kill other streams', async slowSock.destroy(); }); -tap.test('throttled: rapid churn — 5 x 50MB long + 200 x 1MB short streams', async () => { - // 5 long streams (50MB each) running alongside 200 short streams (1MB each) - const longPayload = crypto.randomBytes(50 * 1024 * 1024); +tap.test('throttled: rapid churn — 3 x 20MB long + 50 x 1MB short streams', async () => { + // 3 long streams (20MB each) running alongside 50 short streams (1MB each) + const longPayload = crypto.randomBytes(20 * 1024 * 1024); const longHash = sha256(longPayload); - const longPromises = Array.from({ length: 5 }, () => + const longPromises = Array.from({ length: 3 }, () => sendAndReceive(edgePort, longPayload, 300000).then((r) => ({ hash: sha256(r), sizeOk: r.length === longPayload.length, @@ -330,7 +339,7 @@ tap.test('throttled: rapid churn — 5 x 50MB long + 200 x 1MB short streams', a const shortPayload = crypto.randomBytes(1024 * 1024); const shortHash = sha256(shortPayload); - const shortPromises = Array.from({ length: 200 }, () => + const shortPromises = Array.from({ length: 50 }, () => sendAndReceive(edgePort, shortPayload, 300000).then((r) => ({ hash: sha256(r), sizeOk: r.length === shortPayload.length, @@ -351,10 +360,10 @@ tap.test('throttled: rapid churn — 5 x 50MB long + 200 x 1MB short streams', a expect(status.connected).toBeTrue(); }); -tap.test('throttled: 5 burst waves of 20 streams x 50MB each', async () => { - for (let wave = 0; wave < 5; wave++) { - const streamCount = 20; - const payloadSize = 50 * 1024 * 1024; // 50MB per stream = 1GB per wave +tap.test('throttled: 3 burst waves of 5 streams x 20MB each', async () => { + for (let wave = 0; wave < 3; wave++) { + const streamCount = 5; + const payloadSize = 20 * 1024 * 1024; // 20MB per stream = 100MB per wave const promises = Array.from({ length: streamCount }, () => { const data = crypto.randomBytes(payloadSize);