timvw_rust_runtime/
retry.rs

1//! Retries for HTTP APIs: [`RetryPolicy`], a tower retry policy, and [`RetryLayer`], which adds to it a look at the
2//! body of a `429` to tell a rate limit (retried) from an exhausted quota (not retried).
3//!
4//! `Client::default()` already retries with `RetryLayer::new(RetryPolicy::default())`. For another policy, start from
5//! the bare service:
6//!
7//! ```no_run
8//! # #[cfg(feature = "reqwest")]
9//! # fn demo() {
10//! use timvw_rust_runtime::retry::{self, RetryLayer, RetryPolicy};
11//! use timvw_rust_runtime::{Client, ReqwestService};
12//!
13//! let client = Client::from_service(ReqwestService::default())
14//!     .layer(RetryLayer::new(RetryPolicy::new(5)).permanent_429(retry::openai_insufficient_quota));
15//! # }
16//! ```
17
18use std::fmt;
19use std::future::Future;
20use std::pin::Pin;
21use std::sync::Arc;
22use std::task::{Context, Poll};
23use std::time::Duration;
24
25use bytes::{Bytes, BytesMut};
26use http::header::RETRY_AFTER;
27use http::{HeaderMap, StatusCode};
28use http_body_util::{BodyDataStream, BodyExt};
29use tower::retry::Policy;
30use tower::{Layer, Service};
31
32use crate::{Body, BoxError};
33
34/// The header OpenAI (and others) send next to `Retry-After`, in milliseconds.
35const RETRY_AFTER_MS: &str = "retry-after-ms";
36
37/// How much of a `429` body [`RetryLayer`] reads by default to classify it: 64 KiB.
38pub const DEFAULT_BODY_LIMIT: usize = 64 * 1024;
39
40/// Retries `429 Too Many Requests`, `408 Request Timeout` and `5xx` responses, and failures to connect.
41///
42/// - It waits as long as the server asks, with `retry-after-ms` (milliseconds) or `Retry-After` (seconds), up to
43///   [`RetryPolicy::max_retry_after`] (60 s by default). A server that asks for longer gets its response returned at
44///   once: waiting less than it asked would likely fail again, and waiting an hour is rarely what a caller wants.
45/// - Without those headers it backs off exponentially from [`RetryPolicy::base_delay`] (200 ms), doubling on each retry
46///   up to [`RetryPolicy::max_delay`] (8 s), with jitter: each delay is picked at random between half and all of it, so
47///   that clients that failed together do not retry together. [`RetryPolicy::jitter`] turns that off.
48/// - It never retries a response marked [`Permanent`] (see [`RetryLayer::permanent_429`]), nor a request whose body is
49///   a stream (a streamed file upload), which cannot be sent again.
50#[derive(Clone, Debug)]
51pub struct RetryPolicy {
52    max_retries: usize,
53    attempts: usize,
54    base_delay: Duration,
55    max_delay: Duration,
56    max_retry_after: Duration,
57    jitter: bool,
58}
59
60impl RetryPolicy {
61    /// A policy that retries at most `max_retries` times (after the first attempt).
62    pub fn new(max_retries: usize) -> Self {
63        Self {
64            max_retries,
65            attempts: 0,
66            base_delay: Duration::from_millis(200),
67            max_delay: Duration::from_secs(8),
68            max_retry_after: Duration::from_secs(60),
69            jitter: true,
70        }
71    }
72
73    /// The first backoff delay, doubled on each retry (default 200 ms).
74    pub fn base_delay(mut self, delay: Duration) -> Self {
75        self.base_delay = delay;
76        self
77    }
78
79    /// The longest backoff delay (default 8 s). It does not limit how long a server can ask to wait with
80    /// `Retry-After`: that is [`RetryPolicy::max_retry_after`].
81    pub fn max_delay(mut self, delay: Duration) -> Self {
82        self.max_delay = delay;
83        self
84    }
85
86    /// The longest wait a server can ask for with `Retry-After` or `retry-after-ms` (default 60 s). A response that
87    /// asks for longer is not retried but returned.
88    pub fn max_retry_after(mut self, delay: Duration) -> Self {
89        self.max_retry_after = delay;
90        self
91    }
92
93    /// Whether each backoff delay is picked at random between half and all of it (default `true`). The waits a
94    /// server asks for are never jittered.
95    pub fn jitter(mut self, jitter: bool) -> Self {
96        self.jitter = jitter;
97        self
98    }
99
100    /// The number of retries allowed.
101    pub fn max_retries(&self) -> usize {
102        self.max_retries
103    }
104
105    /// How long to wait before the next retry, or `None` when the server asks for longer than `max_retry_after`.
106    fn delay(&self, retry_after: Option<Duration>) -> Option<Duration> {
107        if let Some(retry_after) = retry_after {
108            return (retry_after <= self.max_retry_after).then_some(retry_after);
109        }
110        let backoff = self
111            .base_delay
112            .saturating_mul(1u32 << self.attempts.min(16))
113            .min(self.max_delay);
114        Some(if self.jitter {
115            jittered(backoff, random_unit())
116        } else {
117            backoff
118        })
119    }
120}
121
122/// Three retries.
123impl Default for RetryPolicy {
124    fn default() -> Self {
125        Self::new(3)
126    }
127}
128
129/// `backoff` scaled to between half and all of it, by `unit` in `[0, 1)`.
130fn jittered(backoff: Duration, unit: f64) -> Duration {
131    backoff.mul_f64(0.5 + 0.5 * unit)
132}
133
134/// A random number in `[0, 1)`, from the standard library's randomly keyed hasher: good enough to spread retries.
135fn random_unit() -> f64 {
136    use std::hash::{BuildHasher, Hasher};
137    let hash = std::collections::hash_map::RandomState::new()
138        .build_hasher()
139        .finish();
140    (hash >> 11) as f64 / (1u64 << 53) as f64
141}
142
143/// Whether a response or error is worth retrying, and how long the server asked to wait.
144fn retryable<B>(result: &Result<http::Response<B>, BoxError>) -> Option<Option<Duration>> {
145    match result {
146        Ok(response) => {
147            let status = response.status();
148            let transient = status == StatusCode::TOO_MANY_REQUESTS
149                || status == StatusCode::REQUEST_TIMEOUT
150                || status.is_server_error();
151            let permanent = response.extensions().get::<Permanent>().is_some();
152            (transient && !permanent).then(|| retry_after(response.headers()))
153        }
154        Err(error) => is_connect(error.as_ref()).then_some(None),
155    }
156}
157
158/// The wait a server asks for: `retry-after-ms` in milliseconds, else `Retry-After` in seconds. Both may have a
159/// fraction. `Retry-After` as an HTTP date is ignored (the policy backs off instead).
160fn retry_after(headers: &HeaderMap) -> Option<Duration> {
161    let parse = |name, scale: f64| {
162        let value = headers
163            .get(name)?
164            .to_str()
165            .ok()?
166            .trim()
167            .parse::<f64>()
168            .ok()?;
169        Duration::try_from_secs_f64(value * scale).ok()
170    };
171    parse(RETRY_AFTER_MS, 0.001).or_else(|| parse(RETRY_AFTER.as_str(), 1.0))
172}
173
174#[allow(unused_variables)]
175fn is_connect(error: &(dyn std::error::Error + 'static)) -> bool {
176    #[cfg(feature = "reqwest")]
177    {
178        let mut source = Some(error);
179        while let Some(error) = source {
180            if error
181                .downcast_ref::<reqwest::Error>()
182                .is_some_and(reqwest::Error::is_connect)
183            {
184                return true;
185            }
186            source = error.source();
187        }
188    }
189    false
190}
191
192impl<B> Policy<http::Request<Body>, http::Response<B>, BoxError> for RetryPolicy {
193    type Future = Pin<Box<dyn Future<Output = ()> + Send>>;
194
195    fn retry(
196        &mut self,
197        _request: &mut http::Request<Body>,
198        result: &mut Result<http::Response<B>, BoxError>,
199    ) -> Option<Self::Future> {
200        if self.attempts >= self.max_retries {
201            return None;
202        }
203        let retry_after = retryable(result)?;
204        let Some(delay) = self.delay(retry_after) else {
205            #[cfg(feature = "tracing")]
206            tracing::debug!(
207                target: "timvw_rust_runtime::trace",
208                retry_after_ms = retry_after.map_or(0, |delay| delay.as_millis() as u64),
209                max_retry_after_ms = self.max_retry_after.as_millis() as u64,
210                status = result.as_ref().ok().map(|response| response.status().as_u16()),
211                "not retrying: the server asks to wait longer than max_retry_after"
212            );
213            return None;
214        };
215        self.attempts += 1;
216        #[cfg(feature = "tracing")]
217        tracing::debug!(
218            target: "timvw_rust_runtime::trace",
219            attempt = self.attempts,
220            max_retries = self.max_retries,
221            delay_ms = delay.as_millis() as u64,
222            status = result.as_ref().ok().map(|response| response.status().as_u16()),
223            "retrying"
224        );
225        Some(Box::pin(tokio::time::sleep(delay)))
226    }
227
228    fn clone_request(&mut self, request: &http::Request<Body>) -> Option<http::Request<Body>> {
229        let body = request.body().try_clone()?;
230        let mut clone = http::Request::new(body);
231        *clone.method_mut() = request.method().clone();
232        *clone.uri_mut() = request.uri().clone();
233        *clone.version_mut() = request.version();
234        *clone.headers_mut() = request.headers().clone();
235        Some(clone)
236    }
237}
238
239/// A response extension that tells [`RetryPolicy`] not to retry a response it would otherwise retry.
240///
241/// [`RetryLayer::permanent_429`] sets it on a `429` whose body says the error is permanent. A layer of your own, inside
242/// the retries, can set it too: `response.extensions_mut().insert(Permanent)`.
243#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
244pub struct Permanent;
245
246/// Whether the body of a `429` says the error is permanent, such as [`openai_insufficient_quota`].
247type Classifier = Arc<dyn Fn(&[u8]) -> bool + Send + Sync>;
248
249/// OpenAI's "quota exhausted": a JSON body whose `error.code` or `error.type` is `insufficient_quota`.
250///
251/// OpenAI answers `429` both when requests come too fast, which passes, and when the account's quota or credit is
252/// used up, which does not pass by waiting. Use it with [`RetryLayer::permanent_429`].
253pub fn openai_insufficient_quota(body: &[u8]) -> bool {
254    const INSUFFICIENT_QUOTA: &str = "insufficient_quota";
255    let Ok(value) = serde_json::from_slice::<serde_json::Value>(body) else {
256        return false;
257    };
258    ["code", "type"].iter().any(|field| {
259        value
260            .pointer(&format!("/error/{field}"))
261            .and_then(serde_json::Value::as_str)
262            == Some(INSUFFICIENT_QUOTA)
263    })
264}
265
266/// Retries with a [`RetryPolicy`], and optionally reads the body of a `429` to tell a permanent error from a rate
267/// limit first.
268///
269/// `RetryLayer::new(policy)` retries exactly like `tower::retry::RetryLayer::new(policy)`. With
270/// [`RetryLayer::permanent_429`], the body of each `429` is read (up to [`RetryLayer::body_limit`]) and given to a
271/// function; when it returns `true`, the response is marked [`Permanent`] and returned at once. The body is put back,
272/// so the caller (and a generated client's error) still sees all of it.
273///
274/// ```no_run
275/// # #[cfg(feature = "reqwest")]
276/// # fn demo() {
277/// use timvw_rust_runtime::retry::{openai_insufficient_quota, RetryLayer, RetryPolicy};
278/// use timvw_rust_runtime::{Client, ReqwestService};
279///
280/// let client = Client::from_service(ReqwestService::default())
281///     .layer(RetryLayer::new(RetryPolicy::default()).permanent_429(openai_insufficient_quota));
282/// # }
283/// ```
284#[derive(Clone)]
285pub struct RetryLayer {
286    policy: RetryPolicy,
287    classify: Option<Classifier>,
288    body_limit: usize,
289}
290
291impl RetryLayer {
292    /// Retries with `policy`, without reading any body.
293    pub fn new(policy: RetryPolicy) -> Self {
294        Self {
295            policy,
296            classify: None,
297            body_limit: DEFAULT_BODY_LIMIT,
298        }
299    }
300
301    /// Reads the body of each `429` and does not retry it when `is_permanent` returns `true` for the body's bytes.
302    ///
303    /// A body longer than [`RetryLayer::body_limit`], or one that fails to read, is not given to `is_permanent`: the
304    /// response is retried as a rate limit, and the caller still reads the whole body (or its error).
305    pub fn permanent_429<F>(mut self, is_permanent: F) -> Self
306    where
307        F: Fn(&[u8]) -> bool + Send + Sync + 'static,
308    {
309        self.classify = Some(Arc::new(is_permanent));
310        self
311    }
312
313    /// How many bytes of a `429` body are read to classify it (default [`DEFAULT_BODY_LIMIT`], 64 KiB).
314    pub fn body_limit(mut self, limit: usize) -> Self {
315        self.body_limit = limit;
316        self
317    }
318
319    /// The retry policy.
320    pub fn policy(&self) -> &RetryPolicy {
321        &self.policy
322    }
323}
324
325/// Retries with [`RetryPolicy::default`].
326impl Default for RetryLayer {
327    fn default() -> Self {
328        Self::new(RetryPolicy::default())
329    }
330}
331
332impl fmt::Debug for RetryLayer {
333    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
334        f.debug_struct("RetryLayer")
335            .field("policy", &self.policy)
336            .field("permanent_429", &self.classify.is_some())
337            .field("body_limit", &self.body_limit)
338            .finish()
339    }
340}
341
342impl<S> Layer<S> for RetryLayer {
343    type Service = tower::retry::Retry<RetryPolicy, Classify429<S>>;
344
345    fn layer(&self, inner: S) -> Self::Service {
346        tower::retry::Retry::new(
347            self.policy.clone(),
348            Classify429 {
349                inner,
350                classify: self.classify.clone(),
351                body_limit: self.body_limit,
352            },
353        )
354    }
355}
356
357/// The service inside the retries of a [`RetryLayer`]: marks a `429` [`Permanent`] when its body says so.
358#[derive(Clone)]
359pub struct Classify429<S> {
360    inner: S,
361    classify: Option<Classifier>,
362    body_limit: usize,
363}
364
365impl<S: fmt::Debug> fmt::Debug for Classify429<S> {
366    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
367        f.debug_struct("Classify429")
368            .field("inner", &self.inner)
369            .field("permanent_429", &self.classify.is_some())
370            .field("body_limit", &self.body_limit)
371            .finish()
372    }
373}
374
375impl<S, R, B> Service<R> for Classify429<S>
376where
377    S: Service<R, Response = http::Response<B>>,
378    S::Future: Send + 'static,
379    B: http_body::Body + Send + Sync + 'static,
380    B::Data: Into<Bytes>,
381    B::Error: Into<BoxError>,
382{
383    type Response = http::Response<Body>;
384    type Error = S::Error;
385    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
386
387    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
388        self.inner.poll_ready(cx)
389    }
390
391    fn call(&mut self, request: R) -> Self::Future {
392        let response = self.inner.call(request);
393        let classify = self.classify.clone();
394        let body_limit = self.body_limit;
395        Box::pin(async move {
396            let response = response.await?.map(Body::wrap);
397            Ok(match classify {
398                Some(classify) if response.status() == StatusCode::TOO_MANY_REQUESTS => {
399                    classify_429(response, &*classify, body_limit).await
400                }
401                _ => response,
402            })
403        })
404    }
405}
406
407/// Reads a `429` body, up to `limit` bytes, marks the response [`Permanent`] if `classify` says so, and puts the body
408/// back.
409async fn classify_429(
410    response: http::Response<Body>,
411    classify: &(dyn Fn(&[u8]) -> bool + Send + Sync),
412    limit: usize,
413) -> http::Response<Body> {
414    let (mut parts, mut body) = response.into_parts();
415    let mut read = BytesMut::new();
416    let body = loop {
417        match body.frame().await {
418            None => {
419                let read = read.freeze();
420                if classify(&read) {
421                    parts.extensions.insert(Permanent);
422                    #[cfg(feature = "tracing")]
423                    tracing::debug!(
424                        target: "timvw_rust_runtime::trace",
425                        status = parts.status.as_u16(),
426                        "not retrying: the body says the error is permanent"
427                    );
428                }
429                break Body::from(read);
430            }
431            // Trailers on a 429 are dropped.
432            Some(Ok(frame)) => {
433                let Ok(data) = frame.into_data() else {
434                    continue;
435                };
436                read.extend_from_slice(&data);
437                if read.len() > limit {
438                    let read = futures_util::stream::once(async move { Ok(read.freeze()) });
439                    break Body::wrap_stream(futures_util::StreamExt::chain(
440                        read,
441                        BodyDataStream::new(body),
442                    ));
443                }
444            }
445            Some(Err(error)) => {
446                break Body::wrap_stream(futures_util::stream::iter([
447                    Ok(read.freeze()),
448                    Err::<Bytes, BoxError>(error),
449                ]));
450            }
451        }
452    };
453    http::Response::from_parts(parts, body)
454}
455
456#[cfg(test)]
457mod tests {
458    use super::*;
459
460    fn response(status: u16, headers: &[(&str, &str)]) -> Result<http::Response<()>, BoxError> {
461        let mut builder = http::Response::builder().status(status);
462        for (name, value) in headers {
463            builder = builder.header(*name, *value);
464        }
465        Ok(builder.body(()).unwrap())
466    }
467
468    #[test]
469    fn retries_transient_statuses_only() {
470        assert_eq!(retryable(&response(429, &[])), Some(None));
471        assert_eq!(
472            retryable(&response(503, &[("retry-after", "2")])),
473            Some(Some(Duration::from_secs(2)))
474        );
475        assert_eq!(retryable(&response(408, &[])), Some(None));
476        assert_eq!(retryable(&response(400, &[])), None);
477        assert_eq!(retryable(&response(200, &[])), None);
478        assert_eq!(retryable::<()>(&Err("boom".into())), None);
479    }
480
481    #[test]
482    fn permanent_responses_are_not_retried() {
483        let mut result = response(429, &[]);
484        result.as_mut().unwrap().extensions_mut().insert(Permanent);
485        assert_eq!(retryable(&result), None);
486    }
487
488    #[test]
489    fn parses_retry_after_and_retry_after_ms() {
490        let wait = |headers: &[(&str, &str)]| retryable(&response(429, headers)).unwrap();
491        assert_eq!(
492            wait(&[("retry-after", " 20 ")]),
493            Some(Duration::from_secs(20))
494        );
495        assert_eq!(
496            wait(&[("retry-after", "1.5")]),
497            Some(Duration::from_millis(1500))
498        );
499        assert_eq!(
500            wait(&[("retry-after-ms", "250"), ("retry-after", "1")]),
501            Some(Duration::from_millis(250))
502        );
503        assert_eq!(
504            wait(&[("retry-after-ms", "12.5")]),
505            Some(Duration::from_micros(12_500))
506        );
507        assert_eq!(
508            wait(&[("retry-after", "Wed, 21 Oct 2015 07:28:00 GMT")]),
509            None
510        );
511        assert_eq!(wait(&[("retry-after", "-1")]), None);
512        assert_eq!(
513            wait(&[("retry-after-ms", "nan"), ("retry-after", "3")]),
514            Some(Duration::from_secs(3))
515        );
516    }
517
518    #[test]
519    fn backs_off_and_caps() {
520        let mut policy = RetryPolicy::new(10)
521            .base_delay(Duration::from_millis(100))
522            .max_delay(Duration::from_secs(1))
523            .jitter(false);
524        assert_eq!(policy.delay(None), Some(Duration::from_millis(100)));
525        policy.attempts = 2;
526        assert_eq!(policy.delay(None), Some(Duration::from_millis(400)));
527        policy.attempts = 8;
528        assert_eq!(policy.delay(None), Some(Duration::from_secs(1)));
529    }
530
531    #[test]
532    fn honours_retry_after_beyond_max_delay_up_to_max_retry_after() {
533        let policy = RetryPolicy::new(3).max_delay(Duration::from_secs(1));
534        assert_eq!(
535            policy.delay(Some(Duration::from_secs(30))),
536            Some(Duration::from_secs(30))
537        );
538        assert_eq!(
539            policy.delay(Some(Duration::from_secs(60))),
540            Some(Duration::from_secs(60))
541        );
542        assert_eq!(policy.delay(Some(Duration::from_secs(61))), None);
543        let policy = policy.max_retry_after(Duration::from_secs(5));
544        assert_eq!(policy.delay(Some(Duration::from_secs(6))), None);
545    }
546
547    #[test]
548    fn jitter_stays_between_half_and_all_of_the_backoff() {
549        let backoff = Duration::from_millis(800);
550        assert_eq!(jittered(backoff, 0.0), Duration::from_millis(400));
551        assert_eq!(jittered(backoff, 0.5), Duration::from_millis(600));
552        assert!(jittered(backoff, 0.999_999) <= backoff);
553
554        let mut policy = RetryPolicy::new(10)
555            .base_delay(Duration::from_millis(100))
556            .max_delay(Duration::from_secs(1));
557        policy.attempts = 3; // 800 ms before jitter
558        let delays: Vec<_> = (0..200).map(|_| policy.delay(None).unwrap()).collect();
559        for delay in &delays {
560            assert!(
561                (Duration::from_millis(400)..=backoff).contains(delay),
562                "{delay:?}"
563            );
564        }
565        assert!(delays.iter().any(|delay| *delay != delays[0]), "random");
566        // The waits a server asks for are not jittered.
567        assert_eq!(
568            policy.delay(Some(Duration::from_secs(2))),
569            Some(Duration::from_secs(2))
570        );
571    }
572
573    #[test]
574    fn random_unit_is_in_range() {
575        for _ in 0..1000 {
576            assert!((0.0..1.0).contains(&random_unit()));
577        }
578    }
579
580    #[test]
581    fn recognises_openai_insufficient_quota() {
582        assert!(openai_insufficient_quota(
583            br#"{"error":{"message":"You exceeded your current quota","type":"insufficient_quota","param":null,"code":"insufficient_quota"}}"#
584        ));
585        assert!(openai_insufficient_quota(
586            br#"{"error":{"type":"insufficient_quota"}}"#
587        ));
588        assert!(openai_insufficient_quota(
589            br#"{"error":{"type":"requests","code":"insufficient_quota"}}"#
590        ));
591        assert!(!openai_insufficient_quota(
592            br#"{"error":{"message":"Rate limit reached","type":"requests","code":"rate_limit_exceeded"}}"#
593        ));
594        assert!(!openai_insufficient_quota(b"insufficient_quota"));
595        assert!(!openai_insufficient_quota(b""));
596    }
597
598    #[test]
599    fn only_bodies_in_memory_are_cloned() {
600        let mut policy = RetryPolicy::default();
601        let mut clone_request = |request: &http::Request<Body>| {
602            Policy::<_, http::Response<()>, BoxError>::clone_request(&mut policy, request)
603        };
604        let request = http::Request::post("https://example.com")
605            .header("x", "1")
606            .body(Body::from("a"))
607            .unwrap();
608        let clone = clone_request(&request).unwrap();
609        assert_eq!(clone.uri(), request.uri());
610        assert_eq!(clone.headers()["x"], "1");
611        let chunks =
612            futures_util::stream::iter(vec![Ok::<_, std::io::Error>(bytes::Bytes::from("a"))]);
613        let streamed = http::Request::post("https://example.com")
614            .body(Body::wrap_stream(chunks))
615            .unwrap();
616        assert!(clone_request(&streamed).is_none());
617    }
618
619    fn body_429(chunks: Vec<Result<&'static str, &'static str>>) -> http::Response<Body> {
620        let stream = futures_util::stream::iter(
621            chunks
622                .into_iter()
623                .map(|chunk| chunk.map(Bytes::from).map_err(BoxError::from)),
624        );
625        http::Response::builder()
626            .status(429)
627            .body(Body::wrap_stream(stream))
628            .unwrap()
629    }
630
631    #[tokio::test]
632    async fn classifies_the_whole_body_and_puts_it_back() {
633        let response = body_429(vec![
634            Ok(r#"{"error":{"type":"#),
635            Ok(r#""insufficient_quota"}}"#),
636        ]);
637        let response = classify_429(response, &openai_insufficient_quota, 1024).await;
638        assert!(response.extensions().get::<Permanent>().is_some());
639        let body = response.into_body().collect_bytes().await.unwrap();
640        assert_eq!(body, r#"{"error":{"type":"insufficient_quota"}}"#);
641    }
642
643    #[tokio::test]
644    async fn a_body_over_the_limit_is_not_classified_but_kept_whole() {
645        let response = body_429(vec![Ok("0123"), Ok("4567"), Ok("89")]);
646        let response = classify_429(response, &|_: &[u8]| true, 5).await;
647        assert!(response.extensions().get::<Permanent>().is_none());
648        let body = response.into_body().collect_bytes().await.unwrap();
649        assert_eq!(body, "0123456789");
650    }
651
652    #[tokio::test]
653    async fn a_body_that_fails_is_not_classified_and_keeps_its_error() {
654        let response = body_429(vec![Ok("01"), Err("reset")]);
655        let response = classify_429(response, &|_: &[u8]| true, 1024).await;
656        assert!(response.extensions().get::<Permanent>().is_none());
657        let error = response.into_body().collect_bytes().await.unwrap_err();
658        assert_eq!(error.to_string(), "reset");
659    }
660}
timvw_rust_runtime/retry.rs

timvw_rust_runtime/
retry.rs
