ddidderr/lanspread
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(peer): enforce absolute transfer send deadlines

Browse Source 
Use one ten-minute authority window for all raw chunk writes and FIN, and one window for Stream Install producer waits, frames, and FIN. Successful incremental writes can no longer renew a public bulk slot indefinitely.

Test Plan:
- just test
- focused paused-time cumulative-write regressions
- git diff --check
This commit is contained in:
ddidderr
2026-09-12 12:36:48 +02:00
parent 24f573d82a
commit 06cd8b93ed
2 changed files with 132 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lanspread-peer/src/peer.rs
+48 -4
View File
@@ -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<ChunkCloseOutcome, String> {
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);
crates/lanspread-peer/src/stream_install.rs
+84 -7
View File
@@ -882,20 +882,23 @@ async fn forward_stream_install_frames_with_timeout<W>(
framed_tx: &mut FramedWrite<W, LengthDelimitedCodec>,
frame_rx: &mut mpsc::Receiver<StreamInstallFrame>,
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<Notify>,
}
struct DelayedReadyWriter {
delay: Duration,
pending_write: Option<Pin<Box<time::Sleep>>>,
successful_writes: Arc<AtomicUsize>,
}
struct DelayedFileWriter {
gate: Arc<(Mutex<bool>, Condvar)>,
entered: Option<tokio::sync::oneshot::Sender<()>>,
@@ -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<std::io::Result<usize>> {
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<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
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);