From 13070d966b5755f3705ce98730516a2ab84e41a2 Mon Sep 17 00:00:00 2001 From: Chris O'Neil Date: Fri, 22 May 2026 18:14:18 +0100 Subject: [PATCH 1/4] fix(download): residential saturation + transient failure hardening MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hardens the download path against two distinct failure modes observed running `ant file download` against the production network: residential link saturation and per-peer / DHT transient errors that previously fatally aborted multi-hundred-chunk downloads. End to end, this takes a residential `ant file download` from "aborts on the first 256-wide concurrent batch's saturation event" to "completes 11/11 files including 2+ GB downloads," and on a fat-pipe droplet from "matches baseline" to "matches baseline" — no regression on the warm-start path that production downloaders actually exercise. Six related changes: 1. retry-on-Ok(None) with unanimous-NotFound threshold in chunk_get. When the close group returns Ok(None) (no peer has the chunk), retry once with a fresh find_closest_peers lookup, unless every queried peer responded with an authoritative NotFound (the only safe stop for genuine data absence). The previous behaviour treated Ok(None) as fatal on first occurrence, which on a saturated link meant any single chunk's transient close-group exhaustion aborted the whole download. 2. rebucketed_unordered in file.rs instead of buffer_unordered for the in-flight chunk fetches. The adaptive limiter's cap can now shrink the in-flight count mid-batch under sustained pressure; buffer_unordered snapshotted the cap once at pipeline build and ignored later Decrease decisions. 3. observe-outer with Ok(None) -> Outcome::Timeout instead of observe-per-peer. The controller sees one observation per chunk_get (not one per peer attempt), classified via a new chunk_get_outcome helper that treats Ok(None) as a load-shedding signal. Avoids the per-peer noise floor on the production network where some peers in any K=7 close group are unreachable from any given client even on a healthy link — that noise was driving spurious Decrease decisions on the droplet and pinning steady-state cap low. 4. ChannelStart::fetch: 64 -> 4 cold-start. The 64-wide initial burst saturated residential connections before the controller had any observation to act on. 4 is the value confirmed safe on a real residential link. On droplets the cost is a one-off cold-start warm-up of ~16 min on the first 2.5 GB file; subsequent files warm-start from the persisted client_adaptive.json snapshot (which the controller cleanly grows to cap=256, the channel ceiling). 5. Deferred-retry pass in streaming_decrypt's consumer. When chunk_get returns Ok(None) for a chunk during a batch, the chunk is deferred rather than aborting the batch. After the main batch settles, the deferred chunks are retried serially with sleeps of 10/30/60 s. This rides out transient saturation events that hit multiple in-flight chunks at once — by the time the batch has drained and the first sleep elapses, the link has usually settled. A chunk only becomes fatal after all 3 deferred attempts fail. 6. Per-peer protocol-error tolerance and deferred-retry transient-error tolerance. A single peer returning Error::Protocol (e.g. "Chunk verification failed" from a corrupted local copy) no longer aborts the close-group sweep — the loop counts it and continues to the next peer. Similarly, an Err(_) from chunk_get_observed during a deferred-retry attempt logs and falls through to the next attempt's longer backoff rather than escalating. Also: latency_inflation_factor default 2.0 -> 4.0. Natural close-group fallback latency on the production network routinely doubles vs the EWMA baseline (a single peer hitting fallback adds ~10 s on top of a sub-second median), and was firing spurious Decrease decisions even on the droplet. 4.0 is the value validated on the previously-merged tune-latency-inflation-factor branch. Test plan: - 296 ant-core unit tests pass. - End-to-end residential download (PROD-LOCAL-DL-04): 11/11 files completed including 2.51 GB (42m 43s) and 2.76 GB (46m 26s). During an earlier residential run 14 chunks went through the deferred-retry path and every one recovered on attempt 1/3 after the 10 s sleep. - End-to-end droplet download (PROD-DL-05): 20/20 files completed. The first 2.5 GB file paid 16m 9s of cold-start cost; subsequent multi-GB files ran in 3-6 min each, near the pre-change ~5 min baseline on PROD-DL-02. No retry mechanism fired across the 20-file run — healthy-network success. Co-Authored-By: Claude Opus 4.7 (1M context) --- ant-core/src/data/client/adaptive.rs | 116 +++++++--- ant-core/src/data/client/chunk.rs | 307 ++++++++++++++++++++++++++- ant-core/src/data/client/data.rs | 13 +- ant-core/src/data/client/file.rs | 282 ++++++++++++++++-------- 4 files changed, 589 insertions(+), 129 deletions(-) diff --git a/ant-core/src/data/client/adaptive.rs b/ant-core/src/data/client/adaptive.rs index dac6895e..e2259058 100644 --- a/ant-core/src/data/client/adaptive.rs +++ b/ant-core/src/data/client/adaptive.rs @@ -74,6 +74,17 @@ pub enum Outcome { ApplicationError, } +/// Lower bound on the `fetch` channel's adaptive cap. +/// +/// AIMD will not shrink fetch concurrency below this even under +/// sustained timeout pressure. Specific to fetch because residential +/// downloads exhibit a noise floor of peer-side timeouts (NAT path +/// issues, peers in the close group not storing the chunk) that look +/// like client saturation to the controller, causing it to fully +/// serialize and collapse throughput. Quote and store channels keep +/// the global `min_concurrency` floor of 1. +const FETCH_MIN_FLOOR: usize = 4; + /// Per-channel concurrency ceilings. Each channel has its own cap so /// that constraining one (e.g. user pinned a low store concurrency for /// a slow uplink) never bleeds into another (download). @@ -152,7 +163,7 @@ impl AdaptiveConfig { } self.timeout_ceiling = self.timeout_ceiling.clamp(0.0, 1.0); if !self.latency_inflation_factor.is_finite() || self.latency_inflation_factor <= 0.0 { - self.latency_inflation_factor = 2.0; + self.latency_inflation_factor = 4.0; } self.min_concurrency = self.min_concurrency.max(1); self.window_ops = self.window_ops.max(1); @@ -173,23 +184,38 @@ impl Default for AdaptiveConfig { min_window_ops: 8, success_target: 0.95, timeout_ceiling: 0.10, - latency_inflation_factor: 2.0, + // p95 doubling is the normal signal on a per-chunk fetch with + // close-group fallback (one slow peer in a chunk's close group + // adds ~10s on top of a sub-second median); 2.0 mis-classified + // that as stress and halved the fetch cap mid-download. 4.0 + // means p95 has to quadruple before we treat the network as + // degraded. + latency_inflation_factor: 4.0, latency_ewma_alpha: 0.2, } } } /// Suggested starting concurrency per channel for a brand-new client -/// with no persisted state. Intentionally matches or exceeds the prior -/// static defaults so the cold path is not slower: +/// with no persisted state: /// /// - quote was statically 32 — start at 32. /// - store was statically 8 — start at 8. -/// - fetch was previously unbounded (the entire self_encryption batch -/// was fired at once via `FuturesUnordered`). Typical batches are -/// small (handful of chunks); occasional larger ones are capped at -/// `max_concurrency`. Start at 64 to keep small/medium downloads -/// indistinguishable from the prior unbounded behavior. +/// - fetch was previously 64. Dropped to 8 because the cold-start +/// burst dominates residential download outcomes: the first batch +/// of `fetch` concurrent chunk_gets fires before any per-peer +/// observation has landed, so if `fetch` exceeds what the link can +/// sustain, the entire first burst saturates the connection and +/// typically fails before the AIMD controller can shrink the cap. +/// Reproduced on PROD-LOCAL-DL-03 with fetch=64: 60 of 64 +/// in-flight chunks failed both first attempt and retry; only the +/// first ~13 seconds of the download (4 chunks) saw any +/// successful completions. With fetch=8, only 8 chunks compete in +/// the initial burst; on a fat pipe with healthy outcomes the +/// AIMD increase logic grows it back to the channel's +/// `max_concurrency` (256) within a window or two — measured +/// cost on the droplet is a one-off ~30-60s slow-start delay on +/// the very first download. #[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct ChannelStart { pub quote: usize, @@ -202,7 +228,18 @@ impl Default for ChannelStart { Self { quote: 32, store: 8, - fetch: 64, + // 4 is the residential-saturation floor we have validated: + // 8 still firehose-saturated a real home link enough to + // disrupt the operator's general internet usage on the + // initial burst (the controller has not yet seen any + // observation by the time those 8 concurrent chunk_gets + // launch). 16, 32, 64 are progressively worse. Per the + // PROD-DL-03 run on 83bcbdb, the droplet pays the cost + // for cold-start once — a one-off ~8 min warm-up on the + // first 2.5 GB file — and persisted-snapshot warm_start + // then runs subsequent downloads at cap=256 (the + // ChannelMax::fetch ceiling). + fetch: 4, } } } @@ -266,7 +303,7 @@ impl LimiterConfig { } self.timeout_ceiling = self.timeout_ceiling.clamp(0.0, 1.0); if !self.latency_inflation_factor.is_finite() || self.latency_inflation_factor <= 0.0 { - self.latency_inflation_factor = 2.0; + self.latency_inflation_factor = 4.0; } self.min_concurrency = self.min_concurrency.max(1); self.window_ops = self.window_ops.max(1); @@ -567,7 +604,28 @@ impl AdaptiveController { config.sanitize(); let quote_cfg = LimiterConfig::from_adaptive(&config, config.max.quote); let store_cfg = LimiterConfig::from_adaptive(&config, config.max.store); - let fetch_cfg = LimiterConfig::from_adaptive(&config, config.max.fetch); + let mut fetch_cfg = LimiterConfig::from_adaptive(&config, config.max.fetch); + // Lift the fetch channel's floor above the global + // `min_concurrency`. Reasoning is specific to download: on + // residential links, residual peer-side timeouts (NAT path + // issues, peers in the close group that don't store the chunk, + // peers under temporary load) continuously push the + // controller's timeout_rate above ceiling. A global floor of 1 + // means the controller fully serializes chunk fetches on that + // noise floor and gets stuck — observed on PROD-LOCAL-DL-03 + // where the download stayed stable but throughput collapsed to + // ~330 KB/s on a multi-MB/s link. + // + // 4 is the smallest floor that keeps the download from fully + // serializing while staying well below the cold-start + // (ChannelStart::fetch = 8) that the home retest tolerated. + // Floor `quote` and `store` separately if a corresponding + // pathology is identified for them; today's evidence is + // download-only. + fetch_cfg.min_concurrency = fetch_cfg.min_concurrency.max(FETCH_MIN_FLOOR); + // Re-establish max >= min after the bump in case the channel + // ceiling was somehow lower than the new floor. + fetch_cfg.max_concurrency = fetch_cfg.max_concurrency.max(fetch_cfg.min_concurrency); Self { quote: Limiter::new(start.quote, quote_cfg), store: Limiter::new(start.store, store_cfg), @@ -1256,10 +1314,17 @@ mod tests { // These exist primarily to prove the controller never silently regresses // upload/download throughput and never panics under hostile workloads. - /// Cold-start defaults must equal-or-exceed the prior static knobs so - /// the very first batch on a fresh install is no slower than before - /// the adaptive controller existed. Hard-coded literals are intentional - /// — this is a guard against future commits accidentally lowering them. + /// Cold-start defaults must equal-or-exceed the values we have + /// deliberately committed to. Hard-coded literals are intentional + /// — this is a guard against future commits accidentally drifting + /// the cold-start values away from the policy decisions documented + /// on `ChannelStart`'s comment. + /// + /// `fetch` was historically 64, lowered to 8 after PROD-LOCAL-DL-03 + /// showed a 64-wide initial burst saturated residential links + /// before the AIMD controller could shrink the cap. Do NOT raise + /// this back without a network-side justification — see the + /// `ChannelStart` doc. #[test] fn no_regression_cold_start_at_least_static_defaults() { let s = ChannelStart::default(); @@ -1274,8 +1339,9 @@ mod tests { s.store, ); assert!( - s.fetch >= 64, - "fetch cold-start regressed: got {}, prior static was 64 (unbounded before)", + s.fetch >= 4, + "fetch cold-start regressed below the residential-saturation floor: \ + got {}, current policy floor is 4 (see ChannelStart doc)", s.fetch, ); } @@ -1943,18 +2009,18 @@ mod tests { /// Cold-start equals the prior static defaults so the FIRST batch /// on a fresh install behaves identically. Guards against future - /// commits silently dropping cold-start values below the prior - /// statics. + /// commits silently dropping cold-start values below the current + /// policy floor. #[test] fn cold_start_at_least_prior_static_defaults() { let cs = ChannelStart::default(); - // Prior statics: quote=32, store=8. Fetch was effectively - // unbounded (the entire self_encryption batch was fired at - // once); we picked 64 as a conservative substitute so the - // first batch of <= 64 chunks is indistinguishable. + // Policy floors: quote=32, store=8 (both match the pre-adaptive + // statics). Fetch was 64 historically; lowered to 4 to keep the + // initial burst from saturating residential downlinks (see + // `ChannelStart` doc). assert!(cs.quote >= 32, "quote cold-start regressed: {}", cs.quote); assert!(cs.store >= 8, "store cold-start regressed: {}", cs.store); - assert!(cs.fetch >= 64, "fetch cold-start regressed: {}", cs.fetch); + assert!(cs.fetch >= 4, "fetch cold-start regressed: {}", cs.fetch); } /// Reviewer N-M5 guard: with the new gated-decrease semantics diff --git a/ant-core/src/data/client/chunk.rs b/ant-core/src/data/client/chunk.rs index 9bf65d62..6424da3d 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -3,6 +3,7 @@ //! Chunks are immutable, content-addressed data blocks where the address //! is the BLAKE3 hash of the content. +use crate::data::client::adaptive::Outcome; use crate::data::client::batch::{finalize_batch_payment, PreparedChunk}; use crate::data::client::peer_cache::record_peer_outcome; use crate::data::client::Client; @@ -24,6 +25,50 @@ use tracing::{debug, info, warn}; /// Data type identifier for chunks (used in quote requests). const CHUNK_DATA_TYPE: u32 = 0; +/// Result of one sweep over a chunk's close group. +/// +/// Either we got the chunk from some peer, or every peer in the group +/// returned NotFound, timed out, or hit a transport / protocol error. +/// The counts are used to decide whether to retry: only a *unanimous* +/// NotFound response from every queried peer counts as authoritative +/// data absence — anything else leaves room for the actual storer to +/// be in the timeout / network-error / protocol-error bucket and is +/// worth one retry against a freshly re-walked close group. +struct CloseGroupOutcome { + chunk: Option, + queried: usize, + not_found: usize, + timeout: usize, + network_err: usize, + /// Counts peers that responded with a remote `Error` (e.g. + /// "Chunk verification failed") or any other protocol-level error + /// that classifies as `Error::Protocol`. Treated the same as + /// `timeout` / `network_err` for retry decisions: one peer's bad + /// response must not abort the whole close-group sweep — the + /// remaining peers might still have a clean copy. + protocol_err: usize, +} + +/// `true` if every peer we managed to query responded with an +/// authoritative NotFound. Only then is it safe to conclude the chunk +/// is genuinely absent from this close group. +/// +/// An earlier version of this check used a majority quorum +/// (`not_found >= close_group_size / 2 + 1`), but production traffic +/// disproved that threshold: on the production network the storage +/// replication target is `CLOSE_GROUP_MAJORITY` (4) of the K=7 +/// close-group peers — so up to 3 peers legitimately don't store any +/// given chunk. A saturated client that sees `not_found=4 timeout=3` +/// is almost certainly looking at "3 storers we couldn't reach" plus +/// "4 non-storers that responded promptly," not authoritative data +/// loss. Reproduced on PROD-LOCAL-DL-03: a chunk that failed at the +/// quorum threshold during `ant file download` was successfully +/// retrieved on the same host moments later via a single +/// `ant chunk get`. Only `not_found == queried` is a safe stop. +fn is_authoritative_not_found(not_found: usize, queried: usize) -> bool { + queried > 0 && not_found == queried +} + /// Store-response timeout for non-merkle chunk PUTs. const STORE_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10); @@ -34,6 +79,54 @@ fn store_response_timeout_for_proof(proof: &[u8], merkle_timeout_secs: u64) -> D } } +impl Client { + /// Run `chunk_get` and feed one observation per call to the + /// adaptive fetch limiter. Use this from any consumer that drives + /// chunk-fetch concurrency from `controller().fetch.current()` — + /// the controller's window relies on every call along the hot + /// path producing an observation. + /// + /// Classifier semantics: see `chunk_get_outcome`. Most importantly, + /// `Ok(None)` is treated as `Outcome::Timeout`, not Success, so a + /// sustained run of close-group exhaustions correctly drives the + /// cap down rather than silently inflating it. + pub(crate) async fn chunk_get_observed(&self, address: &XorName) -> Result> { + let started = Instant::now(); + let result = self.chunk_get(address).await; + let latency = started.elapsed(); + self.controller() + .fetch + .observe(chunk_get_outcome(&result), latency); + result + } +} + +/// Map a `chunk_get` outcome to an adaptive controller `Outcome`. +/// +/// This is the result-aware classifier used by the file-download paths. +/// It differs from `classify_error` in one critical way: an `Ok(None)` +/// from `chunk_get` is `Outcome::Timeout`, not `Outcome::Success`. By +/// the time `chunk_get` returns `Ok(None)` it has already exhausted +/// the close group across its first attempt + retry sweep, so +/// `Ok(None)` is the controller's load-shedding signal — a sustained +/// run of them on a saturated home link is exactly the case where the +/// cap should shrink. +/// +/// Healthy returns (`Ok(Some(_))`) are Success regardless of how many +/// internal peer attempts the chunk_get had to make. The controller +/// does not need to see internal peer noise; that's noise about the +/// production network's natural peer-side variability, not about the +/// client's effective capacity. +pub(crate) fn chunk_get_outcome(result: &Result>) -> Outcome { + match result { + Ok(Some(_)) => Outcome::Success, + Ok(None) => Outcome::Timeout, + Err(Error::Timeout(_)) => Outcome::Timeout, + Err(Error::Network(_)) => Outcome::NetworkError, + Err(_) => Outcome::ApplicationError, + } +} + impl Client { /// Store a chunk on the Autonomi network with payment. /// @@ -239,6 +332,20 @@ impl Client { /// where the storing peer differs from the first peer returned by /// DHT routing. /// + /// ## Adaptive controller feedback + /// + /// Each per-peer GET attempt is fed individually to the adaptive + /// fetch limiter via `controller().fetch.observe(...)`. This is + /// deliberately finer-grained than wrapping the outer `chunk_get` + /// with `observe_op`: when a chunk takes 6 peer tries to land, + /// 5 of them are real capacity signals (timeouts / network errors) + /// that should pull the cap down even if the chunk eventually + /// succeeds. The outer `Ok(_)` would mask all five as a single + /// `Outcome::Success`. See `adaptive::Outcome` for the per-attempt + /// classification rules used below. + /// + /// Callers should therefore NOT wrap `chunk_get` in `observe_op`. + /// /// # Errors /// /// Returns an error if the network operation fails. @@ -258,19 +365,115 @@ impl Client { self.chunk_cache().remove(address); } - let peers = self.close_group_peers(address).await?; let addr_hex = hex::encode(address); + // First attempt against the current close-group view. + let first = self.chunk_get_try_close_group(address).await?; + if let Some(chunk) = first.chunk { + self.chunk_cache().put(chunk.address, chunk.content.clone()); + return Ok(Some(chunk)); + } + + // Only treat as authoritative absence when *every* queried peer + // responded NotFound. Anything less leaves the actual storer + // possibly in the timeout / network-error bucket, which a retry + // could reach. + if is_authoritative_not_found(first.not_found, first.queried) { + info!( + "chunk_get giving up on {addr_hex} (unanimous NotFound): \ + queried={} not_found={} timeout={} network_err={} protocol_err={}", + first.queried, + first.not_found, + first.timeout, + first.network_err, + first.protocol_err, + ); + return Ok(None); + } + + // Otherwise the failure looks like reachability (most peers timed out + // or hit transport errors). The chunk is most likely still on the + // network but the current close-group view either (a) caught a + // transient transport blip or (b) converged on the wrong neighbourhood + // because the routing table is thin. One retry against a freshly + // re-walked close group is the cheapest defence against both. + info!( + "chunk_get retrying {addr_hex} after reachability failure: \ + queried={} not_found={} timeout={} network_err={} protocol_err={}", + first.queried, first.not_found, first.timeout, first.network_err, first.protocol_err, + ); + + // Brief settle so any in-flight transport state can quiesce before + // we re-walk the DHT. Keep this small so we don't add meaningful + // latency to the genuinely-lost case (we already paid for one full + // close-group sweep before getting here). + tokio::time::sleep(Duration::from_secs(1)).await; + + // If the retry's DHT lookup itself fails, treat that as "still + // couldn't find" rather than escalating the error — matches the + // semantics of the first attempt when peers are unreachable. + let retry = match self.chunk_get_try_close_group(address).await { + Ok(o) => o, + Err(e) => { + info!( + "chunk_get retry close-group lookup failed for {addr_hex}: {e}; \ + first(queried={} not_found={} timeout={} network_err={} protocol_err={})", + first.queried, + first.not_found, + first.timeout, + first.network_err, + first.protocol_err, + ); + return Ok(None); + } + }; + if let Some(chunk) = retry.chunk { + info!("chunk_get retry succeeded for {addr_hex}"); + self.chunk_cache().put(chunk.address, chunk.content.clone()); + return Ok(Some(chunk)); + } + + info!( + "chunk_get exhausted close group after retry for {addr_hex}: \ + first(queried={} not_found={} timeout={} network_err={} protocol_err={}) \ + retry(queried={} not_found={} timeout={} network_err={} protocol_err={})", + first.queried, + first.not_found, + first.timeout, + first.network_err, + first.protocol_err, + retry.queried, + retry.not_found, + retry.timeout, + retry.network_err, + retry.protocol_err, + ); + Ok(None) + } + + /// One sweep of the current close group: fetch the K closest peers + /// for `address` from the DHT and ask each for the chunk in turn, + /// returning on the first success. + async fn chunk_get_try_close_group(&self, address: &XorName) -> Result { + let peers = self.close_group_peers(address).await?; + let addr_hex = hex::encode(address); let queried = peers.len(); let mut not_found = 0usize; let mut timeout = 0usize; let mut network_err = 0usize; + let mut protocol_err = 0usize; for (peer, addrs) in &peers { match self.chunk_get_from_peer(address, peer, addrs).await { Ok(Some(chunk)) => { - self.chunk_cache().put(chunk.address, chunk.content.clone()); - return Ok(Some(chunk)); + return Ok(CloseGroupOutcome { + chunk: Some(chunk), + queried, + not_found, + timeout, + network_err, + protocol_err, + }); } Ok(None) => { not_found += 1; @@ -284,18 +487,32 @@ impl Client { network_err += 1; debug!("Peer {peer} unreachable for chunk {addr_hex}, trying next"); } + // A `Protocol` error here is the storer responding with + // `ChunkGetResponse::Error(...)` — e.g. "Chunk verification + // failed" from a peer that has a corrupted local copy. + // That's a per-peer problem, not a per-chunk one: the + // remaining peers might still have a clean copy, so + // continue the sweep rather than aborting it. Counted + // separately from network_err so the summary log still + // distinguishes "peer corrupted" from "peer unreachable". + Err(Error::Protocol(ref e)) => { + protocol_err += 1; + debug!( + "Peer {peer} returned protocol error for chunk {addr_hex} ({e}), trying next" + ); + } Err(e) => return Err(e), } } - // None of the close group peers had the chunk. Emit a single summary - // so operators can distinguish data loss (all peers responded NotFound) - // from a reachability problem (most peers timed out / errored). - info!( - "chunk_get exhausted close group for {addr_hex}: \ - queried={queried} not_found={not_found} timeout={timeout} network_err={network_err}" - ); - Ok(None) + Ok(CloseGroupOutcome { + chunk: None, + queried, + not_found, + timeout, + network_err, + protocol_err, + }) } /// Fetch a chunk from a specific peer. @@ -434,6 +651,74 @@ mod tests { /// Sentinel byte used to represent an unknown/unrecognized proof tag. const UNKNOWN_PROOF_TAG: u8 = 0xff; + #[test] + fn authoritative_not_found_requires_unanimous_response() { + // Unanimous: every queried peer (7 of 7) said NotFound. This is + // the only safe stop. + assert!(is_authoritative_not_found(7, 7)); + + // 4-of-7 looks like authoritative absence on paper, but with + // CLOSE_GROUP_MAJORITY=4 replicas on the production network the + // expected response shape for a healthy chunk fetched from a + // saturated client is 3 NotFound (the non-storers) + 4 Timeout + // (the storers we couldn't reach in time). The mirror — 4 + // NotFound + 3 Timeout — is the same shape with one extra + // storer rotated out and one extra non-storer happening to be + // reachable. Refusing to retry that case was the regression + // PROD-LOCAL-DL-03 surfaced; it must NOT be treated as + // authoritative. + assert!(!is_authoritative_not_found(4, 7)); + assert!(!is_authoritative_not_found(6, 7)); + + // Pure-reachability failure — must retry. + assert!(!is_authoritative_not_found(0, 7)); + + // Defensive: queried == 0 should never reach this check + // (close_group_peers would have errored earlier), but if it + // does, return false so we don't claim authoritative absence + // based on no evidence. + assert!(!is_authoritative_not_found(0, 0)); + } + + #[test] + fn chunk_get_outcome_classifies_each_result_kind() { + // Success: chunk_get returned a chunk, regardless of how many + // internal peer attempts it took. + let chunk = DataChunk::new([0u8; 32], Bytes::from_static(b"x")); + assert_eq!( + chunk_get_outcome(&Ok(Some(chunk))), + Outcome::Success, + "found-chunk must be Success", + ); + + // Ok(None): chunk_get exhausted the close group across first + // attempt + retry. This is the load-shedding signal — count it + // as Timeout so a sustained run of them on a saturated link + // shrinks the cap. + assert_eq!( + chunk_get_outcome(&Ok(None)), + Outcome::Timeout, + "Ok(None) must be Timeout — that's the controller's load-shedding signal", + ); + + // Capacity signals from explicit error variants. + assert_eq!( + chunk_get_outcome(&Err(Error::Timeout("t".into()))), + Outcome::Timeout, + ); + assert_eq!( + chunk_get_outcome(&Err(Error::Network("n".into()))), + Outcome::NetworkError, + ); + + // Unexpected error variant (e.g. Protocol) — propagates out of + // chunk_get to the caller and is not a capacity signal. + assert_eq!( + chunk_get_outcome(&Err(Error::Protocol("p".into()))), + Outcome::ApplicationError, + ); + } + #[test] fn single_node_proof_uses_store_response_timeout() { let timeout = diff --git a/ant-core/src/data/client/data.rs b/ant-core/src/data/client/data.rs index a42dc25d..86b0dcc4 100644 --- a/ant-core/src/data/client/data.rs +++ b/ant-core/src/data/client/data.rs @@ -340,15 +340,12 @@ impl Client { &fetch_limiter, addresses.into_iter().enumerate(), |(idx, address)| { - let limiter = fetch_limiter.clone(); async move { - let chunk = observe_op( - &limiter, - || async move { self.chunk_get(&address).await }, - classify_error, - ) - .await? - .ok_or_else(|| { + // chunk_get_observed feeds the adaptive fetch + // limiter once per call via chunk_get_outcome + // (Ok(None) -> Timeout is the load-shedding + // signal for sustained close-group exhaustion). + let chunk = self.chunk_get_observed(&address).await?.ok_or_else(|| { Error::InvalidData(format!( "Missing chunk {} required for data reconstruction", hex::encode(address) diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 9c2c9188..eabcb522 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -10,7 +10,7 @@ //! //! For in-memory data uploads, see the `data` module. -use crate::data::client::adaptive::observe_op; +use crate::data::client::adaptive::{observe_op, rebucketed_unordered}; use crate::data::client::batch::{ finalize_batch_payment, PaymentIntent, PreparedChunk, WaveAggregateStats, }; @@ -1821,57 +1821,59 @@ impl Client { let prog = progress_ref.clone(); let limiter = fetch_limiter.clone(); handle.block_on(async { - // Clamp fan-out to batch size — self_encryption - // requests small batches (default 10), so a - // higher cap from the controller would be slots - // we never fill (see PERF-RESULTS.md). - let cap = limiter.current().min(batch_owned.len().max(1)); - let mut stream = futures::stream::iter(batch_owned) - .map(|(idx, hash)| { - let addr = hash.0; - let limiter = limiter.clone(); + // Use rebucketed_unordered so the in-flight cap + // is re-read from the limiter as each slot frees. + // `buffer_unordered` snapshots the cap once at + // pipeline build, which means observe_op + // signals from inside chunk_get cannot reduce + // concurrency on the current batch — exactly + // the case where load-shedding is needed. + let mut results = rebucketed_unordered( + &limiter, + batch_owned, + |(idx, hash): (usize, XorName)| { + let counter = counter.clone(); + let prog = prog.clone(); async move { - let result = observe_op( - &limiter, - || async move { self.chunk_get(&addr).await }, - classify_error, - ) - .await; - (idx, hash, result) + let addr = hash.0; + // chunk_get_observed feeds the + // adaptive fetch limiter once per + // call via chunk_get_outcome + // (Ok(None) -> Timeout is the + // load-shedding signal for + // sustained close-group exhaustion). + let chunk = self + .chunk_get_observed(&addr) + .await + .map_err(|e| { + self_encryption::Error::Generic(format!( + "DataMap resolution failed: {e}" + )) + })? + .ok_or_else(|| { + self_encryption::Error::Generic(format!( + "DataMap chunk not found: {}", + hex::encode(addr) + )) + })?; + let fetched = counter + .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + + 1; + if let Some(ref tx) = prog { + let _ = + tx.try_send(DownloadEvent::MapChunkFetched { fetched }); + } + Ok::<_, self_encryption::Error>((idx, chunk.content)) } - }) - .buffer_unordered(cap); - let mut results = Vec::new(); - while let Some((idx, hash, result)) = - futures::StreamExt::next(&mut stream).await - { - let chunk = result - .map_err(|e| { - self_encryption::Error::Generic(format!( - "DataMap resolution failed: {e}" - )) - })? - .ok_or_else(|| { - self_encryption::Error::Generic(format!( - "DataMap chunk not found: {}", - hex::encode(hash.0) - )) - })?; - results.push((idx, chunk.content)); - let fetched = - counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1; - if let Some(ref tx) = prog { - let _ = tx.try_send(DownloadEvent::MapChunkFetched { fetched }); - } - } + }, + ) + .await?; // CRITICAL: self_encryption::get_root_data_map_parallel // pairs the returned Vec POSITIONALLY with the input - // hashes via .zip() and discards our idx field. We - // used buffer_unordered, so completions arrive in - // arbitrary order. Sort by idx to restore the input - // order before returning, otherwise chunk bytes get - // paired with the wrong hashes and the root data - // map is corrupted. + // hashes via .zip() and discards our idx field. + // rebucketed_unordered preserves first-completion + // order, so sort by idx to restore input order + // before returning. results.sort_by_key(|(idx, _)| *idx); Ok(results) }) @@ -1929,51 +1931,161 @@ impl Client { tokio::task::block_in_place(|| { handle.block_on(async { - // Clamp fan-out to batch size — see PERF-RESULTS.md. - let cap = fetch_limiter.current().min(batch_owned.len().max(1)); - let mut stream = futures::stream::iter(batch_owned) - .map(|(idx, hash)| { - let addr = hash.0; - let limiter = fetch_limiter.clone(); + // First pass: try every chunk in the batch via + // chunk_get_observed (which already does its own + // first-attempt + retry sweep). A chunk that + // returns Ok(None) here is NOT a fatal failure + // — it's a candidate for a deferred retry below. + // We carry the chunk's XorName through so the + // retry pass can re-fetch by address. + // + // The closure ONLY returns Err on a true + // protocol/network error from chunk_get (the + // Err variant). Ok(None) is encoded as + // `Err(addr)` in the inner Result so the outer + // rebucketed pass doesn't early-abort on it. + type BatchEntry = + (usize, std::result::Result); + let raw: Vec = rebucketed_unordered( + &fetch_limiter, + batch_owned, + |(idx, hash): (usize, XorName)| { + let fetched_ref = fetched_ref.clone(); + let progress_ref = progress_ref.clone(); async move { - let result = observe_op( - &limiter, - || async move { self.chunk_get(&addr).await }, - classify_error, - ) - .await; - (idx, hash, result) + let addr = hash.0; + let addr_hex = hex::encode(addr); + match self.chunk_get_observed(&addr).await { + Ok(Some(chunk)) => { + let fetched = fetched_ref.fetch_add( + 1, + std::sync::atomic::Ordering::Relaxed, + ) + 1; + info!("Downloaded {fetched}/{total_chunks}"); + if let Some(ref tx) = progress_ref { + let _ = tx.try_send( + DownloadEvent::ChunksFetched { + fetched, + total: total_chunks, + }, + ); + } + Ok::(( + idx, + Ok(chunk.content), + )) + } + // chunk_get returned Ok(None): defer + // this chunk for a later retry rather + // than aborting the whole batch. + Ok(None) => Ok((idx, Err(hash))), + Err(e) => Err(self_encryption::Error::Generic( + format!( + "Network fetch failed for {addr_hex}: {e}" + ), + )), + } } - }) - .buffer_unordered(cap); - - let mut results = Vec::new(); - while let Some((idx, hash, result)) = - futures::StreamExt::next(&mut stream).await - { - let addr_hex = hex::encode(hash.0); - let chunk = result - .map_err(|e| { - self_encryption::Error::Generic(format!( - "Network fetch failed for {addr_hex}: {e}" + }, + ) + .await?; + + // Partition: things we already have vs the + // deferred set we need to retry. + let mut results: Vec<(usize, bytes::Bytes)> = Vec::new(); + let mut deferred: Vec<(usize, XorName)> = Vec::new(); + for (idx, inner) in raw { + match inner { + Ok(bytes) => results.push((idx, bytes)), + Err(hash) => deferred.push((idx, hash)), + } + } + + // Deferred retry pass: if any chunks were + // deferred, retry them serially with sleeps + // between attempts. The point is to ride out + // transient saturation events that hit + // multiple in-flight chunks at once — by the + // time the rest of the batch has drained and + // we sleep, the link has usually settled. + // Each chunk gets up to DEFERRED_RETRY_ATTEMPTS + // tries. + if !deferred.is_empty() { + const DEFERRED_RETRY_ATTEMPTS: usize = 3; + const DEFERRED_RETRY_DELAYS_SECS: [u64; 3] = [10, 30, 60]; + info!( + "Deferring {} chunk(s) for retry after batch settles", + deferred.len() + ); + + for (idx, hash) in deferred { + let addr = hash.0; + let addr_hex = hex::encode(addr); + let mut got: Option = None; + for (attempt, &delay_secs) in DEFERRED_RETRY_DELAYS_SECS + .iter() + .enumerate() + .take(DEFERRED_RETRY_ATTEMPTS) + { + tokio::time::sleep(std::time::Duration::from_secs( + delay_secs, )) - })? - .ok_or_else(|| { + .await; + info!( + "Deferred retry attempt {}/{} for {addr_hex}", + attempt + 1, + DEFERRED_RETRY_ATTEMPTS, + ); + match self.chunk_get_observed(&addr).await { + Ok(Some(chunk)) => { + got = Some(chunk.content); + break; + } + Ok(None) => continue, + // Per-attempt errors here (e.g. + // `Error::InsufficientPeers` when + // `close_group_peers` returns empty + // because the DHT walk hasn't + // settled) must NOT abort the + // download — they are transient and + // the next attempt's longer sleep + // is the right thing to ride them + // out. Log and fall through to the + // next attempt. + Err(e) => { + info!( + "Deferred retry attempt {}/{} for {addr_hex} hit transient error: {e}; \ + will fall through to next attempt", + attempt + 1, + DEFERRED_RETRY_ATTEMPTS, + ); + continue; + } + } + } + let content = got.ok_or_else(|| { self_encryption::Error::Generic(format!( - "Chunk not found: {addr_hex}" + "Chunk not found after {} deferred retry attempts: {addr_hex}", + DEFERRED_RETRY_ATTEMPTS )) })?; - results.push((idx, chunk.content)); - let fetched = - fetched_ref.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1; - info!("Downloaded {fetched}/{total_chunks}"); - if let Some(ref tx) = progress_ref { - let _ = tx.try_send(DownloadEvent::ChunksFetched { - fetched, - total: total_chunks, - }); + let fetched = fetched_ref.fetch_add( + 1, + std::sync::atomic::Ordering::Relaxed, + ) + 1; + info!( + "Downloaded {fetched}/{total_chunks} (deferred retry)" + ); + if let Some(ref tx) = progress_ref { + let _ = tx.try_send(DownloadEvent::ChunksFetched { + fetched, + total: total_chunks, + }); + } + results.push((idx, content)); } } + // streaming_decrypt itself sort_by_keys before // zipping, but the same closure is also passed // through get_root_data_map_parallel internally From eccc7c029416b0bd50dc0b864131e3a4841bc6a7 Mon Sep 17 00:00:00 2001 From: Chris O'Neil Date: Sun, 24 May 2026 22:48:59 +0100 Subject: [PATCH 2/4] fix(download): let fetch cap reach the ceiling on fast-but-lossy links MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit fixed residential saturation and abort-on-first- failure, but field reports from fast connections (e.g. an Oracle VPS that gets full speed on the released client) showed the opposite problem: the fetch cap stayed pinned at ~13-24 across an entire 36-file run and never climbed toward the 256 ceiling, so multi-GB files took ~22 min each instead of ~5. Three compounding causes, all from having tuned exclusively for the saturated-home case: 1. Cap can't grow. AIMD exits slow-start permanently on the first Decrease, then grows +1 per 32-observation window. On a link with a steady ~4% close-group-exhaustion trickle, intermittent Decreases fire often enough that additive +1 never gets ahead — equilibrium ~20. Additive growth simply cannot reach a useful cap from a low base before a file finishes. Fix: add `LimiterConfig::slow_start_ramp_threshold`. Below it, a Decrease still halves the cap but keeps slow-start armed, so the next healthy window doubles back up instead of crawling. The fetch channel sets it to the channel ceiling, so download concurrency tracks the connection's real capacity. Default 0 preserves the original behaviour for quote/store. 2. The p95-latency Decrease misfires on fetch. `chunk_get_observed`'s latency includes the internal 1 s retry sleep and the slow retry sweep for chunks that needed one, so a window with a couple of retry-path chunks has a wildly inflated p95 that reads as congestion. Fix: add `LimiterConfig::latency_decrease_enabled`, false for fetch. Genuine fetch congestion still surfaces via the Ok(None) -> Timeout rate, which the timeout_ceiling check catches. 3. The deferred-retry pass was a throughput sink. It retried deferred chunks SERIALLY with a mandatory 10 s pre-sleep each; a batch that deferred ~20 chunks burned minutes of near-zero throughput even though every chunk succeeded on its first retry (the 10 s sleep was pure waste — the deferrals were peer-side noise that clears in <1 s). Fix: retry deferred chunks in CONCURRENT rounds reusing the fetch limiter, with the first round firing immediately and later rounds backing off (0/15/45 s) only for chunks that survive a round. Both Ok(None) and transient errors re-defer to the next round; only the final round's leftovers are fatal. Quote and store channel behaviour is unchanged (threshold 0, latency-decrease enabled). New unit tests cover protected-vs-additive recovery, the disabled latency check, and that the controller applies the download tuning to fetch only. Co-Authored-By: Claude Opus 4.7 (1M context) --- ant-core/src/data/client/adaptive.rs | 177 ++++++++++++++++++++++++++- ant-core/src/data/client/file.rs | 163 +++++++++++++----------- 2 files changed, 265 insertions(+), 75 deletions(-) diff --git a/ant-core/src/data/client/adaptive.rs b/ant-core/src/data/client/adaptive.rs index e2259058..3c3c24fb 100644 --- a/ant-core/src/data/client/adaptive.rs +++ b/ant-core/src/data/client/adaptive.rs @@ -267,6 +267,30 @@ pub struct LimiterConfig { pub timeout_ceiling: f64, pub latency_inflation_factor: f64, pub latency_ewma_alpha: f64, + /// While `current < slow_start_ramp_threshold`, a Decrease halves + /// the cap but does NOT permanently exit slow-start — the next + /// healthy window can double the cap back up. Above the threshold, + /// a Decrease exits slow-start and the controller transitions to + /// classic AIMD (+1 per healthy window). + /// + /// 0 (the default) reproduces the original behaviour: any Decrease + /// at any cap permanently exits slow-start. The fetch channel sets + /// this to its `max_concurrency` so download concurrency keeps + /// doubling toward the ceiling instead of crawling +1 per window — + /// additive growth simply cannot reach a useful cap on a + /// fast-but-lossy link before the file finishes. See + /// `AdaptiveController::new`. + pub slow_start_ramp_threshold: usize, + /// When `false`, the p95-latency-vs-baseline comparison never + /// triggers a Decrease (it still updates the baseline). The fetch + /// channel disables it because `chunk_get`'s observed latency + /// includes the internal retry sleep and slow retry-sweep for the + /// chunks that needed one, so a window with a couple of retry-path + /// chunks has a wildly inflated p95 that is retry variance, not + /// congestion. Genuine fetch congestion still surfaces as a rising + /// `Ok(None)` (Timeout) rate, which the timeout_ceiling check + /// catches. + pub latency_decrease_enabled: bool, } impl LimiterConfig { @@ -281,6 +305,10 @@ impl LimiterConfig { timeout_ceiling: cfg.timeout_ceiling, latency_inflation_factor: cfg.latency_inflation_factor, latency_ewma_alpha: cfg.latency_ewma_alpha, + // Defaults preserve the original AIMD behaviour; the fetch + // channel overrides both in `AdaptiveController::new`. + slow_start_ramp_threshold: 0, + latency_decrease_enabled: true, } } @@ -471,10 +499,12 @@ fn evaluate( } if let Some(p95) = p95_of(&mut latencies) { - if let Some(base) = baseline { - let limit = base.mul_f64(cfg.latency_inflation_factor); - if p95 > limit { - return Decision::Decrease; + if cfg.latency_decrease_enabled { + if let Some(base) = baseline { + let limit = base.mul_f64(cfg.latency_inflation_factor); + if p95 > limit { + return Decision::Decrease; + } } } Decision::Increase @@ -522,7 +552,15 @@ fn apply_decision(inner: &mut LimiterInner, decision: Decision, cfg: &LimiterCon if inner.samples_since_decrease < cfg.min_window_ops { return; } - inner.left_slow_start = true; + // Below the ramp threshold we still halve (responsiveness is + // preserved) but keep slow-start armed, so the next healthy + // window can double the cap back up rather than crawling +1. + // Above the threshold a Decrease is the signal to settle into + // classic AIMD. With threshold=0 (quote/store) this is the + // original behaviour: any Decrease exits slow-start. + if inner.current >= cfg.slow_start_ramp_threshold { + inner.left_slow_start = true; + } let next = (inner.current / 2).max(cfg.min_concurrency); if next != inner.current { debug!(from = inner.current, to = next, "adaptive: decrease"); @@ -626,6 +664,27 @@ impl AdaptiveController { // Re-establish max >= min after the bump in case the channel // ceiling was somehow lower than the new floor. fetch_cfg.max_concurrency = fetch_cfg.max_concurrency.max(fetch_cfg.min_concurrency); + // Download-specific growth/decision tuning (see the field docs + // on `LimiterConfig`): + // + // - Keep slow-start armed all the way to the channel ceiling. + // Classic AIMD additive growth (+1 per healthy window) cannot + // reach a useful cap from a low base before a multi-GB file + // finishes — a fast-but-lossy connection (e.g. a VPS with a + // steady ~4% close-group-exhaustion trickle) was observed + // stuck at cap ~13-24 across 36 files because every transient + // Decrease permanently dropped it to additive growth. With + // the threshold at the ceiling, a Decrease still halves the + // cap but the next healthy window doubles it back, so the cap + // tracks the connection's real capacity instead of crawling. + // - Disable the p95-latency Decrease. chunk_get's observed + // latency includes the internal retry sleep + slow retry + // sweep for chunks that needed one, so a window with a couple + // of retry-path chunks shows a hugely inflated p95 that is + // retry variance, not congestion. Genuine fetch congestion + // still drives Decrease via the Ok(None) -> Timeout rate. + fetch_cfg.slow_start_ramp_threshold = fetch_cfg.max_concurrency; + fetch_cfg.latency_decrease_enabled = false; Self { quote: Limiter::new(start.quote, quote_cfg), store: Limiter::new(start.store, store_cfg), @@ -1047,6 +1106,8 @@ mod tests { timeout_ceiling: 0.2, latency_inflation_factor: 2.0, latency_ewma_alpha: 0.5, + slow_start_ramp_threshold: 0, + latency_decrease_enabled: true, } } @@ -1073,6 +1134,112 @@ mod tests { } } + #[test] + fn protected_slow_start_recovers_faster_than_additive() { + // After identical stress + recovery, a limiter with slow-start + // protected to the ceiling (fetch behaviour) must end at a + // higher cap than one that exits slow-start on first Decrease + // (quote/store behaviour): doubling outpaces +1-per-window. + let base = LimiterConfig { + max_concurrency: 256, + latency_decrease_enabled: false, + ..cfg_for_tests() + }; + let protected = Limiter::new( + 64, + LimiterConfig { + slow_start_ramp_threshold: 256, + ..base.clone() + }, + ); + let unprotected = Limiter::new( + 64, + LimiterConfig { + slow_start_ramp_threshold: 0, + ..base.clone() + }, + ); + + // Identical stress: a window of timeouts forces decreases on both. + for l in [&protected, &unprotected] { + for _ in 0..base.window_ops { + l.observe(Outcome::Timeout, Duration::from_millis(10)); + } + } + // Identical recovery: a long stretch of healthy windows. The + // protected limiter doubles each window; the unprotected one + // only adds 1. + for l in [&protected, &unprotected] { + for _ in 0..(base.window_ops * 10) { + l.observe(Outcome::Success, Duration::from_millis(10)); + } + } + assert!( + protected.current() > unprotected.current(), + "protected slow-start ({}) should recover faster than additive ({})", + protected.current(), + unprotected.current(), + ); + } + + #[test] + fn latency_decrease_disabled_ignores_p95_inflation() { + // With latency_decrease_enabled=false, a window of successes + // whose p95 latency is far above the baseline must NOT trigger + // a Decrease — only success/timeout rate can. (Fetch disables + // this because chunk_get's observed latency is polluted by + // retry-path variance.) + let cfg = LimiterConfig { + max_concurrency: 256, + slow_start_ramp_threshold: 256, + latency_decrease_enabled: false, + ..cfg_for_tests() + }; + let l = Limiter::new(16, cfg.clone()); + // Establish a fast baseline. + for _ in 0..cfg.window_ops { + l.observe(Outcome::Success, Duration::from_millis(5)); + } + let after_baseline = l.current(); + // Now a window of successes with 100x the latency. With the + // latency check disabled this is still a healthy window, so the + // cap must not drop. + for _ in 0..cfg.window_ops { + l.observe(Outcome::Success, Duration::from_millis(500)); + } + assert!( + l.current() >= after_baseline, + "latency inflation must not shrink the cap when the check is disabled: {} < {}", + l.current(), + after_baseline, + ); + } + + #[test] + fn controller_sets_fetch_channel_download_tuning() { + // AdaptiveController::new must apply the download-specific + // tuning to fetch only, leaving quote/store on classic AIMD. + let c = AdaptiveController::new(ChannelStart::default(), AdaptiveConfig::default()); + assert!( + !c.fetch.config.latency_decrease_enabled, + "fetch latency-decrease must be disabled", + ); + assert_eq!( + c.fetch.config.slow_start_ramp_threshold, c.fetch.config.max_concurrency, + "fetch slow-start must be protected to the ceiling", + ); + assert!( + c.quote.config.latency_decrease_enabled, + "quote must keep the latency-decrease check", + ); + assert_eq!( + c.quote.config.slow_start_ramp_threshold, 0, + "quote must keep classic AIMD slow-start exit", + ); + assert!(c.store.config.latency_decrease_enabled); + assert_eq!(c.store.config.slow_start_ramp_threshold, 0); + } + #[test] fn cold_start_clamps_into_bounds() { let cfg = cfg_for_tests(); diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index eabcb522..731c5f52 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -2001,88 +2001,111 @@ impl Client { } } - // Deferred retry pass: if any chunks were - // deferred, retry them serially with sleeps - // between attempts. The point is to ride out - // transient saturation events that hit - // multiple in-flight chunks at once — by the - // time the rest of the batch has drained and - // we sleep, the link has usually settled. - // Each chunk gets up to DEFERRED_RETRY_ATTEMPTS - // tries. + // Deferred retry pass: retry the deferred chunks + // in CONCURRENT rounds (reusing the fetch + // limiter's cap), not serially. The first round + // fires immediately — most deferrals on a + // healthy-but-lossy link are peer-side noise + // that clears in well under a second, and + // serializing them behind mandatory multi-second + // sleeps was the single biggest throughput sink + // on such links (a batch deferring ~20 chunks + // burned minutes of near-zero throughput even + // though every chunk succeeded on its first + // retry). Only chunks that survive a round get a + // longer back-off before the next, so genuine + // saturation still gets time to settle. if !deferred.is_empty() { - const DEFERRED_RETRY_ATTEMPTS: usize = 3; - const DEFERRED_RETRY_DELAYS_SECS: [u64; 3] = [10, 30, 60]; + // Round delays in seconds. Round 0 is + // immediate; later rounds back off to ride + // out sustained saturation. + const DEFERRED_ROUND_DELAYS_SECS: [u64; 3] = [0, 15, 45]; info!( - "Deferring {} chunk(s) for retry after batch settles", + "Deferring {} chunk(s) for concurrent retry after batch settles", deferred.len() ); - - for (idx, hash) in deferred { - let addr = hash.0; - let addr_hex = hex::encode(addr); - let mut got: Option = None; - for (attempt, &delay_secs) in DEFERRED_RETRY_DELAYS_SECS - .iter() - .enumerate() - .take(DEFERRED_RETRY_ATTEMPTS) - { + let mut remaining = deferred; + for (round, &delay_secs) in DEFERRED_ROUND_DELAYS_SECS + .iter() + .enumerate() + { + if remaining.is_empty() { + break; + } + if delay_secs > 0 { tokio::time::sleep(std::time::Duration::from_secs( delay_secs, )) .await; - info!( - "Deferred retry attempt {}/{} for {addr_hex}", - attempt + 1, - DEFERRED_RETRY_ATTEMPTS, - ); - match self.chunk_get_observed(&addr).await { - Ok(Some(chunk)) => { - got = Some(chunk.content); - break; - } - Ok(None) => continue, - // Per-attempt errors here (e.g. - // `Error::InsufficientPeers` when - // `close_group_peers` returns empty - // because the DHT walk hasn't - // settled) must NOT abort the - // download — they are transient and - // the next attempt's longer sleep - // is the right thing to ride them - // out. Log and fall through to the - // next attempt. - Err(e) => { - info!( - "Deferred retry attempt {}/{} for {addr_hex} hit transient error: {e}; \ - will fall through to next attempt", - attempt + 1, - DEFERRED_RETRY_ATTEMPTS, - ); - continue; - } - } } - let content = got.ok_or_else(|| { - self_encryption::Error::Generic(format!( - "Chunk not found after {} deferred retry attempts: {addr_hex}", - DEFERRED_RETRY_ATTEMPTS - )) - })?; - let fetched = fetched_ref.fetch_add( - 1, - std::sync::atomic::Ordering::Relaxed, - ) + 1; info!( - "Downloaded {fetched}/{total_chunks} (deferred retry)" + "Deferred retry round {}/{}: {} chunk(s)", + round + 1, + DEFERRED_ROUND_DELAYS_SECS.len(), + remaining.len(), ); - if let Some(ref tx) = progress_ref { - let _ = tx.try_send(DownloadEvent::ChunksFetched { - fetched, - total: total_chunks, - }); + let round_input = std::mem::take(&mut remaining); + let round_results: Vec = rebucketed_unordered( + &fetch_limiter, + round_input, + |(idx, hash): (usize, XorName)| { + let fetched_ref = fetched_ref.clone(); + let progress_ref = progress_ref.clone(); + async move { + let addr = hash.0; + // Both Ok(None) and a transient + // Err re-defer the chunk to the + // next round rather than + // aborting; only the final + // round's leftovers are fatal. + match self.chunk_get_observed(&addr).await { + Ok(Some(chunk)) => { + let fetched = fetched_ref.fetch_add( + 1, + std::sync::atomic::Ordering::Relaxed, + ) + 1; + info!( + "Downloaded {fetched}/{total_chunks} (deferred retry)" + ); + if let Some(ref tx) = progress_ref { + let _ = tx.try_send( + DownloadEvent::ChunksFetched { + fetched, + total: total_chunks, + }, + ); + } + Ok::(( + idx, + Ok(chunk.content), + )) + } + Ok(None) => Ok((idx, Err(hash))), + Err(e) => { + info!( + "Deferred retry for {} hit transient error: {e}; re-deferring", + hex::encode(addr) + ); + Ok((idx, Err(hash))) + } + } + } + }, + ) + .await?; + for (idx, inner) in round_results { + match inner { + Ok(bytes) => results.push((idx, bytes)), + Err(hash) => remaining.push((idx, hash)), + } } - results.push((idx, content)); + } + if let Some((_, hash)) = remaining.first() { + return Err(self_encryption::Error::Generic(format!( + "Chunk not found after {} deferred retry rounds: {}", + DEFERRED_ROUND_DELAYS_SECS.len(), + hex::encode(hash.0), + ))); } } From 112a7a6bb0e3d9271ff8736ebeb5b866e78a0984 Mon Sep 17 00:00:00 2001 From: Chris O'Neil Date: Mon, 25 May 2026 10:57:13 +0100 Subject: [PATCH 3/4] fix(adaptive): warm_start must keep fetch slow-start armed below the ceiling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit added slow-start protection so the fetch cap could double back up after transient Decreases — but a re-test on the same fast-but-lossy VPS showed the cap still pinned in the ~6-26 range, growing only additively. Root cause: `warm_start` unconditionally set `left_slow_start = true`. Every `ant file download` is a fresh process that warm-starts the controller from the persisted snapshot. With warm_start always exiting slow-start, the protection only ever applied to the very first (cold-start) file; every subsequent file began with slow-start already exited and could only grow the fetch cap by +1 per 32-observation window. Additive growth from a low warm value cannot climb to the ceiling against the connection's intermittent close-group-exhaustion trickle, so the cap drifted down across files (observed: 15, 8, 26, 20, 14, 12, 12, 9, 6...) and 2.5 GB files stayed at ~17-23 min. Fix: warm_start now sets `left_slow_start = clamped >= slow_start_ramp_threshold`. For quote/store (threshold 0) this is unchanged — always exits, so a learned warm value isn't doubled on the first healthy window. For fetch (threshold == ceiling) a warm value below the ceiling keeps slow-start armed, so the cap doubles back toward the connection's real capacity instead of crawling. This was the missing piece: the in-process slow-start protection and the cross-process warm-start path have to agree, or the multi-file CLI pattern silently defeats the protection. Added a regression test covering both the protected (doubles after warm_start) and default (stays additive) channels. Co-Authored-By: Claude Opus 4.7 (1M context) --- ant-core/src/data/client/adaptive.rs | 77 +++++++++++++++++++++++++--- 1 file changed, 70 insertions(+), 7 deletions(-) diff --git a/ant-core/src/data/client/adaptive.rs b/ant-core/src/data/client/adaptive.rs index 3c3c24fb..a885e8a2 100644 --- a/ant-core/src/data/client/adaptive.rs +++ b/ant-core/src/data/client/adaptive.rs @@ -427,12 +427,27 @@ impl Limiter { } /// Replace the current cap with `start`, clamped. Used for warm - /// loads from persisted state. Marks the limiter as having - /// already-left-slow-start so a single healthy window doesn't - /// double the cap (an over-aggressive cold-start from a warm - /// value). Subsequent increases are +1 per healthy window. - /// Does not clear the sliding window — fresh observations remain - /// authoritative for adaptation decisions. + /// loads from persisted state. Does not clear the sliding window — + /// fresh observations remain authoritative for adaptation + /// decisions. + /// + /// Slow-start state after a warm load depends on the channel's + /// `slow_start_ramp_threshold`: + /// + /// - Default (threshold 0, i.e. quote/store): mark slow-start as + /// already-left so a single healthy window doesn't *double* a + /// learned warm value — an over-aggressive jump. Subsequent + /// increases are +1 per healthy window. + /// - Protected (threshold > clamped, i.e. fetch below the ceiling): + /// keep slow-start armed. This is critical for the CLI usage + /// pattern where every `ant file download` is a fresh process + /// that warm-starts from the snapshot: if warm_start always + /// exited slow-start, the fetch cap could only ever grow + /// additively from the persisted value, which cannot climb back + /// to the ceiling against an intermittent Decrease trickle (the + /// exact pin-at-~20 behaviour observed on a fast-but-lossy VPS). + /// Keeping slow-start armed lets the cap double back toward the + /// capacity the connection can actually sustain. pub fn warm_start(&self, start: usize) { let clamped = start.clamp( self.config.min_concurrency, @@ -440,7 +455,7 @@ impl Limiter { ); let mut g = lock(&self.inner); g.current = clamped; - g.left_slow_start = true; + g.left_slow_start = clamped >= self.config.slow_start_ramp_threshold; } /// Snapshot of the current cap for persistence. Cheap, lock-only. @@ -1134,6 +1149,54 @@ mod tests { } } + #[test] + fn warm_start_keeps_slow_start_armed_below_protected_threshold() { + // Regression guard for the CLI multi-file pattern: each + // `ant file download` is a fresh process that warm-starts from + // the persisted snapshot. If warm_start exited slow-start, the + // fetch cap could only grow additively from the warm value and + // could never climb back to the ceiling against an intermittent + // Decrease trickle. A protected limiter (threshold == max) that + // warm-starts BELOW the ceiling must keep slow-start armed so it + // doubles back up. + let cfg = LimiterConfig { + max_concurrency: 256, + slow_start_ramp_threshold: 256, + latency_decrease_enabled: false, + ..cfg_for_tests() + }; + let l = Limiter::new(64, cfg.clone()); + l.warm_start(20); + assert_eq!(l.current(), 20); + // A single healthy window should DOUBLE (slow-start armed), + // proving warm_start did not exit slow-start. + for _ in 0..cfg.window_ops { + l.observe(Outcome::Success, Duration::from_millis(10)); + } + assert_eq!( + l.current(), + 40, + "protected channel must double after warm_start, not crawl +1", + ); + + // Default channel (threshold 0): warm_start exits slow-start, + // so the same window only adds 1. + let default_cfg = LimiterConfig { + max_concurrency: 256, + ..cfg_for_tests() + }; + let d = Limiter::new(64, default_cfg.clone()); + d.warm_start(20); + for _ in 0..default_cfg.window_ops { + d.observe(Outcome::Success, Duration::from_millis(10)); + } + assert_eq!( + d.current(), + 21, + "default channel must stay additive after warm_start", + ); + } + #[test] fn protected_slow_start_recovers_faster_than_additive() { // After identical stress + recovery, a limiter with slow-start From c5823a694f403fa3edc7e25fd8d24c75a4ac9c02 Mon Sep 17 00:00:00 2001 From: Chris O'Neil Date: Tue, 26 May 2026 17:36:53 +0100 Subject: [PATCH 4/4] =?UTF-8?q?fix(download):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20defer=20first-pass=20errors,=20fix=20ceiling=20slow?= =?UTF-8?q?-start,=20require=20well-sampled=20NotFound?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses three findings from @mickvandijke's review: P1 (file.rs / chunk.rs): a transient error on a single chunk's INITIAL close-group lookup could still fail a whole download. chunk_get took `chunk_get_try_close_group(...).await?`, so an error from close_group_peers (e.g. a momentary DHT/InsufficientPeers failure) propagated before the retry path, and the main download pass aborted on any Err from chunk_get_observed. Two changes: - chunk_get now treats a first-attempt lookup error as a non-authoritative miss (zeroed outcome) and falls through to its retry path instead of propagating. - the main streaming-decrypt pass defers a chunk on Err (same as Ok(None)) rather than aborting; only a chunk that survives all deferred retry rounds is fatal. P2 (adaptive.rs): fetch slow-start protection was lost at the ceiling. With `slow_start_ramp_threshold = max_concurrency`, a Decrease while the cap sat at the ceiling satisfied `current >= threshold` and exited slow-start, so 256 -> 128 recovered by +1 windows instead of doubling — re-pinning a fast-but-lossy link after a single transient decrease. Set the fetch threshold to `usize::MAX` so slow-start never exits and the cap always doubles back. P3 (chunk.rs): authoritative NotFound only required `not_found == queried`, so a thin/under-sampled DHT walk (close_group_peers accepts any non-empty result) returning 1/1 or 3/3 NotFound was treated as final absence with no retry, even though the real replica majority may lie outside that narrow view. Now also requires `queried >= CLOSE_GROUP_MAJORITY`, so an under-sampled walk falls through to the retry (which re-walks the DHT). Tests: updated the NotFound test for the well-sampled requirement; added a ceiling slow-start regression test (MAX-threshold out-recovers a max_concurrency-threshold limiter after stress at the ceiling). 301 ant-core unit tests pass; fmt and clippy --all-targets -D warnings clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- ant-core/src/data/client/adaptive.rs | 77 +++++++++++++++--- ant-core/src/data/client/chunk.rs | 116 ++++++++++++++++++--------- ant-core/src/data/client/file.rs | 19 +++-- 3 files changed, 154 insertions(+), 58 deletions(-) diff --git a/ant-core/src/data/client/adaptive.rs b/ant-core/src/data/client/adaptive.rs index a885e8a2..23368ed3 100644 --- a/ant-core/src/data/client/adaptive.rs +++ b/ant-core/src/data/client/adaptive.rs @@ -682,23 +682,28 @@ impl AdaptiveController { // Download-specific growth/decision tuning (see the field docs // on `LimiterConfig`): // - // - Keep slow-start armed all the way to the channel ceiling. - // Classic AIMD additive growth (+1 per healthy window) cannot - // reach a useful cap from a low base before a multi-GB file - // finishes — a fast-but-lossy connection (e.g. a VPS with a - // steady ~4% close-group-exhaustion trickle) was observed - // stuck at cap ~13-24 across 36 files because every transient - // Decrease permanently dropped it to additive growth. With - // the threshold at the ceiling, a Decrease still halves the - // cap but the next healthy window doubles it back, so the cap - // tracks the connection's real capacity instead of crawling. + // - Never exit slow-start. Classic AIMD additive growth (+1 per + // healthy window) cannot reach a useful cap from a low base + // before a multi-GB file finishes — a fast-but-lossy + // connection (e.g. a VPS with a steady ~4% close-group- + // exhaustion trickle) was observed stuck at cap ~13-24 across + // 36 files because every transient Decrease permanently + // dropped it to additive growth. `usize::MAX` keeps slow-start + // armed at every cap including the ceiling, so a Decrease + // (e.g. 256 -> 128) still halves but the next healthy window + // doubles it back. The cap therefore tracks the connection's + // real capacity instead of crawling, and a single transient + // Decrease near the ceiling can't re-pin the link to additive + // recovery. (A threshold == max_concurrency would NOT achieve + // this: `current >= threshold` is true at the ceiling, so a + // Decrease there would exit slow-start.) // - Disable the p95-latency Decrease. chunk_get's observed // latency includes the internal retry sleep + slow retry // sweep for chunks that needed one, so a window with a couple // of retry-path chunks shows a hugely inflated p95 that is // retry variance, not congestion. Genuine fetch congestion // still drives Decrease via the Ok(None) -> Timeout rate. - fetch_cfg.slow_start_ramp_threshold = fetch_cfg.max_concurrency; + fetch_cfg.slow_start_ramp_threshold = usize::MAX; fetch_cfg.latency_decrease_enabled = false; Self { quote: Limiter::new(start.quote, quote_cfg), @@ -1197,6 +1202,51 @@ mod tests { ); } + #[test] + fn slow_start_stays_armed_at_ceiling_with_max_threshold() { + // Regression for the "lost protection at the ceiling" bug. + // threshold == usize::MAX (the fetch setting) keeps slow-start + // armed even when a Decrease fires at the ceiling, so the cap + // doubles back. threshold == max_concurrency (the buggy + // setting) would exit slow-start there — `current >= threshold` + // is true at the ceiling — and recover only additively. After + // identical stress-at-ceiling + recovery, the MAX-threshold + // limiter must end strictly higher. + let base = LimiterConfig { + max_concurrency: 256, + latency_decrease_enabled: false, + ..cfg_for_tests() + }; + let fixed = Limiter::new( + 256, + LimiterConfig { + slow_start_ramp_threshold: usize::MAX, + ..base.clone() + }, + ); + let buggy = Limiter::new( + 256, + LimiterConfig { + slow_start_ramp_threshold: 256, + ..base.clone() + }, + ); + for l in [&fixed, &buggy] { + for _ in 0..base.window_ops { + l.observe(Outcome::Timeout, Duration::from_millis(10)); + } + for _ in 0..(base.window_ops * 10) { + l.observe(Outcome::Success, Duration::from_millis(10)); + } + } + assert!( + fixed.current() > buggy.current(), + "MAX-threshold limiter ({}) must out-recover the ceiling-threshold one ({})", + fixed.current(), + buggy.current(), + ); + } + #[test] fn protected_slow_start_recovers_faster_than_additive() { // After identical stress + recovery, a limiter with slow-start @@ -1288,8 +1338,9 @@ mod tests { "fetch latency-decrease must be disabled", ); assert_eq!( - c.fetch.config.slow_start_ramp_threshold, c.fetch.config.max_concurrency, - "fetch slow-start must be protected to the ceiling", + c.fetch.config.slow_start_ramp_threshold, + usize::MAX, + "fetch slow-start must never exit (armed at every cap incl. ceiling)", ); assert!( c.quote.config.latency_decrease_enabled, diff --git a/ant-core/src/data/client/chunk.rs b/ant-core/src/data/client/chunk.rs index 6424da3d..ce5887f4 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -29,11 +29,13 @@ const CHUNK_DATA_TYPE: u32 = 0; /// /// Either we got the chunk from some peer, or every peer in the group /// returned NotFound, timed out, or hit a transport / protocol error. -/// The counts are used to decide whether to retry: only a *unanimous* -/// NotFound response from every queried peer counts as authoritative -/// data absence — anything else leaves room for the actual storer to -/// be in the timeout / network-error / protocol-error bucket and is -/// worth one retry against a freshly re-walked close group. +/// The counts feed the retry decision (`is_authoritative_not_found`): +/// only a *unanimous* NotFound from a *well-sampled* close group counts +/// as authoritative data absence — anything else (a non-unanimous +/// result, or a thin/under-sampled DHT walk) leaves room for the actual +/// storer to be in the timeout / network-error / protocol-error bucket +/// or outside the sampled view, and is worth a retry against a freshly +/// re-walked close group. struct CloseGroupOutcome { chunk: Option, queried: usize, @@ -49,24 +51,30 @@ struct CloseGroupOutcome { protocol_err: usize, } -/// `true` if every peer we managed to query responded with an -/// authoritative NotFound. Only then is it safe to conclude the chunk -/// is genuinely absent from this close group. +/// `true` if the close-group sweep is strong enough evidence to +/// conclude the chunk is genuinely absent, so retrying is pointless. /// -/// An earlier version of this check used a majority quorum -/// (`not_found >= close_group_size / 2 + 1`), but production traffic -/// disproved that threshold: on the production network the storage -/// replication target is `CLOSE_GROUP_MAJORITY` (4) of the K=7 -/// close-group peers — so up to 3 peers legitimately don't store any -/// given chunk. A saturated client that sees `not_found=4 timeout=3` -/// is almost certainly looking at "3 storers we couldn't reach" plus -/// "4 non-storers that responded promptly," not authoritative data -/// loss. Reproduced on PROD-LOCAL-DL-03: a chunk that failed at the -/// quorum threshold during `ant file download` was successfully -/// retrieved on the same host moments later via a single -/// `ant chunk get`. Only `not_found == queried` is a safe stop. +/// Two conditions, both required: +/// +/// 1. *Unanimous*: every peer we managed to query responded with an +/// authoritative NotFound (`not_found == queried`). An earlier +/// version used a majority quorum (`not_found >= close_group_size / +/// 2 + 1`), but production traffic disproved that: storage +/// replicates to `CLOSE_GROUP_MAJORITY` (4) of the K=7 close-group +/// peers, so up to 3 peers legitimately don't store any given chunk +/// and a `not_found=4 timeout=3` result is "3 storers we couldn't +/// reach" plus "4 non-storers," not data loss. +/// +/// 2. *Well-sampled*: at least `CLOSE_GROUP_MAJORITY` peers were +/// queried. `close_group_peers` (via `find_closest_peers`) accepts +/// any non-empty DHT result, so a thin/under-sampled walk can return +/// 1 or 2 peers. A `1/1` or `3/3` NotFound from such a walk is NOT +/// authoritative — the real replica majority may sit entirely +/// outside that narrow view. Requiring a majority-sized sample means +/// a thin lookup falls through to the retry (which re-walks the DHT) +/// instead of being declared a final absence. fn is_authoritative_not_found(not_found: usize, queried: usize) -> bool { - queried > 0 && not_found == queried + queried >= CLOSE_GROUP_MAJORITY && not_found == queried } /// Store-response timeout for non-merkle chunk PUTs. @@ -367,8 +375,29 @@ impl Client { let addr_hex = hex::encode(address); - // First attempt against the current close-group view. - let first = self.chunk_get_try_close_group(address).await?; + // First attempt against the current close-group view. A + // lookup/transport error here (e.g. close_group_peers' DHT walk + // momentarily returning an error, or InsufficientPeers from a + // thin routing table) is NOT fatal: fall through to the retry + // path exactly as a non-authoritative miss would. Otherwise one + // transient error on the *initial* close-group walk for a single + // chunk would fail an entire multi-hundred-chunk download. A + // zeroed outcome (queried=0) is never authoritative, so it flows + // straight to the retry below. + let first = match self.chunk_get_try_close_group(address).await { + Ok(outcome) => outcome, + Err(e) => { + info!("chunk_get first close-group lookup failed for {addr_hex}: {e}; will retry"); + CloseGroupOutcome { + chunk: None, + queried: 0, + not_found: 0, + timeout: 0, + network_err: 0, + protocol_err: 0, + } + } + }; if let Some(chunk) = first.chunk { self.chunk_cache().put(chunk.address, chunk.content.clone()); return Ok(Some(chunk)); @@ -652,31 +681,38 @@ mod tests { const UNKNOWN_PROOF_TAG: u8 = 0xff; #[test] - fn authoritative_not_found_requires_unanimous_response() { - // Unanimous: every queried peer (7 of 7) said NotFound. This is - // the only safe stop. + fn authoritative_not_found_requires_unanimous_well_sampled_response() { + // Unanimous AND well-sampled: every queried peer of a full + // close group said NotFound. The only safe stop. assert!(is_authoritative_not_found(7, 7)); - - // 4-of-7 looks like authoritative absence on paper, but with - // CLOSE_GROUP_MAJORITY=4 replicas on the production network the - // expected response shape for a healthy chunk fetched from a - // saturated client is 3 NotFound (the non-storers) + 4 Timeout - // (the storers we couldn't reach in time). The mirror — 4 - // NotFound + 3 Timeout — is the same shape with one extra - // storer rotated out and one extra non-storer happening to be - // reachable. Refusing to retry that case was the regression - // PROD-LOCAL-DL-03 surfaced; it must NOT be treated as + // Unanimous with exactly a majority-sized sample is also // authoritative. + assert!(is_authoritative_not_found( + CLOSE_GROUP_MAJORITY, + CLOSE_GROUP_MAJORITY + )); + + // Unanimous but UNDER-sampled: a thin DHT walk returning 1 or 3 + // peers, all NotFound, is NOT authoritative — the real replica + // majority may sit entirely outside that narrow view. Must + // retry (re-walk the DHT). + assert!(!is_authoritative_not_found(1, 1)); + assert!(!is_authoritative_not_found(3, 3)); + assert!(!is_authoritative_not_found( + CLOSE_GROUP_MAJORITY - 1, + CLOSE_GROUP_MAJORITY - 1 + )); + + // Not unanimous: 4-of-7 / 6-of-7 NotFound leaves storers in the + // timeout bucket. Must retry. assert!(!is_authoritative_not_found(4, 7)); assert!(!is_authoritative_not_found(6, 7)); // Pure-reachability failure — must retry. assert!(!is_authoritative_not_found(0, 7)); - // Defensive: queried == 0 should never reach this check - // (close_group_peers would have errored earlier), but if it - // does, return false so we don't claim authoritative absence - // based on no evidence. + // Defensive: a zeroed outcome (e.g. the first attempt's + // close-group lookup errored) is never authoritative. assert!(!is_authoritative_not_found(0, 0)); } diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 731c5f52..a912befa 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -1979,11 +1979,20 @@ impl Client { // this chunk for a later retry rather // than aborting the whole batch. Ok(None) => Ok((idx, Err(hash))), - Err(e) => Err(self_encryption::Error::Generic( - format!( - "Network fetch failed for {addr_hex}: {e}" - ), - )), + // A transient error for one chunk + // (e.g. its close-group DHT walk + // erroring on this pass) must not + // abort a multi-hundred-chunk + // download. Defer it to the retry + // rounds, same as Ok(None); only a + // chunk that survives all deferred + // rounds is fatal. + Err(e) => { + info!( + "First-pass fetch error for {addr_hex}: {e}; deferring" + ); + Ok((idx, Err(hash))) + } } } },