fix(peer): cap aggregate chunk retry time

Start one twenty-minute retry deadline only after an initial chunk failure, retain it across authenticated source and IP changes, and give each in-flight retry the earlier of that deadline and its ten-minute attempt limit. Expiry remains a non-quarantining transport failure.

Test Plan:
- just test
- just clippy
- three paused-time retry-deadline regressions
- git diff --check

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<Instant>,
 }
 
 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<Instant> {
+        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;

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<DownloadTransferError> {
+        (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<PeerId>,
     attempted_source_ips: HashSet<IpAddr>,
+    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<PeerEndpoint, Vec<RetryChunk>> = 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);

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<Instant>) -> 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);