v4.8.11

changelog.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## 2026-03-17 - 4.8.11 - fix(remoteingress-core)
+stop data frame send loops promptly when stream cancellation is triggered
+
+- Use cancellation-aware tokio::select! around data channel sends in both edge and hub stream forwarding paths
+- Prevent stalled or noisy shutdown behavior when stream or client cancellation happens while awaiting frame delivery
+
+## 2026-03-17 - 4.8.10 - fix(remoteingress-core)
+guard tunnel frame sends with cancellation to prevent async send deadlocks
+
+- Wrap OPEN, CLOSE, CLOSE_BACK, WINDOW_UPDATE, and cleanup channel sends in cancellation-aware tokio::select! blocks.
+- Avoid indefinite blocking when tunnel, stream, or writer tasks are cancelled while awaiting channel capacity.
+- Improve shutdown reliability for edge and hub stream handling under tunnel failure conditions.
+
+## 2026-03-17 - 4.8.9 - fix(repo)
+no changes to commit
+
+
 ## 2026-03-17 - 4.8.8 - fix(remoteingress-core)
 cancel stale edge connections when an edge reconnects
 

package.json

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

@@ -739,7 +739,11 @@ async fn handle_client_connection(
     // Send OPEN frame with PROXY v1 header via control channel
     let proxy_header = build_proxy_v1_header(&client_ip, edge_ip, client_port, dest_port);
     let open_frame = encode_frame(stream_id, FRAME_OPEN, proxy_header.as_bytes());
-    if tunnel_ctrl_tx.send(open_frame).await.is_err() {
+    let send_ok = tokio::select! {
+        result = tunnel_ctrl_tx.send(open_frame) => result.is_ok(),
+        _ = client_token.cancelled() => false,
+    };
+    if !send_ok {
         return;
     }
 
@@ -814,7 +818,10 @@ async fn handle_client_connection(
         // Send final window update for any remaining consumed bytes
         if consumed_since_update > 0 {
             let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
-            let _ = wu_tx.send(frame).await;
+            tokio::select! {
+                _ = wu_tx.send(frame) => {}
+                _ = hub_to_client_token.cancelled() => {}
+            }
         }
         let _ = client_write.shutdown().await;
     });
@@ -866,10 +873,11 @@ 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 = buf[..FRAME_HEADER_SIZE + n].to_vec();
-                        if tunnel_data_tx.send(data_frame).await.is_err() {
+                        let sent = tokio::select! {
-                            log::warn!("Stream {} data channel closed, closing", stream_id);
+                            result = tunnel_data_tx.send(data_frame) => result.is_ok(),
-                            break;
+                            _ = client_token.cancelled() => false,
-                        }
+                        };
+                        if !sent { break; }
                     }
                     Err(_) => break,
                 }
@@ -890,9 +898,13 @@ async fn handle_client_connection(
     ).await;
 
     // NOW send CLOSE — the response has been fully delivered (or timed out).
+    // select! with cancellation guard prevents indefinite blocking if tunnel dies.
     if !client_token.is_cancelled() {
         let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
-        let _ = tunnel_data_tx.send(close_frame).await;
+        tokio::select! {
+            _ = tunnel_data_tx.send(close_frame) => {}
+            _ = client_token.cancelled() => {}
+        }
     }
 
     // Clean up

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

@@ -445,7 +445,10 @@ async fn handle_hub_frame(
                         // 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.send(frame).await;
+                            tokio::select! {
+                                _ = wub_tx.send(frame) => {}
+                                _ = writer_token.cancelled() => {}
+                            }
                         }
                         let _ = up_write.shutdown().await;
                     });
@@ -498,10 +501,11 @@ 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 = buf[..FRAME_HEADER_SIZE + n].to_vec();
-                                        if data_writer_tx.send(frame).await.is_err() {
+                                        let sent = tokio::select! {
-                                            log::warn!("Stream {} data channel closed, closing", stream_id);
+                                            result = data_writer_tx.send(frame) => result.is_ok(),
-                                            break;
+                                            _ = stream_token.cancelled() => false,
-                                        }
+                                        };
+                                        if !sent { break; }
                                     }
                                     Err(_) => break,
                                 }
@@ -511,10 +515,13 @@ async fn handle_hub_frame(
                     }
 
                     // Send CLOSE_BACK via DATA channel (must arrive AFTER last DATA_BACK).
-                    // Use send().await to guarantee delivery (try_send silently drops if full).
+                    // 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 _ = data_writer_tx.send(close_frame).await;
+                        tokio::select! {
+                            _ = data_writer_tx.send(close_frame) => {}
+                            _ = stream_token.cancelled() => {}
+                        }
                     }
 
                     writer_for_edge_data.abort();
@@ -525,15 +532,21 @@ 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).
-                    // Use send().await to guarantee delivery.
                     if !stream_token.is_cancelled() {
                         let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
-                        let _ = data_writer_tx.send(close_frame).await;
+                        tokio::select! {
+                            _ = data_writer_tx.send(close_frame) => {}
+                            _ = stream_token.cancelled() => {}
+                        }
                     }
                 }
 
-                // Signal main loop to remove stream from the map
+                // Signal main loop to remove stream from the map.
-                let _ = cleanup.send(stream_id).await;
+                // Cancellation guard prevents indefinite blocking if cleanup channel is full.
+                tokio::select! {
+                    _ = cleanup.send(stream_id) => {}
+                    _ = stream_token.cancelled() => {}
+                }
                 stream_counter.fetch_sub(1, Ordering::Relaxed);
             });
         }

ts/00_commitinfo_data.ts

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