diff --git a/crates/lanspread-peer/src/download/planning.rs b/crates/lanspread-peer/src/download/planning.rs index 463ae78..2372e6c 100644 --- a/crates/lanspread-peer/src/download/planning.rs +++ b/crates/lanspread-peer/src/download/planning.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use lanspread_db::content_manifest::{Blake3Digest, CanonicalCatalogPath, ContentId}; use lanspread_proto::PeerEndpoint; +use tokio::time::Instant; use super::{ manifest::{ExpectedCatalogBlake3, ValidatedDownloadManifest, ValidatedDownloadPath}, @@ -17,6 +18,8 @@ pub(super) struct DownloadChunk { pub(super) offset: u64, pub(super) length: u64, pub(super) expected_blake3: ExpectedCatalogBlake3, + /// Local scheduling metadata. Initial plans have no aggregate retry cap. + pub(super) retry_deadline: Option, } impl DownloadChunk { @@ -31,6 +34,17 @@ impl DownloadChunk { pub(super) const fn canonical_path(&self) -> &CanonicalCatalogPath { self.destination.catalog_path() } + + #[must_use] + pub(super) fn with_retry_deadline(mut self, retry_deadline: Instant) -> Self { + self.retry_deadline = Some(retry_deadline); + self + } + + #[must_use] + pub(super) const fn retry_deadline(&self) -> Option { + self.retry_deadline + } } /// Download plan for a single peer. @@ -139,6 +153,7 @@ pub(super) fn build_peer_plans( offset: 0, length: 0, expected_blake3: manifest.expected_blake3(desc, None)?, + retry_deadline: None, }); continue; } @@ -155,6 +170,7 @@ pub(super) fn build_peer_plans( offset, length, expected_blake3: manifest.expected_blake3(desc, Some(chunk_index))?, + retry_deadline: None, }); offset += length; chunk_index += 1; diff --git a/crates/lanspread-peer/src/download/retry.rs b/crates/lanspread-peer/src/download/retry.rs index 6bd232e..96f8367 100644 --- a/crates/lanspread-peer/src/download/retry.rs +++ b/crates/lanspread-peer/src/download/retry.rs @@ -2,11 +2,13 @@ use std::{ collections::{HashMap, HashSet, VecDeque}, net::IpAddr, sync::Arc, + time::Duration, }; use futures::stream::FuturesUnordered; use lanspread_db::content_manifest::ContentId; use lanspread_proto::PeerEndpoint; +use tokio::time::Instant; use tokio_util::sync::CancellationToken; use super::{ @@ -30,6 +32,32 @@ use crate::{ /// Bounds the number of distinct authenticated identities that one failed /// chunk may make responsible for another full transfer deadline. const MAX_PEER_ATTEMPTS_PER_CHUNK: usize = 8; +/// Aggregate wall-clock budget shared by every retry source for one failed +/// chunk. It starts only after the initial attempt has failed. +const ORDINARY_CHUNK_AGGREGATE_RETRY_TIMEOUT: Duration = Duration::from_mins(20); + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct ChunkRetryDeadline { + expires_at: Instant, +} + +impl ChunkRetryDeadline { + fn ordinary() -> Self { + Self { + expires_at: Instant::now() + ORDINARY_CHUNK_AGGREGATE_RETRY_TIMEOUT, + } + } + + fn expiration_error(self, chunk: &DownloadChunk) -> Option { + (Instant::now() >= self.expires_at).then(|| { + DownloadTransferError::transport(format!( + "catalog chunk aggregate retry deadline expired after {ORDINARY_CHUNK_AGGREGATE_RETRY_TIMEOUT:?} for {} at offset {}", + chunk.canonical_path(), + chunk.offset + )) + }) + } +} /// One failed chunk plus the exact authenticated sources already attempted. /// @@ -40,6 +68,7 @@ pub(super) struct RetryChunk { chunk: DownloadChunk, attempted_peer_ids: HashSet, attempted_source_ips: HashSet, + deadline: ChunkRetryDeadline, last_source: PeerEndpoint, last_error: DownloadTransferError, } @@ -54,6 +83,7 @@ impl RetryChunk { chunk, attempted_peer_ids: HashSet::from([source.peer_id]), attempted_source_ips: HashSet::from([source.addr.ip()]), + deadline: ChunkRetryDeadline::ordinary(), last_source: source, last_error: error, } @@ -62,6 +92,12 @@ impl RetryChunk { fn source_attempt_budget_exhausted(&self) -> bool { self.attempted_peer_ids.len() >= MAX_PEER_ATTEMPTS_PER_CHUNK } + + fn record_source_attempt(&mut self, source: PeerEndpoint) { + self.attempted_peer_ids.insert(source.peer_id); + self.attempted_source_ips.insert(source.addr.ip()); + self.last_source = source; + } } pub(super) struct RetryContext<'a> { @@ -114,6 +150,14 @@ fn plan_retry_batch( let mut retry_plans: HashMap> = HashMap::new(); while let Some(mut retry) = queue.pop_front() { + if let Some(error) = retry.deadline.expiration_error(&retry.chunk) { + final_results.push(ChunkDownloadResult { + chunk: retry.chunk, + result: Err(error), + peer_endpoint: retry.last_source, + }); + continue; + } if retry.source_attempt_budget_exhausted() { final_results.push(ChunkDownloadResult { chunk: retry.chunk, @@ -143,9 +187,7 @@ fn plan_retry_batch( .set_activity(DownloadVerificationActivity::RetryingInvalidSource); } - retry.attempted_peer_ids.insert(source.peer_id); - retry.attempted_source_ips.insert(source.addr.ip()); - retry.last_source = *source; + retry.record_source_attempt(*source); retry_plans.entry(*source).or_default().push(retry); } @@ -164,7 +206,15 @@ async fn run_retry_batch( } let plan = PeerDownloadPlan { - chunks: chunks.iter().map(|retry| retry.chunk.clone()).collect(), + chunks: chunks + .iter() + .map(|retry| { + retry + .chunk + .clone() + .with_retry_deadline(retry.deadline.expires_at) + }) + .collect(), }; let game_root = ctx.game_root.clone(); let game_id = ctx.game_id.to_string(); @@ -416,6 +466,7 @@ mod tests { expected_blake3: manifest .expected_blake3(entry, Some(0)) .expect("test chunk digest should exist"), + retry_deadline: None, } } @@ -483,6 +534,120 @@ mod tests { assert!(!retry.source_attempt_budget_exhausted()); } + #[tokio::test(start_paused = true)] + async fn successful_source_selection_does_not_renew_retry_deadline() { + let initial = source_at("initial", [192, 0, 2, 20], 12_000); + let alternate = source_at("alternate", [192, 0, 2, 21], 12_001); + let mut retry = RetryChunk::after_failure( + chunk(), + initial, + DownloadTransferError::transport("initial failed"), + ); + let expires_at = retry.deadline.expires_at; + + tokio::time::advance(Duration::from_mins(5)).await; + let selected = select_retry_source( + &[alternate], + &retry.attempted_peer_ids, + &retry.attempted_source_ips, + retry.chunk.content_id, + &ContentQuarantine::default(), + ) + .copied() + .expect("an untried alternate source should be selected"); + retry.record_source_attempt(selected); + + assert_eq!(retry.deadline.expires_at, expires_at); + assert_eq!( + expires_at.duration_since(Instant::now()), + Duration::from_mins(15) + ); + } + + #[tokio::test(start_paused = true)] + async fn failed_source_change_retains_deadline_and_expiry_stops_retry() { + let initial = source_at("initial", [192, 0, 2, 30], 12_000); + let failed_retry = source_at("failed-retry", [192, 0, 2, 31], 12_001); + let next_source = source_at("next", [192, 0, 2, 32], 12_002); + let mut retry = RetryChunk::after_failure( + chunk(), + initial, + DownloadTransferError::transport("initial failed"), + ); + retry.record_source_attempt(failed_retry); + let expires_at = retry.deadline.expires_at; + + tokio::time::advance(Duration::from_mins(10)).await; + let quarantine = ContentQuarantine::default(); + let policy = RetryFailurePolicy { + content_id: retry.chunk.content_id, + quarantine: &quarantine, + }; + let mut queue = VecDeque::new(); + let mut final_results = Vec::new(); + handle_retry_attempt_error( + &failed_retry, + vec![retry], + &DownloadTransferError::transport("retry transport failed"), + &policy, + &mut queue, + &mut final_results, + ); + assert!(final_results.is_empty()); + assert_eq!( + queue + .front() + .expect("retry should be requeued") + .deadline + .expires_at, + expires_at + ); + + tokio::time::advance(Duration::from_mins(10)).await; + let temp = TempDir::new("lanspread-retry-deadline"); + let game_root = ConfinedGameRoot::open_or_create(temp.path(), "game") + .expect("test game root should open"); + let version_buffer = Arc::new( + VersionIniBuffer::new("version.ini", 1).expect("test version buffer should validate"), + ); + let progress_tracker = DownloadProgressTracker::new(1); + let cancel_token = CancellationToken::new(); + let quic = QuicConnector::unavailable(); + let (events, _event_rx) = crate::peer_event_channel(); + let status = crate::transfer_status::DownloadAttemptStatus::new( + crate::DownloadAttemptKey::next("game".to_owned()), + cancel_token.clone(), + events, + ); + let sources = [next_source]; + let ctx = RetryContext { + sources: &sources, + content_id: policy.content_id, + quarantine: &quarantine, + game_root: &game_root, + game_id: "game", + cancel_token: &cancel_token, + quic: &quic, + version_buffer, + progress_tracker, + attempt: status.reporter(), + }; + + let plans = plan_retry_batch(&mut queue, &ctx, &mut final_results); + assert!(plans.is_empty()); + let result = final_results + .pop() + .expect("expired retry should become a terminal chunk result"); + let error = result.result.expect_err("expired retry must fail"); + assert_eq!(error.kind(), DownloadTransferErrorKind::Transport); + assert!( + error + .to_string() + .contains("aggregate retry deadline expired") + ); + assert!(!quarantine.is_quarantined(&failed_retry, policy.content_id)); + } + #[test] fn newly_quarantined_source_is_skipped_before_the_next_selection() { let bad = source("bad", 12000); diff --git a/crates/lanspread-peer/src/download/transport.rs b/crates/lanspread-peer/src/download/transport.rs index 43ee685..b9d2f42 100644 --- a/crates/lanspread-peer/src/download/transport.rs +++ b/crates/lanspread-peer/src/download/transport.rs @@ -51,26 +51,44 @@ fn ensure_download_not_cancelled( #[derive(Clone, Copy, Debug)] struct ChunkDeadline { expires_at: Instant, - timeout: Duration, + cause: ChunkDeadlineCause, +} + +#[derive(Clone, Copy, Debug)] +enum ChunkDeadlineCause { + Attempt(Duration), + AggregateRetry, } impl ChunkDeadline { - fn ordinary() -> Self { - Self::after(ORDINARY_CHUNK_TRANSFER_TIMEOUT) + fn ordinary(retry_deadline: Option) -> Self { + let attempt = Self::after(ORDINARY_CHUNK_TRANSFER_TIMEOUT); + match retry_deadline { + Some(expires_at) if expires_at <= attempt.expires_at => Self { + expires_at, + cause: ChunkDeadlineCause::AggregateRetry, + }, + _ => attempt, + } } fn after(timeout: Duration) -> Self { Self { expires_at: Instant::now() + timeout, - timeout, + cause: ChunkDeadlineCause::Attempt(timeout), } } fn timeout_error(self, canonical_path: &str, offset: u64) -> DownloadTransferError { - let timeout = self.timeout; - DownloadTransferError::transport(format!( - "catalog chunk transfer timed out after {timeout:?} for {canonical_path} at offset {offset}" - )) + let message = match self.cause { + ChunkDeadlineCause::Attempt(timeout) => format!( + "catalog chunk transfer timed out after {timeout:?} for {canonical_path} at offset {offset}" + ), + ChunkDeadlineCause::AggregateRetry => format!( + "catalog chunk aggregate retry deadline expired for {canonical_path} at offset {offset}" + ), + }; + DownloadTransferError::transport(message) } fn ensure_active( @@ -632,7 +650,7 @@ async fn download_chunk_plan( ctx.peer_endpoint.addr ); - let deadline = ChunkDeadline::ordinary(); + let deadline = ChunkDeadline::ordinary(chunk.retry_deadline()); match open_chunk_stream(conn, ctx.game_id, &chunk, ctx.cancel_token, deadline).await { Ok(rx) => { in_flight.push(receive_chunk_result( @@ -779,6 +797,7 @@ mod tests { CatalogFileEntry, }; use lanspread_proto::{PeerEndpoint, PeerId}; + use tokio::time::Instant; use super::{ CancellationToken, @@ -1039,6 +1058,36 @@ mod tests { ); } + #[tokio::test(start_paused = true)] + async fn aggregate_retry_deadline_caps_a_later_live_attempt() { + let cancellation = CancellationToken::new(); + let retry_started = Instant::now(); + let retry_deadline = retry_started + Duration::from_mins(20); + tokio::time::advance(Duration::from_mins(15)).await; + let attempt_started = Instant::now(); + let deadline = ChunkDeadline::ordinary(Some(retry_deadline)); + + let error = deadline + .run( + &cancellation, + "game", + "archive.eti", + 0, + future::pending::<()>(), + ) + .await + .expect_err("the retained retry deadline must cap a later transport attempt"); + + assert_eq!(error.kind(), DownloadTransferErrorKind::Transport); + assert!( + error + .to_string() + .contains("aggregate retry deadline expired") + ); + assert_eq!(attempt_started.elapsed(), Duration::from_mins(5)); + assert_eq!(retry_started.elapsed(), Duration::from_mins(20)); + } + #[test] fn integrity_failure_stops_pending_work_but_drains_the_controlled_window() { let peer = loopback_addr(12030);