fix(peer): bound Stream Install receive retries

Share one catalog-derived deadline across at most four distinct endpoint-IP attempts: ten minutes plus two catalog copies at 1 MiB/s. Keep the ten-minute inactivity timer, but renew it only after a nonempty catalog-checked file chunk is written to staging.

Cancellation, rollback, integrity quarantine, and public exhaustion states retain their existing semantics.

Test Plan:
- just test
- just clippy
- shared-deadline, source-IP budget, metadata-inactivity, and useful-byte regressions
- git diff --check

crates/lanspread-peer/src/handlers.rs

@@ -4,6 +4,7 @@ use std::{
     collections::{HashMap, HashSet},
     fmt,
     future::Future,
+    net::IpAddr,
     path::{Path, PathBuf},
     sync::Arc,
     time::Duration,
@@ -59,6 +60,7 @@ use crate::{
         ReceiveStreamedInstallRequest,
         StreamInstallReceiveError,
         StreamInstallReceiveErrorKind,
+        StreamInstallTotalDeadline,
         receive_streamed_install,
     },
     transfer_status::{DownloadAttemptReporter, DownloadAttemptStatus},
@@ -166,6 +168,53 @@ const fn stream_install_failure_disposition(
     }
 }
 
+const MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS: usize = 4;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum StreamInstallSourceAdmission {
+    Admitted,
+    AlreadyAttempted,
+    Exhausted,
+}
+
+#[derive(Default)]
+struct StreamInstallSourceAttempts {
+    source_ips: HashSet<IpAddr>,
+}
+
+impl StreamInstallSourceAttempts {
+    fn admit(&mut self, source_ip: IpAddr) -> StreamInstallSourceAdmission {
+        if self.source_ips.contains(&source_ip) {
+            return StreamInstallSourceAdmission::AlreadyAttempted;
+        }
+        if self.source_ips.len() >= MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS {
+            return StreamInstallSourceAdmission::Exhausted;
+        }
+        self.source_ips.insert(source_ip);
+        StreamInstallSourceAdmission::Admitted
+    }
+}
+
+fn admit_stream_install_source(
+    attempts: &mut StreamInstallSourceAttempts,
+    source: &PeerEndpoint,
+    game_id: &str,
+) -> StreamInstallSourceAdmission {
+    let admission = attempts.admit(source.addr.ip());
+    match admission {
+        StreamInstallSourceAdmission::Admitted => {}
+        StreamInstallSourceAdmission::AlreadyAttempted => log::debug!(
+            "Skipping streamed-install source {} at {} for {game_id}: endpoint IP was already attempted",
+            source.peer_id,
+            source.addr
+        ),
+        StreamInstallSourceAdmission::Exhausted => log::warn!(
+            "Streamed install for {game_id} exhausted its {MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS}-source-IP attempt budget"
+        ),
+    }
+    admission
+}
+
 #[derive(Debug)]
 struct StreamDownloadError {
     reason: Option<DownloadFailureReason>,
@@ -1079,6 +1128,57 @@ fn settle_failed_stream_receive(
     Ok((disposition, error))
 }
 
+enum StreamInstallFailureAction {
+    Retry {
+        error: StreamInstallReceiveError,
+        invalid_source: bool,
+    },
+    Exhausted(StreamInstallReceiveError),
+    Stop(StreamDownloadError),
+}
+
+fn stream_install_failure_action(
+    disposition: StreamInstallFailureDisposition,
+    error_kind: StreamInstallReceiveErrorKind,
+    error: StreamInstallReceiveError,
+    total_deadline: StreamInstallTotalDeadline,
+    source: &PeerEndpoint,
+    game_id: &str,
+) -> StreamInstallFailureAction {
+    if disposition != StreamInstallFailureDisposition::Stop && total_deadline.is_elapsed() {
+        log::warn!("Streamed install for {game_id} exhausted its shared total receive deadline");
+        return StreamInstallFailureAction::Exhausted(error);
+    }
+
+    match disposition {
+        StreamInstallFailureDisposition::RetryAndQuarantine
+        | StreamInstallFailureDisposition::Retry => {
+            log::warn!(
+                "Streamed install attempt from {} at {} failed for {game_id}; trying another peer if available: {error}",
+                source.peer_id,
+                source.addr
+            );
+            StreamInstallFailureAction::Retry {
+                error,
+                invalid_source: disposition == StreamInstallFailureDisposition::RetryAndQuarantine,
+            }
+        }
+        StreamInstallFailureDisposition::Stop => {
+            let error = eyre::Report::new(error);
+            StreamInstallFailureAction::Stop(match error_kind {
+                StreamInstallReceiveErrorKind::Cancelled => StreamDownloadError::cancelled(error),
+                StreamInstallReceiveErrorKind::Setup => {
+                    StreamDownloadError::operation_failed(error)
+                }
+                StreamInstallReceiveErrorKind::Integrity
+                | StreamInstallReceiveErrorKind::Transport => {
+                    unreachable!("retryable stream failures cannot stop immediately")
+                }
+            })
+        }
+    }
+}
+
 fn exhausted_stream_receive_error(
     id: &str,
     last_receive_error: Option<StreamInstallReceiveError>,
@@ -1126,11 +1226,16 @@ async fn receive_streamed_install_from_peers(
     let mut last_receive_error = None;
     let mut retry_invalid_source = false;
     let content_id = manifest.content_id();
+    let total_deadline = StreamInstallTotalDeadline::for_manifest(manifest)
+        .map_err(StreamDownloadError::operation_failed)?;
+    let mut source_attempts = StreamInstallSourceAttempts::default();
     for source in sources {
-        if cancel_token.is_cancelled() {
-            return Err(StreamDownloadError::cancelled(eyre::eyre!(
-                "streamed install for {id} was cancelled"
-            )));
+        if let Err(error) = total_deadline.ensure_active(id, cancel_token) {
+            if error.kind() == StreamInstallReceiveErrorKind::Cancelled {
+                return Err(StreamDownloadError::cancelled(eyre::Report::new(error)));
+            }
+            last_receive_error = Some(error);
+            break;
         }
         if ctx.content_quarantine.is_quarantined(source, content_id) {
             log::debug!(
@@ -1140,6 +1245,11 @@ async fn receive_streamed_install_from_peers(
             );
             continue;
         }
+        match admit_stream_install_source(&mut source_attempts, source, id) {
+            StreamInstallSourceAdmission::Admitted => {}
+            StreamInstallSourceAdmission::AlreadyAttempted => continue,
+            StreamInstallSourceAdmission::Exhausted => break,
+        }
 
         let transaction = begin_stream_receive_attempt(
             &game_root,
@@ -1157,6 +1267,7 @@ async fn receive_streamed_install_from_peers(
             tx_notify_ui: tx_notify_ui.clone(),
             quic,
             cancel_token: cancel_token.clone(),
+            total_deadline,
         })
         .await;
 
@@ -1166,40 +1277,26 @@ async fn receive_streamed_install_from_peers(
                 let error_kind = err.kind();
                 let (disposition, err) =
                     settle_failed_stream_receive(ctx, source, content_id, id, transaction, err)?;
-
-                match disposition {
-                    StreamInstallFailureDisposition::RetryAndQuarantine => {
-                        log::warn!(
-                            "Streamed install attempt from {} at {} failed for {id}; trying another peer if available: {err}",
-                            source.peer_id,
-                            source.addr
-                        );
-                        last_receive_error = Some(err);
-                        retry_invalid_source = true;
+                match stream_install_failure_action(
+                    disposition,
+                    error_kind,
+                    err,
+                    total_deadline,
+                    source,
+                    id,
+                ) {
+                    StreamInstallFailureAction::Retry {
+                        error,
+                        invalid_source,
+                    } => {
+                        last_receive_error = Some(error);
+                        retry_invalid_source = invalid_source;
                     }
-                    StreamInstallFailureDisposition::Retry => {
-                        log::warn!(
-                            "Streamed install attempt from {} at {} failed for {id}; trying another peer if available: {err}",
-                            source.peer_id,
-                            source.addr
-                        );
-                        last_receive_error = Some(err);
-                    }
-                    StreamInstallFailureDisposition::Stop => {
-                        let error = eyre::Report::new(err);
-                        return Err(match error_kind {
-                            StreamInstallReceiveErrorKind::Cancelled => {
-                                StreamDownloadError::cancelled(error)
-                            }
-                            StreamInstallReceiveErrorKind::Setup => {
-                                StreamDownloadError::operation_failed(error)
-                            }
-                            StreamInstallReceiveErrorKind::Integrity
-                            | StreamInstallReceiveErrorKind::Transport => {
-                                unreachable!("retryable stream failures cannot stop immediately")
-                            }
-                        });
+                    StreamInstallFailureAction::Exhausted(error) => {
+                        last_receive_error = Some(error);
+                        break;
                     }
+                    StreamInstallFailureAction::Stop(error) => return Err(error),
                 }
             }
         }
@@ -3536,6 +3633,10 @@ mod tests {
         PeerEndpoint::new(peer_id(seed), addr(port))
     }
 
+    fn endpoint_at(seed: u8, ip: [u8; 4], port: u16) -> PeerEndpoint {
+        PeerEndpoint::new(peer_id(seed), SocketAddr::from((ip, port)))
+    }
+
     fn upsert(db: &mut PeerGameDB, endpoint: PeerEndpoint, game: Option<(&str, ContentId)>) {
         let ticket = db
             .begin_candidate_negotiation(endpoint)
@@ -3637,6 +3738,47 @@ mod tests {
         );
     }
 
+    #[test]
+    fn streamed_install_same_ip_identities_and_ports_share_one_attempt() {
+        let mut attempts = StreamInstallSourceAttempts::default();
+        let first = endpoint_at(1, [192, 168, 1, 10], 12_000);
+        let same_ip_sybil = endpoint_at(2, [192, 168, 1, 10], 13_000);
+
+        assert_eq!(
+            attempts.admit(first.addr.ip()),
+            StreamInstallSourceAdmission::Admitted
+        );
+        assert_eq!(
+            attempts.admit(same_ip_sybil.addr.ip()),
+            StreamInstallSourceAdmission::AlreadyAttempted
+        );
+        assert_eq!(attempts.source_ips.len(), 1);
+    }
+
+    #[test]
+    fn streamed_install_stops_after_four_distinct_source_ip_attempts() {
+        let mut attempts = StreamInstallSourceAttempts::default();
+        for seed in 1..=u8::try_from(MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS)
+            .expect("attempt limit should fit u8")
+        {
+            let source = endpoint_at(seed, [10, 0, 0, seed], 12_000 + u16::from(seed));
+            assert_eq!(
+                attempts.admit(source.addr.ip()),
+                StreamInstallSourceAdmission::Admitted
+            );
+        }
+
+        let excess = endpoint_at(100, [10, 0, 0, 100], 13_000);
+        assert_eq!(
+            attempts.admit(excess.addr.ip()),
+            StreamInstallSourceAdmission::Exhausted
+        );
+        assert_eq!(
+            attempts.source_ips.len(),
+            MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS
+        );
+    }
+
     #[test]
     fn streamed_install_retries_and_quarantines_only_typed_integrity_failures() {
         assert_eq!(