s3/request/
mod.rs

1#[cfg(feature = "with-async-std")]
2pub mod async_std_backend;
3#[cfg(feature = "sync")]
4pub mod blocking;
5pub mod request_trait;
6#[cfg(feature = "with-tokio")]
7pub mod tokio_backend;
8
9pub use request_trait::*;
10
11#[cfg(test)]
12mod retry_tests;
13
14use crate::command::Command;
15use crate::error::S3Error;
16
17const RETRYABLE_HTTP_STATUSES: [u16; 6] = [408, 429, 500, 502, 503, 504];
18
19pub(crate) struct RequestAttemptError {
20    pub(crate) error: S3Error,
21    retryable: Option<bool>,
22}
23
24impl RequestAttemptError {
25    pub(crate) fn do_not_retry(error: S3Error) -> Self {
26        Self {
27            error,
28            retryable: Some(false),
29        }
30    }
31
32    #[cfg(feature = "with-async-std")]
33    pub(crate) fn surf_send_error(error: surf::Error) -> Self {
34        // Surf's transport stack can report connection-close failures without
35        // retaining an io::Error source. At this boundary errors come only
36        // from send(); replay remains restricted to safe operations.
37        Self {
38            error: S3Error::Surf(error.to_string()),
39            retryable: Some(true),
40        }
41    }
42
43    fn should_retry(&self, command: &Command<'_>) -> bool {
44        command.is_retry_safe()
45            && self
46                .retryable
47                .unwrap_or_else(|| retryable_request_error(command, &self.error))
48    }
49}
50
51impl From<S3Error> for RequestAttemptError {
52    fn from(error: S3Error) -> Self {
53        Self {
54            error,
55            retryable: None,
56        }
57    }
58}
59
60#[cfg(feature = "with-tokio")]
61impl From<reqwest::Error> for RequestAttemptError {
62    fn from(error: reqwest::Error) -> Self {
63        S3Error::Reqwest(error).into()
64    }
65}
66
67#[cfg(feature = "sync")]
68impl From<attohttpc::Error> for RequestAttemptError {
69    fn from(error: attohttpc::Error) -> Self {
70        S3Error::Atto(error).into()
71    }
72}
73
74fn retryable_request_error(command: &Command<'_>, error: &S3Error) -> bool {
75    if !command.is_retry_safe() {
76        return false;
77    }
78
79    match error {
80        S3Error::HttpFailWithBody(status, _) => RETRYABLE_HTTP_STATUSES.contains(status),
81        #[cfg(feature = "with-tokio")]
82        S3Error::Reqwest(error) => {
83            error.is_connect()
84                || error.is_timeout()
85                || (error.is_request() && !error.is_builder() && error.url().is_some())
86                || reqwest_has_transient_io(error)
87        }
88        #[cfg(feature = "with-async-std")]
89        S3Error::Surf(_) => false,
90        #[cfg(feature = "sync")]
91        S3Error::Atto(error) => match error.kind() {
92            attohttpc::ErrorKind::Io(error) => is_transient_io_kind(error.kind()),
93            _ => false,
94        },
95        _ => false,
96    }
97}
98
99#[cfg(any(feature = "with-tokio", feature = "sync"))]
100fn is_transient_io_kind(kind: std::io::ErrorKind) -> bool {
101    matches!(
102        kind,
103        std::io::ErrorKind::ConnectionRefused
104            | std::io::ErrorKind::ConnectionReset
105            | std::io::ErrorKind::ConnectionAborted
106            | std::io::ErrorKind::TimedOut
107            | std::io::ErrorKind::UnexpectedEof
108            | std::io::ErrorKind::BrokenPipe
109            | std::io::ErrorKind::NotConnected
110            | std::io::ErrorKind::NetworkUnreachable
111            | std::io::ErrorKind::HostUnreachable
112            | std::io::ErrorKind::AddrNotAvailable
113            | std::io::ErrorKind::Interrupted
114            | std::io::ErrorKind::WouldBlock
115    )
116}
117
118#[cfg(feature = "with-tokio")]
119fn reqwest_has_transient_io(error: &reqwest::Error) -> bool {
120    let mut source = std::error::Error::source(error);
121    while let Some(current) = source {
122        if let Some(io_error) = current.downcast_ref::<std::io::Error>() {
123            return is_transient_io_kind(io_error.kind());
124        }
125        source = current.source();
126    }
127    false
128}
129
130#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
131pub(crate) async fn retry_request<T, F, Fut>(
132    command: &Command<'_>,
133    mut attempt: F,
134) -> Result<T, S3Error>
135where
136    F: FnMut() -> Fut,
137    Fut: std::future::Future<Output = Result<T, RequestAttemptError>>,
138{
139    let mut retry_count = 0u64;
140    let max_retries = crate::get_retries() as u64;
141
142    loop {
143        match attempt().await {
144            Ok(value) => return Ok(value),
145            Err(error) if retry_count < max_retries && error.should_retry(command) => {
146                retry_count += 1;
147                log::warn!("Retrying safe S3 request after transient failure");
148                let delay = std::time::Duration::from_secs(retry_count.pow(2));
149                #[cfg(feature = "with-tokio")]
150                tokio::time::sleep(delay).await;
151                #[cfg(all(not(feature = "with-tokio"), feature = "with-async-std"))]
152                async_std::task::sleep(delay).await;
153            }
154            Err(error) => return Err(error.error),
155        }
156    }
157}
158
159#[cfg(feature = "sync")]
160pub(crate) fn retry_request_sync<T, F>(command: &Command<'_>, mut attempt: F) -> Result<T, S3Error>
161where
162    F: FnMut() -> Result<T, RequestAttemptError>,
163{
164    let mut retry_count = 0u64;
165    let max_retries = crate::get_retries() as u64;
166
167    loop {
168        match attempt() {
169            Ok(value) => return Ok(value),
170            Err(error) if retry_count < max_retries && error.should_retry(command) => {
171                retry_count += 1;
172                log::warn!("Retrying safe S3 request after transient failure");
173                std::thread::sleep(std::time::Duration::from_secs(retry_count.pow(2)));
174            }
175            Err(error) => return Err(error.error),
176        }
177    }
178}
179
180#[cfg(test)]
181mod retry_policy_tests {
182    use super::*;
183
184    #[test]
185    fn request_retry_policy_requires_safe_operation_and_transient_status() {
186        for status in [408, 429, 500, 502, 503, 504] {
187            assert!(retryable_request_error(
188                &Command::GetObject,
189                &S3Error::HttpFailWithBody(status, "retry".to_owned())
190            ));
191        }
192        for status in [200, 400, 401, 403, 404, 5000, 501] {
193            assert!(!retryable_request_error(
194                &Command::GetObject,
195                &S3Error::HttpFailWithBody(status, "permanent".to_owned())
196            ));
197        }
198        assert!(!retryable_request_error(
199            &Command::DeleteObject,
200            &S3Error::HttpFailWithBody(503, "mutation".to_owned())
201        ));
202        assert!(!retryable_request_error(
203            &Command::InitiateMultipartUpload {
204                content_type: "application/octet-stream",
205            },
206            &S3Error::HttpFailWithBody(503, "mutation".to_owned())
207        ));
208        assert!(!retryable_request_error(
209            &Command::UploadPart {
210                part_number: 1,
211                content: b"part",
212                upload_id: "upload-id",
213            },
214            &S3Error::HttpFailWithBody(503, "legacy command".to_owned())
215        ));
216    }
217
218    #[test]
219    fn request_retry_upload_part_is_safe_but_ordinary_put_is_not() {
220        use crate::command::Multipart;
221
222        let part = Command::PutObject {
223            content: b"part",
224            content_type: "application/octet-stream",
225            custom_headers: None,
226            multipart: Some(Multipart::new(1, "upload-id")),
227        };
228        let ordinary_put = Command::PutObject {
229            content: b"object",
230            content_type: "application/octet-stream",
231            custom_headers: None,
232            multipart: None,
233        };
234
235        assert!(part.is_retry_safe());
236        assert!(!ordinary_put.is_retry_safe());
237    }
238
239    #[cfg(feature = "sync")]
240    #[test]
241    fn request_retry_atto_policy_only_accepts_transport_io_errors() {
242        let transport = attohttpc::Error::from(std::io::Error::new(
243            std::io::ErrorKind::ConnectionReset,
244            "connection reset",
245        ));
246        let invalid_url = attohttpc::Error::from(attohttpc::ErrorKind::InvalidBaseUrl);
247
248        assert!(retryable_request_error(
249            &Command::GetObject,
250            &S3Error::Atto(transport)
251        ));
252        assert!(!retryable_request_error(
253            &Command::GetObject,
254            &S3Error::Atto(invalid_url)
255        ));
256    }
257
258    #[cfg(feature = "with-tokio")]
259    #[tokio::test]
260    async fn request_retry_safe_request_retries_transient_status_once() {
261        use std::sync::Arc;
262        use std::sync::atomic::{AtomicUsize, Ordering};
263
264        let attempts = Arc::new(AtomicUsize::new(0));
265        let observed = attempts.clone();
266        let result = retry_request(&Command::GetObject, || {
267            let current = observed.fetch_add(1, Ordering::SeqCst);
268            async move {
269                if current == 0 {
270                    Err(S3Error::HttpFailWithBody(503, "temporary".to_owned()).into())
271                } else {
272                    Ok(200)
273                }
274            }
275        })
276        .await
277        .unwrap();
278
279        assert_eq!(result, 200);
280        assert_eq!(attempts.load(Ordering::SeqCst), 2);
281    }
282
283    #[cfg(feature = "with-tokio")]
284    #[tokio::test]
285    async fn request_retry_mutation_and_body_read_error_are_not_replayed() {
286        use std::sync::Arc;
287        use std::sync::atomic::{AtomicUsize, Ordering};
288
289        let attempts = Arc::new(AtomicUsize::new(0));
290        let observed = attempts.clone();
291        let result: Result<(), S3Error> = retry_request(&Command::DeleteObject, || {
292            observed.fetch_add(1, Ordering::SeqCst);
293            async {
294                Err::<(), RequestAttemptError>(
295                    S3Error::HttpFailWithBody(503, "service error".to_owned()).into(),
296                )
297            }
298        })
299        .await;
300        assert!(
301            matches!(result, Err(S3Error::HttpFailWithBody(503, body)) if body == "service error")
302        );
303        assert_eq!(attempts.load(Ordering::SeqCst), 1);
304
305        let attempts = AtomicUsize::new(0);
306        let result: Result<(), S3Error> = retry_request(&Command::GetObject, || {
307            attempts.fetch_add(1, Ordering::SeqCst);
308            async {
309                Err::<(), RequestAttemptError>(RequestAttemptError::do_not_retry(S3Error::Io(
310                    std::io::Error::other("original body read diagnostic"),
311                )))
312            }
313        })
314        .await;
315        assert!(
316            matches!(result, Err(S3Error::Io(error)) if error.to_string() == "original body read diagnostic")
317        );
318        assert_eq!(attempts.load(Ordering::SeqCst), 1);
319    }
320
321    #[cfg(feature = "with-async-std")]
322    #[async_std::test]
323    async fn request_retry_safe_request_retries_transient_status_once() {
324        use std::sync::Arc;
325        use std::sync::atomic::{AtomicUsize, Ordering};
326
327        let attempts = Arc::new(AtomicUsize::new(0));
328        let observed = attempts.clone();
329        let result = retry_request(&Command::GetObject, || {
330            let current = observed.fetch_add(1, Ordering::SeqCst);
331            async move {
332                if current == 0 {
333                    Err(S3Error::HttpFailWithBody(503, "temporary".to_owned()).into())
334                } else {
335                    Ok(200)
336                }
337            }
338        })
339        .await
340        .unwrap();
341
342        assert_eq!(result, 200);
343        assert_eq!(attempts.load(Ordering::SeqCst), 2);
344    }
345
346    #[cfg(feature = "with-async-std")]
347    #[async_std::test]
348    async fn request_retry_mutation_and_body_read_error_are_not_replayed() {
349        use std::sync::Arc;
350        use std::sync::atomic::{AtomicUsize, Ordering};
351
352        let attempts = Arc::new(AtomicUsize::new(0));
353        let observed = attempts.clone();
354        let result: Result<(), S3Error> = retry_request(&Command::DeleteObject, || {
355            observed.fetch_add(1, Ordering::SeqCst);
356            async {
357                Err::<(), RequestAttemptError>(
358                    S3Error::HttpFailWithBody(503, "service error".to_owned()).into(),
359                )
360            }
361        })
362        .await;
363        assert!(
364            matches!(result, Err(S3Error::HttpFailWithBody(503, body)) if body == "service error")
365        );
366        assert_eq!(attempts.load(Ordering::SeqCst), 1);
367
368        let attempts = AtomicUsize::new(0);
369        let result: Result<(), S3Error> = retry_request(&Command::GetObject, || {
370            attempts.fetch_add(1, Ordering::SeqCst);
371            async {
372                Err::<(), RequestAttemptError>(RequestAttemptError::do_not_retry(S3Error::Io(
373                    std::io::Error::other("original body read diagnostic"),
374                )))
375            }
376        })
377        .await;
378        assert!(
379            matches!(result, Err(S3Error::Io(error)) if error.to_string() == "original body read diagnostic")
380        );
381        assert_eq!(attempts.load(Ordering::SeqCst), 1);
382    }
383
384    #[cfg(feature = "sync")]
385    #[test]
386    fn request_retry_mutation_does_not_retry_transient_status() {
387        use std::sync::atomic::{AtomicUsize, Ordering};
388
389        let attempts = AtomicUsize::new(0);
390        let result: Result<(), S3Error> = retry_request_sync(&Command::DeleteObject, || {
391            attempts.fetch_add(1, Ordering::SeqCst);
392            Err(S3Error::HttpFailWithBody(503, "do not replay".to_owned()).into())
393        });
394
395        assert!(
396            matches!(result, Err(S3Error::HttpFailWithBody(503, body)) if body == "do not replay")
397        );
398        assert_eq!(attempts.load(Ordering::SeqCst), 1);
399    }
400}
s3/request/mod.rs

s3/request/
mod.rs
