From b3174deabfdbe185c4bf03475dfa516552fcd59d Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 12 Sep 2026 12:53:54 +0200 Subject: [PATCH] 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 | 74 ++++++++++++++++++--- 1 file changed, 65 insertions(+), 9 deletions(-) diff --git a/crates/lanspread-peer/src/download/retry.rs b/crates/lanspread-peer/src/download/retry.rs index b9618dc..4ae5acf 100644 --- a/crates/lanspread-peer/src/download/retry.rs +++ b/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, + attempted_source_ips: HashSet, 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, + attempted_source_ips: &HashSet, 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 -/// cap: terminal failure means the complete eligible source set was exhausted. +/// Each identity and endpoint IP is attempted at most once per chunk, and the +/// initial failed source counts toward the eight-attempt ceiling. pub(super) async fn retry_failed_chunks( failed_chunks: Vec, 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::>(); 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);