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

fix(remoteingress-core): prevent stream stalls by guaranteeing flow-control updates and avoiding bounded per-stream channel overflows

Browse Source
This commit is contained in:
Jürgen Kunz
2026-03-17 11:47:33 +00:00
parent d57d6395dd
commit e8431c0174
4 changed files with 51 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,10 +14,6 @@ use remoteingress_protocol::*;
type HubTlsStream = tokio_rustls::server::TlsStream<TcpStream>;
/// Per-stream data channel capacity. With 4MB window and 32KB frames,
/// at most ~128 frames are in-flight. 256 provides comfortable headroom.
const PER_STREAM_DATA_CAPACITY: usize = 256;
/// Result of processing a frame.
#[allow(dead_code)]
enum FrameAction {
@@ -27,8 +23,10 @@ enum FrameAction {
/// Per-stream state tracked in the hub's stream map.
struct HubStreamState {
/// Channel to deliver FRAME_DATA payloads to the upstream writer task.
data_tx: mpsc::Sender<Vec<u8>>,
/// Unbounded channel to deliver FRAME_DATA payloads to the upstream writer task.
/// Unbounded because flow control (WINDOW_UPDATE) already limits bytes-in-flight.
/// A bounded channel would kill streams instead of applying backpressure.
data_tx: mpsc::UnboundedSender<Vec<u8>>,
/// Cancellation token for this stream.
cancel_token: CancellationToken,
/// Send window for FRAME_DATA_BACK (download direction).
@@ -348,7 +346,7 @@ async fn handle_hub_frame(
});
// Create channel for data from edge to this stream
let (stream_data_tx, mut stream_data_rx) = mpsc::channel::<Vec<u8>>(PER_STREAM_DATA_CAPACITY);
let (stream_data_tx, mut stream_data_rx) = mpsc::unbounded_channel::<Vec<u8>>();
// 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(
@@ -426,10 +424,16 @@ async fn handle_hub_frame(
if consumed_since_update >= threshold {
let increment = consumed_since_update.min(adaptive_window);
let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, increment);
if wub_tx.try_send(frame).is_ok() {
consumed_since_update -= increment;
// Use send().await for guaranteed delivery — dropping WINDOW_UPDATEs
// causes permanent flow stalls. Safe: runs in per-stream task, not main loop.
tokio::select! {
result = wub_tx.send(frame) => {
if result.is_ok() {
consumed_since_update -= increment;
}
}
_ = writer_token.cancelled() => break,
}
// If try_send fails, keep accumulating — retry on next threshold
}
}
None => break,
@@ -441,7 +445,7 @@ 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.try_send(frame);
let _ = wub_tx.send(frame).await;
}
let _ = up_write.shutdown().await;
});
@@ -534,14 +538,13 @@ async fn handle_hub_frame(
});
}
FRAME_DATA => {
// Non-blocking dispatch to per-stream channel.
// With flow control, the sender should rarely exceed the channel capacity.
// Dispatch to per-stream unbounded channel. Flow control (WINDOW_UPDATE)
// limits bytes-in-flight, so the channel won't grow unbounded. send() only
// fails if the receiver is dropped (stream handler already exited).
if let Some(state) = streams.get(&frame.stream_id) {
if state.data_tx.try_send(frame.payload).is_err() {
log::warn!("Stream {} data channel full, closing stream", frame.stream_id);
if let Some(state) = streams.remove(&frame.stream_id) {
state.cancel_token.cancel();
}
if state.data_tx.send(frame.payload).is_err() {
// Receiver dropped — stream handler already exited, clean up
streams.remove(&frame.stream_id);
}
}
}