From 675ca62fa9ea30f86a52e62889657923a9668578 Mon Sep 17 00:00:00 2001 From: Aman Kumar Date: Wed, 14 Oct 2026 18:53:20 -0400 Subject: [PATCH 1/6] Grow and jitter consecutive throttle waits in the retrier --- nativelink-store/src/azure_blob_store.rs | 3 +- nativelink-util/src/retry.rs | 44 +++++- nativelink-util/tests/retry_test.rs | 179 +++++++++++++++++++++++ 3 files changed, 223 insertions(+), 3 deletions(-) diff --git a/nativelink-store/src/azure_blob_store.rs b/nativelink-store/src/azure_blob_store.rs index 0de1fd508b9..2056767eaf2 100644 --- a/nativelink-store/src/azure_blob_store.rs +++ b/nativelink-store/src/azure_blob_store.rs @@ -227,7 +227,8 @@ where spec.common.retry.clone(), ) .with_client("azure") - .with_throttle_budget(throttle_retry_budget), + .with_throttle_budget(throttle_retry_budget) + .with_throttle_max_wait(throttle_retry_max_delay), consider_expired_after_s: i64::from(spec.common.consider_expired_after_s), max_retry_buffer_per_request: spec .common diff --git a/nativelink-util/src/retry.rs b/nativelink-util/src/retry.rs index e1a67e53499..0b50ed13e87 100644 --- a/nativelink-util/src/retry.rs +++ b/nativelink-util/src/retry.rs @@ -99,6 +99,10 @@ pub struct Retrier { /// How long one `retry` call may spend on `RetryAfter` waits before /// they start spending attempts. Zero means they always do. throttle_budget: Duration, + /// Longest a throttle wait may grow to. Set, each consecutive throttle + /// doubles the wait the store asked for, capped here, and jitters it. + /// Unset, the wait is what was asked. + throttle_max_wait: Option, } impl core::fmt::Debug for Retrier { @@ -140,6 +144,7 @@ impl Retrier { client: "unknown", outcome_fn: None, throttle_budget: Duration::ZERO, + throttle_max_wait: None, } } @@ -167,6 +172,33 @@ impl Retrier { self } + /// Lets consecutive throttles wait longer each time: the n-th in a row + /// waits what the server asked times `2^(n-1)`, no more than + /// `max_wait`, then jittered, and never less than what was asked. A + /// store that is told to slow down with no `Retry-After` otherwise has + /// every client retry in lockstep at the configured floor. + #[must_use] + pub const fn with_throttle_max_wait(mut self, max_wait: Duration) -> Self { + self.throttle_max_wait = Some(max_wait); + self + } + + /// The wait for the `n`-th consecutive throttle that asked for + /// `min_wait`. + fn throttle_wait(&self, min_wait: Duration, throttles_in_a_row: u32) -> Duration { + let Some(max_wait) = self.throttle_max_wait else { + return min_wait; + }; + let grown = min_wait + .checked_mul( + 1u32.checked_shl(throttles_in_a_row.saturating_sub(1)) + .unwrap_or(u32::MAX), + ) + .unwrap_or(Duration::MAX) + .min(max_wait); + (self.jitter_fn)(grown).max(min_wait) + } + fn report(&self, outcome: RetryOutcome) { record_retry_outcome(self.client, outcome.as_str()); if let Some(outcome_fn) = &self.outcome_fn { @@ -225,6 +257,7 @@ impl Retrier { tokio::pin!(operation); let mut attempt = 0; let mut throttled = Duration::ZERO; + let mut throttles_in_a_row: u32 = 0; loop { attempt += 1; let (err, min_wait, is_throttle) = match operation.next().await { @@ -245,8 +278,15 @@ impl Retrier { Some(RetryResult::Err(e)) => { return Err(e.append(format!("On attempt {attempt}"))); } - Some(RetryResult::Retry(err)) => (err, Duration::ZERO, false), - Some(RetryResult::RetryAfter(err, min_wait)) => (err, min_wait, true), + Some(RetryResult::Retry(err)) => { + throttles_in_a_row = 0; + (err, Duration::ZERO, false) + } + Some(RetryResult::RetryAfter(err, min_wait)) => { + throttles_in_a_row = throttles_in_a_row.saturating_add(1); + let wait = self.throttle_wait(min_wait, throttles_in_a_row); + (err, wait, true) + } }; if !self.should_retry(err.code) { if err.code == Code::NotFound { diff --git a/nativelink-util/tests/retry_test.rs b/nativelink-util/tests/retry_test.rs index daa3d43010c..b43551e2433 100644 --- a/nativelink-util/tests/retry_test.rs +++ b/nativelink-util/tests/retry_test.rs @@ -535,3 +535,182 @@ async fn throttles_within_the_budget_report_retries_and_recovery() -> Result<(), ); Ok(()) } + +fn growing_retrier( + max_retries: usize, + budget: Duration, + cap: Duration, + sleeps: &Arc>>, +) -> Retrier { + let sleeps = Arc::clone(sleeps); + Retrier::new( + Arc::new(move |duration| { + sleeps.lock().unwrap().push(duration); + Box::pin(ready(())) + }), + Arc::new(|delay| delay), + Retry { + max_retries, + ..Default::default() + }, + ) + .with_throttle_budget(budget) + .with_throttle_max_wait(cap) +} + +/// With a throttle cap the n-th consecutive throttle waits the ask times +/// `2^(n-1)`, no more than the cap, and the budget counts the grown waits. +#[nativelink_test] +async fn consecutive_throttles_double_their_wait_up_to_the_cap() -> Result<(), Error> { + let sleeps = Arc::new(Mutex::new(Vec::new())); + let retrier = growing_retrier( + 1, + Duration::from_secs(100), + Duration::from_secs(30), + &sleeps, + ); + let run_count = Arc::new(AtomicI32::new(0)); + + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + if run_count.fetch_add(1, Ordering::Relaxed) < 6 { + return RetryResult::RetryAfter( + make_err!(Code::Unavailable, "Throttled"), + Duration::from_secs(2), + ); + } + RetryResult::Ok(true) + })) + .await?; + + assert_eq!(result, true); + assert_eq!(run_count.load(Ordering::Relaxed), 7); + assert_eq!( + *sleeps.lock().unwrap(), + [2, 4, 8, 16, 30, 30] + .into_iter() + .map(Duration::from_secs) + .collect::>(), + "each throttle doubles the last, capped" + ); + Ok(()) +} + +/// The grown waits are what the budget is spent on: 2 + 4 + 8 fits a 15 s +/// budget and the fourth throttle (16 s) falls through to the backoff +/// ladder, where it spends the one allowed attempt. +#[nativelink_test] +async fn grown_waits_spend_the_budget() -> Result<(), Error> { + let sleeps = Arc::new(Mutex::new(Vec::new())); + let retrier = growing_retrier(1, Duration::from_secs(15), Duration::from_secs(60), &sleeps); + let run_count = Arc::new(AtomicI32::new(0)); + + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + run_count.fetch_add(1, Ordering::Relaxed); + RetryResult::::RetryAfter( + make_err!(Code::Unavailable, "Throttled"), + Duration::from_secs(2), + ) + })) + .await; + + assert!(result.is_err()); + assert_eq!( + run_count.load(Ordering::Relaxed), + 5, + "three budgeted throttles, then one attempt and its retry" + ); + assert_eq!( + *sleeps.lock().unwrap(), + [2, 4, 8, 16] + .into_iter() + .map(Duration::from_secs) + .collect::>(), + "the fourth wait is the grown ask, longer than the 2 s backoff" + ); + Ok(()) +} + +/// A transient error that is not a throttle ends the streak: the next +/// throttle waits the plain ask again. +#[nativelink_test] +async fn a_plain_retry_resets_the_throttle_streak() -> Result<(), Error> { + let sleeps = Arc::new(Mutex::new(Vec::new())); + let retrier = growing_retrier( + 5, + Duration::from_secs(100), + Duration::from_secs(60), + &sleeps, + ); + let run_count = Arc::new(AtomicI32::new(0)); + + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + match run_count.fetch_add(1, Ordering::Relaxed) { + 0 | 1 | 3 => RetryResult::RetryAfter( + make_err!(Code::Unavailable, "Throttled"), + Duration::from_secs(2), + ), + 2 => RetryResult::Retry(make_err!(Code::Unavailable, "Blip")), + _ => RetryResult::Ok(true), + } + })) + .await?; + + assert_eq!(result, true); + let sleeps = sleeps.lock().unwrap(); + assert_eq!(sleeps[0], Duration::from_secs(2)); + assert_eq!(sleeps[1], Duration::from_secs(4)); + // sleeps[2] is the backoff ladder's first step (delay 0 by default). + assert_eq!( + sleeps[3], + Duration::from_secs(2), + "the streak restarted after the plain retry" + ); + Ok(()) +} + +/// Jitter never lands below what the server asked for. +#[nativelink_test] +async fn throttle_jitter_is_floored_at_the_ask() -> Result<(), Error> { + let sleeps = Arc::new(Mutex::new(Vec::new())); + let retrier = Retrier::new( + Arc::new({ + let sleeps = Arc::clone(&sleeps); + move |duration| { + sleeps.lock().unwrap().push(duration); + Box::pin(ready(())) + } + }), + // A jitter that always lands low. + Arc::new(|delay| delay / 10), + Retry { + max_retries: 1, + ..Default::default() + }, + ) + .with_throttle_budget(Duration::from_secs(100)) + .with_throttle_max_wait(Duration::from_secs(60)); + let run_count = Arc::new(AtomicI32::new(0)); + + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + if run_count.fetch_add(1, Ordering::Relaxed) < 2 { + return RetryResult::RetryAfter( + make_err!(Code::Unavailable, "Throttled"), + Duration::from_secs(10), + ); + } + RetryResult::Ok(true) + })) + .await?; + + assert_eq!(result, true); + assert_eq!( + *sleeps.lock().unwrap(), + vec![Duration::from_secs(10); 2], + "10 s asked, 1 s and 2 s jittered, 10 s waited" + ); + Ok(()) +} From 608d4c3ea35282cf6c7aba0c531a61947cf52500 Mon Sep 17 00:00:00 2001 From: Aman Kumar Date: Wed, 14 Oct 2026 19:57:41 -0400 Subject: [PATCH 2/6] Use from_mins for the minute-long waits in the retry tests --- nativelink-util/tests/retry_test.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/nativelink-util/tests/retry_test.rs b/nativelink-util/tests/retry_test.rs index b43551e2433..74a688ca145 100644 --- a/nativelink-util/tests/retry_test.rs +++ b/nativelink-util/tests/retry_test.rs @@ -602,7 +602,7 @@ async fn consecutive_throttles_double_their_wait_up_to_the_cap() -> Result<(), E #[nativelink_test] async fn grown_waits_spend_the_budget() -> Result<(), Error> { let sleeps = Arc::new(Mutex::new(Vec::new())); - let retrier = growing_retrier(1, Duration::from_secs(15), Duration::from_secs(60), &sleeps); + let retrier = growing_retrier(1, Duration::from_secs(15), Duration::from_mins(1), &sleeps); let run_count = Arc::new(AtomicI32::new(0)); let result = Pin::new(&retrier) @@ -640,7 +640,7 @@ async fn a_plain_retry_resets_the_throttle_streak() -> Result<(), Error> { let retrier = growing_retrier( 5, Duration::from_secs(100), - Duration::from_secs(60), + Duration::from_mins(1), &sleeps, ); let run_count = Arc::new(AtomicI32::new(0)); @@ -691,7 +691,7 @@ async fn throttle_jitter_is_floored_at_the_ask() -> Result<(), Error> { }, ) .with_throttle_budget(Duration::from_secs(100)) - .with_throttle_max_wait(Duration::from_secs(60)); + .with_throttle_max_wait(Duration::from_mins(1)); let run_count = Arc::new(AtomicI32::new(0)); let result = Pin::new(&retrier) From 7363e8ba69cfc0719544863b9a6ced45a430a938 Mon Sep 17 00:00:00 2001 From: Aman Kumar Date: Wed, 14 Oct 2026 20:00:39 -0400 Subject: [PATCH 3/6] Format the retry tests --- nativelink-util/tests/retry_test.rs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/nativelink-util/tests/retry_test.rs b/nativelink-util/tests/retry_test.rs index 74a688ca145..ca2f3d8be73 100644 --- a/nativelink-util/tests/retry_test.rs +++ b/nativelink-util/tests/retry_test.rs @@ -637,12 +637,7 @@ async fn grown_waits_spend_the_budget() -> Result<(), Error> { #[nativelink_test] async fn a_plain_retry_resets_the_throttle_streak() -> Result<(), Error> { let sleeps = Arc::new(Mutex::new(Vec::new())); - let retrier = growing_retrier( - 5, - Duration::from_secs(100), - Duration::from_mins(1), - &sleeps, - ); + let retrier = growing_retrier(5, Duration::from_secs(100), Duration::from_mins(1), &sleeps); let run_count = Arc::new(AtomicI32::new(0)); let result = Pin::new(&retrier) From e2b981d0e7112acd08ba7c4abab3ee4703e693aa Mon Sep 17 00:00:00 2001 From: Aman Kumar Date: Thu, 15 Oct 2026 07:53:05 -0400 Subject: [PATCH 4/6] Spell out that the throttle cap is a ceiling, not the wait --- nativelink-util/src/retry.rs | 34 ++++++++++++++--------------- nativelink-util/tests/retry_test.rs | 29 ++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 17 deletions(-) diff --git a/nativelink-util/src/retry.rs b/nativelink-util/src/retry.rs index 0b50ed13e87..06f47c9e4b8 100644 --- a/nativelink-util/src/retry.rs +++ b/nativelink-util/src/retry.rs @@ -172,31 +172,31 @@ impl Retrier { self } - /// Lets consecutive throttles wait longer each time: the n-th in a row - /// waits what the server asked times `2^(n-1)`, no more than - /// `max_wait`, then jittered, and never less than what was asked. A - /// store that is told to slow down with no `Retry-After` otherwise has - /// every client retry in lockstep at the configured floor. + /// Lets consecutive throttles wait longer each time. The first throttle + /// in a row waits what the server asked, the second twice that, the + /// third four times, and so on; `max_wait` is the ceiling that growth + /// stops at, not a wait of its own. The grown wait is jittered and never + /// less than what was asked. A store that is told to slow down with no + /// `Retry-After` otherwise has every client retry in lockstep at the + /// configured floor. #[must_use] pub const fn with_throttle_max_wait(mut self, max_wait: Duration) -> Self { self.throttle_max_wait = Some(max_wait); self } - /// The wait for the `n`-th consecutive throttle that asked for - /// `min_wait`. - fn throttle_wait(&self, min_wait: Duration, throttles_in_a_row: u32) -> Duration { + /// The wait for the `n`-th consecutive throttle whose server asked for + /// `asked`: `asked` itself for the first, doubled once per throttle + /// before it, capped at the configured ceiling. + fn throttle_wait(&self, asked: Duration, throttles_in_a_row: u32) -> Duration { let Some(max_wait) = self.throttle_max_wait else { - return min_wait; + return asked; }; - let grown = min_wait - .checked_mul( - 1u32.checked_shl(throttles_in_a_row.saturating_sub(1)) - .unwrap_or(u32::MAX), - ) - .unwrap_or(Duration::MAX) - .min(max_wait); - (self.jitter_fn)(grown).max(min_wait) + // One doubling per earlier throttle in the row; past 31 the + // multiplier would overflow and the ceiling applies anyway. + let doublings = throttles_in_a_row.saturating_sub(1).min(31); + let grown = asked.saturating_mul(1 << doublings).min(max_wait); + (self.jitter_fn)(grown).max(asked) } fn report(&self, outcome: RetryOutcome) { diff --git a/nativelink-util/tests/retry_test.rs b/nativelink-util/tests/retry_test.rs index ca2f3d8be73..929c566b58f 100644 --- a/nativelink-util/tests/retry_test.rs +++ b/nativelink-util/tests/retry_test.rs @@ -596,6 +596,35 @@ async fn consecutive_throttles_double_their_wait_up_to_the_cap() -> Result<(), E Ok(()) } +/// The cap is a ceiling, not the wait: a single throttle under a large cap +/// waits only what was asked. +#[nativelink_test] +async fn the_first_throttle_waits_the_ask_not_the_cap() -> Result<(), Error> { + let sleeps = Arc::new(Mutex::new(Vec::new())); + let retrier = growing_retrier(1, Duration::from_secs(100), Duration::from_mins(1), &sleeps); + let run_count = Arc::new(AtomicI32::new(0)); + + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + if run_count.fetch_add(1, Ordering::Relaxed) < 1 { + return RetryResult::RetryAfter( + make_err!(Code::Unavailable, "Throttled"), + Duration::from_secs(2), + ); + } + RetryResult::Ok(true) + })) + .await?; + + assert_eq!(result, true); + assert_eq!( + *sleeps.lock().unwrap(), + vec![Duration::from_secs(2)], + "one throttle waits the ask, not the 60 s cap" + ); + Ok(()) +} + /// The grown waits are what the budget is spent on: 2 + 4 + 8 fits a 15 s /// budget and the fourth throttle (16 s) falls through to the backoff /// ladder, where it spends the one allowed attempt. From 38313ba2c407878f9ac5eda580e95ae5e840f3ca Mon Sep 17 00:00:00 2001 From: Aman Kumar Date: Thu, 15 Oct 2026 09:15:52 -0400 Subject: [PATCH 5/6] Keep the throttle growth when the jitter lands below the ask, and test it through the Azure store --- .../tests/azure_blob_store_test.rs | 43 +++++++++++++++ nativelink-util/src/retry.rs | 9 +++- nativelink-util/tests/retry_test.rs | 52 +++++++++++++++++-- 3 files changed, 99 insertions(+), 5 deletions(-) diff --git a/nativelink-store/tests/azure_blob_store_test.rs b/nativelink-store/tests/azure_blob_store_test.rs index 685786b329a..3d919b3c9d3 100644 --- a/nativelink-store/tests/azure_blob_store_test.rs +++ b/nativelink-store/tests/azure_blob_store_test.rs @@ -553,6 +553,49 @@ async fn throttles_do_not_spend_the_retry_budget() -> Result<(), Error> { Ok(()) } +/// Consecutive throttles with no header double from the floor up to the +/// cap: 2, 4, 8, 16, 32 s for five of them under the 60 s default cap, 62 s +/// in all, where five floors would be 10 s. +#[nativelink_test(start_paused = true)] +async fn consecutive_throttles_double_from_the_floor_up_to_the_cap() -> Result<(), Error> { + const VALUE: &str = "test_content"; + + let mock = MockTransport::new(); + for _ in 0..5 { + mock.push_download(error_response(StatusCode::ServiceUnavailable)); + } + mock.push_download(download_ok(StatusCode::Ok, VALUE.as_bytes().to_vec())); + + let retry = Retry { + max_retries: 0, + delay: 0.0, + jitter: 0.0, + ..Default::default() + }; + let store = create_test_store(Arc::clone(&mock), Some(retry), None)?; + let floor = Duration::from_secs_f32(CommonObjectSpec::default().throttle_retry_delay_s); + let cap = Duration::from_secs_f32(CommonObjectSpec::default().throttle_retry_max_delay_s); + let doubled = floor + floor * 2 + floor * 4 + floor * 8 + floor * 16; + assert!( + floor * 16 < cap, + "the fifth wait must still be under the cap" + ); + + let started = tokio::time::Instant::now(); + let result = store + .get_part_unchunked(create_test_digest()?, 0, None) + .await?; + + assert_eq!(result, VALUE.as_bytes()); + assert_eq!(mock.request_count(), 6); + assert!( + started.elapsed() >= doubled && started.elapsed() < doubled + floor, + "waited {:?}, expected the doubled waits of {doubled:?}, not five floors", + started.elapsed() + ); + Ok(()) +} + #[nativelink_test(start_paused = true)] async fn a_throttled_download_without_a_header_waits_the_configured_floor() -> Result<(), Error> { const VALUE: &str = "test_content"; diff --git a/nativelink-util/src/retry.rs b/nativelink-util/src/retry.rs index 06f47c9e4b8..379e361476b 100644 --- a/nativelink-util/src/retry.rs +++ b/nativelink-util/src/retry.rs @@ -187,7 +187,7 @@ impl Retrier { /// The wait for the `n`-th consecutive throttle whose server asked for /// `asked`: `asked` itself for the first, doubled once per throttle - /// before it, capped at the configured ceiling. + /// before it, capped at the configured ceiling, then jittered. fn throttle_wait(&self, asked: Duration, throttles_in_a_row: u32) -> Duration { let Some(max_wait) = self.throttle_max_wait else { return asked; @@ -196,7 +196,12 @@ impl Retrier { // multiplier would overflow and the ceiling applies anyway. let doublings = throttles_in_a_row.saturating_sub(1).min(31); let grown = asked.saturating_mul(1 << doublings).min(max_wait); - (self.jitter_fn)(grown).max(asked) + // The jitter spreads the grown wait a little either way, as + // `Retry::make_jitter_fn` does. It may not undo the growth: a + // result below the ask (a jitter function that zeroes delays, as + // tests install) leaves the grown wait as it is. + let jittered = (self.jitter_fn)(grown); + if jittered < asked { grown } else { jittered } } fn report(&self, outcome: RetryOutcome) { diff --git a/nativelink-util/tests/retry_test.rs b/nativelink-util/tests/retry_test.rs index 929c566b58f..a10f9aa2d7a 100644 --- a/nativelink-util/tests/retry_test.rs +++ b/nativelink-util/tests/retry_test.rs @@ -625,6 +625,51 @@ async fn the_first_throttle_waits_the_ask_not_the_cap() -> Result<(), Error> { Ok(()) } +/// A jitter function that zeroes every delay, as tests install to keep +/// backoffs instant, cannot flatten the growth back to the ask. +#[nativelink_test] +async fn a_zeroing_jitter_does_not_undo_the_growth() -> Result<(), Error> { + let sleeps = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&sleeps); + let retrier = Retrier::new( + Arc::new(move |duration| { + recorded.lock().unwrap().push(duration); + Box::pin(ready(())) + }), + Arc::new(|_| Duration::ZERO), + Retry { + max_retries: 1, + ..Default::default() + }, + ) + .with_throttle_budget(Duration::from_secs(100)) + .with_throttle_max_wait(Duration::from_secs(30)); + let run_count = Arc::new(AtomicI32::new(0)); + + let result = Pin::new(&retrier) + .retry(repeat_with(|| { + if run_count.fetch_add(1, Ordering::Relaxed) < 4 { + return RetryResult::RetryAfter( + make_err!(Code::Unavailable, "Throttled"), + Duration::from_secs(2), + ); + } + RetryResult::Ok(true) + })) + .await?; + + assert_eq!(result, true); + assert_eq!( + *sleeps.lock().unwrap(), + [2, 4, 8, 16] + .into_iter() + .map(Duration::from_secs) + .collect::>(), + "the growth stands when the jitter returns nothing" + ); + Ok(()) +} + /// The grown waits are what the budget is spent on: 2 + 4 + 8 fits a 15 s /// budget and the fourth throttle (16 s) falls through to the backoff /// ladder, where it spends the one allowed attempt. @@ -695,7 +740,8 @@ async fn a_plain_retry_resets_the_throttle_streak() -> Result<(), Error> { Ok(()) } -/// Jitter never lands below what the server asked for. +/// A jitter that lands below what the server asked for is not applied: the +/// first throttle waits the ask, the second the doubled ask, not the tenth. #[nativelink_test] async fn throttle_jitter_is_floored_at_the_ask() -> Result<(), Error> { let sleeps = Arc::new(Mutex::new(Vec::new())); @@ -733,8 +779,8 @@ async fn throttle_jitter_is_floored_at_the_ask() -> Result<(), Error> { assert_eq!(result, true); assert_eq!( *sleeps.lock().unwrap(), - vec![Duration::from_secs(10); 2], - "10 s asked, 1 s and 2 s jittered, 10 s waited" + vec![Duration::from_secs(10), Duration::from_secs(20)], + "10 s asked, 1 s and 2 s jittered, the grown 10 s and 20 s waited" ); Ok(()) } From 1d160a839a050173ea05447240f37425960398d9 Mon Sep 17 00:00:00 2001 From: Aman Kumar Date: Fri, 16 Oct 2026 00:45:27 -0400 Subject: [PATCH 6/6] Grow only the budgeted throttle waits and spread the jitter upward --- nativelink-util/src/retry.rs | 52 ++++++++++++++++++----------- nativelink-util/tests/retry_test.rs | 19 ++++++----- 2 files changed, 42 insertions(+), 29 deletions(-) diff --git a/nativelink-util/src/retry.rs b/nativelink-util/src/retry.rs index 379e361476b..560bff3a4a9 100644 --- a/nativelink-util/src/retry.rs +++ b/nativelink-util/src/retry.rs @@ -175,10 +175,13 @@ impl Retrier { /// Lets consecutive throttles wait longer each time. The first throttle /// in a row waits what the server asked, the second twice that, the /// third four times, and so on; `max_wait` is the ceiling that growth - /// stops at, not a wait of its own. The grown wait is jittered and never - /// less than what was asked. A store that is told to slow down with no - /// `Retry-After` otherwise has every client retry in lockstep at the - /// configured floor. + /// stops at, not a wait of its own. The grown wait is spread upward by + /// the jitter and never less than what was asked. Growth applies to the + /// waits the throttle budget pays for; once the budget is spent the + /// backoff ladder takes over and waits at least the plain ask, so the + /// ladder's total stays what the retry config says. A store that is + /// told to slow down with no `Retry-After` otherwise has every client + /// retry in lockstep at the configured floor. #[must_use] pub const fn with_throttle_max_wait(mut self, max_wait: Duration) -> Self { self.throttle_max_wait = Some(max_wait); @@ -196,12 +199,17 @@ impl Retrier { // multiplier would overflow and the ceiling applies anyway. let doublings = throttles_in_a_row.saturating_sub(1).min(31); let grown = asked.saturating_mul(1 << doublings).min(max_wait); - // The jitter spreads the grown wait a little either way, as - // `Retry::make_jitter_fn` does. It may not undo the growth: a - // result below the ask (a jitter function that zeroes delays, as - // tests install) leaves the grown wait as it is. + // `Retry::make_jitter_fn` spreads a delay either way, and on the + // first throttle half of every fleet would land at or below the + // ask and retry in lockstep. A low result is mirrored above the + // grown wait instead, so every step spreads upward only; a jitter + // that zeroes delays (as tests install) leaves the grown wait. let jittered = (self.jitter_fn)(grown); - if jittered < asked { grown } else { jittered } + if jittered.is_zero() { + grown + } else { + grown + jittered.abs_diff(grown) + } } fn report(&self, outcome: RetryOutcome) { @@ -289,8 +297,7 @@ impl Retrier { } Some(RetryResult::RetryAfter(err, min_wait)) => { throttles_in_a_row = throttles_in_a_row.saturating_add(1); - let wait = self.throttle_wait(min_wait, throttles_in_a_row); - (err, wait, true) + (err, min_wait, true) } }; if !self.should_retry(err.code) { @@ -306,15 +313,20 @@ impl Retrier { // not an attempt: the server said when to come back, and // the budget bounds how long that is listened to. A zero // wait would spin, so it goes through the backoff instead. - if is_throttle - && !min_wait.is_zero() - && throttled.saturating_add(min_wait) <= self.throttle_budget - { - throttled = throttled.saturating_add(min_wait); - attempt -= 1; - self.report(RetryOutcome::Retried); - (self.sleep_fn)(min_wait).await; - continue; + // The grown wait is what the budget pays for. Past the + // budget the backoff ladder below waits at least the plain + // ask, never the grown one, so the ladder's total stays what + // the retry config says and the caller's own deadline (a + // worker's download window, Bazel's remote timeout) holds. + if is_throttle && !min_wait.is_zero() { + let wait = self.throttle_wait(min_wait, throttles_in_a_row); + if throttled.saturating_add(wait) <= self.throttle_budget { + throttled = throttled.saturating_add(wait); + attempt -= 1; + self.report(RetryOutcome::Retried); + (self.sleep_fn)(wait).await; + continue; + } } let Some(backoff) = iter.next() else { self.report(RetryOutcome::Exhausted); diff --git a/nativelink-util/tests/retry_test.rs b/nativelink-util/tests/retry_test.rs index a10f9aa2d7a..947fcc437b2 100644 --- a/nativelink-util/tests/retry_test.rs +++ b/nativelink-util/tests/retry_test.rs @@ -671,8 +671,9 @@ async fn a_zeroing_jitter_does_not_undo_the_growth() -> Result<(), Error> { } /// The grown waits are what the budget is spent on: 2 + 4 + 8 fits a 15 s -/// budget and the fourth throttle (16 s) falls through to the backoff -/// ladder, where it spends the one allowed attempt. +/// budget and the fourth throttle (16 s grown) falls through to the backoff +/// ladder, where it spends the one allowed attempt and waits the plain ask, +/// not the grown one, so the ladder's total stays what the config says. #[nativelink_test] async fn grown_waits_spend_the_budget() -> Result<(), Error> { let sleeps = Arc::new(Mutex::new(Vec::new())); @@ -697,11 +698,11 @@ async fn grown_waits_spend_the_budget() -> Result<(), Error> { ); assert_eq!( *sleeps.lock().unwrap(), - [2, 4, 8, 16] + [2, 4, 8, 2] .into_iter() .map(Duration::from_secs) .collect::>(), - "the fourth wait is the grown ask, longer than the 2 s backoff" + "the fourth wait is the backoff against the plain 2 s ask" ); Ok(()) } @@ -740,10 +741,10 @@ async fn a_plain_retry_resets_the_throttle_streak() -> Result<(), Error> { Ok(()) } -/// A jitter that lands below what the server asked for is not applied: the -/// first throttle waits the ask, the second the doubled ask, not the tenth. +/// A jitter that lands low is mirrored above the grown wait, so no throttle +/// ever sits on the floor: a tenth of the wait becomes nine tenths on top. #[nativelink_test] -async fn throttle_jitter_is_floored_at_the_ask() -> Result<(), Error> { +async fn a_low_jitter_is_mirrored_above_the_grown_wait() -> Result<(), Error> { let sleeps = Arc::new(Mutex::new(Vec::new())); let retrier = Retrier::new( Arc::new({ @@ -779,8 +780,8 @@ async fn throttle_jitter_is_floored_at_the_ask() -> Result<(), Error> { assert_eq!(result, true); assert_eq!( *sleeps.lock().unwrap(), - vec![Duration::from_secs(10), Duration::from_secs(20)], - "10 s asked, 1 s and 2 s jittered, the grown 10 s and 20 s waited" + vec![Duration::from_secs(19), Duration::from_secs(38)], + "10 s asked, jittered to a tenth, mirrored above: 19 s, then 38 s on the doubled wait" ); Ok(()) }