diff --git a/crates/lanspread-peer/src/peer.rs b/crates/lanspread-peer/src/peer.rs index 13c2cbd..61d18a4 100644 --- a/crates/lanspread-peer/src/peer.rs +++ b/crates/lanspread-peer/src/peer.rs @@ -123,6 +123,10 @@ async fn stream_file_bytes_with_timeout( remote_addr: &str, send_timeout: std::time::Duration, ) -> eyre::Result<()> { + let started = Instant::now(); + // One application deadline covers every send and the final FIN. A peer + // cannot retain transfer authority by accepting each buffer just in time. + let deadline = started + send_timeout; log::debug!( "{remote_addr} streaming file bytes for peer: {}, offset: {offset}, length: {length}", display_path.display() @@ -138,7 +142,6 @@ async fn stream_file_bytes_with_timeout( let mut remaining = length; let mut total_bytes = 0u64; let mut last_total_bytes = 0u64; - let started = Instant::now(); let mut timestamp = Instant::now(); let mut buf = vec![0u8; FILE_TRANSFER_BUFFER_SIZE]; @@ -193,7 +196,7 @@ async fn stream_file_bytes_with_timeout( cancel_send_stream(tx, remote_addr, display_path); return Err(eyre::eyre!("File transfer cancelled by user")); } - () = tokio::time::sleep(send_timeout) => { + () = tokio::time::sleep_until(deadline) => { cancel_send_stream(tx, remote_addr, display_path); return Err(eyre::eyre!( "catalog chunk send timed out after {send_timeout:?} for {}", @@ -249,7 +252,7 @@ async fn stream_file_bytes_with_timeout( cancel_send_stream(tx, remote_addr, display_path); return Err(eyre::eyre!("File transfer cancelled by user")); } - () = tokio::time::sleep(send_timeout) => { + () = tokio::time::sleep_until(deadline) => { cancel_send_stream(tx, remote_addr, display_path); return Err(eyre::eyre!( "catalog chunk close timed out after {send_timeout:?} for {}", @@ -319,6 +322,7 @@ mod tests { #[derive(Clone, Copy)] enum ControlledIo { Ready, + Delayed(Duration), Fail, Pending, } @@ -326,6 +330,7 @@ mod tests { struct ControlledSender { send: ControlledIo, close: ControlledIo, + successful_sends: usize, reset_count: usize, } @@ -334,6 +339,7 @@ mod tests { Self { send, close, + successful_sends: 0, reset_count: 0, } } @@ -342,7 +348,15 @@ mod tests { impl RawChunkSend for ControlledSender { async fn send_chunk(&mut self, _bytes: Bytes) -> Result<(), String> { match self.send { - ControlledIo::Ready => Ok(()), + ControlledIo::Ready => { + self.successful_sends += 1; + Ok(()) + } + ControlledIo::Delayed(delay) => { + tokio::time::sleep(delay).await; + self.successful_sends += 1; + Ok(()) + } ControlledIo::Fail => Err("controlled send failure".to_string()), ControlledIo::Pending => pending().await, } @@ -351,6 +365,10 @@ mod tests { async fn close_chunk(&mut self) -> Result { match self.close { ControlledIo::Ready => Ok(ChunkCloseOutcome::Closed), + ControlledIo::Delayed(delay) => { + tokio::time::sleep(delay).await; + Ok(ChunkCloseOutcome::Closed) + } ControlledIo::Fail => Err("controlled close failure".to_string()), ControlledIo::Pending => pending().await, } @@ -483,6 +501,32 @@ mod tests { assert_eq!(close_sender.reset_count, 1); } + #[tokio::test(start_paused = true)] + async fn successful_chunk_writes_share_one_absolute_deadline() { + let timeout = Duration::from_secs(25); + let mut sender = ControlledSender::new( + ControlledIo::Delayed(Duration::from_secs(10)), + ControlledIo::Ready, + ); + let payload = vec![0_u8; FILE_TRANSFER_BUFFER_SIZE * 3]; + let started = Instant::now(); + + let error = controlled_stream( + &mut sender, + ControlledReader::bytes(&payload), + 0, + u64::try_from(payload.len()).expect("test payload length should fit u64"), + timeout, + ) + .await + .expect_err("cumulative successful sends must not renew the chunk deadline"); + + assert!(error.to_string().contains("send timed out after 25s")); + assert_eq!(sender.successful_sends, 2); + assert_eq!(sender.reset_count, 1); + assert_eq!(started.elapsed(), timeout); + } + #[tokio::test] async fn short_file_and_close_failure_reset_instead_of_finishing() { let mut short_sender = ControlledSender::new(ControlledIo::Ready, ControlledIo::Ready); diff --git a/crates/lanspread-peer/src/stream_install.rs b/crates/lanspread-peer/src/stream_install.rs index 580351a..887304f 100644 --- a/crates/lanspread-peer/src/stream_install.rs +++ b/crates/lanspread-peer/src/stream_install.rs @@ -882,20 +882,23 @@ async fn forward_stream_install_frames_with_timeout( framed_tx: &mut FramedWrite, frame_rx: &mut mpsc::Receiver, cancel_token: &CancellationToken, - inactivity_timeout: Duration, + egress_timeout: Duration, ) -> StreamInstallEgressOutcome 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; loop { let frame = tokio::select! { biased; () = cancel_token.cancelled() => { return StreamInstallEgressOutcome::Cancelled; } - () = tokio::time::sleep(inactivity_timeout) => { + () = time::sleep_until(deadline) => { return StreamInstallEgressOutcome::Failed(eyre::eyre!( - "streamed install producer timed out after {inactivity_timeout:?} without a frame" + "streamed install producer timed out after {egress_timeout:?} without a frame" )); } frame = frame_rx.recv() => frame, @@ -909,9 +912,9 @@ where () = cancel_token.cancelled() => { return StreamInstallEgressOutcome::Cancelled; } - () = tokio::time::sleep(inactivity_timeout) => { + () = time::sleep_until(deadline) => { return StreamInstallEgressOutcome::Failed(eyre::eyre!( - "streamed install frame send timed out after {inactivity_timeout:?}" + "streamed install frame send timed out after {egress_timeout:?}" )); } result = framed_tx.send(frame.encode()) => result, @@ -926,9 +929,9 @@ where tokio::select! { biased; () = cancel_token.cancelled() => StreamInstallEgressOutcome::Cancelled, - () = tokio::time::sleep(inactivity_timeout) => { + () = time::sleep_until(deadline) => { StreamInstallEgressOutcome::Failed(eyre::eyre!( - "streamed install close timed out after {inactivity_timeout:?}" + "streamed install close timed out after {egress_timeout:?}" )) } result = framed_tx.close() => match result { @@ -1950,6 +1953,12 @@ mod tests { first_shutdown: Arc, } + struct DelayedReadyWriter { + delay: Duration, + pending_write: Option>>, + successful_writes: Arc, + } + struct DelayedFileWriter { gate: Arc<(Mutex, Condvar)>, entered: Option>, @@ -2059,6 +2068,34 @@ mod tests { } } + impl tokio::io::AsyncWrite for DelayedReadyWriter { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let delay = self.delay; + let sleep = self + .pending_write + .get_or_insert_with(|| Box::pin(time::sleep(delay))); + if sleep.as_mut().poll(cx).is_pending() { + return Poll::Pending; + } + + self.pending_write = None; + self.successful_writes.fetch_add(1, Ordering::SeqCst); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn aborted_incoming_write_quiesces_before_rollback_can_start() { let gate = Arc::new((Mutex::new(false), Condvar::new())); @@ -2546,6 +2583,46 @@ mod tests { ); } + #[tokio::test(start_paused = true)] + async fn successful_frame_writes_share_one_absolute_egress_deadline() { + let (frame_tx, mut frame_rx) = mpsc::channel(3); + for path in ["one", "two", "three"] { + frame_tx + .try_send(StreamInstallFrame::Directory { + relative_path: canonical_path(path), + }) + .expect("test frame should fit"); + } + drop(frame_tx); + + let successful_writes = Arc::new(AtomicUsize::new(0)); + let mut framed_tx = FramedWrite::new( + DelayedReadyWriter { + delay: Duration::from_secs(10), + pending_write: None, + successful_writes: Arc::clone(&successful_writes), + }, + LengthDelimitedCodec::new(), + ); + let timeout = Duration::from_secs(25); + let started = TokioInstant::now(); + + let outcome = forward_stream_install_frames_with_timeout( + &mut framed_tx, + &mut frame_rx, + &CancellationToken::new(), + timeout, + ) + .await; + + let StreamInstallEgressOutcome::Failed(error) = outcome else { + panic!("cumulative successful frame writes must hit the shared egress deadline"); + }; + assert!(error.to_string().contains("frame send timed out after 25s")); + assert_eq!(successful_writes.load(Ordering::SeqCst), 2); + assert_eq!(started.elapsed(), timeout); + } + #[tokio::test] async fn successful_close_is_the_egress_completion_point() { let (frame_tx, mut frame_rx) = mpsc::channel(1);