From c490e35a8fc8b44143a11e754477ff21d9546ce0 Mon Sep 17 00:00:00 2001 From: Juergen Kunz Date: Sun, 15 Mar 2026 21:06:44 +0000 Subject: [PATCH] fix(remoteingress-core): preserve stream close ordering and add flow-control stall timeouts --- changelog.md | 7 ++++++ rust/crates/remoteingress-core/src/edge.rs | 10 ++++++--- rust/crates/remoteingress-core/src/hub.rs | 26 ++++++++++++++++------ ts/00_commitinfo_data.ts | 2 +- 4 files changed, 34 insertions(+), 11 deletions(-) diff --git a/changelog.md b/changelog.md index 3fb27f1..d7d0249 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,12 @@ # Changelog +## 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 diff --git a/rust/crates/remoteingress-core/src/edge.rs b/rust/crates/remoteingress-core/src/edge.rs index 652d50a..a7453f6 100644 --- a/rust/crates/remoteingress-core/src/edge.rs +++ b/rust/crates/remoteingress-core/src/edge.rs @@ -703,13 +703,17 @@ async fn handle_client_connection( // Task: client -> hub (upload direction) with per-stream flow control let mut buf = vec![0u8; 32768]; loop { - // Wait for send window to have capacity + // 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; } @@ -737,10 +741,10 @@ async fn handle_client_connection( } } - // Send CLOSE frame via control channel (only if not cancelled) + // Send CLOSE frame via DATA channel (must arrive AFTER last DATA for this stream) if !client_token.is_cancelled() { let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]); - let _ = tunnel_ctrl_tx.try_send(close_frame); + let _ = tunnel_data_tx.try_send(close_frame); } // Cleanup diff --git a/rust/crates/remoteingress-core/src/hub.rs b/rust/crates/remoteingress-core/src/hub.rs index b652fd9..6941c95 100644 --- a/rust/crates/remoteingress-core/src/hub.rs +++ b/rust/crates/remoteingress-core/src/hub.rs @@ -537,8 +537,16 @@ async fn handle_edge_connection( match data { Some(data) => { let len = data.len() as u32; - if up_write.write_all(&data).await.is_err() { - break; + match tokio::time::timeout( + Duration::from_secs(60), + up_write.write_all(&data), + ).await { + 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; @@ -568,13 +576,17 @@ async fn handle_edge_connection( // with per-stream flow control (check send_window before reading) let mut buf = vec![0u8; 32768]; loop { - // Wait for send window to have capacity + // 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; } @@ -603,10 +615,10 @@ 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) 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.try_send(close_frame); } writer_for_edge_data.abort(); @@ -616,10 +628,10 @@ 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) 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.try_send(close_frame); } } diff --git a/ts/00_commitinfo_data.ts b/ts/00_commitinfo_data.ts index 7ee5020..98327bb 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.5.3', + version: '4.5.4', 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.' }