v4.6.0

changelog.md

@@ -1,5 +1,28 @@
 # Changelog
 
+## 2026-03-16 - 4.6.0 - feat(remoteingress-core)
+add adaptive per-stream flow control based on active stream counts
+
+- Track active stream counts on edge and hub connections to size per-stream flow control windows dynamically.
+- Cap WINDOW_UPDATE increments and read sizes to the adaptive window so bandwidth is shared more evenly across concurrent streams.
+- Apply the adaptive logic to both upload and download paths on edge and hub stream handlers.
+
+## 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
 

package.json

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

@@ -494,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");
-                                    break EdgeLoopResult::Reconnect;
+                                    // Control channel full (WINDOW_UPDATE burst from many streams).
+                                    // 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");
                             }
@@ -588,6 +590,15 @@ 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_ctrl_tx = tunnel_ctrl_tx.clone();
                                 let tunnel_data_tx = tunnel_data_tx.clone();
@@ -609,6 +620,7 @@ fn apply_port_config(
                                         tunnel_data_tx,
                                         client_writers,
                                         client_token,
+                                        Arc::clone(&active_streams),
                                     )
                                     .await;
                                     active_streams.fetch_sub(1, Ordering::Relaxed);
@@ -640,6 +652,7 @@ async fn handle_client_connection(
     tunnel_data_tx: mpsc::Sender<Vec<u8>>,
     client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
     client_token: CancellationToken,
+    active_streams: Arc<AtomicU32>,
 ) {
     let client_ip = client_addr.ip().to_string();
     let client_port = client_addr.port();
@@ -673,6 +686,7 @@ async fn handle_client_connection(
     // After writing to client TCP, send WINDOW_UPDATE to hub so it can send more
     let hub_to_client_token = client_token.clone();
     let wu_tx = tunnel_ctrl_tx.clone();
+    let active_streams_h2c = Arc::clone(&active_streams);
     let mut hub_to_client = tokio::spawn(async move {
         let mut consumed_since_update: u32 = 0;
         loop {
@@ -684,12 +698,20 @@ async fn handle_client_connection(
                             if client_write.write_all(&data).await.is_err() {
                                 break;
                             }
-                            // Track consumption for flow control
+                            // Track consumption for adaptive flow control.
+                            // The increment is capped to the adaptive window so the sender's
+                            // effective window shrinks to match current demand (fewer streams
+                            // = larger window, more streams = smaller window per stream).
                             consumed_since_update += len;
-                            if consumed_since_update >= WINDOW_UPDATE_THRESHOLD {
-                                let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
+                            let adaptive_window = remoteingress_protocol::compute_window_for_stream_count(
+                                active_streams_h2c.load(Ordering::Relaxed),
+                            );
+                            let threshold = adaptive_window / 2;
+                            if consumed_since_update >= threshold {
+                                let increment = consumed_since_update.min(adaptive_window);
+                                let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, increment);
                                 if wu_tx.try_send(frame).is_ok() {
-                                    consumed_since_update = 0;
+                                    consumed_since_update -= increment;
                                 }
                                 // If try_send fails, keep accumulating — retry on next threshold
                             }
@@ -726,9 +748,20 @@ async fn handle_client_connection(
         }
         if client_token.is_cancelled() { break; }
 
-        // Limit read size to available window
+        // 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;
-        let max_read = w.min(buf.len());
+        if w == 0 {
+            log::warn!("Stream {} upload: window still 0 after stall timeout, closing", stream_id);
+            break;
+        }
+        // Adaptive: cap read to current per-stream target window
+        let adaptive_cap = remoteingress_protocol::compute_window_for_stream_count(
+            active_streams.load(Ordering::Relaxed),
+        ) as usize;
+        let max_read = w.min(buf.len()).min(adaptive_cap);
 
         tokio::select! {
             read_result = client_read.read(&mut buf[..max_read]) => {

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

@@ -373,6 +373,9 @@ async fn handle_edge_connection(
         );
     }
 
+    // Per-edge active stream counter for adaptive flow control
+    let edge_stream_count = Arc::new(AtomicU32::new(0));
+
     // QoS dual-channel tunnel writer: control frames (PING/PONG/WINDOW_UPDATE/CLOSE)
     // have priority over data frames (DATA_BACK). This prevents PING starvation under load.
     let (ctrl_tx, mut ctrl_rx) = mpsc::channel::<Vec<u8>>(64);
@@ -509,8 +512,10 @@ async fn handle_edge_connection(
                                 }
 
                                 // Spawn task: connect to SmartProxy, send PROXY header, pipe data
+                                let stream_counter = Arc::clone(&edge_stream_count);
                                 tokio::spawn(async move {
                                     let _permit = permit; // hold semaphore permit until stream completes
+                                    stream_counter.fetch_add(1, Ordering::Relaxed);
 
                                     let result = async {
                                         // A2: Connect to SmartProxy with timeout
@@ -533,6 +538,7 @@ async fn handle_edge_connection(
                                         // After writing to upstream, send WINDOW_UPDATE_BACK to edge
                                         let writer_token = stream_token.clone();
                                         let wub_tx = writer_tx.clone();
+                                        let stream_counter_w = Arc::clone(&stream_counter);
                                         let writer_for_edge_data = tokio::spawn(async move {
                                             let mut consumed_since_update: u32 = 0;
                                             loop {
@@ -558,12 +564,18 @@ async fn handle_edge_connection(
                                                                         break;
                                                                     }
                                                                 }
-                                                                // Track consumption for flow control
+                                                                // Track consumption for adaptive flow control.
+                                                                // Increment capped to adaptive window to limit per-stream in-flight data.
                                                                 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);
+                                                                let adaptive_window = remoteingress_protocol::compute_window_for_stream_count(
+                                                                    stream_counter_w.load(Ordering::Relaxed),
+                                                                );
+                                                                let threshold = adaptive_window / 2;
+                                                                if consumed_since_update >= threshold {
+                                                                    let increment = consumed_since_update.min(adaptive_window);
+                                                                    let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, increment);
                                                                     if wub_tx.try_send(frame).is_ok() {
-                                                                        consumed_since_update = 0;
+                                                                        consumed_since_update -= increment;
                                                                     }
                                                                     // If try_send fails, keep accumulating — retry on next threshold
                                                                 }
@@ -601,9 +613,20 @@ async fn handle_edge_connection(
                                             }
                                             if stream_token.is_cancelled() { break; }
 
-                                            // Limit read size to available window
+                                            // 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;
-                                            let max_read = w.min(buf.len());
+                                            if w == 0 {
+                                                log::warn!("Stream {} download: window still 0 after stall timeout, closing", stream_id);
+                                                break;
+                                            }
+                                            // Adaptive: cap read to current per-stream target window
+                                            let adaptive_cap = remoteingress_protocol::compute_window_for_stream_count(
+                                                stream_counter.load(Ordering::Relaxed),
+                                            ) as usize;
+                                            let max_read = w.min(buf.len()).min(adaptive_cap);
 
                                             tokio::select! {
                                                 read_result = up_read.read(&mut buf[..max_read]) => {
@@ -658,6 +681,7 @@ async fn handle_edge_connection(
                                             stream_id,
                                         });
                                     }
+                                    stream_counter.fetch_sub(1, Ordering::Relaxed);
                                 });
                             }
                             FRAME_DATA => {
@@ -719,8 +743,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);
-                    break;
+                    // Control channel full — skip this PING cycle.
+                    // 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

@@ -32,6 +32,15 @@ pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> V
     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 {

ts/00_commitinfo_data.ts

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