v4.5.12

changelog.md

@@ -1,5 +1,93 @@
 # Changelog
 
+## 2026-03-16 - 4.5.12 - fix(remoteingress-core)
+improve tunnel liveness handling and enable TCP keepalive for accepted client sockets
+
+- Avoid disconnecting edges when PING or PONG frames cannot be queued because the control channel is temporarily full.
+- Enable TCP_NODELAY and TCP keepalive on accepted client connections to help detect stale or dropped clients.
+
+## 2026-03-16 - 4.5.11 - fix(repo)
+no changes to commit
+
+
+## 2026-03-16 - 4.5.10 - fix(remoteingress-core)
+guard zero-window reads to avoid false EOF handling on stalled streams
+
+- Prevent upload and download loops from calling read on an empty buffer when flow-control window remains at 0 after stall timeout
+- Log a warning and close the affected stream instead of misinterpreting Ok(0) as end-of-file
+
+## 2026-03-16 - 4.5.9 - fix(remoteingress-core)
+delay stream close until downstream response draining finishes to prevent truncated transfers
+
+- Waits for the hub-to-client download task to finish before sending the stream CLOSE frame
+- Prevents upstream reads from being cancelled mid-response during asymmetric transfers such as git fetch
+- Retains the existing timeout so stalled downloads still clean up safely
+
+## 2026-03-16 - 4.5.8 - fix(remoteingress-core)
+ensure upstream writes cancel promptly and reliably deliver CLOSE_BACK frames
+
+- listen for stream cancellation while waiting on upstream write timeouts so FRAME_CLOSE does not block for up to 60 seconds
+- replace try_send with send().await when emitting CLOSE_BACK frames to avoid silently dropping close notifications when the data channel is full
+
+## 2026-03-16 - 4.5.7 - fix(remoteingress-core)
+improve tunnel reconnect and frame write efficiency
+
+- Reuse the TLS connector across edge reconnections to preserve session resumption state and reduce reconnect latency.
+- Buffer hub and edge frame writes to coalesce small control and data frames into fewer TLS records and syscalls while still flushing each frame promptly.
+
+## 2026-03-16 - 4.5.6 - fix(remoteingress-core)
+disable Nagle's algorithm on edge, hub, and upstream TCP sockets to reduce control-frame latency
+
+- Enable TCP_NODELAY on the edge connection to the hub for faster PING/PONG and WINDOW_UPDATE delivery
+- Apply TCP_NODELAY on accepted hub streams before TLS handling
+- Enable TCP_NODELAY on SmartProxy upstream connections before sending the PROXY header
+
+## 2026-03-16 - 4.5.5 - fix(remoteingress-core)
+wait for hub-to-client draining before cleanup and reliably send close frames
+
+- switch CLOSE frame delivery on the data channel from try_send to send().await to avoid dropping it when the channel is full
+- delay stream cleanup until the hub-to-client task finishes or times out so large downstream responses continue after upload EOF
+- add a bounded 5-minute wait for download draining to prevent premature termination of asymmetric transfers such as git fetch
+
+## 2026-03-15 - 4.5.4 - fix(remoteingress-core)
+preserve stream close ordering and add flow-control stall timeouts
+
+- Send CLOSE and CLOSE_BACK frames on the data channel so they arrive after the final stream data frames.
+- Log and abort stalled upload and download paths when flow-control windows stay empty for 120 seconds.
+- Apply a 60-second timeout when writing buffered stream data to the upstream connection to prevent hung streams.
+
+## 2026-03-15 - 4.5.3 - fix(remoteingress-core)
+prioritize control frames over data in edge and hub tunnel writers
+
+- Split tunnel/frame writers into separate control and data channels in edge and hub
+- Use biased select loops so PING, PONG, WINDOW_UPDATE, OPEN, and CLOSE frames are sent before data frames
+- Route stream data through dedicated data channels while keeping OPEN, CLOSE, and flow-control updates on control channels to prevent keepalive starvation under load
+
+## 2026-03-15 - 4.5.2 - fix(remoteingress-core)
+improve stream flow control retries and increase channel buffer capacity
+
+- increase per-stream mpsc channel capacity from 128 to 256 on both edge and hub paths
+- only reset accumulated window update bytes after a successful try_send to avoid dropping flow-control credits when the update channel is busy
+
+## 2026-03-15 - 4.5.1 - fix(protocol)
+increase per-stream flow control window and channel buffers to improve high-RTT throughput
+
+- raise the initial stream window from 256 KB to 4 MB to allow more in-flight data per stream
+- increase edge and hub mpsc channel capacities from 16 to 128 to better absorb throughput under flow control
+
+## 2026-03-15 - 4.5.0 - feat(remoteingress-core)
+add per-stream flow control for edge and hub tunnel data transfer
+
+- introduce WINDOW_UPDATE frame types and protocol helpers for per-stream flow control
+- track per-stream send windows on both edge and hub to limit reads based on available capacity
+- send window updates after downstream writes to reduce channel pressure during large transfers
+
+## 2026-03-15 - 4.4.1 - fix(remoteingress-core)
+prevent stream data loss by applying backpressure and closing saturated channels
+
+- replace non-blocking frame writes with awaited sends in per-stream tasks so large transfers respect backpressure instead of dropping data
+- close and remove streams when back-channel or data channels fill up to avoid TCP stream corruption from silently dropped frames
+
 ## 2026-03-03 - 4.4.0 - feat(remoteingress)
 add heartbeat PING/PONG and liveness timeouts; implement fast-reconnect/backoff reset and JS crash-recovery auto-restart
 

package.json

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

@@ -558,6 +558,7 @@ dependencies = [
  "rustls-pemfile",
  "serde",
  "serde_json",
+ "socket2 0.5.10",
  "tokio",
  "tokio-rustls",
  "tokio-util",
@@ -701,6 +702,16 @@ version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
 
+[[package]]
+name = "socket2"
+version = "0.5.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678"
+dependencies = [
+ "libc",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "socket2"
 version = "0.6.2"
@@ -765,7 +776,7 @@ dependencies = [
  "parking_lot",
  "pin-project-lite",
  "signal-hook-registry",
- "socket2",
+ "socket2 0.6.2",
  "tokio-macros",
  "windows-sys 0.61.2",
 ]

rust/crates/remoteingress-core/Cargo.toml

@@ -14,3 +14,4 @@ serde_json = "1"
 log = "0.4"
 rustls-pemfile = "2"
 tokio-util = "0.7"
+socket2 = "0.5"

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

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpListener, TcpStream};
-use tokio::sync::{mpsc, Mutex, RwLock};
+use tokio::sync::{mpsc, Mutex, Notify, RwLock};
 use tokio::task::JoinHandle;
 use tokio::time::{Instant, sleep_until};
 use tokio_rustls::TlsConnector;
@@ -13,6 +13,17 @@ use serde::{Deserialize, Serialize};
 
 use remoteingress_protocol::*;
 
+/// Per-stream state tracked in the edge's client_writers map.
+struct EdgeStreamState {
+    /// Channel to deliver FRAME_DATA_BACK payloads to the hub_to_client task.
+    back_tx: mpsc::Sender<Vec<u8>>,
+    /// Send window for FRAME_DATA (upload direction).
+    /// Decremented by the client reader, incremented by FRAME_WINDOW_UPDATE_BACK from hub.
+    send_window: Arc<AtomicU32>,
+    /// Notifier to wake the client reader when the window opens.
+    window_notify: Arc<Notify>,
+}
+
 /// Edge configuration (hub-host + credentials only; ports come from hub).
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -183,6 +194,14 @@ async fn edge_main_loop(
     let mut backoff_ms: u64 = 1000;
     let max_backoff_ms: u64 = 30000;
 
+    // Build TLS config ONCE outside the reconnect loop — preserves session
+    // cache across reconnections for TLS session resumption (saves 1 RTT).
+    let tls_config = rustls::ClientConfig::builder()
+        .dangerous()
+        .with_custom_certificate_verifier(Arc::new(NoCertVerifier))
+        .with_no_client_auth();
+    let connector = TlsConnector::from(Arc::new(tls_config));
+
     loop {
         // Create a per-connection child token
         let connection_token = cancel_token.child_token();
@@ -198,6 +217,7 @@ async fn edge_main_loop(
             &listen_ports,
             &mut shutdown_rx,
             &connection_token,
+            &connector,
         )
         .await;
 
@@ -248,18 +268,16 @@ async fn connect_to_hub_and_run(
     listen_ports: &Arc<RwLock<Vec<u16>>>,
     shutdown_rx: &mut mpsc::Receiver<()>,
     connection_token: &CancellationToken,
+    connector: &TlsConnector,
 ) -> EdgeLoopResult {
-    // Build TLS connector that skips cert verification (auth is via secret)
-    let tls_config = rustls::ClientConfig::builder()
-        .dangerous()
-        .with_custom_certificate_verifier(Arc::new(NoCertVerifier))
-        .with_no_client_auth();
-
-    let connector = TlsConnector::from(Arc::new(tls_config));
 
     let addr = format!("{}:{}", config.hub_host, config.hub_port);
     let tcp = match TcpStream::connect(&addr).await {
-        Ok(s) => s,
+        Ok(s) => {
+            // Disable Nagle's algorithm for low-latency control frames (PING/PONG, WINDOW_UPDATE)
+            let _ = s.set_nodelay(true);
+            s
+        }
         Err(e) => {
             log::error!("Failed to connect to hub at {}: {}", addr, e);
             return EdgeLoopResult::Reconnect;
@@ -351,22 +369,38 @@ async fn connect_to_hub_and_run(
         }
     });
 
-    // Client socket map: stream_id -> sender for writing data back to client
+    // Client socket map: stream_id -> per-stream state (back channel + flow control)
-    let client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
+    let client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>> =
         Arc::new(Mutex::new(HashMap::new()));
 
-    // A5: Channel-based tunnel writer replaces Arc<Mutex<WriteHalf>>
+    // QoS dual-channel tunnel writer: control frames (PONG/WINDOW_UPDATE/CLOSE/OPEN)
-    let (tunnel_writer_tx, mut tunnel_writer_rx) = mpsc::channel::<Vec<u8>>(4096);
+    // have priority over data frames (DATA). Prevents PING starvation under load.
+    let (tunnel_ctrl_tx, mut tunnel_ctrl_rx) = mpsc::channel::<Vec<u8>>(64);
+    let (tunnel_data_tx, mut tunnel_data_rx) = mpsc::channel::<Vec<u8>>(4096);
+    // Legacy alias — control channel for PONG, CLOSE, WINDOW_UPDATE, OPEN
+    let tunnel_writer_tx = tunnel_ctrl_tx.clone();
     let tw_token = connection_token.clone();
     let tunnel_writer_handle = tokio::spawn(async move {
+        // BufWriter coalesces small writes (frame headers, control frames) into fewer
+        // TLS records and syscalls. Flushed after each frame to avoid holding data.
+        let mut writer = tokio::io::BufWriter::with_capacity(65536, write_half);
         loop {
             tokio::select! {
-                data = tunnel_writer_rx.recv() => {
+                biased; // control frames always take priority over data
+                ctrl = tunnel_ctrl_rx.recv() => {
+                    match ctrl {
+                        Some(frame_data) => {
+                            if writer.write_all(&frame_data).await.is_err() { break; }
+                            if writer.flush().await.is_err() { break; }
+                        }
+                        None => break,
+                    }
+                }
+                data = tunnel_data_rx.recv() => {
                     match data {
                         Some(frame_data) => {
-                            if write_half.write_all(&frame_data).await.is_err() {
+                            if writer.write_all(&frame_data).await.is_err() { break; }
-                                break;
+                            if writer.flush().await.is_err() { break; }
-                            }
                         }
                         None => break,
                     }
@@ -382,6 +416,7 @@ async fn connect_to_hub_and_run(
         &handshake.listen_ports,
         &mut port_listeners,
         &tunnel_writer_tx,
+        &tunnel_data_tx,
         &client_writers,
         active_streams,
         next_stream_id,
@@ -407,11 +442,28 @@ async fn connect_to_hub_and_run(
 
                         match frame.frame_type {
                             FRAME_DATA_BACK => {
-                                // A1: Non-blocking send to prevent head-of-line blocking
+                                // Non-blocking dispatch to per-stream channel.
-                                let writers = client_writers.lock().await;
+                                // With flow control, the sender should rarely exceed the channel capacity.
-                                if let Some(tx) = writers.get(&frame.stream_id) {
+                                let mut writers = client_writers.lock().await;
-                                    if tx.try_send(frame.payload).is_err() {
+                                if let Some(state) = writers.get(&frame.stream_id) {
-                                        log::warn!("Stream {} back-channel full, dropping frame", frame.stream_id);
+                                    if state.back_tx.try_send(frame.payload).is_err() {
+                                        log::warn!("Stream {} back-channel full, closing stream", frame.stream_id);
+                                        writers.remove(&frame.stream_id);
+                                    }
+                                }
+                            }
+                            FRAME_WINDOW_UPDATE_BACK => {
+                                // Hub consumed data — increase our send window for this stream (upload direction)
+                                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();
+                                        }
                                     }
                                 }
                             }
@@ -430,6 +482,7 @@ async fn connect_to_hub_and_run(
                                         &update.listen_ports,
                                         &mut port_listeners,
                                         &tunnel_writer_tx,
+                                        &tunnel_data_tx,
                                         &client_writers,
                                         active_streams,
                                         next_stream_id,
@@ -441,8 +494,10 @@ async fn connect_to_hub_and_run(
                             FRAME_PING => {
                                 let pong_frame = encode_frame(0, FRAME_PONG, &[]);
                                 if tunnel_writer_tx.try_send(pong_frame).is_err() {
-                                    log::warn!("Failed to send PONG, writer channel full/closed");
+                                    // Control channel full (WINDOW_UPDATE burst from many streams).
-                                    break EdgeLoopResult::Reconnect;
+                                    // DON'T disconnect — the 45s liveness timeout gives margin
+                                    // for the channel to drain and the next PONG to succeed.
+                                    log::warn!("PONG send failed, control channel full — skipping this cycle");
                                 }
                                 log::trace!("Received PING from hub, sent PONG");
                             }
@@ -491,8 +546,9 @@ async fn connect_to_hub_and_run(
 fn apply_port_config(
     new_ports: &[u16],
     port_listeners: &mut HashMap<u16, JoinHandle<()>>,
-    tunnel_writer_tx: &mpsc::Sender<Vec<u8>>,
+    tunnel_ctrl_tx: &mpsc::Sender<Vec<u8>>,
-    client_writers: &Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
+    tunnel_data_tx: &mpsc::Sender<Vec<u8>>,
+    client_writers: &Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
     active_streams: &Arc<AtomicU32>,
     next_stream_id: &Arc<AtomicU32>,
     edge_id: &str,
@@ -511,7 +567,8 @@ fn apply_port_config(
 
     // Add new ports
     for &port in new_set.difference(&old_set) {
-        let tunnel_writer_tx = tunnel_writer_tx.clone();
+        let tunnel_ctrl_tx = tunnel_ctrl_tx.clone();
+        let tunnel_data_tx = tunnel_data_tx.clone();
         let client_writers = client_writers.clone();
         let active_streams = active_streams.clone();
         let next_stream_id = next_stream_id.clone();
@@ -533,8 +590,18 @@ fn apply_port_config(
                     accept_result = listener.accept() => {
                         match accept_result {
                             Ok((client_stream, client_addr)) => {
+                                // TCP keepalive detects dead clients that disappear without FIN.
+                                // Without this, zombie streams accumulate and never get cleaned up.
+                                let _ = client_stream.set_nodelay(true);
+                                let ka = socket2::TcpKeepalive::new()
+                                    .with_time(Duration::from_secs(60));
+                                #[cfg(target_os = "linux")]
+                                let ka = ka.with_interval(Duration::from_secs(60));
+                                let _ = socket2::SockRef::from(&client_stream).set_tcp_keepalive(&ka);
+
                                 let stream_id = next_stream_id.fetch_add(1, Ordering::Relaxed);
-                                let tunnel_writer_tx = tunnel_writer_tx.clone();
+                                let tunnel_ctrl_tx = tunnel_ctrl_tx.clone();
+                                let tunnel_data_tx = tunnel_data_tx.clone();
                                 let client_writers = client_writers.clone();
                                 let active_streams = active_streams.clone();
                                 let edge_id = edge_id.clone();
@@ -549,7 +616,8 @@ fn apply_port_config(
                                         stream_id,
                                         port,
                                         &edge_id,
-                                        tunnel_writer_tx,
+                                        tunnel_ctrl_tx,
+                                        tunnel_data_tx,
                                         client_writers,
                                         client_token,
                                     )
@@ -579,8 +647,9 @@ async fn handle_client_connection(
     stream_id: u32,
     dest_port: u16,
     edge_id: &str,
-    tunnel_writer_tx: mpsc::Sender<Vec<u8>>,
+    tunnel_ctrl_tx: mpsc::Sender<Vec<u8>>,
-    client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
+    tunnel_data_tx: mpsc::Sender<Vec<u8>>,
+    client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
     client_token: CancellationToken,
 ) {
     let client_ip = client_addr.ip().to_string();
@@ -589,33 +658,52 @@ 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 writer channel
+    // 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_writer_tx.send(open_frame).await.is_err() {
+    if tunnel_ctrl_tx.send(open_frame).await.is_err() {
         return;
     }
 
-    // Set up channel for data coming back from hub
+    // 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>>(256);
+    let send_window = Arc::new(AtomicU32::new(INITIAL_STREAM_WINDOW));
+    let window_notify = Arc::new(Notify::new());
     {
         let mut writers = client_writers.lock().await;
-        writers.insert(stream_id, back_tx);
+        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
+    // Task: hub -> client (download direction)
+    // After writing to client TCP, send WINDOW_UPDATE to hub so it can send more
     let hub_to_client_token = client_token.clone();
-    let hub_to_client = tokio::spawn(async move {
+    let wu_tx = tunnel_ctrl_tx.clone();
+    let mut hub_to_client = tokio::spawn(async move {
+        let mut consumed_since_update: u32 = 0;
         loop {
             tokio::select! {
                 data = back_rx.recv() => {
                     match data {
                         Some(data) => {
+                            let len = data.len() as u32;
                             if client_write.write_all(&data).await.is_err() {
                                 break;
                             }
+                            // Track consumption for flow control
+                            consumed_since_update += len;
+                            if consumed_since_update >= WINDOW_UPDATE_THRESHOLD {
+                                let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
+                                if wu_tx.try_send(frame).is_ok() {
+                                    consumed_since_update = 0;
+                                }
+                                // If try_send fails, keep accumulating — retry on next threshold
+                            }
                         }
                         None => break,
                     }
@@ -623,21 +711,52 @@ async fn handle_client_connection(
                 _ = hub_to_client_token.cancelled() => break,
             }
         }
+        // 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);
+        }
         let _ = client_write.shutdown().await;
     });
 
-    // Task: client -> hub (via writer channel)
+    // Task: client -> hub (upload direction) with per-stream flow control
     let mut buf = vec![0u8; 32768];
     loop {
+        // Wait for send window to have capacity (with stall timeout)
+        loop {
+            let w = send_window.load(Ordering::Acquire);
+            if w > 0 { break; }
+            tokio::select! {
+                _ = window_notify.notified() => continue,
+                _ = client_token.cancelled() => break,
+                _ = tokio::time::sleep(Duration::from_secs(120)) => {
+                    log::warn!("Stream {} upload stalled (window empty for 120s)", stream_id);
+                    break;
+                }
+            }
+        }
+        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;
+        if w == 0 {
+            log::warn!("Stream {} upload: window still 0 after stall timeout, closing", stream_id);
+            break;
+        }
+        let max_read = w.min(buf.len());
+
         tokio::select! {
-            read_result = client_read.read(&mut buf) => {
+            read_result = client_read.read(&mut buf[..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]);
-                        // A5: Use try_send to avoid blocking if writer channel is full
+                        if tunnel_data_tx.send(data_frame).await.is_err() {
-                        if tunnel_writer_tx.try_send(data_frame).is_err() {
+                            log::warn!("Stream {} data channel closed, closing", stream_id);
-                            log::warn!("Stream {} tunnel writer full, closing", stream_id);
                             break;
                         }
                     }
@@ -648,18 +767,29 @@ async fn handle_client_connection(
         }
     }
 
-    // Send CLOSE frame (only if not cancelled)
+    // Wait for the download task (hub → client) to finish BEFORE sending CLOSE.
+    // Upload EOF (client done sending) does NOT mean the response is done.
+    // For asymmetric transfers like git fetch (small request, large response),
+    // the response is still streaming when the upload finishes.
+    // Sending CLOSE before the response finishes would cause the hub to cancel
+    // the upstream reader mid-response, truncating the data.
+    let _ = tokio::time::timeout(
+        Duration::from_secs(300),  // 5 min max wait for download to finish
+        &mut hub_to_client,
+    ).await;
+
+    // NOW send CLOSE — the response has been fully delivered (or timed out).
     if !client_token.is_cancelled() {
         let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
-        let _ = tunnel_writer_tx.try_send(close_frame);
+        let _ = tunnel_data_tx.send(close_frame).await;
     }
 
-    // Cleanup
+    // Clean up
     {
         let mut writers = client_writers.lock().await;
         writers.remove(&stream_id);
     }
-    hub_to_client.abort();
+    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

@@ -1,9 +1,10 @@
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::time::Duration;
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpListener, TcpStream};
-use tokio::sync::{mpsc, Mutex, RwLock, Semaphore};
+use tokio::sync::{mpsc, Mutex, Notify, RwLock, Semaphore};
 use tokio::time::{interval, sleep_until, Instant};
 use tokio_rustls::TlsAcceptor;
 use tokio_util::sync::CancellationToken;
@@ -11,6 +12,19 @@ use serde::{Deserialize, Serialize};
 
 use remoteingress_protocol::*;
 
+/// Per-stream state tracked in the hub's stream map.
+struct HubStreamState {
+    /// Channel to deliver FRAME_DATA payloads to the upstream writer task.
+    data_tx: mpsc::Sender<Vec<u8>>,
+    /// Cancellation token for this stream.
+    cancel_token: CancellationToken,
+    /// Send window for FRAME_DATA_BACK (download direction).
+    /// Decremented by the upstream reader, incremented by FRAME_WINDOW_UPDATE from edge.
+    send_window: Arc<AtomicU32>,
+    /// Notifier to wake the upstream reader when the window opens.
+    window_notify: Arc<Notify>,
+}
+
 /// Hub configuration.
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -109,7 +123,7 @@ pub struct TunnelHub {
 struct ConnectedEdgeInfo {
     connected_at: u64,
     peer_addr: String,
-    active_streams: Arc<Mutex<HashMap<u32, (mpsc::Sender<Vec<u8>>, CancellationToken)>>>,
+    active_streams: Arc<Mutex<HashMap<u32, HubStreamState>>>,
     config_tx: mpsc::Sender<EdgeConfigUpdate>,
     #[allow(dead_code)] // kept alive for Drop — cancels child tokens when edge is removed
     cancel_token: CancellationToken,
@@ -284,6 +298,8 @@ async fn handle_edge_connection(
     edge_token: CancellationToken,
     peer_addr: String,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    // Disable Nagle's algorithm for low-latency control frames (PING/PONG, WINDOW_UPDATE)
+    stream.set_nodelay(true)?;
     let tls_stream = acceptor.accept(stream).await?;
     let (read_half, mut write_half) = tokio::io::split(tls_stream);
     let mut buf_reader = BufReader::new(read_half);
@@ -333,7 +349,7 @@ async fn handle_edge_connection(
     write_half.write_all(handshake_json.as_bytes()).await?;
 
     // Track this edge
-    let streams: Arc<Mutex<HashMap<u32, (mpsc::Sender<Vec<u8>>, CancellationToken)>>> =
+    let streams: Arc<Mutex<HashMap<u32, HubStreamState>>> =
         Arc::new(Mutex::new(HashMap::new()));
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -357,19 +373,34 @@ async fn handle_edge_connection(
         );
     }
 
-    // A5: Channel-based writer replaces Arc<Mutex<WriteHalf>>
+    // QoS dual-channel tunnel writer: control frames (PING/PONG/WINDOW_UPDATE/CLOSE)
-    // All frame writes go through this channel → dedicated writer task serializes them
+    // have priority over data frames (DATA_BACK). This prevents PING starvation under load.
-    let (frame_writer_tx, mut frame_writer_rx) = mpsc::channel::<Vec<u8>>(4096);
+    let (ctrl_tx, mut ctrl_rx) = mpsc::channel::<Vec<u8>>(64);
+    let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(4096);
+    // Legacy alias for code that sends both control and data (will be migrated)
+    let frame_writer_tx = ctrl_tx.clone();
     let writer_token = edge_token.clone();
     let writer_handle = tokio::spawn(async move {
+        // BufWriter coalesces small writes (frame headers, control frames) into fewer
+        // TLS records and syscalls. Flushed after each frame to avoid holding data.
+        let mut writer = tokio::io::BufWriter::with_capacity(65536, write_half);
         loop {
             tokio::select! {
-                data = frame_writer_rx.recv() => {
+                biased; // control frames always take priority over data
+                ctrl = ctrl_rx.recv() => {
+                    match ctrl {
+                        Some(frame_data) => {
+                            if writer.write_all(&frame_data).await.is_err() { break; }
+                            if writer.flush().await.is_err() { break; }
+                        }
+                        None => break,
+                    }
+                }
+                data = data_rx.recv() => {
                     match data {
                         Some(frame_data) => {
-                            if write_half.write_all(&frame_data).await.is_err() {
+                            if writer.write_all(&frame_data).await.is_err() { break; }
-                                break;
+                            if writer.flush().await.is_err() { break; }
-                            }
                         }
                         None => break,
                     }
@@ -453,7 +484,8 @@ async fn handle_edge_connection(
                                 let edge_id_clone = edge_id.clone();
                                 let event_tx_clone = event_tx.clone();
                                 let streams_clone = streams.clone();
-                                let writer_tx = frame_writer_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 target = target_host.clone();
                                 let stream_token = edge_token.child_token();
 
@@ -462,11 +494,18 @@ async fn handle_edge_connection(
                                     stream_id,
                                 });
 
-                                // Create channel for data from edge to this stream
+                                // Create channel for data from edge to this stream (capacity 16 is sufficient with flow control)
                                 let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(256);
+                                let send_window = Arc::new(AtomicU32::new(INITIAL_STREAM_WINDOW));
+                                let window_notify = Arc::new(Notify::new());
                                 {
                                     let mut s = streams.lock().await;
-                                    s.insert(stream_id, (data_tx, stream_token.clone()));
+                                    s.insert(stream_id, HubStreamState {
+                                        data_tx,
+                                        cancel_token: stream_token.clone(),
+                                        send_window: Arc::clone(&send_window),
+                                        window_notify: Arc::clone(&window_notify),
+                                    });
                                 }
 
                                 // Spawn task: connect to SmartProxy, send PROXY header, pipe data
@@ -484,21 +523,49 @@ async fn handle_edge_connection(
                                             format!("connect to SmartProxy {}:{} timed out (10s)", target, dest_port).into()
                                         })??;
 
+                                        upstream.set_nodelay(true)?;
                                         upstream.write_all(proxy_header.as_bytes()).await?;
 
                                         let (mut up_read, mut up_write) =
                                             upstream.into_split();
 
                                         // Forward data from edge (via channel) to SmartProxy
+                                        // After writing to upstream, send WINDOW_UPDATE_BACK to edge
                                         let writer_token = stream_token.clone();
+                                        let wub_tx = writer_tx.clone();
                                         let writer_for_edge_data = tokio::spawn(async move {
+                                            let mut consumed_since_update: u32 = 0;
                                             loop {
                                                 tokio::select! {
                                                     data = data_rx.recv() => {
                                                         match data {
                                                             Some(data) => {
-                                                                if up_write.write_all(&data).await.is_err() {
+                                                                let len = data.len() as u32;
-                                                                    break;
+                                                                // Check cancellation alongside the write so we respond
+                                                                // promptly to FRAME_CLOSE instead of blocking up to 60s.
+                                                                let write_result = tokio::select! {
+                                                                    r = tokio::time::timeout(
+                                                                        Duration::from_secs(60),
+                                                                        up_write.write_all(&data),
+                                                                    ) => r,
+                                                                    _ = writer_token.cancelled() => break,
+                                                                };
+                                                                match write_result {
+                                                                    Ok(Ok(())) => {}
+                                                                    Ok(Err(_)) => break,
+                                                                    Err(_) => {
+                                                                        log::warn!("Stream {} write to upstream timed out (60s)", stream_id);
+                                                                        break;
+                                                                    }
+                                                                }
+                                                                // Track consumption for flow control
+                                                                consumed_since_update += len;
+                                                                if consumed_since_update >= WINDOW_UPDATE_THRESHOLD {
+                                                                    let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, consumed_since_update);
+                                                                    if wub_tx.try_send(frame).is_ok() {
+                                                                        consumed_since_update = 0;
+                                                                    }
+                                                                    // If try_send fails, keep accumulating — retry on next threshold
                                                                 }
                                                             }
                                                             None => break,
@@ -507,22 +574,54 @@ async fn handle_edge_connection(
                                                     _ = writer_token.cancelled() => break,
                                                 }
                                             }
+                                            // Send final window update for remaining consumed bytes
+                                            if consumed_since_update > 0 {
+                                                let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, consumed_since_update);
+                                                let _ = wub_tx.try_send(frame);
+                                            }
                                             let _ = up_write.shutdown().await;
                                         });
 
                                         // Forward data from SmartProxy back to edge via writer channel
+                                        // with per-stream flow control (check send_window before reading)
                                         let mut buf = vec![0u8; 32768];
                                         loop {
+                                            // Wait for send window to have capacity (with stall timeout)
+                                            loop {
+                                                let w = send_window.load(Ordering::Acquire);
+                                                if w > 0 { break; }
+                                                tokio::select! {
+                                                    _ = window_notify.notified() => continue,
+                                                    _ = stream_token.cancelled() => break,
+                                                    _ = tokio::time::sleep(Duration::from_secs(120)) => {
+                                                        log::warn!("Stream {} download stalled (window empty for 120s)", stream_id);
+                                                        break;
+                                                    }
+                                                }
+                                            }
+                                            if stream_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;
+                                            if w == 0 {
+                                                log::warn!("Stream {} download: window still 0 after stall timeout, closing", stream_id);
+                                                break;
+                                            }
+                                            let max_read = w.min(buf.len());
+
                                             tokio::select! {
-                                                read_result = up_read.read(&mut buf) => {
+                                                read_result = up_read.read(&mut buf[..max_read]) => {
                                                     match read_result {
                                                         Ok(0) => break,
                                                         Ok(n) => {
+                                                            send_window.fetch_sub(n as u32, Ordering::Release);
                                                             let frame =
                                                                 encode_frame(stream_id, FRAME_DATA_BACK, &buf[..n]);
-                                                            // A5: Use try_send to avoid blocking if writer channel is full
+                                                            if data_writer_tx.send(frame).await.is_err() {
-                                                            if writer_tx.try_send(frame).is_err() {
+                                                                log::warn!("Stream {} data channel closed, closing", stream_id);
-                                                                log::warn!("Stream {} writer channel full, closing", stream_id);
                                                                 break;
                                                             }
                                                         }
@@ -533,10 +632,11 @@ async fn handle_edge_connection(
                                             }
                                         }
 
-                                        // Send CLOSE_BACK to edge (only if not cancelled)
+                                        // Send CLOSE_BACK via DATA channel (must arrive AFTER last DATA_BACK).
+                                        // Use send().await to guarantee delivery (try_send silently drops if full).
                                         if !stream_token.is_cancelled() {
                                             let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
-                                            let _ = writer_tx.try_send(close_frame);
+                                            let _ = data_writer_tx.send(close_frame).await;
                                         }
 
                                         writer_for_edge_data.abort();
@@ -546,10 +646,11 @@ async fn handle_edge_connection(
 
                                     if let Err(e) = result {
                                         log::error!("Stream {} error: {}", stream_id, e);
-                                        // Send CLOSE_BACK on error (only if not cancelled)
+                                        // Send CLOSE_BACK via DATA channel on error (must arrive after any DATA_BACK).
+                                        // Use send().await to guarantee delivery.
                                         if !stream_token.is_cancelled() {
                                             let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
-                                            let _ = writer_tx.try_send(close_frame);
+                                            let _ = data_writer_tx.send(close_frame).await;
                                         }
                                     }
 
@@ -567,18 +668,37 @@ async fn handle_edge_connection(
                                 });
                             }
                             FRAME_DATA => {
-                                // A1: Non-blocking send to prevent head-of-line blocking
+                                // Non-blocking dispatch to per-stream channel.
-                                let s = streams.lock().await;
+                                // With flow control, the sender should rarely exceed the channel capacity.
-                                if let Some((tx, _)) = s.get(&frame.stream_id) {
+                                let mut s = streams.lock().await;
-                                    if tx.try_send(frame.payload).is_err() {
+                                if let Some(state) = s.get(&frame.stream_id) {
-                                        log::warn!("Stream {} data channel full, dropping frame", frame.stream_id);
+                                    if state.data_tx.try_send(frame.payload).is_err() {
+                                        log::warn!("Stream {} data channel full, closing stream", frame.stream_id);
+                                        if let Some(state) = s.remove(&frame.stream_id) {
+                                            state.cancel_token.cancel();
+                                        }
+                                    }
+                                }
+                            }
+                            FRAME_WINDOW_UPDATE => {
+                                // Edge consumed data — increase our send window for this stream
+                                if let Some(increment) = decode_window_update(&frame.payload) {
+                                    if increment > 0 {
+                                        let s = streams.lock().await;
+                                        if let Some(state) = s.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 => {
                                 let mut s = streams.lock().await;
-                                if let Some((_, token)) = s.remove(&frame.stream_id) {
+                                if let Some(state) = s.remove(&frame.stream_id) {
-                                    token.cancel();
+                                    state.cancel_token.cancel();
                                     let _ = event_tx.try_send(HubEvent::StreamClosed {
                                         edge_id: edge_id.clone(),
                                         stream_id: frame.stream_id,
@@ -606,8 +726,9 @@ async fn handle_edge_connection(
             _ = ping_ticker.tick() => {
                 let ping_frame = encode_frame(0, FRAME_PING, &[]);
                 if frame_writer_tx.try_send(ping_frame).is_err() {
-                    log::warn!("Failed to send PING to edge {}, writer channel full/closed", edge_id);
+                    // Control channel full — skip this PING cycle.
-                    break;
+                    // The 45s liveness timeout gives margin for the channel to drain.
+                    log::warn!("PING send to edge {} failed, control channel full — skipping", edge_id);
                 }
                 log::trace!("Sent PING to edge {}", edge_id);
             }

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

@@ -9,6 +9,8 @@ pub const FRAME_CLOSE_BACK: u8 = 0x05;
 pub const FRAME_CONFIG: u8 = 0x06;     // Hub -> Edge: configuration update
 pub const FRAME_PING: u8 = 0x07;       // Hub -> Edge: heartbeat probe
 pub const FRAME_PONG: u8 = 0x08;       // Edge -> Hub: heartbeat response
+pub const FRAME_WINDOW_UPDATE: u8 = 0x09;      // Edge -> Hub: per-stream flow control
+pub const FRAME_WINDOW_UPDATE_BACK: u8 = 0x0A;  // Hub -> Edge: per-stream flow control
 
 // Frame header size: 4 (stream_id) + 1 (type) + 4 (length) = 9 bytes
 pub const FRAME_HEADER_SIZE: usize = 9;
@@ -16,6 +18,37 @@ pub const FRAME_HEADER_SIZE: usize = 9;
 // Maximum payload size (16 MB)
 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.
+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;
+
+/// Encode a WINDOW_UPDATE frame for a specific stream.
+pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> Vec<u8> {
+    encode_frame(stream_id, frame_type, &increment.to_be_bytes())
+}
+
+/// 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.
+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
+}
+
+/// 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 {
+        return None;
+    }
+    Some(u32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]))
+}
+
 /// A single multiplexed frame.
 #[derive(Debug, Clone)]
 pub struct Frame {

ts/00_commitinfo_data.ts

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