object_store/client/
retry.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! [`RetryConfig`] connection retry policy
19
20use crate::PutPayload;
21use crate::client::backoff::{Backoff, BackoffConfig};
22use crate::client::builder::HttpRequestBuilder;
23use crate::client::{HttpClient, HttpError, HttpErrorKind, HttpRequest, HttpResponse};
24use futures_util::future::BoxFuture;
25use http::StatusCode;
26use http::header::LOCATION;
27use http::{Method, Uri};
28#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
29use std::time::{Duration, Instant};
30use tracing::info;
31#[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
32use web_time::{Duration, Instant};
33
34/// Retry request error
35#[derive(Debug)]
36pub struct RetryError(Box<RetryErrorImpl>);
37
38/// Box error to avoid large error variant
39#[derive(Debug)]
40struct RetryErrorImpl {
41    method: Method,
42    uri: Option<Uri>,
43    retries: usize,
44    max_retries: usize,
45    elapsed: Duration,
46    retry_timeout: Duration,
47    inner: RequestError,
48}
49
50impl std::fmt::Display for RetryError {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        write!(f, "Error performing {} ", self.0.method)?;
53        match &self.0.uri {
54            Some(uri) => write!(f, "{uri} ")?,
55            None => write!(f, "REDACTED ")?,
56        }
57        write!(f, "in {:?}", self.0.elapsed)?;
58        if self.0.retries != 0 {
59            write!(
60                f,
61                ", after {} retries, max_retries: {}, retry_timeout: {:?} ",
62                self.0.retries, self.0.max_retries, self.0.retry_timeout
63            )?;
64        }
65        write!(f, " - {}", self.0.inner)
66    }
67}
68
69impl std::error::Error for RetryError {
70    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
71        Some(&self.0.inner)
72    }
73}
74
75/// Context of the retry loop
76///
77/// Most use-cases should use [`RetryExt`] and [`RetryableRequestBuilder`], however,
78/// [`RetryContext`] allows preserving retry state across multiple [`RetryableRequest`]
79pub(crate) struct RetryContext {
80    backoff: Backoff,
81    retries: usize,
82    max_retries: usize,
83    retry_timeout: Duration,
84    start: Instant,
85}
86
87impl RetryContext {
88    pub(crate) fn new(config: &RetryConfig) -> Self {
89        Self {
90            max_retries: config.max_retries,
91            retry_timeout: config.retry_timeout,
92            backoff: Backoff::new(&config.backoff),
93            retries: 0,
94            start: Instant::now(),
95        }
96    }
97
98    pub(crate) fn exhausted(&self) -> bool {
99        self.retries >= self.max_retries || self.start.elapsed() > self.retry_timeout
100    }
101
102    pub(crate) fn backoff(&mut self) -> Duration {
103        self.retries += 1;
104        self.backoff.next()
105    }
106}
107
108/// The reason a request failed
109#[derive(Debug, thiserror::Error)]
110pub enum RequestError {
111    #[error(
112        "Received redirect without LOCATION, this normally indicates an incorrectly configured region"
113    )]
114    BareRedirect,
115
116    #[error("Server returned non-2xx status code: {status}: {}", body.as_deref().unwrap_or(""))]
117    Status {
118        status: StatusCode,
119        body: Option<String>,
120    },
121
122    #[error("Server returned error response: {body}")]
123    Response { status: StatusCode, body: String },
124
125    // We need to use `{0}` here rather than `error(transparent)` to ensure that
126    // `Error::source` returns `HttpError` rather than the underlying
127    // `reqwest::Error` which would happen with `error(transparent)`.
128    #[error("{0}")]
129    Http(#[from] HttpError),
130}
131
132impl RetryError {
133    /// Returns the underlying [`RequestError`]
134    pub fn inner(&self) -> &RequestError {
135        &self.0.inner
136    }
137
138    /// Returns the status code associated with this error if any
139    pub fn status(&self) -> Option<StatusCode> {
140        match self.inner() {
141            RequestError::Status { status, .. } | RequestError::Response { status, .. } => {
142                Some(*status)
143            }
144            RequestError::BareRedirect | RequestError::Http(_) => None,
145        }
146    }
147
148    /// Returns the error body if any
149    pub fn body(&self) -> Option<&str> {
150        match self.inner() {
151            RequestError::Status { body, .. } => body.as_deref(),
152            RequestError::Response { body, .. } => Some(body),
153            RequestError::BareRedirect | RequestError::Http(_) => None,
154        }
155    }
156
157    pub fn error(self, store: &'static str, path: String) -> crate::Error {
158        match self.status() {
159            Some(StatusCode::NOT_FOUND) => crate::Error::NotFound {
160                path,
161                source: Box::new(self),
162            },
163            Some(StatusCode::NOT_MODIFIED) => crate::Error::NotModified {
164                path,
165                source: Box::new(self),
166            },
167            Some(StatusCode::PRECONDITION_FAILED) => crate::Error::Precondition {
168                path,
169                source: Box::new(self),
170            },
171            Some(StatusCode::CONFLICT) => crate::Error::AlreadyExists {
172                path,
173                source: Box::new(self),
174            },
175            Some(StatusCode::FORBIDDEN) => crate::Error::PermissionDenied {
176                path,
177                source: Box::new(self),
178            },
179            Some(StatusCode::UNAUTHORIZED) => crate::Error::Unauthenticated {
180                path,
181                source: Box::new(self),
182            },
183            _ => crate::Error::Generic {
184                store,
185                source: Box::new(self),
186            },
187        }
188    }
189}
190
191impl From<RetryError> for std::io::Error {
192    fn from(err: RetryError) -> Self {
193        use std::io::ErrorKind;
194        let kind = match err.status() {
195            Some(StatusCode::NOT_FOUND) => ErrorKind::NotFound,
196            Some(StatusCode::BAD_REQUEST) => ErrorKind::InvalidInput,
197            Some(StatusCode::UNAUTHORIZED) | Some(StatusCode::FORBIDDEN) => {
198                ErrorKind::PermissionDenied
199            }
200            _ => match err.inner() {
201                RequestError::Http(h) => match h.kind() {
202                    HttpErrorKind::Timeout => ErrorKind::TimedOut,
203                    HttpErrorKind::Connect => ErrorKind::NotConnected,
204                    _ => ErrorKind::Other,
205                },
206                _ => ErrorKind::Other,
207            },
208        };
209        Self::new(kind, err)
210    }
211}
212
213pub(crate) type Result<T, E = RetryError> = std::result::Result<T, E>;
214
215/// The configuration for how to respond to request errors
216///
217/// The following categories of error will be retried:
218///
219/// * 5xx server errors
220/// * Connection errors
221/// * Dropped connections
222/// * Timeouts for [safe] / read-only requests
223///
224/// Requests will be retried up to some limit, using exponential
225/// backoff with jitter. See [`BackoffConfig`] for more information
226///
227/// [safe]: https://datatracker.ietf.org/doc/html/rfc7231#section-4.2.1
228#[derive(Debug, Clone)]
229pub struct RetryConfig {
230    /// The backoff configuration
231    pub backoff: BackoffConfig,
232
233    /// The maximum number of times to retry a request
234    ///
235    /// Set to 0 to disable retries
236    pub max_retries: usize,
237
238    /// The maximum length of time from the initial request
239    /// after which no further retries will be attempted
240    ///
241    /// This not only bounds the length of time before a server
242    /// error will be surfaced to the application, but also bounds
243    /// the length of time a request's credentials must remain valid.
244    ///
245    /// As requests are retried without renewing credentials or
246    /// regenerating request payloads, this number should be kept
247    /// below 5 minutes to avoid errors due to expired credentials
248    /// and/or request payloads
249    pub retry_timeout: Duration,
250}
251
252impl Default for RetryConfig {
253    fn default() -> Self {
254        Self {
255            backoff: Default::default(),
256            max_retries: 10,
257            retry_timeout: Duration::from_secs(3 * 60),
258        }
259    }
260}
261
262fn body_contains_error(response_body: &str) -> bool {
263    response_body.contains("InternalError") || response_body.contains("SlowDown")
264}
265
266/// Combines a [`RetryableRequest`] with a [`RetryContext`]
267pub(crate) struct RetryableRequestBuilder {
268    request: RetryableRequest,
269    context: RetryContext,
270}
271
272impl RetryableRequestBuilder {
273    /// Set whether this request is idempotent
274    ///
275    /// An idempotent request will be retried on timeout even if the request
276    /// method is not [safe](https://datatracker.ietf.org/doc/html/rfc7231#section-4.2.1)
277    pub(crate) fn idempotent(mut self, idempotent: bool) -> Self {
278        self.request.idempotent = Some(idempotent);
279        self
280    }
281
282    /// Set whether this request should be retried on a 409 Conflict response.
283    #[cfg(feature = "aws-base")]
284    pub(crate) fn retry_on_conflict(mut self, retry_on_conflict: bool) -> Self {
285        self.request.retry_on_conflict = retry_on_conflict;
286        self
287    }
288
289    /// Set whether this request contains sensitive data
290    ///
291    /// This will avoid printing out the URL in error messages
292    #[allow(unused)]
293    pub(crate) fn sensitive(mut self, sensitive: bool) -> Self {
294        self.request.sensitive = sensitive;
295        self
296    }
297
298    /// Provide a [`PutPayload`]
299    pub(crate) fn payload(mut self, payload: Option<PutPayload>) -> Self {
300        self.request.payload = payload;
301        self
302    }
303
304    #[allow(unused)]
305    pub(crate) fn retry_error_body(mut self, retry_error_body: bool) -> Self {
306        self.request.retry_error_body = retry_error_body;
307        self
308    }
309
310    pub(crate) async fn send(mut self) -> Result<HttpResponse> {
311        self.request.send(&mut self.context).await
312    }
313}
314
315/// A retryable request
316pub(crate) struct RetryableRequest {
317    client: HttpClient,
318    http: HttpRequest,
319
320    sensitive: bool,
321    idempotent: Option<bool>,
322    retry_on_conflict: bool,
323    payload: Option<PutPayload>,
324
325    retry_error_body: bool,
326}
327
328impl RetryableRequest {
329    #[allow(unused)]
330    pub(crate) fn sensitive(self, sensitive: bool) -> Self {
331        Self { sensitive, ..self }
332    }
333
334    fn err(&self, error: RequestError, ctx: &RetryContext) -> RetryError {
335        RetryError(Box::new(RetryErrorImpl {
336            uri: (!self.sensitive).then(|| self.http.uri().clone()),
337            method: self.http.method().clone(),
338            retries: ctx.retries,
339            max_retries: ctx.max_retries,
340            elapsed: ctx.start.elapsed(),
341            retry_timeout: ctx.retry_timeout,
342            inner: error,
343        }))
344    }
345
346    pub(crate) async fn send(self, ctx: &mut RetryContext) -> Result<HttpResponse> {
347        loop {
348            let mut request = self.http.clone();
349
350            if let Some(payload) = &self.payload {
351                *request.body_mut() = payload.clone().into();
352            }
353
354            match self.client.execute(request).await {
355                Ok(r) => {
356                    let status = r.status();
357                    if status.is_success() {
358                        // For certain S3 requests, 200 response may contain `InternalError` or
359                        // `SlowDown` in the message. These responses should be handled similarly
360                        // to r5xx errors.
361                        // More info here: https://repost.aws/knowledge-center/s3-resolve-200-internalerror
362                        if !self.retry_error_body {
363                            return Ok(r);
364                        }
365
366                        let (parts, body) = r.into_parts();
367                        let body = match body.text().await {
368                            Ok(body) => body,
369                            Err(e) => return Err(self.err(RequestError::Http(e), ctx)),
370                        };
371
372                        if !body_contains_error(&body) {
373                            // Success response and no error, clone and return response
374                            return Ok(HttpResponse::from_parts(parts, body.into()));
375                        } else {
376                            // Retry as if this was a 5xx response
377                            if ctx.exhausted() {
378                                return Err(self.err(RequestError::Response { body, status }, ctx));
379                            }
380
381                            let sleep = ctx.backoff();
382                            info!(
383                                "Encountered a response status of {} but body contains Error, backing off for {} seconds, retry {} of {}",
384                                status,
385                                sleep.as_secs_f32(),
386                                ctx.retries,
387                                ctx.max_retries,
388                            );
389                            tokio::time::sleep(sleep).await;
390                        }
391                    } else if status == StatusCode::NOT_MODIFIED {
392                        return Err(self.err(RequestError::Status { status, body: None }, ctx));
393                    } else if status.is_redirection() {
394                        let is_bare_redirect = !r.headers().contains_key(LOCATION);
395                        return match is_bare_redirect {
396                            true => Err(self.err(RequestError::BareRedirect, ctx)),
397                            false => Err(self.err(
398                                RequestError::Status {
399                                    body: None,
400                                    status: r.status(),
401                                },
402                                ctx,
403                            )),
404                        };
405                    } else {
406                        let status = r.status();
407                        if ctx.exhausted()
408                            || !(status.is_server_error()
409                                || status == StatusCode::TOO_MANY_REQUESTS
410                                || status == StatusCode::REQUEST_TIMEOUT
411                                || (self.retry_on_conflict && status == StatusCode::CONFLICT))
412                        {
413                            let source = match r.into_body().text().await {
414                                Ok(body) => RequestError::Status {
415                                    status,
416                                    body: Some(body),
417                                },
418                                Err(e) => RequestError::Http(e),
419                            };
420                            return Err(self.err(source, ctx));
421                        };
422
423                        let sleep = ctx.backoff();
424                        info!(
425                            "Encountered server error with status {}, backing off for {} seconds, retry {} of {}",
426                            status,
427                            sleep.as_secs_f32(),
428                            ctx.retries,
429                            ctx.max_retries,
430                        );
431                        tokio::time::sleep(sleep).await;
432                    }
433                }
434                Err(e) => {
435                    let is_idempotent = self
436                        .idempotent
437                        .unwrap_or_else(|| self.http.method().is_safe());
438
439                    let do_retry = match e.kind() {
440                        HttpErrorKind::Connect | HttpErrorKind::Request => true, // Request not sent, can retry
441                        HttpErrorKind::Timeout | HttpErrorKind::Interrupted => is_idempotent,
442                        HttpErrorKind::Unknown | HttpErrorKind::Decode => false,
443                    };
444
445                    if ctx.exhausted() || !do_retry {
446                        return Err(self.err(RequestError::Http(e), ctx));
447                    }
448                    let sleep = ctx.backoff();
449                    info!(
450                        "Encountered transport error of kind {:?}, backing off for {} seconds, retry {} of {}: {}",
451                        e.kind(),
452                        sleep.as_secs_f32(),
453                        ctx.retries,
454                        ctx.max_retries,
455                        e,
456                    );
457                    tokio::time::sleep(sleep).await;
458                }
459            }
460        }
461    }
462}
463
464pub(crate) trait RetryExt {
465    /// Return a [`RetryableRequestBuilder`]
466    fn retryable(self, config: &RetryConfig) -> RetryableRequestBuilder;
467
468    /// Return a [`RetryableRequest`]
469    fn retryable_request(self) -> RetryableRequest;
470
471    /// Dispatch a request with the given retry configuration
472    ///
473    /// # Panic
474    ///
475    /// This will panic if the request body is a stream
476    fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<HttpResponse>>;
477}
478
479impl RetryExt for HttpRequestBuilder {
480    fn retryable(self, config: &RetryConfig) -> RetryableRequestBuilder {
481        RetryableRequestBuilder {
482            request: self.retryable_request(),
483            context: RetryContext::new(config),
484        }
485    }
486
487    fn retryable_request(self) -> RetryableRequest {
488        let (client, request) = self.into_parts();
489        let request = request.expect("request must be valid");
490
491        RetryableRequest {
492            client,
493            http: request,
494            idempotent: None,
495            payload: None,
496            sensitive: false,
497            retry_on_conflict: false,
498            retry_error_body: false,
499        }
500    }
501
502    fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<HttpResponse>> {
503        let request = self.retryable(config);
504        Box::pin(async move { request.send().await })
505    }
506}
507
508#[cfg(not(target_arch = "wasm32"))]
509#[cfg(test)]
510mod tests {
511    use crate::RetryConfig;
512    use crate::client::mock_server::MockServer;
513    use crate::client::retry::{RequestError, RetryContext, RetryExt, body_contains_error};
514    use crate::client::{HttpClient, HttpError, HttpErrorKind, HttpResponse};
515    use http::Method;
516    use http::StatusCode;
517    use hyper::Response;
518    use hyper::header::LOCATION;
519    use hyper::server::conn::http1;
520    use hyper::service::service_fn;
521    use hyper_util::rt::TokioIo;
522    #[cfg(feature = "reqwest")]
523    use reqwest::Client;
524    use std::convert::Infallible;
525    use std::error::Error;
526    use std::time::Duration;
527    use tokio::net::TcpListener;
528    use tokio::time::timeout;
529
530    #[test]
531    fn test_body_contains_error() {
532        // Example error message provided by https://repost.aws/knowledge-center/s3-resolve-200-internalerror
533        let error_response = "AmazonS3Exception: We encountered an internal error. Please try again. (Service: Amazon S3; Status Code: 200; Error Code: InternalError; Request ID: 0EXAMPLE9AAEB265)";
534        assert!(body_contains_error(error_response));
535
536        let error_response_2 = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><Error><Code>SlowDown</Code><Message>Please reduce your request rate.</Message><RequestId>123</RequestId><HostId>456</HostId></Error>";
537        assert!(body_contains_error(error_response_2));
538
539        // Example success response from https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html
540        let success_response = "<CopyObjectResult><LastModified>2009-10-12T17:50:30.000Z</LastModified><ETag>\"9b2cf535f27731c974343645a3985328\"</ETag></CopyObjectResult>";
541        assert!(!body_contains_error(success_response));
542    }
543
544    #[cfg(feature = "reqwest")]
545    #[tokio::test]
546    async fn test_retry() {
547        let mock = MockServer::new().await;
548
549        let retry = RetryConfig {
550            backoff: Default::default(),
551            max_retries: 2,
552            retry_timeout: Duration::from_secs(1000),
553        };
554
555        let client = HttpClient::new(
556            Client::builder()
557                .timeout(Duration::from_millis(100))
558                .build()
559                .unwrap(),
560        );
561
562        let do_request = || client.request(Method::GET, mock.url()).send_retry(&retry);
563
564        // Simple request should work
565        let r = do_request().await.unwrap();
566        assert_eq!(r.status(), StatusCode::OK);
567
568        // Returns client errors immediately with a status message
569        mock.push(
570            Response::builder()
571                .status(StatusCode::BAD_REQUEST)
572                .body("cupcakes".to_string())
573                .unwrap(),
574        );
575
576        let e = do_request().await.unwrap_err();
577        assert_eq!(e.status().unwrap(), StatusCode::BAD_REQUEST);
578        assert_eq!(e.body(), Some("cupcakes"));
579        assert_eq!(
580            e.inner().to_string(),
581            "Server returned non-2xx status code: 400 Bad Request: cupcakes"
582        );
583
584        // Handles client errors with no payload
585        mock.push(
586            Response::builder()
587                .status(StatusCode::BAD_REQUEST)
588                .body("NAUGHTY NAUGHTY".to_string())
589                .unwrap(),
590        );
591
592        let e = do_request().await.unwrap_err();
593        assert_eq!(e.status().unwrap(), StatusCode::BAD_REQUEST);
594        assert_eq!(e.body(), Some("NAUGHTY NAUGHTY"));
595        assert_eq!(
596            e.inner().to_string(),
597            "Server returned non-2xx status code: 400 Bad Request: NAUGHTY NAUGHTY"
598        );
599
600        // Should retry server error request
601        mock.push(
602            Response::builder()
603                .status(StatusCode::BAD_GATEWAY)
604                .body(String::new())
605                .unwrap(),
606        );
607
608        let r = do_request().await.unwrap();
609        assert_eq!(r.status(), StatusCode::OK);
610
611        // Should retry 429 Too Many Requests
612        mock.push(
613            Response::builder()
614                .status(StatusCode::TOO_MANY_REQUESTS)
615                .body(String::new())
616                .unwrap(),
617        );
618
619        let r = do_request().await.unwrap();
620        assert_eq!(r.status(), StatusCode::OK);
621
622        // Should retry 408 Request Timeout
623        mock.push(
624            Response::builder()
625                .status(StatusCode::REQUEST_TIMEOUT)
626                .body(String::new())
627                .unwrap(),
628        );
629
630        let r = do_request().await.unwrap();
631        assert_eq!(r.status(), StatusCode::OK);
632
633        // Accepts 204 status code
634        mock.push(
635            Response::builder()
636                .status(StatusCode::NO_CONTENT)
637                .body(String::new())
638                .unwrap(),
639        );
640
641        let r = do_request().await.unwrap();
642        assert_eq!(r.status(), StatusCode::NO_CONTENT);
643
644        // Follows 402 redirects
645        mock.push(
646            Response::builder()
647                .status(StatusCode::FOUND)
648                .header(LOCATION, "/foo")
649                .body(String::new())
650                .unwrap(),
651        );
652
653        let r = do_request().await.unwrap();
654        assert_eq!(r.status(), StatusCode::OK);
655
656        // Follows 401 redirects
657        mock.push(
658            Response::builder()
659                .status(StatusCode::FOUND)
660                .header(LOCATION, "/bar")
661                .body(String::new())
662                .unwrap(),
663        );
664
665        let r = do_request().await.unwrap();
666        assert_eq!(r.status(), StatusCode::OK);
667
668        // Handles redirect loop
669        for _ in 0..11 {
670            mock.push(
671                Response::builder()
672                    .status(StatusCode::FOUND)
673                    .header(LOCATION, "/bar")
674                    .body(String::new())
675                    .unwrap(),
676            );
677        }
678
679        let e = do_request().await.unwrap_err().to_string();
680        assert!(e.contains("error following redirect"), "{}", e);
681
682        // Handles redirect missing location
683        mock.push(
684            Response::builder()
685                .status(StatusCode::FOUND)
686                .body(String::new())
687                .unwrap(),
688        );
689
690        let e = do_request().await.unwrap_err();
691        assert!(matches!(e.inner(), RequestError::BareRedirect));
692        assert_eq!(
693            e.inner().to_string(),
694            "Received redirect without LOCATION, this normally indicates an incorrectly configured region"
695        );
696
697        // Gives up after the retrying the specified number of times
698        for _ in 0..=retry.max_retries {
699            mock.push(
700                Response::builder()
701                    .status(StatusCode::BAD_GATEWAY)
702                    .body("ignored".to_string())
703                    .unwrap(),
704            );
705        }
706
707        let e = do_request().await.unwrap_err();
708        assert!(
709            e.to_string().contains(" after 2 retries, max_retries: 2, retry_timeout: 1000s  - Server returned non-2xx status code: 502 Bad Gateway: ignored"),
710            "{e}"
711        );
712        // verify e.source() is available as well for users who need programmatic access
713        assert_eq!(
714            e.source().unwrap().to_string(),
715            "Server returned non-2xx status code: 502 Bad Gateway: ignored",
716        );
717        // verify body is accessible
718        assert_eq!(e.body(), Some("ignored"));
719
720        // Panic results in an incomplete message error in the client
721        mock.push_fn::<_, String>(|_| panic!());
722        let r = do_request().await.unwrap();
723        assert_eq!(r.status(), StatusCode::OK);
724
725        // Gives up after retrying multiple panics
726        for _ in 0..=retry.max_retries {
727            mock.push_fn::<_, String>(|_| panic!());
728        }
729        let e = do_request().await.unwrap_err();
730        assert!(
731            e.to_string().contains("after 2 retries, max_retries: 2, retry_timeout: 1000s  - HTTP error: error sending request"),
732            "{e}"
733        );
734        // verify e.source() is available as well for users who need programmatic access
735        assert_eq!(
736            e.source().unwrap().to_string(),
737            "HTTP error: error sending request",
738        );
739        // the HttpError type needs to be directly accessible in the source chain
740        let mut err: &dyn Error = &e;
741        let mut found = false;
742        while let Some(source) = err.source() {
743            err = source;
744            if let Some(http_err) = err.downcast_ref::<HttpError>() {
745                assert_eq!(http_err.kind(), HttpErrorKind::Request);
746                found = true;
747                break;
748            }
749        }
750        assert!(found, "HttpError not found in source chain");
751
752        // Retries on client timeout
753        mock.push_async_fn(|_| async move {
754            tokio::time::sleep(Duration::from_secs(10)).await;
755            panic!()
756        });
757        do_request().await.unwrap();
758
759        // Does not retry PUT request
760        mock.push_async_fn(|_| async move {
761            tokio::time::sleep(Duration::from_secs(10)).await;
762            panic!()
763        });
764        let res = client.request(Method::PUT, mock.url()).send_retry(&retry);
765        let e = res.await.unwrap_err().to_string();
766        assert!(
767            !e.contains("retries") && e.contains("error sending request"),
768            "{e}"
769        );
770
771        let url = format!("{}/SENSITIVE", mock.url());
772        for _ in 0..=retry.max_retries {
773            mock.push(
774                Response::builder()
775                    .status(StatusCode::BAD_GATEWAY)
776                    .body("ignored".to_string())
777                    .unwrap(),
778            );
779        }
780        let res = client.request(Method::GET, url).send_retry(&retry).await;
781        let err = res.unwrap_err().to_string();
782        assert!(err.contains("SENSITIVE"), "{err}");
783
784        let url = format!("{}/SENSITIVE", mock.url());
785        for _ in 0..=retry.max_retries {
786            mock.push(
787                Response::builder()
788                    .status(StatusCode::BAD_GATEWAY)
789                    .body("ignored".to_string())
790                    .unwrap(),
791            );
792        }
793
794        // Sensitive requests should strip URL from error
795        let req = client
796            .request(Method::GET, &url)
797            .retryable(&retry)
798            .sensitive(true);
799        let err = req.send().await.unwrap_err().to_string();
800        assert!(!err.contains("SENSITIVE"), "{err}");
801
802        for _ in 0..=retry.max_retries {
803            mock.push_fn::<_, String>(|_| panic!());
804        }
805
806        let req = client
807            .request(Method::GET, &url)
808            .retryable(&retry)
809            .sensitive(true);
810        let err = req.send().await.unwrap_err().to_string();
811        assert!(!err.contains("SENSITIVE"), "{err}");
812
813        // Success response with error in body is retried
814        mock.push(
815            Response::builder()
816                .status(StatusCode::OK)
817                .body("InternalError".to_string())
818                .unwrap(),
819        );
820        let req = client
821            .request(Method::PUT, &url)
822            .retryable(&retry)
823            .idempotent(true)
824            .retry_error_body(true);
825        let r = req.send().await.unwrap();
826        assert_eq!(r.status(), StatusCode::OK);
827        // Response with InternalError should have been retried
828        let b = r.into_body().text().await.unwrap();
829        assert!(!b.contains("InternalError"));
830
831        // Should not retry success response with no error in body
832        mock.push(
833            Response::builder()
834                .status(StatusCode::OK)
835                .body("success".to_string())
836                .unwrap(),
837        );
838        let req = client
839            .request(Method::PUT, &url)
840            .retryable(&retry)
841            .idempotent(true)
842            .retry_error_body(true);
843        let r = req.send().await.unwrap();
844        assert_eq!(r.status(), StatusCode::OK);
845        let b = r.into_body().text().await.unwrap();
846        assert!(b.contains("success"));
847
848        // Shutdown
849        mock.shutdown().await
850    }
851
852    #[cfg(feature = "reqwest")]
853    #[tokio::test]
854    async fn test_503_error_body_captured() {
855        let mock = MockServer::new().await;
856
857        let retry = RetryConfig {
858            backoff: Default::default(),
859            max_retries: 0,
860            retry_timeout: Duration::from_secs(1000),
861        };
862
863        let client = HttpClient::new(Client::builder().build().unwrap());
864
865        // Test that 503 SlowDown body is captured for throttling detection
866        let slowdown_body = r#"<?xml version="1.0" encoding="UTF-8"?><Error><Code>SlowDown</Code><Message>Please reduce your request rate.</Message></Error>"#;
867        mock.push(
868            Response::builder()
869                .status(StatusCode::SERVICE_UNAVAILABLE)
870                .body(slowdown_body.to_string())
871                .unwrap(),
872        );
873
874        let e = client
875            .request(Method::GET, mock.url())
876            .send_retry(&retry)
877            .await
878            .unwrap_err();
879
880        assert_eq!(e.status().unwrap(), StatusCode::SERVICE_UNAVAILABLE);
881        assert_eq!(e.body(), Some(slowdown_body));
882        assert!(e.body().unwrap().contains("SlowDown"));
883
884        mock.shutdown().await
885    }
886
887    #[cfg(feature = "reqwest")]
888    #[tokio::test]
889    #[expect(
890        deprecated,
891        reason = "SO_LINGER w/ zero timeout doesn't block, see https://github.com/tokio-rs/tokio/issues/7751#issuecomment-3709831265"
892    )]
893    async fn test_connection_reset_is_retried() {
894        let retry = RetryConfig {
895            backoff: Default::default(),
896            max_retries: 2,
897            retry_timeout: Duration::from_secs(1),
898        };
899        assert!(retry.max_retries > 0);
900
901        // Setup server which resets a connection and then quits
902        let listener = TcpListener::bind("::1:0").await.unwrap();
903        let url = format!("http://{}", listener.local_addr().unwrap());
904        let handle = tokio::spawn(async move {
905            // Reset the connection on the first n-1 attempts
906            for _ in 0..retry.max_retries {
907                let (stream, _) = listener.accept().await.unwrap();
908                stream.set_linger(Some(Duration::from_secs(0))).unwrap();
909            }
910            // Succeed on the last attempt
911            let (stream, _) = listener.accept().await.unwrap();
912            http1::Builder::new()
913                // we want the connection to end after responding
914                .keep_alive(false)
915                .serve_connection(
916                    TokioIo::new(stream),
917                    service_fn(move |_req| async {
918                        Ok::<_, Infallible>(HttpResponse::new("Success!".to_string().into()))
919                    }),
920                )
921                .await
922                .unwrap();
923        });
924
925        // Perform the request
926        let client = HttpClient::new(reqwest::Client::new());
927        let ctx = &mut RetryContext::new(&retry);
928        let res = client
929            .get(url)
930            .retryable_request()
931            .send(ctx)
932            .await
933            .expect("request should eventually succeed");
934        assert_eq!(res.status(), StatusCode::OK);
935        assert!(ctx.exhausted());
936
937        // Wait for server to shutdown
938        let _ = timeout(Duration::from_secs(1), handle)
939            .await
940            .expect("shutdown shouldn't hang");
941    }
942}
object_store/client/retry.rs

object_store/client/
retry.rs
