v4.8.1

changelog.md

@@ -1,5 +1,32 @@
 # Changelog
 
+## 2026-03-17 - 4.8.1 - fix(remoteingress-core)
+remove tunnel writer timeouts from edge and hub buffered writes
+
+- Drops the 30 second timeout wrapper around writer.write_all and writer.flush in both edge and hub tunnel writers.
+- Updates error logging to report write failures without referring to stalled writes.
+
+## 2026-03-17 - 4.8.0 - feat(events)
+include disconnect reasons in edge and hub management events
+
+- Add reason fields to tunnelDisconnected and edgeDisconnected events emitted from the Rust core and binary bridge
+- Propagate specific disconnect causes such as EOF, liveness timeout, writer failure, handshake failure, and hub cancellation
+- Update TypeScript edge and hub classes to log and forward disconnect reason data
+- Extend serialization tests to cover the new reason fields
+
+## 2026-03-17 - 4.7.2 - fix(remoteingress-core)
+add tunnel write timeouts and scale initial stream windows by active stream count
+
+- Wrap tunnel frame writes and flushes in a 30-second timeout on both edge and hub to detect stalled writers and trigger faster reconnect or cleanup.
+- Compute each stream's initial send window from the current active stream count instead of using a fixed window to keep total in-flight data within the 32MB budget.
+
+## 2026-03-17 - 4.7.1 - fix(remoteingress-core)
+improve tunnel failure detection and reconnect handling
+
+- Enable TCP keepalive on edge and hub connections to detect silent network failures sooner
+- Trigger immediate reconnect or disconnect when tunnel writer tasks fail instead of waiting for liveness timeouts
+- Prevent active stream counter underflow during concurrent connection cleanup
+
 ## 2026-03-16 - 4.7.0 - feat(edge,protocol,test)
 add configurable edge bind address and expand flow-control test coverage
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "4.7.0",
+  "version": "4.8.1",
   "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-bin/src/main.rs

@@ -173,10 +173,10 @@ async fn handle_request(
                                     serde_json::json!({ "edgeId": edge_id, "peerAddr": peer_addr }),
                                 );
                             }
-                            HubEvent::EdgeDisconnected { edge_id } => {
+                            HubEvent::EdgeDisconnected { edge_id, reason } => {
                                 send_event(
                                     "edgeDisconnected",
-                                    serde_json::json!({ "edgeId": edge_id }),
+                                    serde_json::json!({ "edgeId": edge_id, "reason": reason }),
                                 );
                             }
                             HubEvent::StreamOpened {
@@ -295,8 +295,8 @@ async fn handle_request(
                             EdgeEvent::TunnelConnected => {
                                 send_event("tunnelConnected", serde_json::json!({}));
                             }
-                            EdgeEvent::TunnelDisconnected => {
-                                send_event("tunnelDisconnected", serde_json::json!({}));
+                            EdgeEvent::TunnelDisconnected { reason } => {
+                                send_event("tunnelDisconnected", serde_json::json!({ "reason": reason }));
                             }
                             EdgeEvent::PublicIpDiscovered { ip } => {
                                 send_event(

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

@@ -64,7 +64,8 @@ struct ConfigUpdate {
 #[serde(tag = "type")]
 pub enum EdgeEvent {
     TunnelConnected,
-    TunnelDisconnected,
+    #[serde(rename_all = "camelCase")]
+    TunnelDisconnected { reason: String },
     #[serde(rename_all = "camelCase")]
     PublicIpDiscovered { ip: String },
     #[serde(rename_all = "camelCase")]
@@ -236,10 +237,15 @@ async fn edge_main_loop(
         }
 
         *connected.write().await = false;
+        // Extract reason for disconnect event
+        let reason = match &result {
+            EdgeLoopResult::Reconnect(r) => r.clone(),
+            EdgeLoopResult::Shutdown => "shutdown".to_string(),
+        };
         // Only emit disconnect event on actual disconnection, not on failed reconnects.
         // Failed reconnects never reach line 335 (handshake success), so was_connected is false.
         if was_connected {
-            let _ = event_tx.try_send(EdgeEvent::TunnelDisconnected);
+            let _ = event_tx.try_send(EdgeEvent::TunnelDisconnected { reason: reason.clone() });
         }
         active_streams.store(0, Ordering::Relaxed);
         // Reset stream ID counter for next connection cycle
@@ -248,7 +254,7 @@ async fn edge_main_loop(
 
         match result {
             EdgeLoopResult::Shutdown => break,
-            EdgeLoopResult::Reconnect => {
+            EdgeLoopResult::Reconnect(_) => {
                 log::info!("Reconnecting in {}ms...", backoff_ms);
                 tokio::select! {
                     _ = tokio::time::sleep(Duration::from_millis(backoff_ms)) => {}
@@ -263,7 +269,7 @@ async fn edge_main_loop(
 
 enum EdgeLoopResult {
     Shutdown,
-    Reconnect,
+    Reconnect(String),  // reason for disconnection
 }
 
 async fn connect_to_hub_and_run(
@@ -284,11 +290,18 @@ async fn connect_to_hub_and_run(
         Ok(s) => {
             // Disable Nagle's algorithm for low-latency control frames (PING/PONG, WINDOW_UPDATE)
             let _ = s.set_nodelay(true);
+            // TCP keepalive detects silent network failures (NAT timeout, path change)
+            // faster than the 45s application-level liveness timeout.
+            let ka = socket2::TcpKeepalive::new()
+                .with_time(Duration::from_secs(30));
+            #[cfg(target_os = "linux")]
+            let ka = ka.with_interval(Duration::from_secs(10));
+            let _ = socket2::SockRef::from(&s).set_tcp_keepalive(&ka);
             s
         }
         Err(e) => {
             log::error!("Failed to connect to hub at {}: {}", addr, e);
-            return EdgeLoopResult::Reconnect;
+            return EdgeLoopResult::Reconnect(format!("tcp_connect_failed: {}", e));
         }
     };
 
@@ -299,7 +312,7 @@ async fn connect_to_hub_and_run(
         Ok(s) => s,
         Err(e) => {
             log::error!("TLS handshake failed: {}", e);
-            return EdgeLoopResult::Reconnect;
+            return EdgeLoopResult::Reconnect(format!("tls_handshake_failed: {}", e));
         }
     };
 
@@ -308,7 +321,7 @@ async fn connect_to_hub_and_run(
     // Send auth line
     let auth_line = format!("EDGE {} {}\n", config.edge_id, config.secret);
     if write_half.write_all(auth_line.as_bytes()).await.is_err() {
-        return EdgeLoopResult::Reconnect;
+        return EdgeLoopResult::Reconnect("auth_write_failed".to_string());
     }
 
     // Read handshake response line from hub (JSON with initial config)
@@ -317,12 +330,12 @@ async fn connect_to_hub_and_run(
     match buf_reader.read_line(&mut handshake_line).await {
         Ok(0) => {
             log::error!("Hub rejected connection (EOF before handshake)");
-            return EdgeLoopResult::Reconnect;
+            return EdgeLoopResult::Reconnect("hub_rejected_eof".to_string());
         }
         Ok(_) => {}
         Err(e) => {
             log::error!("Failed to read handshake response: {}", e);
-            return EdgeLoopResult::Reconnect;
+            return EdgeLoopResult::Reconnect(format!("handshake_read_failed: {}", e));
         }
     }
 
@@ -330,7 +343,7 @@ async fn connect_to_hub_and_run(
         Ok(h) => h,
         Err(e) => {
             log::error!("Invalid handshake response: {}", e);
-            return EdgeLoopResult::Reconnect;
+            return EdgeLoopResult::Reconnect(format!("handshake_invalid: {}", e));
         }
     };
 
@@ -388,18 +401,22 @@ async fn connect_to_hub_and_run(
     // Legacy alias — control channel for PONG, CLOSE, WINDOW_UPDATE, OPEN
     let tunnel_writer_tx = tunnel_ctrl_tx.clone();
     let tw_token = connection_token.clone();
+    // Oneshot to signal the reader loop when the writer dies from a write error.
+    // This avoids the 45s liveness timeout delay when the tunnel is already dead.
+    let (writer_dead_tx, mut writer_dead_rx) = tokio::sync::oneshot::channel::<()>();
     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);
+        let mut write_error = false;
         loop {
             tokio::select! {
                 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; }
+                            if writer.write_all(&frame_data).await.is_err() { write_error = true; break; }
+                            if writer.flush().await.is_err() { write_error = true; break; }
                         }
                         None => break,
                     }
@@ -407,8 +424,8 @@ async fn connect_to_hub_and_run(
                 data = tunnel_data_rx.recv() => {
                     match data {
                         Some(frame_data) => {
-                            if writer.write_all(&frame_data).await.is_err() { break; }
-                            if writer.flush().await.is_err() { break; }
+                            if writer.write_all(&frame_data).await.is_err() { write_error = true; break; }
+                            if writer.flush().await.is_err() { write_error = true; break; }
                         }
                         None => break,
                     }
@@ -416,6 +433,10 @@ async fn connect_to_hub_and_run(
                 _ = tw_token.cancelled() => break,
             }
         }
+        if write_error {
+            log::error!("Tunnel writer failed, signalling reader for fast reconnect");
+            let _ = writer_dead_tx.send(());
+        }
     });
 
     // Start TCP listeners for initial ports (hot-reloadable)
@@ -519,18 +540,22 @@ async fn connect_to_hub_and_run(
                     }
                     Ok(None) => {
                         log::info!("Hub disconnected (EOF)");
-                        break EdgeLoopResult::Reconnect;
+                        break EdgeLoopResult::Reconnect("hub_eof".to_string());
                     }
                     Err(e) => {
                         log::error!("Hub frame error: {}", e);
-                        break EdgeLoopResult::Reconnect;
+                        break EdgeLoopResult::Reconnect(format!("hub_frame_error: {}", e));
                     }
                 }
             }
             _ = &mut liveness_deadline => {
                 log::warn!("Hub liveness timeout (no frames for {}s), reconnecting",
                     liveness_timeout_dur.as_secs());
-                break EdgeLoopResult::Reconnect;
+                break EdgeLoopResult::Reconnect("liveness_timeout".to_string());
+            }
+            _ = &mut writer_dead_rx => {
+                log::error!("Tunnel writer died, reconnecting immediately");
+                break EdgeLoopResult::Reconnect("writer_dead".to_string());
             }
             _ = connection_token.cancelled() => {
                 log::info!("Connection cancelled");
@@ -636,7 +661,18 @@ fn apply_port_config(
                                         Arc::clone(&active_streams),
                                     )
                                     .await;
-                                    active_streams.fetch_sub(1, Ordering::Relaxed);
+                                    // Saturating decrement: prevent underflow when
+                                    // edge_main_loop's store(0) races with task cleanup.
+                                    loop {
+                                        let current = active_streams.load(Ordering::Relaxed);
+                                        if current == 0 { break; }
+                                        if active_streams.compare_exchange_weak(
+                                            current, current - 1,
+                                            Ordering::Relaxed, Ordering::Relaxed,
+                                        ).is_ok() {
+                                            break;
+                                        }
+                                    }
                                 });
                             }
                             Err(e) => {
@@ -682,7 +718,12 @@ async fn handle_client_connection(
 
     // 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));
+    // 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());
     {
         let mut writers = client_writers.lock().await;
@@ -921,9 +962,10 @@ mod tests {
 
     #[test]
     fn test_edge_event_tunnel_disconnected() {
-        let event = EdgeEvent::TunnelDisconnected;
+        let event = EdgeEvent::TunnelDisconnected { reason: "hub_eof".to_string() };
         let json = serde_json::to_value(&event).unwrap();
         assert_eq!(json["type"], "tunnelDisconnected");
+        assert_eq!(json["reason"], "hub_eof");
     }
 
     #[test]

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

@@ -92,7 +92,7 @@ pub enum HubEvent {
     #[serde(rename_all = "camelCase")]
     EdgeConnected { edge_id: String, peer_addr: String },
     #[serde(rename_all = "camelCase")]
-    EdgeDisconnected { edge_id: String },
+    EdgeDisconnected { edge_id: String, reason: String },
     #[serde(rename_all = "camelCase")]
     StreamOpened { edge_id: String, stream_id: u32 },
     #[serde(rename_all = "camelCase")]
@@ -300,6 +300,13 @@ async fn handle_edge_connection(
 ) -> 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)?;
+    // TCP keepalive detects silent network failures (NAT timeout, path change)
+    // faster than the 45s application-level liveness timeout.
+    let ka = socket2::TcpKeepalive::new()
+        .with_time(Duration::from_secs(30));
+    #[cfg(target_os = "linux")]
+    let ka = ka.with_interval(Duration::from_secs(10));
+    let _ = socket2::SockRef::from(&stream).set_tcp_keepalive(&ka);
     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);
@@ -383,18 +390,20 @@ async fn handle_edge_connection(
     // 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_dead_tx, mut writer_dead_rx) = tokio::sync::oneshot::channel::<()>();
     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);
+        let mut write_error = false;
         loop {
             tokio::select! {
                 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; }
+                            if writer.write_all(&frame_data).await.is_err() { write_error = true; break; }
+                            if writer.flush().await.is_err() { write_error = true; break; }
                         }
                         None => break,
                     }
@@ -402,8 +411,8 @@ async fn handle_edge_connection(
                 data = data_rx.recv() => {
                     match data {
                         Some(frame_data) => {
-                            if writer.write_all(&frame_data).await.is_err() { break; }
-                            if writer.flush().await.is_err() { break; }
+                            if writer.write_all(&frame_data).await.is_err() { write_error = true; break; }
+                            if writer.flush().await.is_err() { write_error = true; break; }
                         }
                         None => break,
                     }
@@ -411,6 +420,10 @@ async fn handle_edge_connection(
                 _ = writer_token.cancelled() => break,
             }
         }
+        if write_error {
+            log::error!("Tunnel writer to edge failed, signalling reader for fast cleanup");
+            let _ = writer_dead_tx.send(());
+        }
     });
 
     // Spawn task to forward config updates as FRAME_CONFIG frames
@@ -453,6 +466,7 @@ async fn handle_edge_connection(
 
     // Frame reading loop
     let mut frame_reader = FrameReader::new(buf_reader);
+    let mut disconnect_reason = "unknown".to_string();
 
     loop {
         tokio::select! {
@@ -499,7 +513,12 @@ async fn handle_edge_connection(
 
                                 // 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));
+                                // Adaptive initial window: scale with current stream count
+                                // to keep total in-flight data within the 32MB budget.
+                                let initial_window = compute_window_for_stream_count(
+                                    edge_stream_count.load(Ordering::Relaxed),
+                                );
+                                let send_window = Arc::new(AtomicU32::new(initial_window));
                                 let window_notify = Arc::new(Notify::new());
                                 {
                                     let mut s = streams.lock().await;
@@ -732,10 +751,12 @@ async fn handle_edge_connection(
                     }
                     Ok(None) => {
                         log::info!("Edge {} disconnected (EOF)", edge_id);
+                        disconnect_reason = "edge_eof".to_string();
                         break;
                     }
                     Err(e) => {
                         log::error!("Edge {} frame error: {}", edge_id, e);
+                        disconnect_reason = format!("edge_frame_error: {}", e);
                         break;
                     }
                 }
@@ -752,10 +773,17 @@ async fn handle_edge_connection(
             _ = &mut liveness_deadline => {
                 log::warn!("Edge {} liveness timeout (no frames for {}s), disconnecting",
                     edge_id, liveness_timeout_dur.as_secs());
+                disconnect_reason = "liveness_timeout".to_string();
+                break;
+            }
+            _ = &mut writer_dead_rx => {
+                log::error!("Tunnel writer to edge {} died, disconnecting immediately", edge_id);
+                disconnect_reason = "writer_dead".to_string();
                 break;
             }
             _ = edge_token.cancelled() => {
                 log::info!("Edge {} cancelled by hub", edge_id);
+                disconnect_reason = "cancelled_by_hub".to_string();
                 break;
             }
         }
@@ -771,6 +799,7 @@ async fn handle_edge_connection(
     }
     let _ = event_tx.try_send(HubEvent::EdgeDisconnected {
         edge_id: edge_id.clone(),
+        reason: disconnect_reason,
     });
 
     Ok(())
@@ -993,10 +1022,12 @@ mod tests {
     fn test_hub_event_edge_disconnected_serialize() {
         let event = HubEvent::EdgeDisconnected {
             edge_id: "edge-2".to_string(),
+            reason: "liveness_timeout".to_string(),
         };
         let json = serde_json::to_value(&event).unwrap();
         assert_eq!(json["type"], "edgeDisconnected");
         assert_eq!(json["edgeId"], "edge-2");
+        assert_eq!(json["reason"], "liveness_timeout");
     }
 
     #[test]

ts/00_commitinfo_data.ts

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

ts/classes.remoteingressedge.ts

@@ -83,8 +83,10 @@ export class RemoteIngressEdge extends EventEmitter {
     this.bridge.on('management:tunnelConnected', () => {
       this.emit('tunnelConnected');
     });
-    this.bridge.on('management:tunnelDisconnected', () => {
-      this.emit('tunnelDisconnected');
+    this.bridge.on('management:tunnelDisconnected', (data: { reason?: string }) => {
+      const reason = data?.reason ?? 'unknown';
+      console.log(`[RemoteIngressEdge] Tunnel disconnected: ${reason}`);
+      this.emit('tunnelDisconnected', data);
     });
     this.bridge.on('management:publicIpDiscovered', (data: { ip: string }) => {
       this.emit('publicIpDiscovered', data);

ts/classes.remoteingresshub.ts

@@ -93,7 +93,9 @@ export class RemoteIngressHub extends EventEmitter {
     this.bridge.on('management:edgeConnected', (data: { edgeId: string; peerAddr: string }) => {
       this.emit('edgeConnected', data);
     });
-    this.bridge.on('management:edgeDisconnected', (data: { edgeId: string }) => {
+    this.bridge.on('management:edgeDisconnected', (data: { edgeId: string; reason?: string }) => {
+      const reason = data?.reason ?? 'unknown';
+      console.log(`[RemoteIngressHub] Edge ${data.edgeId} disconnected: ${reason}`);
       this.emit('edgeDisconnected', data);
     });
     this.bridge.on('management:streamOpened', (data: { edgeId: string; streamId: number }) => {