serve.zone/remoteingress
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Jürgen Kunz
2026-03-17 15:36:23 +00:00
parent 236d6d16ee
commit 8b5df9a0b7
4 changed files with 100 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
rust/crates/remoteingress-core/src/edge.rs
View File

@@ -519,6 +519,7 @@ async fn connect_to_hub_and_run(
// Single-owner I/O engine — no tokio::io::split, no mutex // Single-owner I/O engine — no tokio::io::split, no mutex
let mut tunnel_io = remoteingress_protocol::TunnelIo::new(tls_stream, Vec::new()); let mut tunnel_io = remoteingress_protocol::TunnelIo::new(tls_stream, Vec::new());
let liveness_timeout_dur = Duration::from_secs(45); let liveness_timeout_dur = Duration::from_secs(45);
let mut last_activity = Instant::now(); let mut last_activity = Instant::now();
let mut liveness_deadline = Box::pin(sleep_until(last_activity + liveness_timeout_dur)); let mut liveness_deadline = Box::pin(sleep_until(last_activity + liveness_timeout_dur));

1
rust/crates/remoteingress-core/src/hub.rs
View File

@@ -755,6 +755,7 @@ async fn handle_edge_connection(
// Single-owner I/O engine — no tokio::io::split, no mutex // Single-owner I/O engine — no tokio::io::split, no mutex
let mut tunnel_io = remoteingress_protocol::TunnelIo::new(tls_stream, Vec::new()); 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. // Assigned in every break path of the hub_loop before use at the end.
#[allow(unused_assignments)] #[allow(unused_assignments)]
let mut disconnect_reason = String::new(); let mut disconnect_reason = String::new();

76
rust/crates/remoteingress-protocol/src/lib.rs
View File

@@ -183,6 +183,11 @@ pub enum TunnelEvent {
Cancelled, 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 /// Write state extracted into a sub-struct so the borrow checker can see
/// disjoint field access between `self.write` and `self.stream`. /// disjoint field access between `self.write` and `self.stream`.
struct WriteState { struct WriteState {
@@ -190,6 +195,7 @@ struct WriteState {
data_queue: VecDeque<Vec<u8>>, // DATA, DATA_BACK — only when ctrl is empty data_queue: VecDeque<Vec<u8>>, // DATA, DATA_BACK — only when ctrl is empty
offset: usize, // progress within current frame being written offset: usize, // progress within current frame being written
flush_needed: bool, flush_needed: bool,
unflushed_bytes: usize, // bytes written to TLS since last successful flush
} }
impl WriteState { impl WriteState {
@@ -231,6 +237,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
data_queue: VecDeque::new(), data_queue: VecDeque::new(),
offset: 0, offset: 0,
flush_needed: false, flush_needed: false,
unflushed_bytes: 0,
}, },
} }
} }
@@ -312,12 +319,15 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
cancel_token: &tokio_util::sync::CancellationToken, cancel_token: &tokio_util::sync::CancellationToken,
) -> Poll<TunnelEvent> { ) -> Poll<TunnelEvent> {
// 1. WRITE: drain ctrl queue first, then data queue. // 1. WRITE: drain ctrl queue first, then data queue.
// Only write when flush is complete — otherwise the TLS session buffer // Allow up to MAX_UNFLUSHED_BYTES in the TLS session buffer before
// grows without bound (poll_write always returns Ready, buffering plaintext // requiring a flush. This keeps the pipe saturated (unlike waiting for
// in the TLS session even when TCP can't keep up). // flush to complete) while preventing unbounded buffer growth (OOM).
// Safe: `self.write` and `self.stream` are disjoint fields. // Safe: `self.write` and `self.stream` are disjoint fields.
let mut writes = 0; 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 from_ctrl = !self.write.ctrl_queue.is_empty();
let frame = if from_ctrl { let frame = if from_ctrl {
self.write.ctrl_queue.front().unwrap() self.write.ctrl_queue.front().unwrap()
@@ -328,6 +338,8 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
match Pin::new(&mut self.stream).poll_write(cx, remaining) { match Pin::new(&mut self.stream).poll_write(cx, remaining) {
Poll::Ready(Ok(0)) => { 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( return Poll::Ready(TunnelEvent::WriteError(
std::io::Error::new(std::io::ErrorKind::WriteZero, "write zero"), std::io::Error::new(std::io::ErrorKind::WriteZero, "write zero"),
)); ));
@@ -335,6 +347,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
Poll::Ready(Ok(n)) => { Poll::Ready(Ok(n)) => {
self.write.offset += n; self.write.offset += n;
self.write.flush_needed = true; self.write.flush_needed = true;
self.write.unflushed_bytes += n;
if self.write.offset >= frame.len() { if self.write.offset >= frame.len() {
if from_ctrl { self.write.ctrl_queue.pop_front(); } if from_ctrl { self.write.ctrl_queue.pop_front(); }
else { self.write.data_queue.pop_front(); } else { self.write.data_queue.pop_front(); }
@@ -342,7 +355,11 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
writes += 1; 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, Poll::Pending => break,
} }
} }
@@ -350,8 +367,14 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
// 2. FLUSH: push encrypted data from TLS session to TCP. // 2. FLUSH: push encrypted data from TLS session to TCP.
if self.write.flush_needed { if self.write.flush_needed {
match Pin::new(&mut self.stream).poll_flush(cx) { match Pin::new(&mut self.stream).poll_flush(cx) {
Poll::Ready(Ok(())) => self.write.flush_needed = false, Poll::Ready(Ok(())) => {
Poll::Ready(Err(e)) => return Poll::Ready(TunnelEvent::WriteError(e)), 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 Poll::Pending => {} // TCP waker will notify us
} }
} }
@@ -387,12 +410,19 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
// Partial data — loop to call poll_read again so the TCP // Partial data — loop to call poll_read again so the TCP
// waker is re-registered when it finally returns Pending. // 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, 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; let mut got_new = false;
loop { loop {
match ctrl_rx.poll_recv(cx) { match ctrl_rx.poll_recv(cx) {
@@ -405,15 +435,17 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
Poll::Pending => break, Poll::Pending => break,
} }
} }
loop { if self.write.data_queue.len() < 64 {
match data_rx.poll_recv(cx) { loop {
Poll::Ready(Some(frame)) => { self.write.data_queue.push_back(frame); got_new = true; } match data_rx.poll_recv(cx) {
Poll::Ready(None) => { Poll::Ready(Some(frame)) => { self.write.data_queue.push_back(frame); got_new = true; }
return Poll::Ready(TunnelEvent::WriteError( Poll::Ready(None) => {
std::io::Error::new(std::io::ErrorKind::BrokenPipe, "data channel closed"), 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<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
} }
// 6. SELF-WAKE: only when flush is complete AND we have work. // 6. SELF-WAKE: only when flush is complete AND we have work.
// If flush is pending, the TCP write-readiness waker will notify us. // When flush is Pending, the TCP write-readiness waker will notify us.
// CRITICAL: do NOT self-wake when flush_needed — this causes unbounded // CRITICAL: do NOT self-wake when flush_needed — poll_write always returns
// TLS session buffer growth (poll_write always accepts plaintext, but TCP // Ready (TLS buffers in-memory), so self-waking causes a tight spin loop
// can't drain it fast enough → OOM → process killed → ECONNRESET). // 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()) { if !self.write.flush_needed && (got_new || self.write.has_work()) {
cx.waker().wake_by_ref(); cx.waker().wake_by_ref();
} }

77
test/test.loadtest.node.ts
View File

@@ -142,7 +142,7 @@ class ThrottleTransform extends stream.Transform {
this.bucket = 0; this.bucket = 0;
const delayMs = Math.min((deficit / this.bytesPerSec) * 1000, 1000); const delayMs = Math.min((deficit / this.bytesPerSec) * 1000, 1000);
setTimeout(() => { setTimeout(() => {
if (this.destroyed_) return; if (this.destroyed_) { callback(); return; }
this.lastRefill = Date.now(); this.lastRefill = Date.now();
this.bucket = 0; this.bucket = 0;
callback(null, chunk); callback(null, chunk);
@@ -179,7 +179,16 @@ async function startThrottleProxy(
clientSock.pipe(throttleUp).pipe(upstream); clientSock.pipe(throttleUp).pipe(upstream);
upstream.pipe(throttleDown).pipe(clientSock); 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(); throttleUp.destroy();
throttleDown.destroy(); throttleDown.destroy();
clientSock.destroy(); clientSock.destroy();
@@ -187,12 +196,12 @@ async function startThrottleProxy(
connections.delete(clientSock); connections.delete(clientSock);
connections.delete(upstream); connections.delete(upstream);
}; };
clientSock.on('error', cleanup); clientSock.on('error', (e) => cleanup('clientSock.error', e));
upstream.on('error', cleanup); upstream.on('error', (e) => cleanup('upstream.error', e));
throttleUp.on('error', cleanup); throttleUp.on('error', (e) => cleanup('throttleUp.error', e));
throttleDown.on('error', cleanup); throttleDown.on('error', (e) => cleanup('throttleDown.error', e));
clientSock.on('close', cleanup); clientSock.on('close', () => cleanup('clientSock.close'));
upstream.on('close', cleanup); upstream.on('close', () => cleanup('upstream.close'));
}); });
await new Promise<void>((resolve) => server.listen(listenPort, '127.0.0.1', resolve)); await new Promise<void>((resolve) => server.listen(listenPort, '127.0.0.1', resolve));
@@ -222,13 +231,13 @@ let edgePort: number;
// Tests // 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); [hubPort, proxyPort, edgePort] = await findFreePorts(3);
echoServer = await startEchoServer(edgePort, '127.0.0.2'); echoServer = await startEchoServer(edgePort, '127.0.0.2');
// Throttle proxy: edge → proxy → hub at 20 Mbit/s (2.5 MB/s) // Throttle proxy: edge → proxy → hub at 100 Mbit/s (12.5 MB/s)
throttle = await startThrottleProxy(proxyPort, '127.0.0.1', hubPort, 2.5 * 1024 * 1024); throttle = await startThrottleProxy(proxyPort, '127.0.0.1', hubPort, 12.5 * 1024 * 1024);
hub = new RemoteIngressHub(); hub = new RemoteIngressHub();
edge = new RemoteIngressEdge(); 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({ await edge.start({
hubHost: '127.0.0.1', hubHost: '127.0.0.1',
hubPort: proxyPort, hubPort: proxyPort,
@@ -262,12 +271,12 @@ tap.test('setup: start throttled tunnel (20 Mbit/s)', async () => {
expect(status.connected).toBeTrue(); expect(status.connected).toBeTrue();
}); });
tap.test('throttled: 10 streams x 50MB each through 10MB/s tunnel', async () => { tap.test('throttled: 5 streams x 20MB each through 100Mbit tunnel', async () => {
const streamCount = 10; const streamCount = 5;
const payloadSize = 50 * 1024 * 1024; // 50MB per stream = 500MB total round-trip const payloadSize = 20 * 1024 * 1024; // 20MB per stream = 100MB total round-trip
const promises = Array.from({ length: streamCount }, () => { const payloads = Array.from({ length: streamCount }, () => crypto.randomBytes(payloadSize));
const data = crypto.randomBytes(payloadSize); const promises = payloads.map((data) => {
const hash = sha256(data); const hash = sha256(data);
return sendAndReceive(edgePort, data, 300000).then((received) => ({ return sendAndReceive(edgePort, data, 300000).then((received) => ({
sent: hash, sent: hash,
@@ -284,23 +293,23 @@ tap.test('throttled: 10 streams x 50MB each through 10MB/s tunnel', async () =>
expect(status.connected).toBeTrue(); expect(status.connected).toBeTrue();
}); });
tap.test('throttled: slow consumer with 50MB does not kill other streams', async () => { tap.test('throttled: slow consumer with 20MB does not kill other streams', async () => {
// Open a connection that creates massive download-direction backpressure: // Open a connection that creates download-direction backpressure:
// send 50MB but DON'T read the response — client TCP receive buffer fills // send 20MB but DON'T read the response — client TCP receive buffer fills
const slowSock = net.createConnection({ host: '127.0.0.1', port: edgePort }); const slowSock = net.createConnection({ host: '127.0.0.1', port: edgePort });
await new Promise<void>((resolve) => slowSock.on('connect', resolve)); await new Promise<void>((resolve) => slowSock.on('connect', resolve));
const slowData = crypto.randomBytes(50 * 1024 * 1024); const slowData = crypto.randomBytes(20 * 1024 * 1024);
slowSock.write(slowData); slowSock.write(slowData);
slowSock.end(); slowSock.end();
// Don't read — backpressure builds on the download path // Don't read — backpressure builds on the download path
// Wait for backpressure to develop // 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 // Meanwhile, 5 normal echo streams with 20MB each must complete
const payload = crypto.randomBytes(50 * 1024 * 1024); const payload = crypto.randomBytes(20 * 1024 * 1024);
const hash = sha256(payload); const hash = sha256(payload);
const promises = Array.from({ length: 10 }, () => const promises = Array.from({ length: 5 }, () =>
sendAndReceive(edgePort, payload, 300000).then((r) => ({ sendAndReceive(edgePort, payload, 300000).then((r) => ({
hash: sha256(r), hash: sha256(r),
sizeOk: r.length === payload.length, sizeOk: r.length === payload.length,
@@ -317,11 +326,11 @@ tap.test('throttled: slow consumer with 50MB does not kill other streams', async
slowSock.destroy(); slowSock.destroy();
}); });
tap.test('throttled: rapid churn — 5 x 50MB long + 200 x 1MB short streams', async () => { tap.test('throttled: rapid churn — 3 x 20MB long + 50 x 1MB short streams', async () => {
// 5 long streams (50MB each) running alongside 200 short streams (1MB each) // 3 long streams (20MB each) running alongside 50 short streams (1MB each)
const longPayload = crypto.randomBytes(50 * 1024 * 1024); const longPayload = crypto.randomBytes(20 * 1024 * 1024);
const longHash = sha256(longPayload); const longHash = sha256(longPayload);
const longPromises = Array.from({ length: 5 }, () => const longPromises = Array.from({ length: 3 }, () =>
sendAndReceive(edgePort, longPayload, 300000).then((r) => ({ sendAndReceive(edgePort, longPayload, 300000).then((r) => ({
hash: sha256(r), hash: sha256(r),
sizeOk: r.length === longPayload.length, 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 shortPayload = crypto.randomBytes(1024 * 1024);
const shortHash = sha256(shortPayload); const shortHash = sha256(shortPayload);
const shortPromises = Array.from({ length: 200 }, () => const shortPromises = Array.from({ length: 50 }, () =>
sendAndReceive(edgePort, shortPayload, 300000).then((r) => ({ sendAndReceive(edgePort, shortPayload, 300000).then((r) => ({
hash: sha256(r), hash: sha256(r),
sizeOk: r.length === shortPayload.length, 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(); expect(status.connected).toBeTrue();
}); });
tap.test('throttled: 5 burst waves of 20 streams x 50MB each', async () => { tap.test('throttled: 3 burst waves of 5 streams x 20MB each', async () => {
for (let wave = 0; wave < 5; wave++) { for (let wave = 0; wave < 3; wave++) {
const streamCount = 20; const streamCount = 5;
const payloadSize = 50 * 1024 * 1024; // 50MB per stream = 1GB per wave const payloadSize = 20 * 1024 * 1024; // 20MB per stream = 100MB per wave
const promises = Array.from({ length: streamCount }, () => { const promises = Array.from({ length: streamCount }, () => {
const data = crypto.randomBytes(payloadSize); const data = crypto.randomBytes(payloadSize);