diff --git a/changelog.md b/changelog.md index 4594a86..2fb4dc0 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,12 @@ # Changelog +## 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 diff --git a/rust/crates/remoteingress-core/src/edge.rs b/rust/crates/remoteingress-core/src/edge.rs index 3985b1e..ed06ffd 100644 --- a/rust/crates/remoteingress-core/src/edge.rs +++ b/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; }); @@ -890,9 +897,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 diff --git a/rust/crates/remoteingress-core/src/hub.rs b/rust/crates/remoteingress-core/src/hub.rs index d586dc1..5f18d14 100644 --- a/rust/crates/remoteingress-core/src/hub.rs +++ b/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; }); @@ -511,10 +514,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 +531,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 - let _ = cleanup.send(stream_id).await; + // Signal main loop to remove stream from the map. + // 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); }); } diff --git a/ts/00_commitinfo_data.ts b/ts/00_commitinfo_data.ts index 9f22104..e0f0b43 100644 --- a/ts/00_commitinfo_data.ts +++ b/ts/00_commitinfo_data.ts @@ -3,6 +3,6 @@ */ export const commitinfo = { name: '@serve.zone/remoteingress', - version: '4.8.9', + version: '4.8.10', 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.' }