serve.zone/remoteingress
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(remoteingress-core): preserve stream close ordering and add flow-control stall timeouts

Browse Source
This commit is contained in:
Jürgen Kunz
2026-03-15 21:06:44 +00:00
parent 579e553da0
commit c490e35a8f
4 changed files with 34 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
rust/crates/remoteingress-core/src/edge.rs
View File

@@ -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

26
rust/crates/remoteingress-core/src/hub.rs
View File

@@ -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);
}
}