diff --git a/crates/lanspread-peer/src/stream_install.rs b/crates/lanspread-peer/src/stream_install.rs index 65d0e3d..ab13d4f 100644 --- a/crates/lanspread-peer/src/stream_install.rs +++ b/crates/lanspread-peer/src/stream_install.rs @@ -848,11 +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_with_timeout( + let egress = forward_stream_install_frames_with_deadlines( &mut framed_tx, &mut frame_rx, &producer_cancel, egress_timeout, + STREAM_INSTALL_INACTIVITY_TIMEOUT, ); tokio::pin!(egress); @@ -1011,18 +1012,42 @@ async fn forward_stream_install_frames_with_timeout( where W: tokio::io::AsyncWrite + Unpin, { - // Producer waits, every frame write, and the final FIN share this deadline. - // Successfully forwarding a frame never renews the transfer authority. - let deadline = TokioInstant::now() + egress_timeout; + forward_stream_install_frames_with_deadlines( + framed_tx, + frame_rx, + cancel_token, + egress_timeout, + egress_timeout, + ) + .await +} + +async fn forward_stream_install_frames_with_deadlines( + framed_tx: &mut FramedWrite, + frame_rx: &mut mpsc::Receiver, + cancel_token: &CancellationToken, + total_timeout: Duration, + inactivity_timeout: Duration, +) -> StreamInstallEgressOutcome +where + W: tokio::io::AsyncWrite + Unpin, +{ + let total_deadline = TokioInstant::now() + total_timeout; + let mut inactivity_deadline = TokioInstant::now() + inactivity_timeout; loop { let frame = tokio::select! { biased; () = cancel_token.cancelled() => { return StreamInstallEgressOutcome::Cancelled; } - () = time::sleep_until(deadline) => { + () = time::sleep_until(total_deadline) => { return StreamInstallEgressOutcome::Failed(eyre::eyre!( - "streamed install producer timed out after {egress_timeout:?} without a frame" + "streamed install producer timed out after {total_timeout:?} without a frame" + )); + } + () = time::sleep_until(inactivity_deadline) => { + return StreamInstallEgressOutcome::Failed(eyre::eyre!( + "streamed install producer timed out after {inactivity_timeout:?} without a frame" )); } frame = frame_rx.recv() => frame, @@ -1036,9 +1061,14 @@ where () = cancel_token.cancelled() => { return StreamInstallEgressOutcome::Cancelled; } - () = time::sleep_until(deadline) => { + () = time::sleep_until(total_deadline) => { return StreamInstallEgressOutcome::Failed(eyre::eyre!( - "streamed install frame send timed out after {egress_timeout:?}" + "streamed install frame send timed out after {total_timeout:?}" + )); + } + () = time::sleep_until(inactivity_deadline) => { + return StreamInstallEgressOutcome::Failed(eyre::eyre!( + "streamed install frame send timed out after {inactivity_timeout:?}" )); } result = framed_tx.send(frame.encode()) => result, @@ -1048,14 +1078,20 @@ where "failed to send streamed install frame: {err}" )); } + inactivity_deadline = TokioInstant::now() + inactivity_timeout; } tokio::select! { biased; () = cancel_token.cancelled() => StreamInstallEgressOutcome::Cancelled, - () = time::sleep_until(deadline) => { + () = time::sleep_until(total_deadline) => { StreamInstallEgressOutcome::Failed(eyre::eyre!( - "streamed install close timed out after {egress_timeout:?}" + "streamed install close timed out after {total_timeout:?}" + )) + } + () = time::sleep_until(inactivity_deadline) => { + StreamInstallEgressOutcome::Failed(eyre::eyre!( + "streamed install close timed out after {inactivity_timeout:?}" )) } result = framed_tx.close() => match result { @@ -2839,10 +2875,11 @@ mod tests { let (_frame_tx, mut frame_rx) = mpsc::channel(1); let mut framed_tx = FramedWrite::new(tokio::io::sink(), LengthDelimitedCodec::new()); - let outcome = forward_stream_install_frames_with_timeout( + let outcome = forward_stream_install_frames_with_deadlines( &mut framed_tx, &mut frame_rx, &CancellationToken::new(), + Duration::from_hours(1), Duration::from_millis(25), ) .await; @@ -2913,10 +2950,11 @@ mod tests { LengthDelimitedCodec::new(), ); - let outcome = forward_stream_install_frames_with_timeout( + let outcome = forward_stream_install_frames_with_deadlines( &mut framed_tx, &mut frame_rx, &CancellationToken::new(), + Duration::from_hours(1), Duration::from_millis(25), ) .await;