From f0a6d5f7cae6a50225d902c2a52cacccd9b209e8 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 12 Sep 2026 13:36:10 +0200 Subject: [PATCH] fix(peer): size Stream Install egress deadlines Give a provider ten setup minutes plus its catalog streamed bytes at 1 MiB/s instead of one fixed ten-minute absolute window. Large healthy games can complete while the sender still has a finite nonrenewable deadline through FIN. Test Plan: - just test - just clippy - large-provider timeout regression - git diff --check --- crates/lanspread-peer/src/stream_install.rs | 35 +++++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/crates/lanspread-peer/src/stream_install.rs b/crates/lanspread-peer/src/stream_install.rs index 7eeed4e..65d0e3d 100644 --- a/crates/lanspread-peer/src/stream_install.rs +++ b/crates/lanspread-peer/src/stream_install.rs @@ -71,7 +71,15 @@ const STREAM_INSTALL_MIN_BYTES_PER_SECOND: u64 = 1024 * 1024; const STREAM_INSTALL_TOTAL_COPY_ALLOWANCE: u64 = 2; fn stream_install_total_timeout(expected_file_bytes: u64) -> Duration { - let budgeted_bytes = expected_file_bytes.saturating_mul(STREAM_INSTALL_TOTAL_COPY_ALLOWANCE); + stream_install_sized_timeout(expected_file_bytes, STREAM_INSTALL_TOTAL_COPY_ALLOWANCE) +} + +fn stream_install_egress_timeout(expected_file_bytes: u64) -> Duration { + stream_install_sized_timeout(expected_file_bytes, 1) +} + +fn stream_install_sized_timeout(expected_file_bytes: u64, copy_allowance: u64) -> Duration { + let budgeted_bytes = expected_file_bytes.saturating_mul(copy_allowance); let transfer_seconds = budgeted_bytes.div_ceil(STREAM_INSTALL_MIN_BYTES_PER_SECOND); STREAM_INSTALL_TOTAL_BASE_TIMEOUT.saturating_add(Duration::from_secs(transfer_seconds)) } @@ -800,6 +808,16 @@ pub(crate) async fn send_game_install_stream( } }; let (frame_tx, mut frame_rx) = mpsc::channel(FRAME_CHANNEL_DEPTH); + let egress_timeout = match expected_streamed_file_bytes(&manifest) { + Ok(bytes) => stream_install_egress_timeout(bytes), + Err(error) => { + let message = error.to_string(); + let tx = + send_stream_install_error_cancellable(tx, message.clone(), game_id, &cancel_token) + .await; + return (tx, Err(eyre::eyre!(message))); + } + }; let producer_cancel = cancel_token.child_token(); let frame_sink = StreamInstallFrameSink::new(frame_tx, producer_cancel.clone()); let game_id_for_producer = game_id.to_string(); @@ -830,7 +848,12 @@ pub(crate) async fn send_game_install_stream( let mut framed_tx = FramedWrite::new(tx, LengthDelimitedCodec::new()); let (egress_outcome, producer_result) = { - let egress = forward_stream_install_frames(&mut framed_tx, &mut frame_rx, &producer_cancel); + let egress = forward_stream_install_frames_with_timeout( + &mut framed_tx, + &mut frame_rx, + &producer_cancel, + egress_timeout, + ); tokio::pin!(egress); tokio::select! { @@ -2367,6 +2390,14 @@ mod tests { stream_install_total_timeout(STREAM_INSTALL_MIN_BYTES_PER_SECOND + 1), STREAM_INSTALL_TOTAL_BASE_TIMEOUT + Duration::from_secs(3) ); + assert_eq!( + stream_install_egress_timeout(STREAM_INSTALL_MIN_BYTES_PER_SECOND + 1), + STREAM_INSTALL_TOTAL_BASE_TIMEOUT + Duration::from_secs(2) + ); + assert!( + stream_install_egress_timeout(100 * 1024 * 1024 * 1024) > Duration::from_mins(10), + "large healthy providers must not inherit the former ten-minute absolute cap" + ); } #[tokio::test(start_paused = true)]