v4.9.0

changelog.md

@@ -1,11 +1,37 @@
 # Changelog
 
-## 2026-03-17 - 4.8.14 - fix(rust-core,protocol)
-eliminate edge stream registration races and reduce frame buffering copies
+## 2026-03-18 - 4.9.0 - feat(protocol)
+add sustained-stream tunnel scheduling to isolate high-throughput traffic
 
-- 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
+- Introduce a third low-priority sustained queue in TunnelIo with a forced drain budget to prevent long-lived high-bandwidth streams from starving control and normal data frames.
+- Classify upload and download streams as sustained after exceeding the throughput threshold for the minimum duration, and route their DATA and CLOSE frames through the sustained channel.
+- Wire the new sustained channel through edge and hub stream handling so sustained traffic is scheduled consistently on both sides of the tunnel.
+
+## 2026-03-18 - 4.8.19 - fix(remoteingress-protocol)
+reduce per-stream flow control windows and increase control channel buffering
+
+- Lower the initial and maximum per-stream window from 16MB to 4MB and scale adaptive windows against a 200MB total budget with a 1MB minimum.
+- Increase edge and hub control frame channel capacity from 256 to 512 to better handle prioritized control traffic.
+- Update flow-control tests and comments to reflect the new window sizing and budget behavior.
+
+## 2026-03-17 - 4.8.18 - fix(rust-protocol)
+switch tunnel frame buffers from Vec<u8> to Bytes to reduce copying and memory overhead
+
+- Add the bytes crate to core and protocol crates
+- Update frame encoding, reader payloads, channel queues, and stream backchannels to use Bytes
+- Adjust edge and hub data/control paths to send framed payloads as Bytes
+
+## 2026-03-17 - 4.8.17 - fix(protocol)
+increase per-stream flow control windows and remove adaptive read caps
+
+- Raise the initial per-stream window from 4MB to 16MB and expand the adaptive window budget to 800MB with a 4MB floor
+- Stop limiting edge and hub reads by the adaptive per-stream target window, keeping reads capped only by the current window and 32KB chunk size
+- Update protocol tests to match the new adaptive window scaling and budget boundaries
+
+## 2026-03-17 - 4.8.16 - fix(release)
+bump package version to 4.8.15
+
+- Updates the package.json version field from 4.8.13 to 4.8.15.
 
 ## 2026-03-17 - 4.8.13 - fix(remoteingress-protocol)
 require a flush after each written frame to bound TLS buffer growth

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "4.8.14",
+  "version": "4.9.0",
   "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

@@ -9,9 +9,9 @@ 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 bytes::Bytes;
 use remoteingress_protocol::*;
 
 type EdgeTlsStream = tokio_rustls::client::TlsStream<TcpStream>;
@@ -23,7 +23,7 @@ enum EdgeFrameAction {
     Disconnect(String),
 }
 
-/// Per-stream state tracked in the edge's stream map.
+/// Per-stream state tracked in the edge's client_writers map.
 struct EdgeStreamState {
     /// 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.
@@ -35,12 +35,6 @@ 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")]
@@ -291,29 +285,40 @@ enum EdgeLoopResult {
 
 /// 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(
+async 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>>,
+    client_writers: &Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
+    listen_ports: &Arc<RwLock<Vec<u16>>>,
+    event_tx: &mpsc::Sender<EdgeEvent>,
+    tunnel_writer_tx: &mpsc::Sender<Bytes>,
+    tunnel_data_tx: &mpsc::Sender<Bytes>,
+    tunnel_sustained_tx: &mpsc::Sender<Bytes>,
+    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).
-            if let Some(state) = edge_streams.get(&frame.stream_id) {
+            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
-                    edge_streams.remove(&frame.stream_id);
+                    writers.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 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);
@@ -324,12 +329,29 @@ fn handle_edge_frame(
             }
         }
         FRAME_CLOSE_BACK => {
-            edge_streams.remove(&frame.stream_id);
+            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_update = Some(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,
+                    tunnel_sustained_tx,
+                    client_writers,
+                    active_streams,
+                    next_stream_id,
+                    edge_id,
+                    connection_token,
+                    bind_address,
+                );
             }
         }
         FRAME_PING => {
@@ -471,17 +493,15 @@ async fn connect_to_hub_and_run(
         }
     });
 
-    // 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);
+    // 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()));
 
     // 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::<Bytes>(256);
+    let (tunnel_ctrl_tx, mut tunnel_ctrl_rx) = mpsc::channel::<Bytes>(512);
     let (tunnel_data_tx, mut tunnel_data_rx) = mpsc::channel::<Bytes>(4096);
+    let (tunnel_sustained_tx, mut tunnel_sustained_rx) = mpsc::channel::<Bytes>(4096);
     let tunnel_writer_tx = tunnel_ctrl_tx.clone();
 
     // Start TCP listeners for initial ports
@@ -492,8 +512,8 @@ async fn connect_to_hub_and_run(
         &mut port_listeners,
         &tunnel_writer_tx,
         &tunnel_data_tx,
-        &register_tx,
-        &cleanup_tx,
+        &tunnel_sustained_tx,
+        &client_writers,
         active_streams,
         next_stream_id,
         &config.edge_id,
@@ -510,18 +530,7 @@ async fn connect_to_hub_and_run(
     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 {
             let frame = match tunnel_io.try_parse_frame() {
                 Some(Ok(f)) => f,
@@ -534,55 +543,28 @@ async fn connect_to_hub_and_run(
             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,
-            ) {
+                frame, &mut tunnel_io, &client_writers, listen_ports, event_tx,
+                &tunnel_writer_tx, &tunnel_data_tx, &tunnel_sustained_tx, &mut port_listeners,
+                active_streams, next_stream_id, &config.edge_id, connection_token, bind_address,
+            ).await {
                 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)
+            tunnel_io.poll_step(cx, &mut tunnel_ctrl_rx, &mut tunnel_data_rx, &mut tunnel_sustained_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);
                 if let EdgeFrameAction::Disconnect(reason) = handle_edge_frame(
-                    frame, &mut tunnel_io, &mut edge_streams, &mut listen_ports_update,
-                ) {
+                    frame, &mut tunnel_io, &client_writers, listen_ports, event_tx,
+                    &tunnel_writer_tx, &tunnel_data_tx, &tunnel_sustained_tx, &mut port_listeners,
+                    active_streams, next_stream_id, &config.edge_id, connection_token, bind_address,
+                ).await {
                     break EdgeLoopResult::Reconnect(reason);
                 }
             }
@@ -609,27 +591,6 @@ 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,
-            );
-        }
     };
 
     // Cancel stream tokens FIRST so stream handlers exit immediately.
@@ -658,8 +619,8 @@ fn apply_port_config(
     port_listeners: &mut HashMap<u16, JoinHandle<()>>,
     tunnel_ctrl_tx: &mpsc::Sender<Bytes>,
     tunnel_data_tx: &mpsc::Sender<Bytes>,
-    register_tx: &mpsc::Sender<StreamRegistration>,
-    cleanup_tx: &mpsc::Sender<u32>,
+    tunnel_sustained_tx: &mpsc::Sender<Bytes>,
+    client_writers: &Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
     active_streams: &Arc<AtomicU32>,
     next_stream_id: &Arc<AtomicU32>,
     edge_id: &str,
@@ -681,8 +642,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 register_tx = register_tx.clone();
-        let cleanup_tx = cleanup_tx.clone();
+        let tunnel_sustained_tx = tunnel_sustained_tx.clone();
+        let client_writers = client_writers.clone();
         let active_streams = active_streams.clone();
         let next_stream_id = next_stream_id.clone();
         let edge_id = edge_id.to_string();
@@ -716,8 +677,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 register_tx = register_tx.clone();
-                                let cleanup_tx = cleanup_tx.clone();
+                                let tunnel_sustained_tx = tunnel_sustained_tx.clone();
+                                let client_writers = client_writers.clone();
                                 let active_streams = active_streams.clone();
                                 let edge_id = edge_id.clone();
                                 let client_token = port_token.child_token();
@@ -733,8 +694,8 @@ fn apply_port_config(
                                         &edge_id,
                                         tunnel_ctrl_tx,
                                         tunnel_data_tx,
-                                        register_tx,
-                                        cleanup_tx,
+                                        tunnel_sustained_tx,
+                                        client_writers,
                                         client_token,
                                         Arc::clone(&active_streams),
                                     )
@@ -777,8 +738,8 @@ async fn handle_client_connection(
     edge_id: &str,
     tunnel_ctrl_tx: mpsc::Sender<Bytes>,
     tunnel_data_tx: mpsc::Sender<Bytes>,
-    register_tx: mpsc::Sender<StreamRegistration>,
-    cleanup_tx: mpsc::Sender<u32>,
+    tunnel_sustained_tx: mpsc::Sender<Bytes>,
+    client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
     client_token: CancellationToken,
     active_streams: Arc<AtomicU32>,
 ) {
@@ -788,36 +749,6 @@ 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";
 
-    // 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(
-        active_streams.load(Ordering::Relaxed),
-    );
-    let send_window = Arc::new(AtomicU32::new(initial_window));
-    let window_notify = Arc::new(Notify::new());
-
-    // 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());
@@ -826,10 +757,29 @@ async fn handle_client_connection(
         _ = client_token.cancelled() => false,
     };
     if !send_ok {
-        let _ = cleanup_tx.try_send(stream_id);
         return;
     }
 
+    // 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 200MB budget. Prevents burst flooding when many streams open.
+    let initial_window = remoteingress_protocol::compute_window_for_stream_count(
+        active_streams.load(Ordering::Relaxed),
+    );
+    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),
+        });
+    }
+
     let (mut client_read, mut client_write) = client_stream.into_split();
 
     // Task: hub -> client (download direction)
@@ -892,6 +842,9 @@ async fn handle_client_connection(
     // 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];
+    let mut stream_bytes_sent: u64 = 0;
+    let stream_start = tokio::time::Instant::now();
+    let mut is_sustained = false;
     loop {
         // Wait for send window to have capacity (with stall timeout).
         // Safe pattern: register notified BEFORE checking the condition
@@ -913,12 +866,11 @@ async fn handle_client_connection(
         }
         if client_token.is_cancelled() { break; }
 
-        // 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;
+        // 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;
         if w == 0 {
             log::warn!("Stream {} upload: window still 0 after stall timeout, closing", stream_id);
             break;
@@ -933,8 +885,21 @@ async fn handle_client_connection(
                         send_window.fetch_sub(n as u32, Ordering::Release);
                         encode_frame_header(&mut buf, stream_id, FRAME_DATA, n);
                         let data_frame = Bytes::copy_from_slice(&buf[..FRAME_HEADER_SIZE + n]);
+                        // Sustained classification: >2.5 MB/s for >10 seconds
+                        stream_bytes_sent += n as u64;
+                        if !is_sustained {
+                            let elapsed = stream_start.elapsed().as_secs();
+                            if elapsed >= remoteingress_protocol::SUSTAINED_MIN_DURATION_SECS
+                                && stream_bytes_sent / elapsed >= remoteingress_protocol::SUSTAINED_THRESHOLD_BPS
+                            {
+                                is_sustained = true;
+                                log::debug!("Stream {} classified as sustained (upload, {} bytes in {}s)",
+                                    stream_id, stream_bytes_sent, elapsed);
+                            }
+                        }
+                        let tx = if is_sustained { &tunnel_sustained_tx } else { &tunnel_data_tx };
                         let sent = tokio::select! {
-                            result = tunnel_data_tx.send(data_frame) => result.is_ok(),
+                            result = tx.send(data_frame) => result.is_ok(),
                             _ = client_token.cancelled() => false,
                         };
                         if !sent { break; }
@@ -961,14 +926,18 @@ async fn handle_client_connection(
     // 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 tx = if is_sustained { &tunnel_sustained_tx } else { &tunnel_data_tx };
         tokio::select! {
-            _ = tunnel_data_tx.send(close_frame) => {}
+            _ = tx.send(close_frame) => {}
             _ = client_token.cancelled() => {}
         }
     }
 
-    // Clean up — notify main loop to remove stream state
-    let _ = cleanup_tx.try_send(stream_id);
+    // Clean up
+    {
+        let mut writers = client_writers.lock().await;
+        writers.remove(&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-core/src/hub.rs

@@ -310,6 +310,7 @@ async fn handle_hub_frame(
     event_tx: &mpsc::Sender<HubEvent>,
     ctrl_tx: &mpsc::Sender<Bytes>,
     data_tx: &mpsc::Sender<Bytes>,
+    sustained_tx: &mpsc::Sender<Bytes>,
     target_host: &str,
     edge_token: &CancellationToken,
     cleanup_tx: &mpsc::Sender<u32>,
@@ -338,6 +339,7 @@ async fn handle_hub_frame(
             let cleanup = cleanup_tx.clone();
             let writer_tx = ctrl_tx.clone();   // control: CLOSE_BACK, WINDOW_UPDATE_BACK
             let data_writer_tx = data_tx.clone(); // data: DATA_BACK
+            let sustained_writer_tx = sustained_tx.clone(); // sustained: DATA_BACK from elephant flows
             let target = target_host.to_string();
             let stream_token = edge_token.child_token();
 
@@ -349,7 +351,7 @@ async fn handle_hub_frame(
             // Create channel for data from edge to this stream
             let (stream_data_tx, mut stream_data_rx) = mpsc::unbounded_channel::<Bytes>();
             // Adaptive initial window: scale with current stream count
-            // to keep total in-flight data within the 32MB budget.
+            // to keep total in-flight data within the 200MB budget.
             let initial_window = compute_window_for_stream_count(
                 edge_stream_count.load(Ordering::Relaxed),
             );
@@ -458,6 +460,9 @@ async fn handle_hub_frame(
                     // with per-stream flow control (check send_window before reading).
                     // Zero-copy: read payload directly after the header, then prepend header.
                     let mut buf = vec![0u8; FRAME_HEADER_SIZE + 32768];
+                    let mut dl_bytes_sent: u64 = 0;
+                    let dl_start = tokio::time::Instant::now();
+                    let mut is_sustained = false;
                     loop {
                         // Wait for send window to have capacity (with stall timeout).
                         // Safe pattern: register notified BEFORE checking the condition
@@ -479,12 +484,11 @@ async fn handle_hub_frame(
                         }
                         if stream_token.is_cancelled() { break; }
 
-                        // 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(
-                            stream_counter.load(Ordering::Relaxed),
-                        );
-                        let w = remoteingress_protocol::clamp_send_window(&send_window, adaptive_target) as usize;
+                        // 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;
                         if w == 0 {
                             log::warn!("Stream {} download: window still 0 after stall timeout, closing", stream_id);
                             break;
@@ -499,8 +503,21 @@ async fn handle_hub_frame(
                                         send_window.fetch_sub(n as u32, Ordering::Release);
                                         encode_frame_header(&mut buf, stream_id, FRAME_DATA_BACK, n);
                                         let frame = Bytes::copy_from_slice(&buf[..FRAME_HEADER_SIZE + n]);
+                                        // Sustained classification: >2.5 MB/s for >10 seconds
+                                        dl_bytes_sent += n as u64;
+                                        if !is_sustained {
+                                            let elapsed = dl_start.elapsed().as_secs();
+                                            if elapsed >= remoteingress_protocol::SUSTAINED_MIN_DURATION_SECS
+                                                && dl_bytes_sent / elapsed >= remoteingress_protocol::SUSTAINED_THRESHOLD_BPS
+                                            {
+                                                is_sustained = true;
+                                                log::debug!("Stream {} classified as sustained (download, {} bytes in {}s)",
+                                                    stream_id, dl_bytes_sent, elapsed);
+                                            }
+                                        }
+                                        let tx = if is_sustained { &sustained_writer_tx } else { &data_writer_tx };
                                         let sent = tokio::select! {
-                                            result = data_writer_tx.send(frame) => result.is_ok(),
+                                            result = tx.send(frame) => result.is_ok(),
                                             _ = stream_token.cancelled() => false,
                                         };
                                         if !sent { break; }
@@ -512,12 +529,13 @@ async fn handle_hub_frame(
                         }
                     }
 
-                    // Send CLOSE_BACK via DATA channel (must arrive AFTER last DATA_BACK).
+                    // Send CLOSE_BACK via same channel as DATA_BACK (must arrive AFTER last DATA_BACK).
                     // select! with cancellation guard prevents indefinite blocking if tunnel dies.
                     if !stream_token.is_cancelled() {
                         let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                        let tx = if is_sustained { &sustained_writer_tx } else { &data_writer_tx };
                         tokio::select! {
-                            _ = data_writer_tx.send(close_frame) => {}
+                            _ = tx.send(close_frame) => {}
                             _ = stream_token.cancelled() => {}
                         }
                     }
@@ -529,7 +547,9 @@ async fn handle_hub_frame(
 
                 if let Err(e) = result {
                     log::error!("Stream {} error: {}", stream_id, e);
-                    // Send CLOSE_BACK via DATA channel on error (must arrive after any DATA_BACK).
+                    // Send CLOSE_BACK on error (must arrive after any DATA_BACK).
+                    // Error path: is_sustained not available here, use data channel (safe —
+                    // if error occurs before classification, no sustained frames were sent).
                     if !stream_token.is_cancelled() {
                         let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
                         tokio::select! {
@@ -709,8 +729,9 @@ async fn handle_edge_connection(
 
     // QoS dual-channel: ctrl frames have priority over data frames.
     // Stream handlers send through these channels -> TunnelIo drains them.
-    let (ctrl_tx, mut ctrl_rx) = mpsc::channel::<Bytes>(256);
+    let (ctrl_tx, mut ctrl_rx) = mpsc::channel::<Bytes>(512);
     let (data_tx, mut data_rx) = mpsc::channel::<Bytes>(4096);
+    let (sustained_tx, mut sustained_rx) = mpsc::channel::<Bytes>(4096);
 
     // Spawn task to forward config updates as FRAME_CONFIG frames
     let config_writer_tx = ctrl_tx.clone();
@@ -784,7 +805,7 @@ async fn handle_edge_connection(
             liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);
             if let FrameAction::Disconnect(reason) = handle_hub_frame(
                 frame, &mut tunnel_io, &mut streams, &stream_semaphore, &edge_stream_count,
-                &edge_id, &event_tx, &ctrl_tx, &data_tx, &target_host, &edge_token,
+                &edge_id, &event_tx, &ctrl_tx, &data_tx, &sustained_tx, &target_host, &edge_token,
                 &cleanup_tx,
             ).await {
                 disconnect_reason = reason;
@@ -798,7 +819,7 @@ async fn handle_edge_connection(
             if ping_ticker.poll_tick(cx).is_ready() {
                 tunnel_io.queue_ctrl(encode_frame(0, FRAME_PING, &[]));
             }
-            tunnel_io.poll_step(cx, &mut ctrl_rx, &mut data_rx, &mut liveness_deadline, &edge_token)
+            tunnel_io.poll_step(cx, &mut ctrl_rx, &mut data_rx, &mut sustained_rx, &mut liveness_deadline, &edge_token)
         }).await;
 
         match event {
@@ -807,7 +828,7 @@ async fn handle_edge_connection(
                 liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);
                 if let FrameAction::Disconnect(reason) = handle_hub_frame(
                     frame, &mut tunnel_io, &mut streams, &stream_semaphore, &edge_stream_count,
-                    &edge_id, &event_tx, &ctrl_tx, &data_tx, &target_host, &edge_token,
+                    &edge_id, &event_tx, &ctrl_tx, &data_tx, &sustained_tx, &target_host, &edge_token,
                     &cleanup_tx,
                 ).await {
                     disconnect_reason = reason;

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

@@ -2,8 +2,10 @@ use std::collections::VecDeque;
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use bytes::{Bytes, BytesMut};
+use std::time::Duration;
+use bytes::{Bytes, BytesMut, BufMut};
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, ReadBuf};
+use tokio::time::Instant;
 
 // Frame type constants
 pub const FRAME_OPEN: u8 = 0x01;
@@ -24,13 +26,22 @@ pub const FRAME_HEADER_SIZE: usize = 9;
 pub const MAX_PAYLOAD_SIZE: u32 = 16 * 1024 * 1024;
 
 // Per-stream flow control constants
-/// Initial per-stream window size (4 MB). Sized for full throughput at high RTT:
-/// at 100ms RTT, this sustains ~40 MB/s per stream.
+/// Initial (and maximum) per-stream window size (4 MB).
 pub const INITIAL_STREAM_WINDOW: u32 = 4 * 1024 * 1024;
 /// Send WINDOW_UPDATE after consuming this many bytes (half the initial window).
 pub const WINDOW_UPDATE_THRESHOLD: u32 = INITIAL_STREAM_WINDOW / 2;
 /// Maximum window size to prevent overflow.
-pub const MAX_WINDOW_SIZE: u32 = 16 * 1024 * 1024;
+pub const MAX_WINDOW_SIZE: u32 = 4 * 1024 * 1024;
+
+// Sustained stream classification constants
+/// Throughput threshold for sustained classification (2.5 MB/s = 20 Mbit/s).
+pub const SUSTAINED_THRESHOLD_BPS: u64 = 2_500_000;
+/// Minimum duration before a stream can be classified as sustained.
+pub const SUSTAINED_MIN_DURATION_SECS: u64 = 10;
+/// Fixed window for sustained streams (1 MB — the floor).
+pub const SUSTAINED_WINDOW: u32 = 1 * 1024 * 1024;
+/// Maximum bytes written from sustained queue per forced drain (1 MB/s guarantee).
+pub const SUSTAINED_FORCED_DRAIN_CAP: usize = 1_048_576;
 
 /// Encode a WINDOW_UPDATE frame for a specific stream.
 pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> Bytes {
@@ -38,36 +49,11 @@ pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> B
 }
 
 /// Compute the target per-stream window size based on the number of active streams.
-/// Total memory budget is ~32MB shared across all streams. As more streams are active,
-/// each gets a smaller window. This adapts to current demand — few streams get high
-/// throughput, many streams save memory and reduce control frame pressure.
+/// Total memory budget is ~200MB shared across all streams. Up to 50 streams get the
+/// full 4MB window; above that the window scales down to a 1MB floor at 200+ streams.
 pub fn compute_window_for_stream_count(active: u32) -> u32 {
-    let per_stream = (32 * 1024 * 1024u64) / (active.max(1) as u64);
-    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,
-        }
-    }
+    let per_stream = (200 * 1024 * 1024u64) / (active.max(1) as u64);
+    per_stream.clamp(1 * 1024 * 1024, INITIAL_STREAM_WINDOW as u64) as u32
 }
 
 /// Decode a WINDOW_UPDATE payload into a byte increment. Returns None if payload is malformed.
@@ -89,12 +75,12 @@ pub struct Frame {
 /// 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]) -> 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);
-    Bytes::from(buf)
+    let mut buf = BytesMut::with_capacity(FRAME_HEADER_SIZE + payload.len());
+    buf.put_slice(&stream_id.to_be_bytes());
+    buf.put_u8(frame_type);
+    buf.put_slice(&len.to_be_bytes());
+    buf.put_slice(payload);
+    buf.freeze()
 }
 
 /// Write a frame header into `buf[0..FRAME_HEADER_SIZE]`.
@@ -169,7 +155,7 @@ impl<R: AsyncRead + Unpin> FrameReader<R> {
             ));
         }
 
-        let mut payload = vec![0u8; length as usize];
+        let mut payload = BytesMut::zeroed(length as usize);
         if length > 0 {
             self.reader.read_exact(&mut payload).await?;
         }
@@ -177,7 +163,7 @@ impl<R: AsyncRead + Unpin> FrameReader<R> {
         Ok(Some(Frame {
             stream_id,
             frame_type,
-            payload: Bytes::from(payload),
+            payload: payload.freeze(),
         }))
     }
 
@@ -211,46 +197,60 @@ pub enum TunnelEvent {
 /// 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
+    ctrl_queue: VecDeque<Bytes>,      // PONG, WINDOW_UPDATE, CLOSE, OPEN — always first
+    data_queue: VecDeque<Bytes>,      // DATA, DATA_BACK — only when ctrl is empty
+    sustained_queue: VecDeque<Bytes>, // DATA, DATA_BACK from sustained streams — lowest priority
+    offset: usize,                    // progress within current frame being written
     flush_needed: bool,
+    // Sustained starvation prevention: guaranteed 1 MB/s drain
+    sustained_last_drain: Instant,
+    sustained_bytes_this_period: usize,
 }
 
 impl WriteState {
     fn has_work(&self) -> bool {
-        !self.ctrl_queue.is_empty() || !self.data_queue.is_empty()
+        !self.ctrl_queue.is_empty() || !self.data_queue.is_empty() || !self.sustained_queue.is_empty()
     }
 }
 
 /// Single-owner I/O engine for the tunnel TLS connection.
 ///
 /// Owns the TLS stream directly — no `tokio::io::split()`, no mutex.
-/// Uses two priority write queues: ctrl frames (PONG, WINDOW_UPDATE, CLOSE, OPEN)
-/// are ALWAYS written before data frames (DATA, DATA_BACK). This prevents
-/// WINDOW_UPDATE starvation that causes flow control deadlocks.
+/// Uses three priority write queues:
+///   1. ctrl  (PONG, WINDOW_UPDATE, CLOSE, OPEN) — always first
+///   2. data  (DATA, DATA_BACK from normal streams) — when ctrl empty
+///   3. sustained (DATA, DATA_BACK from sustained streams) — lowest priority,
+///      drained freely when ctrl+data empty, or forced 1MB/s when they're not
 pub struct TunnelIo<S> {
     stream: S,
-    // Read state: BytesMut accumulates bytes; split_to extracts frames zero-copy.
-    read_buf: BytesMut,
+    // Read state: accumulate bytes, parse frames incrementally
+    read_buf: Vec<u8>,
+    read_pos: usize,
+    parse_pos: usize,
     // 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 mut read_buf = BytesMut::from(&initial_data[..]);
+        let read_pos = initial_data.len();
+        let mut read_buf = initial_data;
         if read_buf.capacity() < 65536 {
             read_buf.reserve(65536 - read_buf.len());
         }
         Self {
             stream,
             read_buf,
+            read_pos,
+            parse_pos: 0,
             write: WriteState {
                 ctrl_queue: VecDeque::new(),
                 data_queue: VecDeque::new(),
+                sustained_queue: VecDeque::new(),
                 offset: 0,
                 flush_needed: false,
+                sustained_last_drain: Instant::now(),
+                sustained_bytes_this_period: 0,
             },
         }
     }
@@ -265,30 +265,37 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         self.write.data_queue.push_back(frame);
     }
 
+    /// Queue a lowest-priority sustained data frame.
+    pub fn queue_sustained(&mut self, frame: Bytes) {
+        self.write.sustained_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.
+    /// 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_buf.len() < 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={}",
@@ -301,48 +308,63 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         }
 
         let total_frame_size = FRAME_HEADER_SIZE + length as usize;
-        if self.read_buf.len() < total_frame_size {
+        if available < total_frame_size {
             return None;
         }
 
-        // 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();
+        let payload = Bytes::copy_from_slice(
+            &self.read_buf[base + FRAME_HEADER_SIZE..base + total_frame_size],
+        );
+        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 }))
     }
 
     /// Poll-based I/O step. Returns Ready on events, Pending when idle.
     ///
-    /// Order: write(ctrl→data) → flush → read → channels → timers
+    /// Order: write(ctrl->data->sustained) -> flush -> read -> channels -> timers
     pub fn poll_step(
         &mut self,
         cx: &mut Context<'_>,
         ctrl_rx: &mut tokio::sync::mpsc::Receiver<Bytes>,
         data_rx: &mut tokio::sync::mpsc::Receiver<Bytes>,
+        sustained_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.
+        // 1. WRITE: 3-tier priority — ctrl first, then data, then sustained.
+        //    Sustained drains freely when ctrl+data are empty.
         //    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.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()
+            // Pick queue: ctrl > data > sustained
+            let queue_id = if !self.write.ctrl_queue.is_empty() {
+                0 // ctrl
+            } else if !self.write.data_queue.is_empty() {
+                1 // data
             } else {
-                self.write.data_queue.front().unwrap()
+                2 // sustained
+            };
+            let frame = match queue_id {
+                0 => self.write.ctrl_queue.front().unwrap(),
+                1 => self.write.data_queue.front().unwrap(),
+                _ => self.write.sustained_queue.front().unwrap(),
             };
             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());
+                    log::error!("TunnelIo: poll_write returned 0 (write zero), ctrl_q={} data_q={} sustained_q={}",
+                        self.write.ctrl_queue.len(), self.write.data_queue.len(), self.write.sustained_queue.len());
                     return Poll::Ready(TunnelEvent::WriteError(
                         std::io::Error::new(std::io::ErrorKind::WriteZero, "write zero"),
                     ));
@@ -351,21 +373,70 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
                     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(); }
+                        match queue_id {
+                            0 => { self.write.ctrl_queue.pop_front(); }
+                            1 => { self.write.data_queue.pop_front(); }
+                            _ => {
+                                self.write.sustained_queue.pop_front();
+                                self.write.sustained_last_drain = Instant::now();
+                                self.write.sustained_bytes_this_period = 0;
+                            }
+                        }
                         self.write.offset = 0;
                         writes += 1;
                     }
                 }
                 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());
+                    log::error!("TunnelIo: poll_write error: {} (ctrl_q={} data_q={} sustained_q={})",
+                        e, self.write.ctrl_queue.len(), self.write.data_queue.len(), self.write.sustained_queue.len());
                     return Poll::Ready(TunnelEvent::WriteError(e));
                 }
                 Poll::Pending => break,
             }
         }
 
+        // 1b. FORCED SUSTAINED DRAIN: when ctrl/data have work but sustained is waiting,
+        //     guarantee at least 1 MB/s by draining up to SUSTAINED_FORCED_DRAIN_CAP
+        //     once per second.
+        if !self.write.sustained_queue.is_empty()
+            && (!self.write.ctrl_queue.is_empty() || !self.write.data_queue.is_empty())
+            && !self.write.flush_needed
+        {
+            let now = Instant::now();
+            if now.duration_since(self.write.sustained_last_drain) >= Duration::from_secs(1) {
+                self.write.sustained_bytes_this_period = 0;
+                self.write.sustained_last_drain = now;
+
+                while !self.write.sustained_queue.is_empty()
+                    && self.write.sustained_bytes_this_period < SUSTAINED_FORCED_DRAIN_CAP
+                    && !self.write.flush_needed
+                {
+                    let frame = self.write.sustained_queue.front().unwrap();
+                    let remaining = &frame[self.write.offset..];
+                    match Pin::new(&mut self.stream).poll_write(cx, remaining) {
+                        Poll::Ready(Ok(0)) => {
+                            return Poll::Ready(TunnelEvent::WriteError(
+                                std::io::Error::new(std::io::ErrorKind::WriteZero, "write zero"),
+                            ));
+                        }
+                        Poll::Ready(Ok(n)) => {
+                            self.write.offset += n;
+                            self.write.flush_needed = true;
+                            self.write.sustained_bytes_this_period += n;
+                            if self.write.offset >= frame.len() {
+                                self.write.sustained_queue.pop_front();
+                                self.write.offset = 0;
+                            }
+                        }
+                        Poll::Ready(Err(e)) => {
+                            return Poll::Ready(TunnelEvent::WriteError(e));
+                        }
+                        Poll::Pending => break,
+                    }
+                }
+            }
+        }
+
         // 2. FLUSH: push encrypted data from TLS session to TCP.
         if self.write.flush_needed {
             match Pin::new(&mut self.stream).poll_flush(cx) {
@@ -385,18 +456,23 @@ 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 {
-            // 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..]);
+            // 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);
+            }
+            let mut rbuf = ReadBuf::new(&mut self.read_buf[self.read_pos..]);
             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)),
@@ -407,14 +483,10 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
                     // waker is re-registered when it finally returns Pending.
                 }
                 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;
-                }
+                Poll::Pending => break,
             }
         }
 
@@ -422,7 +494,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
         //    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.
+        //    data_queue (unbounded VecDeque) grows to hundreds of MB under throttle -> OOM.
         let mut got_new = false;
         loop {
             match ctrl_rx.poll_recv(cx) {
@@ -448,6 +520,16 @@ impl<S: AsyncRead + AsyncWrite + Unpin> TunnelIo<S> {
                 }
             }
         }
+        // Sustained channel: drain when sustained_queue is small (same backpressure pattern).
+        // Channel close is non-fatal — not all connections have sustained streams.
+        if self.write.sustained_queue.len() < 64 {
+            loop {
+                match sustained_rx.poll_recv(cx) {
+                    Poll::Ready(Some(frame)) => { self.write.sustained_queue.push_back(frame); got_new = true; }
+                    Poll::Ready(None) | Poll::Pending => break,
+                }
+            }
+        }
 
         // 5. TIMERS
         if liveness_deadline.as_mut().poll(cx).is_ready() {
@@ -484,14 +566,14 @@ mod tests {
         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)[..]);
+        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, &[])[..]);
+        assert_eq!(buf, &encode_frame(99, FRAME_CLOSE, &[])[..]);
     }
 
     #[test]
@@ -696,90 +778,57 @@ mod tests {
 
     #[test]
     fn test_adaptive_window_zero_streams() {
-        // 0 streams treated as 1: 32MB/1 = 32MB → clamped to 4MB max
+        // 0 streams treated as 1: 200MB/1 -> clamped to 4MB max
         assert_eq!(compute_window_for_stream_count(0), INITIAL_STREAM_WINDOW);
     }
 
     #[test]
     fn test_adaptive_window_one_stream() {
-        // 32MB/1 = 32MB → clamped to 4MB max
         assert_eq!(compute_window_for_stream_count(1), INITIAL_STREAM_WINDOW);
     }
 
     #[test]
-    fn test_adaptive_window_at_max_boundary() {
-        // 32MB/8 = 4MB = exactly INITIAL_STREAM_WINDOW
-        assert_eq!(compute_window_for_stream_count(8), INITIAL_STREAM_WINDOW);
+    fn test_adaptive_window_50_streams_full() {
+        // 200MB/50 = 4MB = exactly INITIAL_STREAM_WINDOW
+        assert_eq!(compute_window_for_stream_count(50), INITIAL_STREAM_WINDOW);
     }
 
     #[test]
-    fn test_adaptive_window_just_below_max() {
-        // 32MB/9 = 3,728,270 — first value below INITIAL_STREAM_WINDOW
-        let w = compute_window_for_stream_count(9);
+    fn test_adaptive_window_51_streams_starts_scaling() {
+        // 200MB/51 < 4MB — first value below max
+        let w = compute_window_for_stream_count(51);
         assert!(w < INITIAL_STREAM_WINDOW);
-        assert_eq!(w, (32 * 1024 * 1024u64 / 9) as u32);
-    }
-
-    #[test]
-    fn test_adaptive_window_16_streams() {
-        // 32MB/16 = 2MB
-        assert_eq!(compute_window_for_stream_count(16), 2 * 1024 * 1024);
+        assert_eq!(w, (200 * 1024 * 1024u64 / 51) as u32);
     }
 
     #[test]
     fn test_adaptive_window_100_streams() {
-        // 32MB/100 = 335,544 bytes (~327KB)
-        let w = compute_window_for_stream_count(100);
-        assert_eq!(w, (32 * 1024 * 1024u64 / 100) as u32);
-        assert!(w > 64 * 1024); // above floor
-        assert!(w < INITIAL_STREAM_WINDOW as u32); // below ceiling
+        // 200MB/100 = 2MB
+        assert_eq!(compute_window_for_stream_count(100), 2 * 1024 * 1024);
     }
 
     #[test]
-    fn test_adaptive_window_200_streams() {
-        // 32MB/200 = 167,772 bytes (~163KB), above 64KB floor
-        let w = compute_window_for_stream_count(200);
-        assert_eq!(w, (32 * 1024 * 1024u64 / 200) as u32);
-        assert!(w > 64 * 1024);
+    fn test_adaptive_window_200_streams_at_floor() {
+        // 200MB/200 = 1MB = exactly the floor
+        assert_eq!(compute_window_for_stream_count(200), 1 * 1024 * 1024);
     }
 
     #[test]
-    fn test_adaptive_window_500_streams() {
-        // 32MB/500 = 67,108 bytes (~65.5KB), just above 64KB floor
-        let w = compute_window_for_stream_count(500);
-        assert_eq!(w, (32 * 1024 * 1024u64 / 500) as u32);
-        assert!(w > 64 * 1024);
-    }
-
-    #[test]
-    fn test_adaptive_window_at_min_boundary() {
-        // 32MB/512 = 65,536 = exactly 64KB floor
-        assert_eq!(compute_window_for_stream_count(512), 64 * 1024);
-    }
-
-    #[test]
-    fn test_adaptive_window_below_min_clamped() {
-        // 32MB/513 = 65,408 → clamped up to 64KB
-        assert_eq!(compute_window_for_stream_count(513), 64 * 1024);
-    }
-
-    #[test]
-    fn test_adaptive_window_1000_streams() {
-        // 32MB/1000 = 33,554 → clamped to 64KB
-        assert_eq!(compute_window_for_stream_count(1000), 64 * 1024);
+    fn test_adaptive_window_500_streams_clamped() {
+        // 200MB/500 = 0.4MB -> clamped up to 1MB floor
+        assert_eq!(compute_window_for_stream_count(500), 1 * 1024 * 1024);
     }
 
     #[test]
     fn test_adaptive_window_max_u32() {
-        // Extreme: u32::MAX streams → tiny value → clamped to 64KB
-        assert_eq!(compute_window_for_stream_count(u32::MAX), 64 * 1024);
+        // Extreme: u32::MAX streams -> tiny value -> clamped to 1MB
+        assert_eq!(compute_window_for_stream_count(u32::MAX), 1 * 1024 * 1024);
     }
 
     #[test]
     fn test_adaptive_window_monotonically_decreasing() {
-        // Window should decrease (or stay same) as stream count increases
         let mut prev = compute_window_for_stream_count(1);
-        for n in [2, 5, 10, 50, 100, 200, 500, 512, 1000] {
+        for n in [2, 10, 50, 51, 100, 200, 500, 1000] {
             let w = compute_window_for_stream_count(n);
             assert!(w <= prev, "window increased from {} to {} at n={}", prev, w, n);
             prev = w;
@@ -788,47 +837,14 @@ mod tests {
 
     #[test]
     fn test_adaptive_window_total_budget_bounded() {
-        // active × per_stream_window should never exceed 32MB (+ clamp overhead for high N)
-        for n in [1, 10, 50, 100, 200, 500] {
+        // active x per_stream_window should never exceed 200MB (+ clamp overhead for high N)
+        for n in [1, 10, 50, 100, 200] {
             let w = compute_window_for_stream_count(n);
             let total = w as u64 * n as u64;
-            assert!(total <= 32 * 1024 * 1024, "total {}MB exceeds budget at n={}", total / (1024*1024), n);
+            assert!(total <= 200 * 1024 * 1024, "total {}MB exceeds budget at n={}", total / (1024*1024), n);
         }
     }
 
-    // --- 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.flowcontrol.node.ts

@@ -324,7 +324,7 @@ tap.test('setup: start echo server and tunnel', async () => {
   expect(tunnel.hub.running).toBeTrue();
 });
 
-tap.test('single stream: 32MB transfer exceeding initial 4MB window', async () => {
+tap.test('single stream: 32MB transfer exceeding initial 4MB window (multiple refills)', async () => {
   const size = 32 * 1024 * 1024;
   const data = crypto.randomBytes(size);
   const expectedHash = sha256(data);
@@ -392,7 +392,7 @@ tap.test('asymmetric transfer: 4KB request -> 4MB response', async () => {
   }
 });
 
-tap.test('100 streams x 1MB each (100MB total exceeding 32MB budget)', async () => {
+tap.test('100 streams x 1MB each (100MB total exceeding 200MB budget)', async () => {
   const streamCount = 100;
   const payloadSize = 1 * 1024 * 1024;
 
@@ -446,7 +446,7 @@ tap.test('active stream counter tracks concurrent connections', async () => {
 });
 
 tap.test('50 streams x 2MB each (forces multiple window refills per stream)', async () => {
-  // At 50 concurrent streams: adaptive window = 32MB/50 = 655KB per stream
+  // At 50 concurrent streams: adaptive window = 200MB/50 = 4MB per stream
   // Each stream sends 2MB → needs ~3 WINDOW_UPDATE refill cycles per stream
   const streamCount = 50;
   const payloadSize = 2 * 1024 * 1024;

ts/00_commitinfo_data.ts

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