fix(peer): bound per-chunk retry source fanout

A failed catalog chunk could try every authenticated peer identity. Since each
source owns a fresh transfer deadline, one host could use Sybil identities to
turn a single unavailable chunk into a multi-hour retry sequence.

Count the initial source and cap one chunk at eight distinct peer identities.
Integrity quarantine, transport classification, cancellation, and explicit
user retries retain their existing behavior.

Test Plan:
- `just test` -- passed outside the sandbox; 495 peer tests and all workspace
  targets passed.
- `just fmt` -- Rust formatting completed; the recipe then hit the pre-existing
  generated `security-report/report.md` Markdown-lint failures.
- `git diff --cached --check` -- passed.
This commit is contained in:
2026-09-12 12:21:36 +02:00
parent 7de373afb6
commit c892838d72
@@ -26,6 +26,10 @@ use crate::{
transfer_status::DownloadAttemptReporter,
};
/// 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;
/// One failed chunk plus the exact authenticated sources already attempted.
///
/// Transport addresses deliberately do not participate in retry identity. A
@@ -51,6 +55,10 @@ impl RetryChunk {
last_error: error,
}
}
fn source_attempt_budget_exhausted(&self) -> bool {
self.attempted_peer_ids.len() >= MAX_PEER_ATTEMPTS_PER_CHUNK
}
}
pub(super) struct RetryContext<'a> {
@@ -101,6 +109,15 @@ fn plan_retry_batch(
let mut retry_plans: HashMap<PeerEndpoint, Vec<RetryChunk>> = HashMap::new();
while let Some(mut retry) = queue.pop_front() {
if retry.source_attempt_budget_exhausted() {
final_results.push(ChunkDownloadResult {
chunk: retry.chunk,
result: Err(retry.last_error),
peer_endpoint: retry.last_source,
});
continue;
}
let Some(source) = select_retry_source(
ctx.sources,
&retry.attempted_peer_ids,
@@ -411,6 +428,31 @@ mod tests {
assert_eq!(selected.last().copied(), Some(source("peer-4", 0).peer_id));
}
#[test]
fn retry_budget_counts_the_initial_failed_source() {
let initial = source("initial", 12_000);
let mut retry = RetryChunk::after_failure(
chunk(),
initial,
DownloadTransferError::transport("initial failed"),
);
for index in 0..MAX_PEER_ATTEMPTS_PER_CHUNK - 1 {
retry.attempted_peer_ids.insert(
source(
&format!("retry-{index}"),
12_001 + u16::try_from(index).expect("retry index fits in port"),
)
.peer_id,
);
}
assert!(retry.source_attempt_budget_exhausted());
retry
.attempted_peer_ids
.remove(&source("retry-0", 12_001).peer_id);
assert!(!retry.source_attempt_budget_exhausted());
}
#[test]
fn newly_quarantined_source_is_skipped_before_the_next_selection() {
let bad = source("bad", 12000);