From cf1d952ee1903b8cfac9d656a340ecefa70898df Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 20:06:59 +0100 Subject: chunks from dropped sources --- src/swarm/blob.rs | 32 +++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 6febc92..8c47c98 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -656,10 +656,11 @@ pub(crate) async fn fetch( }; match answer { Ok(Some((from, BlobBody::Chunk { index, data }))) => { - let Some(source) = sources.iter_mut().find(|source| source.peer == from) else { - continue; - }; - let sent_at = source.in_flight.remove(&index); + // From a source, or from a peer that was one (dropped meanwhile, as no + // longer reached, say, while its chunks were on their way): its hash + // tells whether it is good, not who sent it. + let at = sources.iter().position(|source| source.peer == from); + let sent_at = at.and_then(|at| sources[at].in_flight.remove(&index)); let slot = index as usize; if slot >= total || chunks[slot].is_some() { continue; @@ -668,22 +669,27 @@ pub(crate) async fn fetch( if data.len() != expected || *blake3::hash(&data).as_bytes() != manifest.hashes[slot] { - source.bad += 1; + if let Some(at) = at { + sources[at].bad += 1; + } warn!(peer = %from, blob = %id, index, "Bad blob chunk"); continue; } - if let Some(sent_at) = sent_at { - debug!(peer = %from, blob = %id, index, ms = sent_at.elapsed().as_millis() as u64, "Blob chunk"); - source.came(sent_at.elapsed()); + if let Some(at) = at { + let source = &mut sources[at]; + if let Some(sent_at) = sent_at { + debug!(peer = %from, blob = %id, index, ms = sent_at.elapsed().as_millis() as u64, "Blob chunk"); + source.came(sent_at.elapsed()); + } + // A peer that delivers may have one more in flight. + source.window = (source.window + 1).min(MAX_WINDOW); } chunks[slot] = Some(data); done += 1; last_progress = Instant::now(); - // A peer that delivers may have one more in flight. - source.window = (source.window + 1).min(MAX_WINDOW); - // Asked of two at the end: the other's copy is no longer awaited (nor - // counted against it as late). - if doubled.contains(&index) { + // Asked of two at the end, or come from a peer no longer a source: no + // other copy is awaited (nor counted against its source as late). + if at.is_none() || doubled.contains(&index) { for other in &mut sources { other.in_flight.remove(&index); } -- cgit v1.3.1