fix(peer): deduplicate chunk retries by source IP

Track both authenticated peer IDs and endpoint IPs for each failed chunk. Rotating keys or ports at one host no longer grants another automatic ten-minute transfer attempt, while the existing eight-attempt ceiling remains.

Test Plan:
- just test
- focused same-IP Sybil and distinct-source selection tests
- git diff --check

crates/lanspread-peer/src/download/retry.rs

@@ -1,5 +1,6 @@
 use std::{
     collections::{HashMap, HashSet, VecDeque},
+    net::IpAddr,
     sync::Arc,
 };
 
@@ -38,6 +39,7 @@ const MAX_PEER_ATTEMPTS_PER_CHUNK: usize = 8;
 pub(super) struct RetryChunk {
     chunk: DownloadChunk,
     attempted_peer_ids: HashSet<PeerId>,
+    attempted_source_ips: HashSet<IpAddr>,
     last_source: PeerEndpoint,
     last_error: DownloadTransferError,
 }
@@ -51,6 +53,7 @@ impl RetryChunk {
         Self {
             chunk,
             attempted_peer_ids: HashSet::from([source.peer_id]),
+            attempted_source_ips: HashSet::from([source.addr.ip()]),
             last_source: source,
             last_error: error,
         }
@@ -90,6 +93,7 @@ fn ensure_not_cancelled(cancel_token: &CancellationToken, game_id: &str) -> eyre
 fn select_retry_source<'a>(
     sources: &'a [PeerEndpoint],
     attempted_peer_ids: &HashSet<PeerId>,
+    attempted_source_ips: &HashSet<IpAddr>,
     content_id: ContentId,
     quarantine: &ContentQuarantine,
 ) -> Option<&'a PeerEndpoint> {
@@ -97,6 +101,7 @@ fn select_retry_source<'a>(
     sources.iter().find(|source| {
         seen_peer_ids.insert(source.peer_id)
             && !attempted_peer_ids.contains(&source.peer_id)
+            && !attempted_source_ips.contains(&source.addr.ip())
             && !quarantine.is_quarantined(source, content_id)
     })
 }
@@ -121,6 +126,7 @@ fn plan_retry_batch(
         let Some(source) = select_retry_source(
             ctx.sources,
             &retry.attempted_peer_ids,
+            &retry.attempted_source_ips,
             ctx.content_id,
             ctx.quarantine,
         ) else {
@@ -138,6 +144,7 @@ fn plan_retry_batch(
         }
 
         retry.attempted_peer_ids.insert(source.peer_id);
+        retry.attempted_source_ips.insert(source.addr.ip());
         retry.last_source = *source;
         retry_plans.entry(*source).or_default().push(retry);
     }
@@ -308,10 +315,11 @@ fn handle_retry_attempt(
     Ok(())
 }
 
-/// Retries failed chunks against every eligible, nonquarantined peer identity.
+/// Retries failed chunks against a bounded set of eligible, nonquarantined
+/// identities at distinct observed endpoint IPs.
 ///
-/// Each source is attempted at most once per chunk. There is no numeric retry
+/// Each identity and endpoint IP is attempted at most once per chunk, and the
-/// cap: terminal failure means the complete eligible source set was exhausted.
+/// initial failed source counts toward the eight-attempt ceiling.
 pub(super) async fn retry_failed_chunks(
     failed_chunks: Vec<RetryChunk>,
     ctx: &RetryContext<'_>,
@@ -359,9 +367,13 @@ mod tests {
     use crate::test_support::TempDir;
 
     fn source(peer_id: &str, port: u16) -> PeerEndpoint {
+        source_at(peer_id, [127, 0, 0, 1], port)
+    }
+
+    fn source_at(peer_id: &str, ip: [u8; 4], port: u16) -> PeerEndpoint {
         PeerEndpoint::new(
             PeerId::from_bytes(*blake3::hash(peer_id.as_bytes()).as_bytes()),
-            SocketAddr::from(([127, 0, 0, 1], port)),
+            SocketAddr::from((ip, port)),
         )
     }
 
@@ -410,16 +422,29 @@ mod tests {
     #[test]
     fn source_selection_exhausts_more_than_three_unique_peer_ids() {
         let sources = (0_u16..5)
-            .map(|index| source(&format!("peer-{index}"), 12000 + index))
+            .map(|index| {
+                source_at(
+                    &format!("peer-{index}"),
+                    [192, 0, 2, u8::try_from(index + 1).expect("source index fits")],
+                    12000 + index,
+                )
+            })
             .collect::<Vec<_>>();
         let quarantine = ContentQuarantine::default();
         let content_id = content(1);
         let mut attempted = HashSet::new();
+        let mut attempted_ips = HashSet::new();
         let mut selected = Vec::new();
 
-        while let Some(source) = select_retry_source(&sources, &attempted, content_id, &quarantine)
+        while let Some(source) = select_retry_source(
-        {
+            &sources,
+            &attempted,
+            &attempted_ips,
+            content_id,
+            &quarantine,
+        ) {
             attempted.insert(source.peer_id);
+            attempted_ips.insert(source.addr.ip());
             selected.push(source.peer_id);
         }
 
@@ -462,17 +487,48 @@ mod tests {
         let content_id = content(2);
 
         assert_eq!(
-            select_retry_source(&sources, &HashSet::new(), content_id, &quarantine),
+            select_retry_source(
+                &sources,
+                &HashSet::new(),
+                &HashSet::new(),
+                content_id,
+                &quarantine,
+            ),
             Some(&bad)
         );
         quarantine.record_integrity_failure(&bad, content_id);
 
         assert_eq!(
-            select_retry_source(&sources, &HashSet::new(), content_id, &quarantine),
+            select_retry_source(
+                &sources,
+                &HashSet::new(),
+                &HashSet::new(),
+                content_id,
+                &quarantine,
+            ),
             Some(&good)
         );
     }
 
+    #[test]
+    fn source_selection_does_not_retry_same_ip_sybil_identities() {
+        let initial = source_at("initial", [192, 0, 2, 10], 12_000);
+        let same_ip = source_at("rotated-key", [192, 0, 2, 10], 12_001);
+        let other_ip = source_at("other-host", [192, 0, 2, 11], 12_002);
+        let sources = vec![same_ip, other_ip];
+
+        assert_eq!(
+            select_retry_source(
+                &sources,
+                &HashSet::from([initial.peer_id]),
+                &HashSet::from([initial.addr.ip()]),
+                content(7),
+                &ContentQuarantine::default(),
+            ),
+            Some(&other_ip)
+        );
+    }
+
     #[test]
     fn transport_and_local_errors_never_quarantine_content() {
         let source = source("peer", 12000);