v4.8.14

changelog.md

@@ -1,5 +1,79 @@
 # Changelog
 
+## 2026-03-17 - 4.8.14 - fix(rust-core,protocol)
+eliminate edge stream registration races and reduce frame buffering copies
+
+- replace Vec<u8> tunnel/frame buffers with bytes::Bytes and BytesMut for lower-copy frame parsing and queueing
+- move edge stream ownership into the main I/O loop with explicit register and cleanup channels to ensure streams are registered before OPEN processing
+- add proactive send window clamping so active streams converge immediately to adaptive flow-control targets
+
+## 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)
+prevent tunnel backpressure buffering from exhausting memory and cancel stream handlers before TLS shutdown
+
+- stop self-waking and writing new frames while a flush is pending to avoid unbounded TLS session buffer growth under load
+- reorder edge and hub shutdown cleanup so stream cancellation happens before TLS close_notify, preventing handlers from blocking on dead channels
+- add load tests covering sustained large transfers, burst traffic, and rapid stream churn to verify tunnel stability
+
+## 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.14",
   "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/Cargo.lock

@@ -551,6 +551,7 @@ dependencies = [
 name = "remoteingress-core"
 version = "2.0.0"
 dependencies = [
+ "bytes",
  "log",
  "rcgen",
  "remoteingress-protocol",
@@ -568,6 +569,7 @@ dependencies = [
 name = "remoteingress-protocol"
 version = "2.0.0"
 dependencies = [
+ "bytes",
  "log",
  "tokio",
  "tokio-util",

rust/crates/remoteingress-core/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 remoteingress-protocol = { path = "../remoteingress-protocol" }
 tokio = { version = "1", features = ["full"] }
 tokio-rustls = "0.26"
+bytes = "1"
 rustls = { version = "0.23", default-features = false, features = ["ring", "logging", "std", "tls12"] }
 rcgen = "0.13"
 serde = { version = "1", features = ["derive"] }

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

@@ -9,14 +9,25 @@ use tokio::task::JoinHandle;
 use tokio::time::{Instant, sleep_until};
 use tokio_rustls::TlsConnector;
 use tokio_util::sync::CancellationToken;
+use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 
 use remoteingress_protocol::*;
 
-/// Per-stream state tracked in the edge's client_writers map.
+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 stream 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<Bytes>,
     /// Send window for FRAME_DATA (upload direction).
     /// Decremented by the client reader, incremented by FRAME_WINDOW_UPDATE_BACK from hub.
     send_window: Arc<AtomicU32>,
@@ -24,6 +35,12 @@ struct EdgeStreamState {
     window_notify: Arc<Notify>,
 }
 
+/// Registration message sent from per-stream tasks to the main I/O loop.
+struct StreamRegistration {
+    stream_id: u32,
+    state: EdgeStreamState,
+}
+
 /// Edge configuration (hub-host + credentials only; ports come from hub).
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -272,6 +289,60 @@ 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.
+/// No mutex — edge_streams is owned by the main I/O loop (same pattern as hub.rs).
+fn handle_edge_frame(
+    frame: Frame,
+    tunnel_io: &mut remoteingress_protocol::TunnelIo<EdgeTlsStream>,
+    edge_streams: &mut HashMap<u32, EdgeStreamState>,
+    listen_ports_update: &mut Option<Vec<u16>>,
+) -> 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).
+            if let Some(state) = edge_streams.get(&frame.stream_id) {
+                if state.back_tx.send(frame.payload).is_err() {
+                    // Receiver dropped — hub_to_client task already exited, clean up
+                    edge_streams.remove(&frame.stream_id);
+                }
+            }
+        }
+        FRAME_WINDOW_UPDATE_BACK => {
+            if let Some(increment) = decode_window_update(&frame.payload) {
+                if increment > 0 {
+                    if let Some(state) = edge_streams.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 => {
+            edge_streams.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_update = Some(update.listen_ports);
+            }
+        }
+        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>>,
@@ -400,14 +471,17 @@ async fn connect_to_hub_and_run(
         }
     });
 
-    // Client socket map: stream_id -> per-stream state (back channel + flow control)
-    let client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>> =
-        Arc::new(Mutex::new(HashMap::new()));
+    // Stream map owned by the main I/O loop — no mutex, matching hub.rs pattern.
+    let mut edge_streams: HashMap<u32, EdgeStreamState> = HashMap::new();
+    // Channel for per-stream tasks to register their stream state with the main loop.
+    let (register_tx, mut register_rx) = mpsc::channel::<StreamRegistration>(256);
+    // Channel for per-stream tasks to deregister when done.
+    let (cleanup_tx, mut cleanup_rx) = mpsc::channel::<u32>(256);
 
     // QoS dual-channel: ctrl frames have priority over data frames.
     // Stream handlers send through these channels → TunnelIo drains them.
-    let (tunnel_ctrl_tx, mut tunnel_ctrl_rx) = mpsc::channel::<Vec<u8>>(256);
-    let (tunnel_data_tx, mut tunnel_data_rx) = mpsc::channel::<Vec<u8>>(4096);
+    let (tunnel_ctrl_tx, mut tunnel_ctrl_rx) = mpsc::channel::<Bytes>(256);
+    let (tunnel_data_tx, mut tunnel_data_rx) = mpsc::channel::<Bytes>(4096);
     let tunnel_writer_tx = tunnel_ctrl_tx.clone();
 
     // Start TCP listeners for initial ports
@@ -418,7 +492,8 @@ async fn connect_to_hub_and_run(
         &mut port_listeners,
         &tunnel_writer_tx,
         &tunnel_data_tx,
-        &client_writers,
+        &register_tx,
+        &cleanup_tx,
         active_streams,
         next_stream_id,
         &config.edge_id,
@@ -429,147 +504,86 @@ 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));
 
     let result = 'io_loop: loop {
+        // Drain stream registrations from per-stream tasks (before poll_step so
+        // registrations are processed before OPEN frames are sent to the hub).
+        while let Ok(reg) = register_rx.try_recv() {
+            edge_streams.insert(reg.stream_id, reg.state);
+        }
+        // Drain stream cleanups from per-stream tasks
+        while let Ok(stream_id) = cleanup_rx.try_recv() {
+            edge_streams.remove(&stream_id);
+        }
+
         // Drain any buffered frames
+        let mut listen_ports_update = None;
         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, &mut edge_streams, &mut listen_ports_update,
+            ) {
+                break 'io_loop EdgeLoopResult::Reconnect(reason);
             }
         }
 
+        // Apply port config update if handle_edge_frame signalled one
+        if let Some(new_ports) = listen_ports_update.take() {
+            *listen_ports.write().await = new_ports.clone();
+            let _ = event_tx.try_send(EdgeEvent::PortsUpdated {
+                listen_ports: new_ports.clone(),
+            });
+            apply_port_config(
+                &new_ports,
+                &mut port_listeners,
+                &tunnel_writer_tx,
+                &tunnel_data_tx,
+                &register_tx,
+                &cleanup_tx,
+                active_streams,
+                next_stream_id,
+                &config.edge_id,
+                connection_token,
+                bind_address,
+            );
+        }
+
         // Poll I/O: write(ctrl→data), flush, read, channels, timers
         let event = std::future::poll_fn(|cx| {
             tunnel_io.poll_step(cx, &mut tunnel_ctrl_rx, &mut tunnel_data_rx, &mut liveness_deadline, connection_token)
         }).await;
 
+        // Drain registrations/cleanups before processing the event — registrations
+        // may have arrived while poll_step was running (multiple poll cycles inside .await).
+        while let Ok(reg) = register_rx.try_recv() {
+            edge_streams.insert(reg.stream_id, reg.state);
+        }
+        while let Ok(stream_id) = cleanup_rx.try_recv() {
+            edge_streams.remove(&stream_id);
+        }
+
+        let mut listen_ports_update = None;
         match event {
             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, &mut edge_streams, &mut listen_ports_update,
+                ) {
+                    break EdgeLoopResult::Reconnect(reason);
                 }
             }
             remoteingress_protocol::TunnelEvent::Eof => {
@@ -595,15 +609,46 @@ async fn connect_to_hub_and_run(
                 break EdgeLoopResult::Shutdown;
             }
         }
+
+        // Apply port config update if handle_edge_frame signalled one
+        if let Some(new_ports) = listen_ports_update.take() {
+            *listen_ports.write().await = new_ports.clone();
+            let _ = event_tx.try_send(EdgeEvent::PortsUpdated {
+                listen_ports: new_ports.clone(),
+            });
+            apply_port_config(
+                &new_ports,
+                &mut port_listeners,
+                &tunnel_writer_tx,
+                &tunnel_data_tx,
+                &register_tx,
+                &cleanup_tx,
+                active_streams,
+                next_stream_id,
+                &config.edge_id,
+                connection_token,
+                bind_address,
+            );
+        }
     };
 
-    // Cleanup
+    // Cancel stream tokens FIRST so stream handlers exit immediately.
+    // If we TLS-shutdown first, stream handlers are stuck sending to dead channels
+    // for up to 2 seconds while the shutdown times out on a dead connection.
     connection_token.cancel();
     stun_handle.abort();
     for (_, h) in port_listeners.drain() {
         h.abort();
     }
 
+    // Graceful TLS shutdown: send close_notify so the hub sees a clean disconnect.
+    // Stream handlers are already cancelled, so no new data is being produced.
+    let mut tls_stream = tunnel_io.into_inner();
+    let _ = tokio::time::timeout(
+        Duration::from_secs(2),
+        tls_stream.shutdown(),
+    ).await;
+
     result
 }
 
@@ -611,9 +656,10 @@ async fn connect_to_hub_and_run(
 fn apply_port_config(
     new_ports: &[u16],
     port_listeners: &mut HashMap<u16, JoinHandle<()>>,
-    tunnel_ctrl_tx: &mpsc::Sender<Vec<u8>>,
-    tunnel_data_tx: &mpsc::Sender<Vec<u8>>,
-    client_writers: &Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
+    tunnel_ctrl_tx: &mpsc::Sender<Bytes>,
+    tunnel_data_tx: &mpsc::Sender<Bytes>,
+    register_tx: &mpsc::Sender<StreamRegistration>,
+    cleanup_tx: &mpsc::Sender<u32>,
     active_streams: &Arc<AtomicU32>,
     next_stream_id: &Arc<AtomicU32>,
     edge_id: &str,
@@ -635,7 +681,8 @@ fn apply_port_config(
     for &port in new_set.difference(&old_set) {
         let tunnel_ctrl_tx = tunnel_ctrl_tx.clone();
         let tunnel_data_tx = tunnel_data_tx.clone();
-        let client_writers = client_writers.clone();
+        let register_tx = register_tx.clone();
+        let cleanup_tx = cleanup_tx.clone();
         let active_streams = active_streams.clone();
         let next_stream_id = next_stream_id.clone();
         let edge_id = edge_id.to_string();
@@ -669,7 +716,8 @@ fn apply_port_config(
                                 let stream_id = next_stream_id.fetch_add(1, Ordering::Relaxed);
                                 let tunnel_ctrl_tx = tunnel_ctrl_tx.clone();
                                 let tunnel_data_tx = tunnel_data_tx.clone();
-                                let client_writers = client_writers.clone();
+                                let register_tx = register_tx.clone();
+                                let cleanup_tx = cleanup_tx.clone();
                                 let active_streams = active_streams.clone();
                                 let edge_id = edge_id.clone();
                                 let client_token = port_token.child_token();
@@ -685,7 +733,8 @@ fn apply_port_config(
                                         &edge_id,
                                         tunnel_ctrl_tx,
                                         tunnel_data_tx,
-                                        client_writers,
+                                        register_tx,
+                                        cleanup_tx,
                                         client_token,
                                         Arc::clone(&active_streams),
                                     )
@@ -726,9 +775,10 @@ async fn handle_client_connection(
     stream_id: u32,
     dest_port: u16,
     edge_id: &str,
-    tunnel_ctrl_tx: mpsc::Sender<Vec<u8>>,
-    tunnel_data_tx: mpsc::Sender<Vec<u8>>,
-    client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
+    tunnel_ctrl_tx: mpsc::Sender<Bytes>,
+    tunnel_data_tx: mpsc::Sender<Bytes>,
+    register_tx: mpsc::Sender<StreamRegistration>,
+    cleanup_tx: mpsc::Sender<u32>,
     client_token: CancellationToken,
     active_streams: Arc<AtomicU32>,
 ) {
@@ -738,15 +788,10 @@ async fn handle_client_connection(
     // Determine edge IP (use 0.0.0.0 as placeholder — hub doesn't use it for routing)
     let edge_ip = "0.0.0.0";
 
-    // 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() {
-        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::<Bytes>();
     // 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(
@@ -754,13 +799,35 @@ async fn handle_client_connection(
     );
     let send_window = Arc::new(AtomicU32::new(initial_window));
     let window_notify = Arc::new(Notify::new());
-    {
-        let mut writers = client_writers.lock().await;
-        writers.insert(stream_id, EdgeStreamState {
-            back_tx,
-            send_window: Arc::clone(&send_window),
-            window_notify: Arc::clone(&window_notify),
-        });
+
+    // Register with the main I/O loop BEFORE sending OPEN. The main loop drains
+    // register_rx before poll_step drains ctrl_rx, guaranteeing the stream is
+    // registered before the OPEN frame reaches the hub and DATA_BACK arrives.
+    let reg_ok = tokio::select! {
+        result = register_tx.send(StreamRegistration {
+            stream_id,
+            state: EdgeStreamState {
+                back_tx,
+                send_window: Arc::clone(&send_window),
+                window_notify: Arc::clone(&window_notify),
+            },
+        }) => result.is_ok(),
+        _ = client_token.cancelled() => false,
+    };
+    if !reg_ok {
+        return;
+    }
+
+    // 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());
+    let send_ok = tokio::select! {
+        result = tunnel_ctrl_tx.send(open_frame) => result.is_ok(),
+        _ = client_token.cancelled() => false,
+    };
+    if !send_ok {
+        let _ = cleanup_tx.try_send(stream_id);
+        return;
     }
 
     let (mut client_read, mut client_write) = client_stream.into_split();
@@ -793,10 +860,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 +881,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);
@@ -831,32 +913,31 @@ async fn handle_client_connection(
         }
         if client_token.is_cancelled() { break; }
 
-        // Limit read size to available window.
-        // IMPORTANT: if window is 0 (stall timeout fired), we must NOT
-        // read into an empty buffer — read(&mut buf[..0]) returns Ok(0)
-        // which would be falsely interpreted as EOF.
-        let w = send_window.load(Ordering::Acquire) as usize;
+        // Proactive QoS: clamp send_window to current adaptive target so existing
+        // streams converge immediately when concurrency increases (no drain cycle).
+        let adaptive_target = remoteingress_protocol::compute_window_for_stream_count(
+            active_streams.load(Ordering::Relaxed),
+        );
+        let w = remoteingress_protocol::clamp_send_window(&send_window, adaptive_target) as usize;
         if w == 0 {
             log::warn!("Stream {} upload: window still 0 after stall timeout, closing", stream_id);
             break;
         }
-        // Adaptive: cap read to current per-stream target window
-        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);
 
         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 = Bytes::copy_from_slice(&buf[..FRAME_HEADER_SIZE + n]);
+                        let sent = tokio::select! {
+                            result = tunnel_data_tx.send(data_frame) => result.is_ok(),
+                            _ = client_token.cancelled() => false,
+                        };
+                        if !sent { break; }
                     }
                     Err(_) => break,
                 }
@@ -877,16 +958,17 @@ 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
-    {
-        let mut writers = client_writers.lock().await;
-        writers.remove(&stream_id);
-    }
+    // Clean up — notify main loop to remove stream state
+    let _ = cleanup_tx.try_send(stream_id);
     hub_to_client.abort(); // No-op if already finished; safety net if timeout fired
     let _ = edge_id; // used for logging context
 }

rust/crates/remoteingress-protocol/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 tokio = { version = "1", features = ["io-util", "sync", "time"] }
 tokio-util = "0.7"
+bytes = "1"
 log = "0.4"
 
 [dev-dependencies]

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

@@ -2,6 +2,7 @@ use std::collections::VecDeque;
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use bytes::{Bytes, BytesMut};
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, ReadBuf};
 
 // Frame type constants
@@ -32,7 +33,7 @@ pub const WINDOW_UPDATE_THRESHOLD: u32 = INITIAL_STREAM_WINDOW / 2;
 pub const MAX_WINDOW_SIZE: u32 = 16 * 1024 * 1024;
 
 /// Encode a WINDOW_UPDATE frame for a specific stream.
-pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> Vec<u8> {
+pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> Bytes {
     encode_frame(stream_id, frame_type, &increment.to_be_bytes())
 }
 
@@ -45,6 +46,30 @@ pub fn compute_window_for_stream_count(active: u32) -> u32 {
     per_stream.clamp(64 * 1024, INITIAL_STREAM_WINDOW as u64) as u32
 }
 
+/// Proactively clamp a send_window AtomicU32 down to at most `target`.
+/// CAS loop so concurrent WINDOW_UPDATE additions are not lost.
+/// Returns the value after clamping.
+#[inline]
+pub fn clamp_send_window(
+    send_window: &std::sync::atomic::AtomicU32,
+    target: u32,
+) -> u32 {
+    loop {
+        let current = send_window.load(std::sync::atomic::Ordering::Acquire);
+        if current <= target {
+            return current;
+        }
+        match send_window.compare_exchange_weak(
+            current, target,
+            std::sync::atomic::Ordering::AcqRel,
+            std::sync::atomic::Ordering::Relaxed,
+        ) {
+            Ok(_) => return target,
+            Err(_) => continue,
+        }
+    }
+}
+
 /// Decode a WINDOW_UPDATE payload into a byte increment. Returns None if payload is malformed.
 pub fn decode_window_update(payload: &[u8]) -> Option<u32> {
     if payload.len() != 4 {
@@ -58,18 +83,28 @@ pub fn decode_window_update(payload: &[u8]) -> Option<u32> {
 pub struct Frame {
     pub stream_id: u32,
     pub frame_type: u8,
-    pub payload: Vec<u8>,
+    pub payload: Bytes,
 }
 
 /// Encode a frame into bytes: [stream_id:4][type:1][length:4][payload]
-pub fn encode_frame(stream_id: u32, frame_type: u8, payload: &[u8]) -> Vec<u8> {
+pub fn encode_frame(stream_id: u32, frame_type: u8, payload: &[u8]) -> Bytes {
     let len = payload.len() as u32;
     let mut buf = Vec::with_capacity(FRAME_HEADER_SIZE + payload.len());
     buf.extend_from_slice(&stream_id.to_be_bytes());
     buf.push(frame_type);
     buf.extend_from_slice(&len.to_be_bytes());
     buf.extend_from_slice(payload);
-    buf
+    Bytes::from(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.
@@ -142,7 +177,7 @@ impl<R: AsyncRead + Unpin> FrameReader<R> {
         Ok(Some(Frame {
             stream_id,
             frame_type,
-            payload,
+            payload: Bytes::from(payload),
         }))
     }
 
@@ -173,6 +208,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<Bytes>,   // PONG, WINDOW_UPDATE, CLOSE, OPEN — always first
+    data_queue: VecDeque<Bytes>,   // 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.
@@ -181,62 +231,63 @@ pub enum TunnelEvent {
 /// WINDOW_UPDATE starvation that causes flow control deadlocks.
 pub struct TunnelIo<S> {
     stream: 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,
+    // Read state: BytesMut accumulates bytes; split_to extracts frames zero-copy.
+    read_buf: BytesMut,
+    // Write state: extracted sub-struct for safe disjoint borrows
+    write: WriteState,
 }
 
 impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
     pub fn new(stream: S, initial_data: Vec<u8>) -> Self {
-        let read_pos = initial_data.len();
-        let mut read_buf = initial_data;
+        let mut read_buf = BytesMut::from(&initial_data[..]);
         if read_buf.capacity() < 65536 {
             read_buf.reserve(65536 - read_buf.len());
         }
         Self {
             stream,
             read_buf,
-            read_pos,
-            ctrl_queue: VecDeque::new(),
-            data_queue: VecDeque::new(),
-            write_offset: 0,
-            flush_needed: false,
+            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);
+    pub fn queue_ctrl(&mut self, frame: Bytes) {
+        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);
+    pub fn queue_data(&mut self, frame: Bytes) {
+        self.write.data_queue.push_back(frame);
     }
 
     /// Try to parse a complete frame from the read buffer.
+    /// Zero-copy: uses BytesMut::split_to to extract frames without allocating.
     pub fn try_parse_frame(&mut self) -> Option<Result<Frame, std::io::Error>> {
-        if self.read_pos < FRAME_HEADER_SIZE {
+        if self.read_buf.len() < FRAME_HEADER_SIZE {
             return None;
         }
 
         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[0], self.read_buf[1],
+            self.read_buf[2], self.read_buf[3],
         ]);
         let frame_type = self.read_buf[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[5], self.read_buf[6],
+            self.read_buf[7], self.read_buf[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[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],
             ];
             log::error!(
@@ -250,76 +301,81 @@ 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 self.read_buf.len() < 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;
+        // Zero-copy extraction: split the frame off the read buffer (O(1) pointer adjustment).
+        // split_to removes the first total_frame_size bytes from read_buf.
+        let mut frame_data = self.read_buf.split_to(total_frame_size);
+        // Split off header, keep only payload. freeze() converts BytesMut → Bytes (O(1)).
+        let payload = frame_data.split_off(FRAME_HEADER_SIZE).freeze();
 
         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
     pub fn poll_step(
         &mut self,
         cx: &mut Context<'_>,
-        ctrl_rx: &mut tokio::sync::mpsc::Receiver<Vec<u8>>,
-        data_rx: &mut tokio::sync::mpsc::Receiver<Vec<u8>>,
+        ctrl_rx: &mut tokio::sync::mpsc::Receiver<Bytes>,
+        data_rx: &mut tokio::sync::mpsc::Receiver<Bytes>,
         liveness_deadline: &mut Pin<Box<tokio::time::Sleep>>,
         cancel_token: &tokio_util::sync::CancellationToken,
     ) -> Poll<TunnelEvent> {
         // 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.
+        //    Write one frame, set flush_needed, then flush must complete before
+        //    writing more. This prevents unbounded TLS session buffer growth.
+        //    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 && !self.write.flush_needed {
+            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)) => {
+                    log::error!("TunnelIo: poll_write returned 0 (write zero), ctrl_q={} data_q={}",
+                        self.write.ctrl_queue.len(), self.write.data_queue.len());
                     return Poll::Ready(TunnelEvent::WriteError(
                         std::io::Error::new(std::io::ErrorKind::WriteZero, "write zero"),
                     ));
                 }
                 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;
                     }
                 }
-                Poll::Ready(Err(e)) => return Poll::Ready(TunnelEvent::WriteError(e)),
+                Poll::Ready(Err(e)) => {
+                    log::error!("TunnelIo: poll_write error: {} (ctrl_q={} data_q={})",
+                        e, self.write.ctrl_queue.len(), self.write.data_queue.len());
+                    return Poll::Ready(TunnelEvent::WriteError(e));
+                }
                 Poll::Pending => break,
             }
         }
 
         // 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(Err(e)) => return Poll::Ready(TunnelEvent::WriteError(e)),
+                Poll::Ready(Ok(())) => {
+                    self.write.flush_needed = false;
+                }
+                Poll::Ready(Err(e)) => {
+                    log::error!("TunnelIo: poll_flush error: {}", e);
+                    return Poll::Ready(TunnelEvent::WriteError(e));
+                }
                 Poll::Pending => {} // TCP waker will notify us
             }
         }
@@ -329,17 +385,18 @@ 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 {
-            if self.read_buf.len() < self.read_pos + 32768 {
-                self.read_buf.resize(self.read_pos + 32768, 0);
-            }
-            let mut rbuf = ReadBuf::new(&mut self.read_buf[self.read_pos..]);
+            // Ensure at least 32KB of writable space
+            let len_before = self.read_buf.len();
+            self.read_buf.resize(len_before + 32768, 0);
+            let mut rbuf = ReadBuf::new(&mut self.read_buf[len_before..]);
             match Pin::new(&mut self.stream).poll_read(cx, &mut rbuf) {
                 Poll::Ready(Ok(())) => {
                     let n = rbuf.filled().len();
+                    // Trim back to actual data length
+                    self.read_buf.truncate(len_before + n);
                     if n == 0 {
                         return Poll::Ready(TunnelEvent::Eof);
                     }
-                    self.read_pos += n;
                     if let Some(result) = self.try_parse_frame() {
                         return match result {
                             Ok(frame) => Poll::Ready(TunnelEvent::Frame(frame)),
@@ -349,16 +406,27 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
                     // 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::Pending => break,
+                Poll::Ready(Err(e)) => {
+                    self.read_buf.truncate(len_before);
+                    log::error!("TunnelIo: poll_read error: {}", e);
+                    return Poll::Ready(TunnelEvent::ReadError(e));
+                }
+                Poll::Pending => {
+                    self.read_buf.truncate(len_before);
+                    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) {
-                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"),
@@ -367,15 +435,17 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
                 Poll::Pending => break,
             }
         }
-        loop {
-            match data_rx.poll_recv(cx) {
-                Poll::Ready(Some(frame)) => { self.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,
             }
         }
 
@@ -387,10 +457,12 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
             return Poll::Ready(TunnelEvent::Cancelled);
         }
 
-        // 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()) {
+        // 6. SELF-WAKE: only when flush is complete AND we have work.
+        //    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.
+        if !self.write.flush_needed && (got_new || self.write.has_work()) {
             cx.waker().wake_by_ref();
         }
 
@@ -406,6 +478,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";
@@ -571,7 +659,7 @@ mod tests {
             let frame = reader.next_frame().await.unwrap().unwrap();
             assert_eq!(frame.stream_id, i as u32);
             assert_eq!(frame.frame_type, ft);
-            assert_eq!(frame.payload, format!("payload_{}", i).as_bytes());
+            assert_eq!(&frame.payload[..], format!("payload_{}", i).as_bytes());
         }
 
         assert!(reader.next_frame().await.unwrap().is_none());
@@ -580,7 +668,7 @@ mod tests {
     #[tokio::test]
     async fn test_frame_reader_zero_length_payload() {
         let data = encode_frame(42, FRAME_CLOSE, &[]);
-        let cursor = std::io::Cursor::new(data);
+        let cursor = std::io::Cursor::new(data.to_vec());
         let mut reader = FrameReader::new(cursor);
 
         let frame = reader.next_frame().await.unwrap().unwrap();
@@ -708,6 +796,39 @@ mod tests {
         }
     }
 
+    // --- clamp_send_window tests ---
+
+    #[test]
+    fn test_clamp_send_window_reduces_above_target() {
+        let w = std::sync::atomic::AtomicU32::new(4 * 1024 * 1024); // 4 MB
+        let result = clamp_send_window(&w, 512 * 1024); // target 512 KB
+        assert_eq!(result, 512 * 1024);
+        assert_eq!(w.load(std::sync::atomic::Ordering::Relaxed), 512 * 1024);
+    }
+
+    #[test]
+    fn test_clamp_send_window_noop_below_target() {
+        let w = std::sync::atomic::AtomicU32::new(256 * 1024); // 256 KB
+        let result = clamp_send_window(&w, 512 * 1024); // target 512 KB
+        assert_eq!(result, 256 * 1024);
+        assert_eq!(w.load(std::sync::atomic::Ordering::Relaxed), 256 * 1024);
+    }
+
+    #[test]
+    fn test_clamp_send_window_noop_at_target() {
+        let w = std::sync::atomic::AtomicU32::new(512 * 1024);
+        let result = clamp_send_window(&w, 512 * 1024);
+        assert_eq!(result, 512 * 1024);
+        assert_eq!(w.load(std::sync::atomic::Ordering::Relaxed), 512 * 1024);
+    }
+
+    #[test]
+    fn test_clamp_send_window_zero_value() {
+        let w = std::sync::atomic::AtomicU32::new(0);
+        let result = clamp_send_window(&w, 64 * 1024);
+        assert_eq!(result, 0);
+    }
+
     // --- encode/decode window_update roundtrip ---
 
     #[test]

test/test.loadtest.node.ts

@@ -0,0 +1,402 @@
+import { expect, tap } from '@push.rocks/tapbundle';
+import * as net from 'net';
+import * as stream from 'stream';
+import * as crypto from 'crypto';
+import { RemoteIngressHub, RemoteIngressEdge } from '../ts/index.js';
+
+// ---------------------------------------------------------------------------
+// Helpers (self-contained — same patterns as test.flowcontrol.node.ts)
+// ---------------------------------------------------------------------------
+
+async function findFreePorts(count: number): Promise<number[]> {
+  const servers: net.Server[] = [];
+  const ports: number[] = [];
+  for (let i = 0; i < count; i++) {
+    const server = net.createServer();
+    await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+    ports.push((server.address() as net.AddressInfo).port);
+    servers.push(server);
+  }
+  await Promise.all(servers.map((s) => new Promise<void>((resolve) => s.close(() => resolve()))));
+  return ports;
+}
+
+type TrackingServer = net.Server & { destroyAll: () => void };
+
+function startEchoServer(port: number, host: string): Promise<TrackingServer> {
+  return new Promise((resolve, reject) => {
+    const connections = new Set<net.Socket>();
+    const server = net.createServer((socket) => {
+      connections.add(socket);
+      socket.on('close', () => connections.delete(socket));
+      let proxyHeaderParsed = false;
+      let pendingBuf = Buffer.alloc(0);
+      socket.on('data', (data: Buffer) => {
+        if (!proxyHeaderParsed) {
+          pendingBuf = Buffer.concat([pendingBuf, data]);
+          const idx = pendingBuf.indexOf('\r\n');
+          if (idx !== -1) {
+            proxyHeaderParsed = true;
+            const remainder = pendingBuf.subarray(idx + 2);
+            if (remainder.length > 0) socket.write(remainder);
+          }
+          return;
+        }
+        socket.write(data);
+      });
+      socket.on('error', () => {});
+    }) as TrackingServer;
+    server.destroyAll = () => {
+      for (const conn of connections) conn.destroy();
+      connections.clear();
+    };
+    server.on('error', reject);
+    server.listen(port, host, () => resolve(server));
+  });
+}
+
+function sendAndReceive(port: number, data: Buffer, timeoutMs = 30000): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    let totalReceived = 0;
+    const expectedLength = data.length;
+    let settled = false;
+
+    const client = net.createConnection({ host: '127.0.0.1', port }, () => {
+      client.write(data);
+      client.end();
+    });
+
+    const timer = setTimeout(() => {
+      if (!settled) {
+        settled = true;
+        client.destroy();
+        reject(new Error(`Timeout after ${timeoutMs}ms — received ${totalReceived}/${expectedLength} bytes`));
+      }
+    }, timeoutMs);
+
+    client.on('data', (chunk: Buffer) => {
+      chunks.push(chunk);
+      totalReceived += chunk.length;
+      if (totalReceived >= expectedLength && !settled) {
+        settled = true;
+        clearTimeout(timer);
+        client.destroy();
+        resolve(Buffer.concat(chunks));
+      }
+    });
+
+    client.on('end', () => {
+      if (!settled) {
+        settled = true;
+        clearTimeout(timer);
+        resolve(Buffer.concat(chunks));
+      }
+    });
+
+    client.on('error', (err) => {
+      if (!settled) {
+        settled = true;
+        clearTimeout(timer);
+        reject(err);
+      }
+    });
+  });
+}
+
+function sha256(buf: Buffer): string {
+  return crypto.createHash('sha256').update(buf).digest('hex');
+}
+
+// ---------------------------------------------------------------------------
+// Throttle Proxy: rate-limits TCP traffic between edge and hub
+// ---------------------------------------------------------------------------
+
+class ThrottleTransform extends stream.Transform {
+  private bytesPerSec: number;
+  private bucket: number;
+  private lastRefill: number;
+  private destroyed_: boolean = false;
+
+  constructor(bytesPerSecond: number) {
+    super();
+    this.bytesPerSec = bytesPerSecond;
+    this.bucket = bytesPerSecond;
+    this.lastRefill = Date.now();
+  }
+
+  _transform(chunk: Buffer, _encoding: BufferEncoding, callback: stream.TransformCallback) {
+    if (this.destroyed_) return;
+
+    const now = Date.now();
+    const elapsed = (now - this.lastRefill) / 1000;
+    this.bucket = Math.min(this.bytesPerSec, this.bucket + elapsed * this.bytesPerSec);
+    this.lastRefill = now;
+
+    if (chunk.length <= this.bucket) {
+      this.bucket -= chunk.length;
+      callback(null, chunk);
+    } else {
+      // Not enough budget — delay the entire chunk (don't split)
+      const deficit = chunk.length - this.bucket;
+      this.bucket = 0;
+      const delayMs = Math.min((deficit / this.bytesPerSec) * 1000, 1000);
+      setTimeout(() => {
+        if (this.destroyed_) { callback(); return; }
+        this.lastRefill = Date.now();
+        this.bucket = 0;
+        callback(null, chunk);
+      }, delayMs);
+    }
+  }
+
+  _destroy(err: Error | null, callback: (error: Error | null) => void) {
+    this.destroyed_ = true;
+    callback(err);
+  }
+}
+
+interface ThrottleProxy {
+  server: net.Server;
+  close: () => Promise<void>;
+}
+
+async function startThrottleProxy(
+  listenPort: number,
+  targetHost: string,
+  targetPort: number,
+  bytesPerSecond: number,
+): Promise<ThrottleProxy> {
+  const connections = new Set<net.Socket>();
+  const server = net.createServer((clientSock) => {
+    connections.add(clientSock);
+    const upstream = net.createConnection({ host: targetHost, port: targetPort });
+    connections.add(upstream);
+
+    const throttleUp = new ThrottleTransform(bytesPerSecond);
+    const throttleDown = new ThrottleTransform(bytesPerSecond);
+
+    clientSock.pipe(throttleUp).pipe(upstream);
+    upstream.pipe(throttleDown).pipe(clientSock);
+
+    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();
+      upstream.destroy();
+      connections.delete(clientSock);
+      connections.delete(upstream);
+    };
+    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<void>((resolve) => server.listen(listenPort, '127.0.0.1', resolve));
+  return {
+    server,
+    close: async () => {
+      for (const c of connections) c.destroy();
+      connections.clear();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    },
+  };
+}
+
+// ---------------------------------------------------------------------------
+// Test state
+// ---------------------------------------------------------------------------
+
+let hub: RemoteIngressHub;
+let edge: RemoteIngressEdge;
+let echoServer: TrackingServer;
+let throttle: ThrottleProxy;
+let hubPort: number;
+let proxyPort: number;
+let edgePort: number;
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+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 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();
+
+  await hub.start({ tunnelPort: hubPort, targetHost: '127.0.0.2' });
+  await hub.updateAllowedEdges([
+    { id: 'test-edge', secret: 'test-secret', listenPorts: [edgePort] },
+  ]);
+
+  const connectedPromise = new Promise<void>((resolve, reject) => {
+    const timeout = setTimeout(() => reject(new Error('Edge did not connect within 10s')), 10000);
+    edge.once('tunnelConnected', () => {
+      clearTimeout(timeout);
+      resolve();
+    });
+  });
+
+  // Edge connects through throttle proxy
+  await edge.start({
+    hubHost: '127.0.0.1',
+    hubPort: proxyPort,
+    edgeId: 'test-edge',
+    secret: 'test-secret',
+    bindAddress: '127.0.0.1',
+  });
+
+  await connectedPromise;
+  await new Promise((resolve) => setTimeout(resolve, 500));
+
+  const status = await edge.getStatus();
+  expect(status.connected).toBeTrue();
+});
+
+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 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,
+      received: sha256(received),
+      sizeOk: received.length === payloadSize,
+    }));
+  });
+
+  const results = await Promise.all(promises);
+  const failures = results.filter((r) => !r.sizeOk || r.sent !== r.received);
+  expect(failures.length).toEqual(0);
+
+  const status = await edge.getStatus();
+  expect(status.connected).toBeTrue();
+});
+
+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<void>((resolve) => slowSock.on('connect', resolve));
+  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, 2000));
+
+  // 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: 5 }, () =>
+    sendAndReceive(edgePort, payload, 300000).then((r) => ({
+      hash: sha256(r),
+      sizeOk: r.length === payload.length,
+    }))
+  );
+  const results = await Promise.all(promises);
+  const failures = results.filter((r) => !r.sizeOk || r.hash !== hash);
+  expect(failures.length).toEqual(0);
+
+  // Tunnel still alive
+  const status = await edge.getStatus();
+  expect(status.connected).toBeTrue();
+
+  slowSock.destroy();
+});
+
+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: 3 }, () =>
+    sendAndReceive(edgePort, longPayload, 300000).then((r) => ({
+      hash: sha256(r),
+      sizeOk: r.length === longPayload.length,
+    }))
+  );
+
+  const shortPayload = crypto.randomBytes(1024 * 1024);
+  const shortHash = sha256(shortPayload);
+  const shortPromises = Array.from({ length: 50 }, () =>
+    sendAndReceive(edgePort, shortPayload, 300000).then((r) => ({
+      hash: sha256(r),
+      sizeOk: r.length === shortPayload.length,
+    }))
+  );
+
+  const [longResults, shortResults] = await Promise.all([
+    Promise.all(longPromises),
+    Promise.all(shortPromises),
+  ]);
+
+  const longFails = longResults.filter((r) => !r.sizeOk || r.hash !== longHash);
+  const shortFails = shortResults.filter((r) => !r.sizeOk || r.hash !== shortHash);
+  expect(longFails.length).toEqual(0);
+  expect(shortFails.length).toEqual(0);
+
+  const status = await edge.getStatus();
+  expect(status.connected).toBeTrue();
+});
+
+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);
+      return sendAndReceive(edgePort, data, 300000).then((r) => r.length === payloadSize);
+    });
+
+    const results = await Promise.all(promises);
+    const ok = results.filter(Boolean).length;
+    expect(ok).toEqual(streamCount);
+
+    // Brief pause between waves
+    await new Promise((r) => setTimeout(r, 500));
+
+    const status = await edge.getStatus();
+    expect(status.connected).toBeTrue();
+  }
+});
+
+tap.test('throttled: tunnel still works after all load tests', async () => {
+  const data = crypto.randomBytes(1024);
+  const hash = sha256(data);
+  const received = await sendAndReceive(edgePort, data, 30000);
+  expect(sha256(received)).toEqual(hash);
+
+  const status = await edge.getStatus();
+  expect(status.connected).toBeTrue();
+});
+
+tap.test('teardown: stop tunnel', async () => {
+  await edge.stop();
+  await hub.stop();
+  if (throttle) await throttle.close();
+  await new Promise<void>((resolve) => echoServer.close(() => resolve()));
+});
+
+export default tap.start();

ts/00_commitinfo_data.ts

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