refactor: Use loop instead of futures_retry

Cargo.lock

@@ -1202,17 +1202,6 @@ dependencies = [
  "syn",
 ]
 
-[[package]]
-name = "futures-retry"
-version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fde5a672a61f96552aa5ed9fd9c81c3fbdae4be9b1e205d6eaf17c83705adc0f"
-dependencies = [
- "futures",
- "pin-project-lite",
- "tokio",
-]
-
 [[package]]
 name = "futures-sink"
 version = "0.3.17"
@@ -2495,7 +2484,6 @@ dependencies = [
  "cloud-storage",
  "dotenv",
  "futures",
- "futures-retry",
  "futures-test",
  "indexmap",
  "itertools",

object_store/Cargo.toml

@@ -15,7 +15,6 @@ chrono = "0.4"
 # Google Cloud Storage integration
 cloud-storage = {version = "0.10.2", optional = true}
 futures = "0.3"
-futures-retry = "0.6"
 # https://github.com/tkaitchuck/aHash/issues/95
 indexmap = { version = "~1.6.2", optional = true }
 itertools = "0.10.1"

object_store/src/aws.rs

@@ -12,7 +12,6 @@ use futures::{
     stream::{self, BoxStream},
     Future, Stream, StreamExt, TryStreamExt,
 };
-use futures_retry::{FutureRetry, RetryPolicy};
 use rusoto_core::ByteStream;
 use rusoto_credential::{InstanceMetadataProvider, StaticProvider};
 use rusoto_s3::S3;
@@ -480,20 +479,17 @@ where
     // TODO: make the number of retries configurable
     let n_retries = 10;
 
-    FutureRetry::new(
-        move || {
-            let future_factory = future_factory.clone();
+    loop {
+        let future_factory = future_factory.clone();
+        let request = future_factory().await?;
 
-            async move {
-                let request = future_factory().await?;
+        let result = request.await;
 
-                request.await
-            }
-        },
-        // retry
-        {
-            move |e| {
+        match result {
+            Ok(r) => return Ok(r),
+            Err(e) => {
                 attempts += 1;
+
                 let should_retry = matches!(
                     e,
                     rusoto_core::RusotoError::Unknown(ref response)
@@ -501,17 +497,14 @@ where
                 );
 
                 if attempts > n_retries || !should_retry {
-                    RetryPolicy::ForwardError(e)
+                    return Err(e);
                 } else {
-                    RetryPolicy::WaitRetry(Duration::from_millis(2u64.pow(attempts) * 50))
+                    let wait_time = Duration::from_millis(2u64.pow(attempts) * 50);
+                    tokio::time::sleep(wait_time).await;
                 }
             }
-        },
-    )
-    .await
-    // TODO: log number of attempts?
-    .map(|(response, _attempts)| response)
-    .map_err(|(err, _attempts)| err)
+        }
+    }
 }
 
 impl Error {