typesafe_sdk/retry.rs
1//! When a failed attempt is worth repeating, and how long to wait first.
2//!
3//! A [`RetryPolicy`] says which failures are repeated, how often, how long to
4//! wait between attempts and how long one call may take in all. A client
5//! holds one ([`ClientBuilder::retry`](crate::ClientBuilder::retry)) and a
6//! call can replace it for itself alone
7//! ([`SystemOne::retry`](crate::SystemOne::retry),
8//! [`ListModels::retry`](crate::ListModels::retry)). The defaults are the
9//! Python SDK's: two retries, a backoff from 0.5 s doubling up to 5 s with a
10//! quarter of it jittered away, statuses 408, 429 and 500-599, connection
11//! failures and timeouts, and 30 s for the whole call.
12//!
13//! The delay is a pure function of the attempt number and a random draw, so
14//! the schedule can be asserted exactly rather than observed; the clock, the
15//! sleep and the draw are a seam a test fills with a fake, so a retry test
16//! costs no wall time and cannot be flaky.
17//!
18//! A server that says how long to wait is obeyed however long it asks for:
19//! `retry-after-ms` or `Retry-After` replaces the backoff, and the backoff cap
20//! does not apply to it. Retrying stops when the next delay would carry the
21//! call past its budget, and the failure the caller gets is the last one,
22//! unchanged.
23//!
24//! The loop is written here rather than taken from a crate. `backon` and
25//! `tower::retry` were weighed and declined: the decision sequence has to be
26//! the Python SDK's (which failures, then the server's delay or the backoff,
27//! then the attempt count and the budget), the budget has to be measured on a
28//! clock a test controls, the request body has to be shared rather than
29//! rebuilt, and the loop must not box the attempt's future. Adapting either
30//! crate to all four costs more code than the loop, which is short, and
31//! `fastrand` supplies the one random number a delay needs.
32//!
33//! Nothing here spawns a task: the sleep between attempts is awaited inside
34//! the call's own future, so dropping that future cancels the wait and sends
35//! nothing more.
36
37use std::{
38 fmt,
39 future::Future,
40 io::Write as _,
41 sync::Arc,
42 time::{Duration, Instant, SystemTime},
43};
44
45use http::{HeaderMap, Method, Uri, header::RETRY_AFTER};
46
47use crate::{
48 config::ZERO_TIMEOUT,
49 constants::RETRY_AFTER_MS_HEADER,
50 error::{Error, ErrorKind, parse_retry_after},
51 telemetry,
52};
53
54// ------------------------------------------------------------ RetryPolicy
55
56/// A caller's rule for deciding whether a failed attempt is retried.
57type Predicate = Arc<dyn Fn(&Error) -> bool + Send + Sync>;
58
59/// Which failures a call repeats, how often, and how long it waits first.
60///
61/// Build one from [`RetryPolicy::default`] and its setters, then give it to a
62/// client with [`ClientBuilder::retry`](crate::ClientBuilder::retry) or to one
63/// call with [`SystemOne::retry`](crate::SystemOne::retry) or
64/// [`ListModels::retry`](crate::ListModels::retry). A policy given to a call
65/// replaces the client's for that call only.
66///
67/// A failed attempt is retried when all of these hold, checked in this order:
68///
69/// 1. The failure is retryable: a [timeout](ErrorKind::Timeout) (unless
70/// [`api_timeout_error`](Self::api_timeout_error) is off), a
71/// [connection failure](ErrorKind::Connection) (unless
72/// [`api_connection_error`](Self::api_connection_error) is off), or an
73/// [API error](ErrorKind::Api) whose status is in
74/// [`http_statuses`](Self::http_statuses) - or else the
75/// [`predicate`](Self::predicate) says so. A request that could not be
76/// built, a response that did not decode and a response over the size limit
77/// are never retryable on their own; only the predicate can ask for them.
78/// 2. Fewer than [`max_retries`](Self::max_retries) retries have been made.
79/// 3. The wait before the next attempt, added to the time the call has
80/// already taken, stays below the [`timeout`](Self::timeout) budget.
81///
82/// The wait is what the server asked for in `retry-after-ms` or
83/// `Retry-After`, when [`respect_retry_after`](Self::respect_retry_after) is
84/// on and the failure carries one, however long that is. Otherwise it is the
85/// backoff: [`backoff_initial`](Self::backoff_initial) doubled once per
86/// attempt up to [`backoff_max`](Self::backoff_max), less a random share of up
87/// to [`backoff_jitter`](Self::backoff_jitter) of it, in whole milliseconds.
88///
89/// When retrying stops, the call fails with the last attempt's error exactly
90/// as that attempt produced it.
91///
92/// ```
93/// use std::time::Duration;
94///
95/// use typesafe_sdk::{RetryPolicy, StatusSet};
96///
97/// let policy = RetryPolicy::default()
98/// .max_retries(3)
99/// .backoff_max(Duration::from_secs(2))
100/// .http_statuses([429, 502, 503, 504].into_iter().collect::<StatusSet>())
101/// .timeout(Duration::from_secs(10))?;
102/// # Ok::<(), typesafe_sdk::Error>(())
103/// ```
104#[derive(Clone)]
105pub struct RetryPolicy {
106 max_retries: u32,
107 backoff_initial: Duration,
108 backoff_max: Duration,
109 backoff_jitter: f64,
110 http_statuses: StatusSet,
111 respect_retry_after: bool,
112 api_connection_error: bool,
113 api_timeout_error: bool,
114 predicate: Option<Predicate>,
115 /// The budget of a whole call, or `None` for no budget.
116 timeout: Option<Duration>,
117 /// The clock, sleep and draw a test runs the loop on instead of the real
118 /// ones.
119 #[cfg(test)]
120 time: Option<Arc<tests::FakeTime>>,
121}
122
123impl RetryPolicy {
124 /// The Python SDK's defaults, which [`Default`] also gives: 2 retries, a
125 /// backoff from 500 ms up to 5 s with a jitter of 0.25, the statuses of
126 /// [`StatusSet::DEFAULT`], `Retry-After` respected, connection failures
127 /// and timeouts retried, no predicate, and a budget of 30 s.
128 #[must_use]
129 pub const fn new() -> Self {
130 Self {
131 max_retries: 2,
132 backoff_initial: Duration::from_millis(500),
133 backoff_max: Duration::from_secs(5),
134 backoff_jitter: 0.25,
135 http_statuses: StatusSet::DEFAULT,
136 respect_retry_after: true,
137 api_connection_error: true,
138 api_timeout_error: true,
139 predicate: None,
140 timeout: Some(Duration::from_secs(30)),
141 #[cfg(test)]
142 time: None,
143 }
144 }
145
146 /// The most retries after the first attempt; `0` makes one attempt only.
147 #[must_use]
148 pub fn max_retries(mut self, retries: u32) -> Self {
149 self.max_retries = retries;
150 self
151 }
152
153 /// The wait after the first failed attempt, doubled after each later one
154 /// up to [`backoff_max`](Self::backoff_max). Zero turns the backoff off,
155 /// so retries follow each other at once.
156 #[must_use]
157 pub fn backoff_initial(mut self, delay: Duration) -> Self {
158 self.backoff_initial = delay;
159 self
160 }
161
162 /// The longest backoff. Zero turns the backoff off. A delay the server
163 /// asks for is not capped by it.
164 #[must_use]
165 pub fn backoff_max(mut self, delay: Duration) -> Self {
166 self.backoff_max = delay;
167 self
168 }
169
170 /// The largest share of each backoff that is randomly taken off it, from
171 /// 0 (none) to 1 (up to all of it), so that clients which failed together
172 /// do not retry together.
173 ///
174 /// # Errors
175 ///
176 /// Returns an [`ErrorKind::Config`] error when `jitter` is not a number
177 /// from 0 to 1, NaN and the infinities included.
178 pub fn backoff_jitter(mut self, jitter: f64) -> Result<Self, Error> {
179 // A NaN is outside every range, so it fails the check too.
180 if !(0.0..=1.0).contains(&jitter) {
181 return Err(Error::config("backoff_jitter must be between zero and one."));
182 }
183 self.backoff_jitter = jitter;
184 Ok(self)
185 }
186
187 /// The statuses whose API errors are retried; [`StatusSet::DEFAULT`]
188 /// unless set.
189 ///
190 /// A status counts only for a response the SDK reports as
191 /// [`ErrorKind::Api`]. A success status in the set has no effect: a
192 /// success response whose body does not decode is an
193 /// [`ErrorKind::ResponseValidation`] error and is not retried for its
194 /// status, since the same body would come back; a
195 /// [`predicate`](Self::predicate) can still ask for it.
196 #[must_use]
197 pub fn http_statuses(mut self, statuses: StatusSet) -> Self {
198 self.http_statuses = statuses;
199 self
200 }
201
202 /// Whether a delay the server asks for in `retry-after-ms` or
203 /// `Retry-After` replaces the backoff. On unless turned off.
204 ///
205 /// The budget set with [`timeout`](Self::timeout) is what bounds such a
206 /// delay: without a budget ([`no_timeout`](Self::no_timeout)) a server's
207 /// `Retry-After` is obeyed however long it is. Keep a budget, or turn
208 /// this off, when the server is not trusted.
209 #[must_use]
210 pub fn respect_retry_after(mut self, respect: bool) -> Self {
211 self.respect_retry_after = respect;
212 self
213 }
214
215 /// Whether an attempt that got no response - a refused, failed or broken
216 /// connection, [`ErrorKind::Connection`] - is retried. On unless turned
217 /// off.
218 #[must_use]
219 pub fn api_connection_error(mut self, retry: bool) -> Self {
220 self.api_connection_error = retry;
221 self
222 }
223
224 /// Whether an attempt that ran past its deadline,
225 /// [`ErrorKind::Timeout`], is retried. On unless turned off.
226 #[must_use]
227 pub fn api_timeout_error(mut self, retry: bool) -> Self {
228 self.api_timeout_error = retry;
229 self
230 }
231
232 /// A rule of the caller's own: a failure it returns `true` for is
233 /// retried, in addition to the ones the other settings retry.
234 ///
235 /// It is called once for each failed attempt, with the error that attempt
236 /// ended with, before the attempt count and the budget are checked; a
237 /// failure the other settings already retry may skip it. It sees every
238 /// failure an attempt can end with, a response that did not decode or was
239 /// over the size limit included, and runs on the task that sends the
240 /// call, so it should return quickly.
241 #[must_use]
242 pub fn predicate<F>(mut self, predicate: F) -> Self
243 where
244 F: Fn(&Error) -> bool + Send + Sync + 'static,
245 {
246 self.predicate = Some(Arc::new(predicate));
247 self
248 }
249
250 /// The most time one call may take in all - every attempt and every wait
251 /// between them. Retrying stops before a wait that would reach it, and
252 /// the call fails with the last error. 30 s unless set.
253 ///
254 /// This is not the deadline of one attempt, which the client and each
255 /// call set with their own `timeout`.
256 ///
257 /// # Errors
258 ///
259 /// Returns an [`ErrorKind::Config`] error for a budget of zero.
260 pub fn timeout(mut self, budget: Duration) -> Result<Self, Error> {
261 if budget.is_zero() {
262 return Err(Error::config(ZERO_TIMEOUT));
263 }
264 self.timeout = Some(budget);
265 Ok(self)
266 }
267
268 /// No budget for the whole call: only the attempt count stops retrying.
269 ///
270 /// The budget is also what bounds a delay the server asks for: without
271 /// it, a server's `Retry-After` is obeyed however long it is. Keep a
272 /// budget, or turn [`respect_retry_after`](Self::respect_retry_after)
273 /// off, when the server is not trusted.
274 #[must_use]
275 pub fn no_timeout(mut self) -> Self {
276 self.timeout = None;
277 self
278 }
279
280 /// Whether a call under this policy can make more than one attempt, and
281 /// so has to keep its request body after the first.
282 pub(crate) fn can_retry(&self) -> bool {
283 self.max_retries > 0
284 }
285
286 /// Whether `error` is a failure this policy repeats, before counting
287 /// attempts or time.
288 fn retryable(&self, error: &Error) -> bool {
289 let builtin = match error.kind() {
290 ErrorKind::Timeout { .. } => self.api_timeout_error,
291 ErrorKind::Connection => self.api_connection_error,
292 ErrorKind::Api(api) => self.http_statuses.contains(api.status().as_u16()),
293 // A request that could not be built fails the same way again, a
294 // response that did not decode is not a status, and a response
295 // over the limit will be as large again.
296 ErrorKind::InvalidRequest
297 | ErrorKind::Config
298 | ErrorKind::ResponseValidation(_)
299 | ErrorKind::ResponseTooLarge { .. } => false,
300 };
301 builtin || self.predicate.as_ref().is_some_and(|predicate| predicate(error))
302 }
303
304 /// The wait before the next attempt, after `attempts` attempts, the last
305 /// of which failed with `error`: the server's delay when it gave one and
306 /// this policy respects it, the backoff otherwise.
307 fn delay<T: Time>(&self, time: &T, attempts: u32, error: &Error) -> Duration {
308 if self.respect_retry_after {
309 // The wall clock is read only for a response that names a delay,
310 // since only an HTTP date is measured against it.
311 let names_delay = |headers: &HeaderMap| {
312 headers.contains_key(RETRY_AFTER_MS_HEADER) || headers.contains_key(RETRY_AFTER)
313 };
314 let asked = match error.kind() {
315 ErrorKind::Api(api) if names_delay(api.headers()) => {
316 parse_retry_after(api.headers(), time.system_now())
317 }
318 // A response that did not decode is still a response, and a
319 // predicate may have asked for it to be retried.
320 ErrorKind::ResponseValidation(invalid) if names_delay(invalid.headers()) => {
321 parse_retry_after(invalid.headers(), time.system_now())
322 }
323 _ => None,
324 };
325 if let Some(asked) = asked {
326 return asked;
327 }
328 }
329 seconds_to_duration(backoff_seconds(
330 attempts,
331 self.backoff_initial.as_secs_f64(),
332 self.backoff_max.as_secs_f64(),
333 self.backoff_jitter,
334 time.draw(),
335 ))
336 }
337
338 /// The wait before the next attempt, when `retries` retries came before
339 /// the attempt that just failed with `error`; `None` when the call should
340 /// fail with `error` instead. `started` is when the call began, read only
341 /// when the policy has a budget.
342 fn next_delay<T: Time>(
343 &self,
344 time: &T,
345 retries: u32,
346 error: &Error,
347 started: Option<Instant>,
348 ) -> Option<Duration> {
349 if !self.retryable(error) || retries >= self.max_retries {
350 return None;
351 }
352 // Below `max_retries`, itself a `u32`, so adding one cannot overflow.
353 let attempts = retries + 1;
354 let delay = self.delay(time, attempts, error);
355 if let (Some(budget), Some(started)) = (self.timeout, started) {
356 // Saturating: a delay of `Duration::MAX` reaches any budget
357 // instead of overflowing.
358 let elapsed = time.now().saturating_duration_since(started);
359 if elapsed.saturating_add(delay) >= budget {
360 return None;
361 }
362 }
363 Some(delay)
364 }
365}
366
367impl Default for RetryPolicy {
368 fn default() -> Self {
369 Self::new()
370 }
371}
372
373impl fmt::Debug for RetryPolicy {
374 /// Every setting; a predicate, which has no text of its own, prints as
375 /// `<predicate>`.
376 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
377 formatter
378 .debug_struct("RetryPolicy")
379 .field("max_retries", &self.max_retries)
380 .field("backoff_initial", &self.backoff_initial)
381 .field("backoff_max", &self.backoff_max)
382 .field("backoff_jitter", &self.backoff_jitter)
383 .field("http_statuses", &self.http_statuses)
384 .field("respect_retry_after", &self.respect_retry_after)
385 .field("api_connection_error", &self.api_connection_error)
386 .field("api_timeout_error", &self.api_timeout_error)
387 .field("predicate", &self.predicate.as_ref().map(|_| Opaque))
388 .field("timeout", &self.timeout)
389 .finish()
390 }
391}
392
393/// What a predicate prints as.
394struct Opaque;
395
396impl fmt::Debug for Opaque {
397 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
398 formatter.write_str("<predicate>")
399 }
400}
401
402// -------------------------------------------------------------- StatusSet
403
404/// The number of statuses a [`StatusSet`] can hold: 0 to 639.
405const STATUS_LIMIT: u16 = 640;
406
407/// A set of HTTP status codes, as a fixed bitmap: copying one costs 80 bytes
408/// and looking a status up costs a shift, with no allocation either way.
409///
410/// It holds 0 to 639, which covers every status HTTP defines (100 to 599).
411/// A status of 640 or more is never contained, and inserting one does
412/// nothing and returns `false`.
413///
414/// In a [`RetryPolicy`] the set is asked only about responses reported as
415/// [`ErrorKind::Api`], so a success status in it has no effect; see
416/// [`RetryPolicy::http_statuses`].
417///
418/// ```
419/// use typesafe_sdk::StatusSet;
420///
421/// let mut statuses = StatusSet::DEFAULT;
422/// assert!(statuses.contains(503));
423/// statuses.remove(503);
424/// statuses.insert(409);
425/// assert_eq!(format!("{statuses:?}"), "{408, 409, 429, 500..=502, 504..=599}");
426///
427/// let only = [429, 503].into_iter().collect::<StatusSet>();
428/// assert_eq!(only.iter().collect::<Vec<_>>(), [429, 503]);
429/// ```
430#[derive(Clone, Copy, PartialEq, Eq, Hash)]
431pub struct StatusSet([u64; 10]);
432
433impl StatusSet {
434 /// The statuses retried by default, as in the Python SDK: 408 (request
435 /// timeout), 429 (too many requests) and every 5xx, 500 to 599.
436 pub const DEFAULT: Self = {
437 let mut set = Self::empty();
438 set.insert(408);
439 set.insert(429);
440 let mut status = 500;
441 while status < 600 {
442 set.insert(status);
443 status += 1;
444 }
445 set
446 };
447
448 /// A set with no status in it.
449 #[must_use]
450 pub const fn empty() -> Self {
451 Self([0; 10])
452 }
453
454 /// Whether `status` is in the set; always `false` from 640 up.
455 #[must_use]
456 pub const fn contains(&self, status: u16) -> bool {
457 match Self::slot(status) {
458 Some((word, bit)) => self.0[word] & bit != 0,
459 None => false,
460 }
461 }
462
463 /// Adds `status`, and says whether it was not there before. A status of
464 /// 640 or more cannot be held: nothing changes and the answer is `false`.
465 pub const fn insert(&mut self, status: u16) -> bool {
466 match Self::slot(status) {
467 Some((word, bit)) => {
468 let added = self.0[word] & bit == 0;
469 self.0[word] |= bit;
470 added
471 }
472 None => false,
473 }
474 }
475
476 /// Removes `status`, and says whether it was there.
477 pub const fn remove(&mut self, status: u16) -> bool {
478 match Self::slot(status) {
479 Some((word, bit)) => {
480 let removed = self.0[word] & bit != 0;
481 self.0[word] &= !bit;
482 removed
483 }
484 None => false,
485 }
486 }
487
488 /// Whether the set holds no status.
489 #[must_use]
490 pub const fn is_empty(&self) -> bool {
491 let mut word = 0;
492 while word < self.0.len() {
493 if self.0[word] != 0 {
494 return false;
495 }
496 word += 1;
497 }
498 true
499 }
500
501 /// The statuses in the set, from the lowest up.
502 pub fn iter(&self) -> impl Iterator<Item = u16> + use<> {
503 let words = self.0;
504 (0_u16..).zip(words).flat_map(|(index, word)| {
505 let mut rest = word;
506 std::iter::from_fn(move || {
507 if rest == 0 {
508 return None;
509 }
510 // `trailing_zeros` of a non-zero `u64` is below 64, so the
511 // status fits a `u16` with room to spare.
512 let bit = rest.trailing_zeros() as u16;
513 // Clears the lowest set bit.
514 rest &= rest - 1;
515 Some(index * 64 + bit)
516 })
517 })
518 }
519
520 /// The word and the bit of `status`, or `None` past the last one.
521 const fn slot(status: u16) -> Option<(usize, u64)> {
522 if status >= STATUS_LIMIT {
523 return None;
524 }
525 Some(((status / 64) as usize, 1 << (status % 64)))
526 }
527}
528
529impl Default for StatusSet {
530 /// [`StatusSet::DEFAULT`], the statuses retried unless a policy says
531 /// otherwise.
532 fn default() -> Self {
533 Self::DEFAULT
534 }
535}
536
537impl FromIterator<u16> for StatusSet {
538 /// Exactly the statuses given; one of 640 or more is left out.
539 fn from_iter<I: IntoIterator<Item = u16>>(statuses: I) -> Self {
540 let mut set = Self::empty();
541 set.extend(statuses);
542 set
543 }
544}
545
546impl Extend<u16> for StatusSet {
547 fn extend<I: IntoIterator<Item = u16>>(&mut self, statuses: I) {
548 for status in statuses {
549 self.insert(status);
550 }
551 }
552}
553
554impl fmt::Debug for StatusSet {
555 /// `{408, 429, 500..=599}`: a run of three or more statuses prints as a
556 /// range.
557 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
558 formatter.write_str("{")?;
559 let mut statuses = self.iter().peekable();
560 let mut first = true;
561 while let Some(start) = statuses.next() {
562 let mut end = start;
563 while statuses.next_if_eq(&(end + 1)).is_some() {
564 end += 1;
565 }
566 if !first {
567 formatter.write_str(", ")?;
568 }
569 first = false;
570 match end - start {
571 0 => write!(formatter, "{start}")?,
572 1 => write!(formatter, "{start}, {end}")?,
573 _ => write!(formatter, "{start}..={end}")?,
574 }
575 }
576 formatter.write_str("}")
577 }
578}
579
580// ------------------------------------------------------------- the loop
581
582/// What the retry loop reads from outside the program: the monotonic clock
583/// the budget is measured on, the wall clock an HTTP-date `Retry-After` is
584/// measured against, the sleep between attempts, and the random draw of the
585/// jitter.
586///
587/// A call runs on [`Tokio`]'s; the crate's own tests run it on a fake whose
588/// clock stands still until the fake sleep or the test server moves it.
589trait Time {
590 /// Now, on the clock the budget is measured on.
591 fn now(&self) -> Instant;
592 /// Now, on the wall clock.
593 fn system_now(&self) -> SystemTime;
594 /// Waits `delay`. Dropping the future cancels the wait.
595 fn sleep(&self, delay: Duration) -> impl Future<Output = ()> + Send;
596 /// A random number in `[0, 1)`.
597 fn draw(&self) -> f64;
598}
599
600/// The real clocks, Tokio's timer, and `fastrand`'s thread-local generator.
601struct Tokio;
602
603impl Time for Tokio {
604 fn now(&self) -> Instant {
605 Instant::now()
606 }
607
608 fn system_now(&self) -> SystemTime {
609 SystemTime::now()
610 }
611
612 fn sleep(&self, delay: Duration) -> impl Future<Output = ()> + Send {
613 // Tokio clamps a deadline past its far future instead of panicking,
614 // so even `Duration::MAX` is a valid wait.
615 tokio::time::sleep(delay)
616 }
617
618 fn draw(&self) -> f64 {
619 fastrand::f64()
620 }
621}
622
623/// Runs `attempt` until it succeeds or `policy` stops retrying, and returns
624/// its success or its last error, unchanged.
625///
626/// `attempt` is called with the number of attempts made before it, which is
627/// what `X-TypeSafe-Retry-Count` carries. `method` and `uri` name the request
628/// in the event logged before each retry.
629///
630/// A plain function returning the loop's future rather than an `async fn`
631/// awaiting it: a wrapping `async fn` would keep its own copy of `attempt`
632/// beside the loop's, and the call's future would carry both.
633#[cfg(not(test))]
634pub(crate) fn run<R, F, Fut>(
635 policy: &RetryPolicy,
636 method: &Method,
637 uri: &Uri,
638 attempt: F,
639) -> impl Future<Output = Result<R, Error>>
640where
641 F: FnMut(u32) -> Fut,
642 Fut: Future<Output = Result<R, Error>>,
643{
644 run_on(&Tokio, policy, method, uri, attempt)
645}
646
647/// [`run`], on the fake clock of the crate's own tests when the policy
648/// carries one.
649#[cfg(test)]
650pub(crate) async fn run<R, F, Fut>(
651 policy: &RetryPolicy,
652 method: &Method,
653 uri: &Uri,
654 attempt: F,
655) -> Result<R, Error>
656where
657 F: FnMut(u32) -> Fut,
658 Fut: Future<Output = Result<R, Error>>,
659{
660 if let Some(time) = &policy.time {
661 return run_on(&**time, policy, method, uri, attempt).await;
662 }
663 run_on(&Tokio, policy, method, uri, attempt).await
664}
665
666/// [`run`] on the given clock, sleep and draw.
667async fn run_on<T, R, F, Fut>(
668 time: &T,
669 policy: &RetryPolicy,
670 method: &Method,
671 uri: &Uri,
672 mut attempt: F,
673) -> Result<R, Error>
674where
675 T: Time,
676 F: FnMut(u32) -> Fut,
677 Fut: Future<Output = Result<R, Error>>,
678{
679 // The budget counts from the start of the call, the first attempt
680 // included; with no budget, or no retry to spend it on, the clock is not
681 // read at all.
682 let started = (policy.timeout.is_some() && policy.can_retry()).then(|| time.now());
683 let mut retry = 0_u32;
684 loop {
685 if retry > 0 {
686 telemetry::retrying(telemetry::Exchange::new(method, uri, retry));
687 }
688 let error = match attempt(retry).await {
689 Ok(done) => return Ok(done),
690 Err(error) => error,
691 };
692 let Some(delay) = policy.next_delay(time, retry, &error, started) else {
693 return Err(error);
694 };
695 drop(error);
696 time.sleep(delay).await;
697 // `next_delay` answers only while `retry` is below `max_retries`.
698 retry += 1;
699 }
700}
701
702/// The delay, in seconds, before the attempt after attempt number `attempt`
703/// failed.
704///
705/// The delay starts at `initial`, doubles with each attempt and stops growing
706/// at `max`; `jitter` then removes a random share of it, up to `jitter` of the
707/// whole, so that clients which failed together do not retry together. `draw`
708/// is that random number, in `[0, 1)`, taken as an argument so the delay is a
709/// pure function of its inputs. A zero `initial` or `max` disables backoff.
710///
711/// Attempts are numbered from 1, the first try. `attempt` 0 is not a number
712/// the retry loop produces; it follows the same arithmetic and gives half of
713/// `initial`, which keeps the schedule monotonic from 0 upwards. The type is
714/// `u32` because every value converts exactly into `f64` and `i64`, which the
715/// cap test and the doubling depend on, and no call is retried four billion
716/// times.
717///
718/// The cap is tested in log2 space, before any doubling, so no intermediate
719/// value overflows however large `attempt` is; the doubling itself is an exact
720/// `ldexp`, not `initial * 2^exponent`, whose power of two alone would
721/// overflow before the product came back under `max`.
722///
723/// The result is rounded to milliseconds and never exceeds the delay before
724/// jitter: a delay that rounds up past a sub-millisecond `max` is `max`.
725/// Rounding is exact decimal rounding of the binary value with ties to even,
726/// the rule of Python's `round(delay, 3)`, rather than scaling by 1000 and
727/// rounding the product: that shortcut rounds `1.0005` (stored just below it)
728/// up to `1.001`, and the exact tie `0.0625` up to `0.063`, where this and
729/// Python give `1.0` and `0.062`.
730///
731/// Non-finite or negative inputs are rejected where a retry policy is built,
732/// so none is checked here; none of them panics.
733pub(crate) fn backoff_seconds(attempt: u32, initial: f64, max: f64, jitter: f64, draw: f64) -> f64 {
734 if initial == 0.0 || max == 0.0 {
735 return 0.0;
736 }
737 // The subtraction is exact: every `u32` is representable in an `f64`.
738 let exponential = if f64::from(attempt) - 1.0 >= max.log2() - initial.log2() {
739 max
740 } else {
741 // Reaching here bounds the exponent by the log2 distance between two
742 // finite values, under 2100, so the conversion only saturates when
743 // `max` is infinite, where `scalbn` overflows to infinity either way.
744 let exponent = i32::try_from(i64::from(attempt) - 1).unwrap_or(i32::MAX);
745 scalbn(initial, exponent)
746 };
747 let delay = exponential * (1.0 - draw * jitter);
748 exponential.min(round_to_millis(delay))
749}
750
751/// Converts a delay in seconds into a [`Duration`], saturating instead of
752/// failing.
753///
754/// NaN, zero and negative values become [`Duration::ZERO`]; a value too large
755/// for a `Duration`, infinity included, becomes [`Duration::MAX`].
756fn seconds_to_duration(seconds: f64) -> Duration {
757 if seconds.is_nan() || seconds <= 0.0 {
758 return Duration::ZERO;
759 }
760 // `from_secs_f64` would panic on overflow; the fallible form reports it,
761 // and a positive, non-NaN value can only fail by being too large.
762 Duration::try_from_secs_f64(seconds).unwrap_or(Duration::MAX)
763}
764
765/// `x * 2^n`, correctly rounded: C's `scalbn`, which std does not provide.
766///
767/// This is musl's algorithm. A power of two outside the normal range of an
768/// `f64` cannot be written as one value, so a large `n` is applied in steps of
769/// `2^1023`; while scaling down, each step is `2^-1022` times `2^53`, which
770/// keeps the intermediate value normal so the result is rounded only once.
771/// `n` is clamped after two steps, where any finite nonzero `x` has already
772/// overflowed or underflowed.
773fn scalbn(x: f64, n: i32) -> f64 {
774 const TWO_POW_1023: f64 = f64::from_bits(0x7FE0_0000_0000_0000);
775 const TWO_POW_MINUS_969: f64 = f64::from_bits(0x0360_0000_0000_0000);
776 let mut y = x;
777 let mut n = n;
778 if n > 1023 {
779 y *= TWO_POW_1023;
780 n -= 1023;
781 if n > 1023 {
782 y *= TWO_POW_1023;
783 n = (n - 1023).min(1023);
784 }
785 } else if n < -1022 {
786 y *= TWO_POW_MINUS_969;
787 n += 1022 - 53;
788 if n < -1022 {
789 y *= TWO_POW_MINUS_969;
790 n = (n + 1022 - 53).max(-1022);
791 }
792 }
793 // `n` is now in `[-1022, 1023]`, so the biased exponent `1023 + n` is in
794 // `[1, 2046]`: a normal power of two built directly from its bits.
795 y * f64::from_bits(u64::from((0x3FF + n).unsigned_abs()) << 52)
796}
797
798/// `x` rounded to three decimal places, as Python's `round(x, 3)`.
799///
800/// Formatting with a precision rounds the exact decimal expansion of the
801/// binary value, ties to even, and parsing back picks the nearest `f64`: the
802/// same two steps CPython's `round` takes. From `2^52` up every `f64` is an
803/// integer, so larger values, infinities and NaN are returned unchanged, which
804/// also bounds the text to 21 bytes and keeps it on the stack.
805fn round_to_millis(x: f64) -> f64 {
806 const INTEGRAL_FROM: f64 = 4_503_599_627_370_496.0; // 2^52
807 if x.is_nan() || x.abs() >= INTEGRAL_FROM {
808 return x;
809 }
810 let mut buf = [0_u8; 24];
811 // Writing into `&mut [u8]` advances the slice past what was written, so
812 // the length left over tells how much was.
813 let unused = {
814 let mut rest = &mut buf[..];
815 write!(rest, "{x:.3}")
816 .expect("invariant: a value below 2^52 needs at most 21 bytes at three decimals");
817 rest.len()
818 };
819 let written = buf.len() - unused;
820 std::str::from_utf8(&buf[..written])
821 .expect("invariant: formatted digits are ASCII")
822 .parse()
823 .expect("invariant: a formatted finite f64 parses back")
824}
825
826#[cfg(test)]
827#[path = "retry_tests.rs"]
828mod tests;