v4.8.11

changelog.md

@@ -1,5 +1,58 @@
 # Changelog
 
+## 2026-03-17 - 4.8.11 - fix(remoteingress-core)
+stop data frame send loops promptly when stream cancellation is triggered
+
+- Use cancellation-aware tokio::select! around data channel sends in both edge and hub stream forwarding paths
+- Prevent stalled or noisy shutdown behavior when stream or client cancellation happens while awaiting frame delivery
+
+## 2026-03-17 - 4.8.10 - fix(remoteingress-core)
+guard tunnel frame sends with cancellation to prevent async send deadlocks
+
+- Wrap OPEN, CLOSE, CLOSE_BACK, WINDOW_UPDATE, and cleanup channel sends in cancellation-aware tokio::select! blocks.
+- Avoid indefinite blocking when tunnel, stream, or writer tasks are cancelled while awaiting channel capacity.
+- Improve shutdown reliability for edge and hub stream handling under tunnel failure conditions.
+
+## 2026-03-17 - 4.8.9 - fix(repo)
+no changes to commit
+
+
+## 2026-03-17 - 4.8.8 - fix(remoteingress-core)
+cancel stale edge connections when an edge reconnects
+
+- Remove any existing edge entry before registering a reconnected edge
+- Trigger the previous connection's cancellation token so stale sessions shut down immediately instead of waiting for TCP keepalive
+
+## 2026-03-17 - 4.8.7 - fix(remoteingress-core)
+perform graceful TLS shutdown on edge and hub tunnel streams
+
+- Send TLS close_notify before cleanup to avoid peer disconnect warnings on both tunnel endpoints
+- Wrap stream shutdown in a 2 second timeout so connection teardown does not block cleanup
+
+## 2026-03-17 - 4.8.6 - fix(remoteingress-core)
+initialize disconnect reason only when set in hub loop break paths
+
+- Replace the default "unknown" disconnect reason with an explicitly assigned string and document that all hub loop exits set it before use
+- Add an allow attribute for unused assignments to avoid warnings around the deferred initialization pattern
+
+## 2026-03-17 - 4.8.5 - fix(repo)
+no changes to commit
+
+
+## 2026-03-17 - 4.8.4 - fix(remoteingress-core)
+prevent stream stalls by guaranteeing flow-control updates and avoiding bounded per-stream channel overflows
+
+- Replace bounded per-stream data channels with unbounded channels on edge and hub, relying on existing WINDOW_UPDATE flow control to limit bytes in flight
+- Use awaited sends for FRAME_WINDOW_UPDATE and FRAME_WINDOW_UPDATE_BACK so updates are not dropped and streams do not deadlock under backpressure
+- Clean up stream state when channel receivers have already exited instead of closing active streams because a bounded queue filled
+
+## 2026-03-17 - 4.8.3 - fix(protocol,edge)
+optimize tunnel frame handling and zero-copy uploads in edge I/O
+
+- extract hub frame processing into a shared edge handler to remove duplicated tunnel logic
+- add zero-copy frame header encoding and read payloads directly into framed buffers for client-to-hub uploads
+- refactor TunnelIo read/write state to avoid unsafe queue access and reduce buffer churn with incremental parsing
+
 ## 2026-03-17 - 4.8.2 - fix(rust-edge)
 refactor tunnel I/O to preserve TLS state and prioritize control frames
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "4.8.2",
+  "version": "4.8.11",
   "private": false,
   "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.",
   "main": "dist_ts/index.js",

rust/crates/remoteingress-core/src/edge.rs

@@ -13,10 +13,20 @@ use serde::{Deserialize, Serialize};
 
 use remoteingress_protocol::*;
 
+type EdgeTlsStream = tokio_rustls::client::TlsStream<TcpStream>;
+
+/// Result of processing a frame (shared with hub.rs pattern).
+#[allow(dead_code)]
+enum EdgeFrameAction {
+    Continue,
+    Disconnect(String),
+}
+
 /// Per-stream state tracked in the edge's client_writers map.
 struct EdgeStreamState {
-    /// Channel to deliver FRAME_DATA_BACK payloads to the hub_to_client task.
-    back_tx: mpsc::Sender<Vec<u8>>,
+    /// Unbounded channel to deliver FRAME_DATA_BACK payloads to the hub_to_client task.
+    /// Unbounded because flow control (WINDOW_UPDATE) already limits bytes-in-flight.
+    back_tx: mpsc::UnboundedSender<Vec<u8>>,
     /// Send window for FRAME_DATA (upload direction).
     /// Decremented by the client reader, incremented by FRAME_WINDOW_UPDATE_BACK from hub.
     send_window: Arc<AtomicU32>,
@@ -272,6 +282,86 @@ enum EdgeLoopResult {
     Reconnect(String),  // reason for disconnection
 }
 
+/// Process a single frame received from the hub side of the tunnel.
+/// Handles FRAME_DATA_BACK, FRAME_WINDOW_UPDATE_BACK, FRAME_CLOSE_BACK, FRAME_CONFIG, FRAME_PING.
+async fn handle_edge_frame(
+    frame: Frame,
+    tunnel_io: &mut remoteingress_protocol::TunnelIo<EdgeTlsStream>,
+    client_writers: &Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
+    listen_ports: &Arc<RwLock<Vec<u16>>>,
+    event_tx: &mpsc::Sender<EdgeEvent>,
+    tunnel_writer_tx: &mpsc::Sender<Vec<u8>>,
+    tunnel_data_tx: &mpsc::Sender<Vec<u8>>,
+    port_listeners: &mut HashMap<u16, JoinHandle<()>>,
+    active_streams: &Arc<AtomicU32>,
+    next_stream_id: &Arc<AtomicU32>,
+    edge_id: &str,
+    connection_token: &CancellationToken,
+    bind_address: &str,
+) -> EdgeFrameAction {
+    match frame.frame_type {
+        FRAME_DATA_BACK => {
+            // Dispatch to per-stream unbounded channel. Flow control (WINDOW_UPDATE)
+            // limits bytes-in-flight, so the channel won't grow unbounded. send() only
+            // fails if the receiver is dropped (hub_to_client task already exited).
+            let mut writers = client_writers.lock().await;
+            if let Some(state) = writers.get(&frame.stream_id) {
+                if state.back_tx.send(frame.payload).is_err() {
+                    // Receiver dropped — hub_to_client task already exited, clean up
+                    writers.remove(&frame.stream_id);
+                }
+            }
+        }
+        FRAME_WINDOW_UPDATE_BACK => {
+            if let Some(increment) = decode_window_update(&frame.payload) {
+                if increment > 0 {
+                    let writers = client_writers.lock().await;
+                    if let Some(state) = writers.get(&frame.stream_id) {
+                        let prev = state.send_window.fetch_add(increment, Ordering::Release);
+                        if prev + increment > MAX_WINDOW_SIZE {
+                            state.send_window.store(MAX_WINDOW_SIZE, Ordering::Release);
+                        }
+                        state.window_notify.notify_one();
+                    }
+                }
+            }
+        }
+        FRAME_CLOSE_BACK => {
+            let mut writers = client_writers.lock().await;
+            writers.remove(&frame.stream_id);
+        }
+        FRAME_CONFIG => {
+            if let Ok(update) = serde_json::from_slice::<ConfigUpdate>(&frame.payload) {
+                log::info!("Config update from hub: ports {:?}", update.listen_ports);
+                *listen_ports.write().await = update.listen_ports.clone();
+                let _ = event_tx.try_send(EdgeEvent::PortsUpdated {
+                    listen_ports: update.listen_ports.clone(),
+                });
+                apply_port_config(
+                    &update.listen_ports,
+                    port_listeners,
+                    tunnel_writer_tx,
+                    tunnel_data_tx,
+                    client_writers,
+                    active_streams,
+                    next_stream_id,
+                    edge_id,
+                    connection_token,
+                    bind_address,
+                );
+            }
+        }
+        FRAME_PING => {
+            // Queue PONG directly — no channel round-trip, guaranteed delivery
+            tunnel_io.queue_ctrl(encode_frame(0, FRAME_PONG, &[]));
+        }
+        _ => {
+            log::warn!("Unexpected frame type {} from hub", frame.frame_type);
+        }
+    }
+    EdgeFrameAction::Continue
+}
+
 async fn connect_to_hub_and_run(
     config: &EdgeConfig,
     connected: &Arc<RwLock<bool>>,
@@ -436,73 +526,22 @@ async fn connect_to_hub_and_run(
     let result = 'io_loop: loop {
         // Drain any buffered frames
         loop {
-            match tunnel_io.try_parse_frame() {
-                Some(Ok(frame)) => {
-                    last_activity = Instant::now();
-                    liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);
-                    match frame.frame_type {
-                        FRAME_DATA_BACK => {
-                            let mut writers = client_writers.lock().await;
-                            if let Some(state) = writers.get(&frame.stream_id) {
-                                if state.back_tx.try_send(frame.payload).is_err() {
-                                    log::warn!("Stream {} back-channel full, closing", frame.stream_id);
-                                    writers.remove(&frame.stream_id);
-                                }
-                            }
-                        }
-                        FRAME_WINDOW_UPDATE_BACK => {
-                            if let Some(increment) = decode_window_update(&frame.payload) {
-                                if increment > 0 {
-                                    let writers = client_writers.lock().await;
-                                    if let Some(state) = writers.get(&frame.stream_id) {
-                                        let prev = state.send_window.fetch_add(increment, Ordering::Release);
-                                        if prev + increment > MAX_WINDOW_SIZE {
-                                            state.send_window.store(MAX_WINDOW_SIZE, Ordering::Release);
-                                        }
-                                        state.window_notify.notify_one();
-                                    }
-                                }
-                            }
-                        }
-                        FRAME_CLOSE_BACK => {
-                            let mut writers = client_writers.lock().await;
-                            writers.remove(&frame.stream_id);
-                        }
-                        FRAME_CONFIG => {
-                            if let Ok(update) = serde_json::from_slice::<ConfigUpdate>(&frame.payload) {
-                                log::info!("Config update from hub: ports {:?}", update.listen_ports);
-                                *listen_ports.write().await = update.listen_ports.clone();
-                                let _ = event_tx.try_send(EdgeEvent::PortsUpdated {
-                                    listen_ports: update.listen_ports.clone(),
-                                });
-                                apply_port_config(
-                                    &update.listen_ports,
-                                    &mut port_listeners,
-                                    &tunnel_writer_tx,
-                                    &tunnel_data_tx,
-                                    &client_writers,
-                                    active_streams,
-                                    next_stream_id,
-                                    &config.edge_id,
-                                    connection_token,
-                                    bind_address,
-                                );
-                            }
-                        }
-                        FRAME_PING => {
-                            // Queue PONG directly — no channel round-trip, guaranteed delivery
-                            tunnel_io.queue_ctrl(encode_frame(0, FRAME_PONG, &[]));
-                        }
-                        _ => {
-                            log::warn!("Unexpected frame type {} from hub", frame.frame_type);
-                        }
-                    }
-                }
+            let frame = match tunnel_io.try_parse_frame() {
+                Some(Ok(f)) => f,
                 Some(Err(e)) => {
                     log::error!("Hub frame error: {}", e);
                     break 'io_loop EdgeLoopResult::Reconnect(format!("hub_frame_error: {}", e));
                 }
                 None => break,
+            };
+            last_activity = Instant::now();
+            liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);
+            if let EdgeFrameAction::Disconnect(reason) = handle_edge_frame(
+                frame, &mut tunnel_io, &client_writers, listen_ports, event_tx,
+                &tunnel_writer_tx, &tunnel_data_tx, &mut port_listeners,
+                active_streams, next_stream_id, &config.edge_id, connection_token, bind_address,
+            ).await {
+                break 'io_loop EdgeLoopResult::Reconnect(reason);
             }
         }
 
@@ -515,61 +554,12 @@ async fn connect_to_hub_and_run(
             remoteingress_protocol::TunnelEvent::Frame(frame) => {
                 last_activity = Instant::now();
                 liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);
-                match frame.frame_type {
-                    FRAME_DATA_BACK => {
-                        let mut writers = client_writers.lock().await;
-                        if let Some(state) = writers.get(&frame.stream_id) {
-                            if state.back_tx.try_send(frame.payload).is_err() {
-                                log::warn!("Stream {} back-channel full, closing", frame.stream_id);
-                                writers.remove(&frame.stream_id);
-                            }
-                        }
-                    }
-                    FRAME_WINDOW_UPDATE_BACK => {
-                        if let Some(increment) = decode_window_update(&frame.payload) {
-                            if increment > 0 {
-                                let writers = client_writers.lock().await;
-                                if let Some(state) = writers.get(&frame.stream_id) {
-                                    let prev = state.send_window.fetch_add(increment, Ordering::Release);
-                                    if prev + increment > MAX_WINDOW_SIZE {
-                                        state.send_window.store(MAX_WINDOW_SIZE, Ordering::Release);
-                                    }
-                                    state.window_notify.notify_one();
-                                }
-                            }
-                        }
-                    }
-                    FRAME_CLOSE_BACK => {
-                        let mut writers = client_writers.lock().await;
-                        writers.remove(&frame.stream_id);
-                    }
-                    FRAME_CONFIG => {
-                        if let Ok(update) = serde_json::from_slice::<ConfigUpdate>(&frame.payload) {
-                            log::info!("Config update from hub: ports {:?}", update.listen_ports);
-                            *listen_ports.write().await = update.listen_ports.clone();
-                            let _ = event_tx.try_send(EdgeEvent::PortsUpdated {
-                                listen_ports: update.listen_ports.clone(),
-                            });
-                            apply_port_config(
-                                &update.listen_ports,
-                                &mut port_listeners,
-                                &tunnel_writer_tx,
-                                &tunnel_data_tx,
-                                &client_writers,
-                                active_streams,
-                                next_stream_id,
-                                &config.edge_id,
-                                connection_token,
-                                bind_address,
-                            );
-                        }
-                    }
-                    FRAME_PING => {
-                        tunnel_io.queue_ctrl(encode_frame(0, FRAME_PONG, &[]));
-                    }
-                    _ => {
-                        log::warn!("Unexpected frame type {} from hub", frame.frame_type);
-                    }
+                if let EdgeFrameAction::Disconnect(reason) = handle_edge_frame(
+                    frame, &mut tunnel_io, &client_writers, listen_ports, event_tx,
+                    &tunnel_writer_tx, &tunnel_data_tx, &mut port_listeners,
+                    active_streams, next_stream_id, &config.edge_id, connection_token, bind_address,
+                ).await {
+                    break EdgeLoopResult::Reconnect(reason);
                 }
             }
             remoteingress_protocol::TunnelEvent::Eof => {
@@ -597,6 +587,14 @@ async fn connect_to_hub_and_run(
         }
     };
 
+    // Graceful TLS shutdown: send close_notify so the hub sees a clean disconnect
+    // instead of "peer closed connection without sending TLS close_notify".
+    let mut tls_stream = tunnel_io.into_inner();
+    let _ = tokio::time::timeout(
+        Duration::from_secs(2),
+        tls_stream.shutdown(),
+    ).await;
+
     // Cleanup
     connection_token.cancel();
     stun_handle.abort();
@@ -741,12 +739,18 @@ async fn handle_client_connection(
     // Send OPEN frame with PROXY v1 header via control channel
     let proxy_header = build_proxy_v1_header(&client_ip, edge_ip, client_port, dest_port);
     let open_frame = encode_frame(stream_id, FRAME_OPEN, proxy_header.as_bytes());
-    if tunnel_ctrl_tx.send(open_frame).await.is_err() {
+    let send_ok = tokio::select! {
+        result = tunnel_ctrl_tx.send(open_frame) => result.is_ok(),
+        _ = client_token.cancelled() => false,
+    };
+    if !send_ok {
         return;
     }
 
-    // Set up channel for data coming back from hub (capacity 16 is sufficient with flow control)
-    let (back_tx, mut back_rx) = mpsc::channel::<Vec<u8>>(1024);
+    // Per-stream unbounded back-channel. Flow control (WINDOW_UPDATE) limits
+    // bytes-in-flight, so this won't grow unbounded. Unbounded avoids killing
+    // streams due to channel overflow — backpressure slows streams, never kills them.
+    let (back_tx, mut back_rx) = mpsc::unbounded_channel::<Vec<u8>>();
     // Adaptive initial window: scale with current stream count to keep total in-flight
     // data within the 32MB budget. Prevents burst flooding when many streams open.
     let initial_window = remoteingress_protocol::compute_window_for_stream_count(
@@ -793,10 +797,16 @@ async fn handle_client_connection(
                             if consumed_since_update >= threshold {
                                 let increment = consumed_since_update.min(adaptive_window);
                                 let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, increment);
-                                if wu_tx.try_send(frame).is_ok() {
-                                    consumed_since_update -= increment;
+                                // Use send().await for guaranteed delivery — dropping WINDOW_UPDATEs
+                                // causes permanent flow stalls. Safe: runs in per-stream task, not main loop.
+                                tokio::select! {
+                                    result = wu_tx.send(frame) => {
+                                        if result.is_ok() {
+                                            consumed_since_update -= increment;
+                                        }
+                                    }
+                                    _ = hub_to_client_token.cancelled() => break,
                                 }
-                                // If try_send fails, keep accumulating — retry on next threshold
                             }
                         }
                         None => break,
@@ -808,20 +818,29 @@ async fn handle_client_connection(
         // Send final window update for any remaining consumed bytes
         if consumed_since_update > 0 {
             let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
-            let _ = wu_tx.try_send(frame);
+            tokio::select! {
+                _ = wu_tx.send(frame) => {}
+                _ = hub_to_client_token.cancelled() => {}
+            }
         }
         let _ = client_write.shutdown().await;
     });
 
-    // Task: client -> hub (upload direction) with per-stream flow control
-    let mut buf = vec![0u8; 32768];
+    // Task: client -> hub (upload direction) with per-stream flow control.
+    // Zero-copy: read payload directly after the header, then prepend header.
+    let mut buf = vec![0u8; FRAME_HEADER_SIZE + 32768];
     loop {
-        // Wait for send window to have capacity (with stall timeout)
+        // Wait for send window to have capacity (with stall timeout).
+        // Safe pattern: register notified BEFORE checking the condition
+        // to avoid missing a notify_one that fires between load and select.
         loop {
+            let notified = window_notify.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
             let w = send_window.load(Ordering::Acquire);
             if w > 0 { break; }
             tokio::select! {
-                _ = window_notify.notified() => continue,
+                _ = notified => continue,
                 _ = client_token.cancelled() => break,
                 _ = tokio::time::sleep(Duration::from_secs(120)) => {
                     log::warn!("Stream {} upload stalled (window empty for 120s)", stream_id);
@@ -844,19 +863,21 @@ async fn handle_client_connection(
         let adaptive_cap = remoteingress_protocol::compute_window_for_stream_count(
             active_streams.load(Ordering::Relaxed),
         ) as usize;
-        let max_read = w.min(buf.len()).min(adaptive_cap);
+        let max_read = w.min(32768).min(adaptive_cap);
 
         tokio::select! {
-            read_result = client_read.read(&mut buf[..max_read]) => {
+            read_result = client_read.read(&mut buf[FRAME_HEADER_SIZE..FRAME_HEADER_SIZE + max_read]) => {
                 match read_result {
                     Ok(0) => break,
                     Ok(n) => {
                         send_window.fetch_sub(n as u32, Ordering::Release);
-                        let data_frame = encode_frame(stream_id, FRAME_DATA, &buf[..n]);
-                        if tunnel_data_tx.send(data_frame).await.is_err() {
-                            log::warn!("Stream {} data channel closed, closing", stream_id);
-                            break;
-                        }
+                        encode_frame_header(&mut buf, stream_id, FRAME_DATA, n);
+                        let data_frame = buf[..FRAME_HEADER_SIZE + n].to_vec();
+                        let sent = tokio::select! {
+                            result = tunnel_data_tx.send(data_frame) => result.is_ok(),
+                            _ = client_token.cancelled() => false,
+                        };
+                        if !sent { break; }
                     }
                     Err(_) => break,
                 }
@@ -877,9 +898,13 @@ async fn handle_client_connection(
     ).await;
 
     // NOW send CLOSE — the response has been fully delivered (or timed out).
+    // select! with cancellation guard prevents indefinite blocking if tunnel dies.
     if !client_token.is_cancelled() {
         let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
-        let _ = tunnel_data_tx.send(close_frame).await;
+        tokio::select! {
+            _ = tunnel_data_tx.send(close_frame) => {}
+            _ = client_token.cancelled() => {}
+        }
     }
 
     // Clean up

rust/crates/remoteingress-protocol/src/lib.rs

@@ -72,6 +72,16 @@ pub fn encode_frame(stream_id: u32, frame_type: u8, payload: &[u8]) -> Vec<u8> {
     buf
 }
 
+/// Write a frame header into `buf[0..FRAME_HEADER_SIZE]`.
+/// The caller must ensure payload is already at `buf[FRAME_HEADER_SIZE..FRAME_HEADER_SIZE + payload_len]`.
+/// This enables zero-copy encoding: read directly into `buf[FRAME_HEADER_SIZE..]`, then
+/// prepend the header without copying the payload.
+pub fn encode_frame_header(buf: &mut [u8], stream_id: u32, frame_type: u8, payload_len: usize) {
+    buf[0..4].copy_from_slice(&stream_id.to_be_bytes());
+    buf[4] = frame_type;
+    buf[5..9].copy_from_slice(&(payload_len as u32).to_be_bytes());
+}
+
 /// Build a PROXY protocol v1 header line.
 /// Format: `PROXY TCP4 <client_ip> <edge_ip> <client_port> <dest_port>\r\n`
 pub fn build_proxy_v1_header(
@@ -173,6 +183,21 @@ pub enum TunnelEvent {
     Cancelled,
 }
 
+/// Write state extracted into a sub-struct so the borrow checker can see
+/// disjoint field access between `self.write` and `self.stream`.
+struct WriteState {
+    ctrl_queue: VecDeque<Vec<u8>>,   // PONG, WINDOW_UPDATE, CLOSE, OPEN — always first
+    data_queue: VecDeque<Vec<u8>>,   // DATA, DATA_BACK — only when ctrl is empty
+    offset: usize,                   // progress within current frame being written
+    flush_needed: bool,
+}
+
+impl WriteState {
+    fn has_work(&self) -> bool {
+        !self.ctrl_queue.is_empty() || !self.data_queue.is_empty()
+    }
+}
+
 /// Single-owner I/O engine for the tunnel TLS connection.
 ///
 /// Owns the TLS stream directly — no `tokio::io::split()`, no mutex.
@@ -184,11 +209,9 @@ pub struct TunnelIo<S> {
     // Read state: accumulate bytes, parse frames incrementally
     read_buf: Vec<u8>,
     read_pos: usize,
-    // Write state: dual priority queues
-    ctrl_queue: VecDeque<Vec<u8>>,   // PONG, WINDOW_UPDATE, CLOSE, OPEN — always first
-    data_queue: VecDeque<Vec<u8>>,   // DATA, DATA_BACK — only when ctrl is empty
-    write_offset: usize,             // progress within current frame being written
-    flush_needed: bool,
+    parse_pos: usize,
+    // Write state: extracted sub-struct for safe disjoint borrows
+    write: WriteState,
 }
 
 impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
@@ -202,42 +225,52 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
             stream,
             read_buf,
             read_pos,
-            ctrl_queue: VecDeque::new(),
-            data_queue: VecDeque::new(),
-            write_offset: 0,
-            flush_needed: false,
+            parse_pos: 0,
+            write: WriteState {
+                ctrl_queue: VecDeque::new(),
+                data_queue: VecDeque::new(),
+                offset: 0,
+                flush_needed: false,
+            },
         }
     }
 
     /// Queue a high-priority control frame (PONG, WINDOW_UPDATE, CLOSE, OPEN).
     pub fn queue_ctrl(&mut self, frame: Vec<u8>) {
-        self.ctrl_queue.push_back(frame);
+        self.write.ctrl_queue.push_back(frame);
     }
 
     /// Queue a lower-priority data frame (DATA, DATA_BACK).
     pub fn queue_data(&mut self, frame: Vec<u8>) {
-        self.data_queue.push_back(frame);
+        self.write.data_queue.push_back(frame);
     }
 
     /// Try to parse a complete frame from the read buffer.
+    /// Uses a parse_pos cursor to avoid drain() on every frame.
     pub fn try_parse_frame(&mut self) -> Option<Result<Frame, std::io::Error>> {
-        if self.read_pos < FRAME_HEADER_SIZE {
+        let available = self.read_pos - self.parse_pos;
+        if available < FRAME_HEADER_SIZE {
             return None;
         }
 
+        let base = self.parse_pos;
         let stream_id = u32::from_be_bytes([
-            self.read_buf[0], self.read_buf[1], self.read_buf[2], self.read_buf[3],
+            self.read_buf[base], self.read_buf[base + 1],
+            self.read_buf[base + 2], self.read_buf[base + 3],
         ]);
-        let frame_type = self.read_buf[4];
+        let frame_type = self.read_buf[base + 4];
         let length = u32::from_be_bytes([
-            self.read_buf[5], self.read_buf[6], self.read_buf[7], self.read_buf[8],
+            self.read_buf[base + 5], self.read_buf[base + 6],
+            self.read_buf[base + 7], self.read_buf[base + 8],
         ]);
 
         if length > MAX_PAYLOAD_SIZE {
             let header = [
-                self.read_buf[0], self.read_buf[1], self.read_buf[2], self.read_buf[3],
-                self.read_buf[4], self.read_buf[5], self.read_buf[6], self.read_buf[7],
-                self.read_buf[8],
+                self.read_buf[base], self.read_buf[base + 1],
+                self.read_buf[base + 2], self.read_buf[base + 3],
+                self.read_buf[base + 4], self.read_buf[base + 5],
+                self.read_buf[base + 6], self.read_buf[base + 7],
+                self.read_buf[base + 8],
             ];
             log::error!(
                 "CORRUPT FRAME HEADER: raw={:02x?} stream_id={} type=0x{:02x} length={}",
@@ -250,21 +283,23 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         }
 
         let total_frame_size = FRAME_HEADER_SIZE + length as usize;
-        if self.read_pos < total_frame_size {
+        if available < total_frame_size {
             return None;
         }
 
-        let payload = self.read_buf[FRAME_HEADER_SIZE..total_frame_size].to_vec();
-        self.read_buf.drain(..total_frame_size);
-        self.read_pos -= total_frame_size;
+        let payload = self.read_buf[base + FRAME_HEADER_SIZE..base + total_frame_size].to_vec();
+        self.parse_pos += total_frame_size;
+
+        // Compact when parse_pos > half the data to reclaim memory
+        if self.parse_pos > self.read_pos / 2 && self.parse_pos > 0 {
+            self.read_buf.drain(..self.parse_pos);
+            self.read_pos -= self.parse_pos;
+            self.parse_pos = 0;
+        }
 
         Some(Ok(Frame { stream_id, frame_type, payload }))
     }
 
-    fn has_write_work(&self) -> bool {
-        !self.ctrl_queue.is_empty() || !self.data_queue.is_empty()
-    }
-
     /// Poll-based I/O step. Returns Ready on events, Pending when idle.
     ///
     /// Order: write(ctrl→data) → flush → read → channels → timers
@@ -279,20 +314,16 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         // 1. WRITE: drain ctrl queue first, then data queue.
         //    TLS poll_write writes plaintext to session buffer (always Ready).
         //    Batch up to 16 frames per poll cycle.
+        //    Safe: `self.write` and `self.stream` are disjoint fields.
         let mut writes = 0;
-        while self.has_write_work() && writes < 16 {
-            // Determine which queue to write from and the frame data.
-            // We access the queues via raw pointers to avoid borrow conflicts with self.stream.
-            let from_ctrl = !self.ctrl_queue.is_empty();
-            let frame_ptr: *const Vec<u8> = if from_ctrl {
-                self.ctrl_queue.front().unwrap()
+        while self.write.has_work() && writes < 16 {
+            let from_ctrl = !self.write.ctrl_queue.is_empty();
+            let frame = if from_ctrl {
+                self.write.ctrl_queue.front().unwrap()
             } else {
-                self.data_queue.front().unwrap()
+                self.write.data_queue.front().unwrap()
             };
-            // SAFETY: the frame is not modified while we hold the pointer — poll_write
-            // only writes to self.stream, and advance_write only runs after poll_write returns.
-            let frame = unsafe { &*frame_ptr };
-            let remaining = &frame[self.write_offset..];
+            let remaining = &frame[self.write.offset..];
 
             match Pin::new(&mut self.stream).poll_write(cx, remaining) {
                 Poll::Ready(Ok(0)) => {
@@ -301,12 +332,12 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
                     ));
                 }
                 Poll::Ready(Ok(n)) => {
-                    self.write_offset += n;
-                    self.flush_needed = true;
-                    if self.write_offset >= frame.len() {
-                        if from_ctrl { self.ctrl_queue.pop_front(); }
-                        else { self.data_queue.pop_front(); }
-                        self.write_offset = 0;
+                    self.write.offset += n;
+                    self.write.flush_needed = true;
+                    if self.write.offset >= frame.len() {
+                        if from_ctrl { self.write.ctrl_queue.pop_front(); }
+                        else { self.write.data_queue.pop_front(); }
+                        self.write.offset = 0;
                         writes += 1;
                     }
                 }
@@ -316,9 +347,9 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         }
 
         // 2. FLUSH: push encrypted data from TLS session to TCP.
-        if self.flush_needed {
+        if self.write.flush_needed {
             match Pin::new(&mut self.stream).poll_flush(cx) {
-                Poll::Ready(Ok(())) => self.flush_needed = false,
+                Poll::Ready(Ok(())) => self.write.flush_needed = false,
                 Poll::Ready(Err(e)) => return Poll::Ready(TunnelEvent::WriteError(e)),
                 Poll::Pending => {} // TCP waker will notify us
             }
@@ -329,6 +360,12 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         //    the waker without re-registering it, causing the task to sleep until a
         //    timer or channel wakes it (potentially 15+ seconds of lost reads).
         loop {
+            // Compact if needed to make room for reads
+            if self.parse_pos > 0 && self.read_buf.len() - self.read_pos < 32768 {
+                self.read_buf.drain(..self.parse_pos);
+                self.read_pos -= self.parse_pos;
+                self.parse_pos = 0;
+            }
             if self.read_buf.len() < self.read_pos + 32768 {
                 self.read_buf.resize(self.read_pos + 32768, 0);
             }
@@ -358,7 +395,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         let mut got_new = false;
         loop {
             match ctrl_rx.poll_recv(cx) {
-                Poll::Ready(Some(frame)) => { self.ctrl_queue.push_back(frame); got_new = true; }
+                Poll::Ready(Some(frame)) => { self.write.ctrl_queue.push_back(frame); got_new = true; }
                 Poll::Ready(None) => {
                     return Poll::Ready(TunnelEvent::WriteError(
                         std::io::Error::new(std::io::ErrorKind::BrokenPipe, "ctrl channel closed"),
@@ -369,7 +406,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         }
         loop {
             match data_rx.poll_recv(cx) {
-                Poll::Ready(Some(frame)) => { self.data_queue.push_back(frame); got_new = true; }
+                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"),
@@ -390,7 +427,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         // 6. SELF-WAKE: only when we have frames AND flush is done.
         //    If flush is pending, the TCP write-readiness waker will notify us.
         //    If we got new channel frames, wake to write them.
-        if got_new || (!self.flush_needed && self.has_write_work()) {
+        if got_new || (!self.write.flush_needed && self.write.has_work()) {
             cx.waker().wake_by_ref();
         }
 
@@ -406,6 +443,22 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_encode_frame_header() {
+        let payload = b"hello";
+        let mut buf = vec![0u8; FRAME_HEADER_SIZE + payload.len()];
+        buf[FRAME_HEADER_SIZE..].copy_from_slice(payload);
+        encode_frame_header(&mut buf, 42, FRAME_DATA, payload.len());
+        assert_eq!(buf, encode_frame(42, FRAME_DATA, payload));
+    }
+
+    #[test]
+    fn test_encode_frame_header_empty_payload() {
+        let mut buf = vec![0u8; FRAME_HEADER_SIZE];
+        encode_frame_header(&mut buf, 99, FRAME_CLOSE, 0);
+        assert_eq!(buf, encode_frame(99, FRAME_CLOSE, &[]));
+    }
+
     #[test]
     fn test_encode_frame() {
         let data = b"hello";

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@serve.zone/remoteingress',
-  version: '4.8.2',
+  version: '4.8.11',
   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.'
 }