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

fix(remoteingress-protocol): require a flush after each written frame to bound TLS buffer growth

Browse Source
This commit is contained in:
Jürgen Kunz
2026-03-17 15:50:47 +00:00
parent 8b5df9a0b7
commit 3c2299430a
3 changed files with 17 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
changelog.md
View File

@@ -1,5 +1,12 @@
# Changelog # Changelog
## 2026-03-17 - 4.8.13 - fix(remoteingress-protocol)
require a flush after each written frame to bound TLS buffer growth
- Remove the unflushed byte threshold and stop queueing additional writes while a flush is pending
- Simplify write and flush error logging after dropping unflushed byte tracking
- Update tunnel I/O comments to reflect the stricter flush behavior that avoids OOM and connection resets
## 2026-03-17 - 4.8.12 - fix(tunnel) ## 2026-03-17 - 4.8.12 - fix(tunnel)
prevent tunnel backpressure buffering from exhausting memory and cancel stream handlers before TLS shutdown prevent tunnel backpressure buffering from exhausting memory and cancel stream handlers before TLS shutdown

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

@@ -183,11 +183,6 @@ 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 {
@@ -195,7 +190,6 @@ 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 {
@@ -237,7 +231,6 @@ 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,
}, },
} }
} }
@@ -319,15 +312,11 @@ 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.
// Allow up to MAX_UNFLUSHED_BYTES in the TLS session buffer before // Write one frame, set flush_needed, then flush must complete before
// requiring a flush. This keeps the pipe saturated (unlike waiting for // writing more. This prevents unbounded TLS session buffer growth.
// 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 while self.write.has_work() && writes < 16 && !self.write.flush_needed {
&& 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()
@@ -338,8 +327,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={}", log::error!("TunnelIo: poll_write returned 0 (write zero), ctrl_q={} data_q={}",
self.write.ctrl_queue.len(), self.write.data_queue.len(), self.write.unflushed_bytes); self.write.ctrl_queue.len(), self.write.data_queue.len());
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"),
)); ));
@@ -347,7 +336,6 @@ 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(); }
@@ -356,8 +344,8 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
} }
} }
Poll::Ready(Err(e)) => { Poll::Ready(Err(e)) => {
log::error!("TunnelIo: poll_write error: {} (ctrl_q={} data_q={} unflushed={})", log::error!("TunnelIo: poll_write error: {} (ctrl_q={} data_q={})",
e, self.write.ctrl_queue.len(), self.write.data_queue.len(), self.write.unflushed_bytes); e, self.write.ctrl_queue.len(), self.write.data_queue.len());
return Poll::Ready(TunnelEvent::WriteError(e)); return Poll::Ready(TunnelEvent::WriteError(e));
} }
Poll::Pending => break, Poll::Pending => break,
@@ -369,10 +357,9 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
match Pin::new(&mut self.stream).poll_flush(cx) { match Pin::new(&mut self.stream).poll_flush(cx) {
Poll::Ready(Ok(())) => { Poll::Ready(Ok(())) => {
self.write.flush_needed = false; self.write.flush_needed = false;
self.write.unflushed_bytes = 0;
} }
Poll::Ready(Err(e)) => { Poll::Ready(Err(e)) => {
log::error!("TunnelIo: poll_flush error: {} (unflushed={})", e, self.write.unflushed_bytes); log::error!("TunnelIo: poll_flush error: {}", e);
return Poll::Ready(TunnelEvent::WriteError(e)); return Poll::Ready(TunnelEvent::WriteError(e));
} }
Poll::Pending => {} // TCP waker will notify us Poll::Pending => {} // TCP waker will notify us
@@ -461,9 +448,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
// When 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 — poll_write always returns // 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 // Ready (TLS buffers in-memory), so self-waking causes a tight spin loop
// that fills the TLS session buffer unboundedly → OOM → 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();
} }

2
ts/00_commitinfo_data.ts
View File

@@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@serve.zone/remoteingress', name: '@serve.zone/remoteingress',
version: '4.8.12', version: '4.8.13',
description: 'Edge ingress tunnel for DcRouter - accepts incoming TCP connections at network edge and tunnels them to DcRouter SmartProxy preserving client IP via PROXY protocol v1.' description: 'Edge ingress tunnel for DcRouter - accepts incoming TCP connections at network edge and tunnels them to DcRouter SmartProxy preserving client IP via PROXY protocol v1.'
} }