storage: delete_prefix, more retries, reduce concurrency

crates/bin/docs_rs_watcher/src/main.rs

@@ -4,13 +4,14 @@ use docs_rs_config::AppConfig as _;
 use docs_rs_context::Context;
 use docs_rs_types::{KrateName, Version};
 use docs_rs_watcher::{Config, Index, index_watcher};
+use futures_util::FutureExt as _;
 use std::sync::Arc;
 
 #[tokio::main]
 async fn main() -> Result<()> {
     let _guard = docs_rs_logging::init_from_environment().context("error initializing logging")?;
 
-    if let Err(err) = CommandLine::parse().handle_args().await {
+    if let Err(err) = CommandLine::parse().handle_args().boxed().await {
         eprintln!("error running watcher: {:?}", err);
         drop(_guard);
         std::process::exit(1);

crates/lib/docs_rs_storage/src/backends/s3.rs

@@ -557,9 +557,7 @@ impl StorageBackendMethods for S3Backend {
         stream
             .chunks(S3_DELETE_OBJECTS_LIMIT)
             .map(|batch| batch.into_iter().collect::<Result<Vec<_>, Error>>())
-            .try_for_each_concurrent(Some(self.network_parallelism), |batch| {
-                self.delete_batch_with_retry(batch)
-            })
+            .try_for_each(|batch| self.delete_batch_with_retry(batch))
             .await
     }
 }
@@ -568,13 +566,30 @@ impl S3Backend {
     async fn delete_batch_with_retry(&self, keys: Vec<String>) -> Result<(), Error> {
         let mut remaining = keys;
         for attempt in 1.. {
-            let resp = self
+            let resp = match self
                 .client
                 .delete_objects()
                 .bucket(&self.bucket)
-                .delete(Self::delete_request(remaining)?)
+                .delete(Self::delete_request(&remaining)?)
                 .send()
-                .await?;
+                .await
+            {
+                Ok(resp) => resp,
+                Err(err)
+                    if attempt <= self.max_retries && Self::is_retryable_delete_error(&err) =>
+                {
+                    let backoff = retry_backoff(attempt);
+                    warn!(
+                        attempt,
+                        ?backoff,
+                        ?err,
+                        "retrying s3 delete_objects request after transient error",
+                    );
+                    time::sleep(backoff).await;
+                    continue;
+                }
+                Err(err) => return Err(err.into()),
+            };
 
             let Some(errs) = resp.errors.filter(|e| !e.is_empty()) else {
                 return Ok(());
@@ -625,12 +640,31 @@ impl S3Backend {
         unreachable!("unbounded retry loop should return or fail internally")
     }
 
-    fn delete_request(keys: Vec<String>) -> Result<Delete, Error> {
+    fn is_retryable_delete_error<E>(err: &SdkError<E>) -> bool
+    where
+        E: ProvideErrorMetadata + std::error::Error + Send + Sync + 'static,
+    {
+        if let Some(err_code) = err.code()
+            && RETRYABLE_DELETE_OBJECT_ERROR_CODES.contains(&err_code)
+        {
+            return true;
+        }
+
+        if let SdkError::ServiceError(err) = err
+            && err.raw().status().is_server_error()
+        {
+            return true;
+        }
+
+        false
+    }
+
+    fn delete_request(keys: &[String]) -> Result<Delete, Error> {
         let objects: Vec<_> = keys
-            .into_iter()
+            .iter()
             .map(|key| {
                 ObjectIdentifier::builder()
-                    .key(key)
+                    .key(key.clone())
                     .build()
                     .expect("can never fail, the keys come from list_prefix")
             })