diff --git a/ant-core/src/data/client/adaptive.rs b/ant-core/src/data/client/adaptive.rs index dac6895e..23368ed3 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, } } } @@ -230,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 { @@ -244,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, } } @@ -266,7 +331,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); @@ -362,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, @@ -375,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. @@ -434,10 +514,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 @@ -485,7 +567,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"); @@ -567,7 +657,54 @@ 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); + // Download-specific growth/decision tuning (see the field docs + // on `LimiterConfig`): + // + // - 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 = usize::MAX; + fetch_cfg.latency_decrease_enabled = false; Self { quote: Limiter::new(start.quote, quote_cfg), store: Limiter::new(start.store, store_cfg), @@ -989,6 +1126,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, } } @@ -1015,6 +1154,206 @@ 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 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 + // 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, + usize::MAX, + "fetch slow-start must never exit (armed at every cap incl. 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(); @@ -1256,10 +1595,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 +1620,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 +2290,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..ce5887f4 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,58 @@ 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 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, + 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 the close-group sweep is strong enough evidence to +/// conclude the chunk is genuinely absent, so retrying is pointless. +/// +/// 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 >= CLOSE_GROUP_MAJORITY && not_found == queried +} + /// Store-response timeout for non-merkle chunk PUTs. const STORE_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10); @@ -34,6 +87,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 +340,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 +373,136 @@ 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. 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)); + } + + // 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 +516,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 +680,81 @@ 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_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)); + // 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: a zeroed outcome (e.g. the first attempt's + // close-group lookup errored) is never authoritative. + 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..a912befa 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,193 @@ 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))), + // 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))) + } + } } - }) - .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}" - )) - })? - .ok_or_else(|| { - self_encryption::Error::Generic(format!( - "Chunk not found: {addr_hex}" + }, + ) + .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: 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() { + // 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 concurrent retry after batch settles", + deferred.len() + ); + 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, )) - })?; - 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, - }); + .await; + } + info!( + "Deferred retry round {}/{}: {} chunk(s)", + round + 1, + DEFERRED_ROUND_DELAYS_SECS.len(), + remaining.len(), + ); + 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)), + } + } + } + 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), + ))); } } + // streaming_decrypt itself sort_by_keys before // zipping, but the same closure is also passed // through get_root_data_map_parallel internally