[ENH] Explicitly convert SlowDown from S3 to StorageError::Backoff (#5377)

## Description of changes

Amazon periodically tells us to slow down and that manifests as a
gruesome error trace.  Make a specific error type so callers can handle
the error.

## Test plan

CI

## Migration plan

N/A

## Observability plan

N/A

## Documentation Changes

N/A

Author: rescrv

rust/storage/src/lib.rs

@@ -145,6 +145,10 @@ pub enum StorageError {
         /// The configuration key used
         key: String,
     },
+
+    // Back off and retry---usually indicates an explicit 429/SlowDown.
+    #[error("Back off and retry---usually indicates an explicit 429/SlowDown.")]
+    Backoff,
 }
 
 impl ChromaError for StorageError {
@@ -163,6 +167,7 @@ impl ChromaError for StorageError {
             StorageError::PermissionDenied { .. } => ErrorCodes::PermissionDenied,
             StorageError::Unauthenticated { .. } => ErrorCodes::Unauthenticated,
             StorageError::UnknownConfigurationKey { .. } => ErrorCodes::InvalidArgument,
+            StorageError::Backoff { .. } => ErrorCodes::ResourceExhausted,
         }
     }
 }

rust/storage/src/s3.rs

@@ -238,36 +238,40 @@ impl S3Storage {
                     res.e_tag.map(ETag),
                 ))
             }
-            Err(e) => {
-                match e {
-                    SdkError::ServiceError(err) => {
-                        let inner = err.into_err();
-                        match &inner {
-                            aws_sdk_s3::operation::get_object::GetObjectError::NoSuchKey(_) => {
-                                Err(StorageError::NotFound {
-                                    path: key.to_string(),
-                                    source: Arc::new(inner),
-                                })
-                            }
-                            aws_sdk_s3::operation::get_object::GetObjectError::InvalidObjectState(msg) => {
-                                tracing::error!("invalid object state: {}", msg);
-                                Err(StorageError::Generic {
-                                    source: Arc::new(inner),
-                                })
-                            }
-                            _ => {
+            Err(e) => match e {
+                SdkError::ServiceError(err) => {
+                    let inner = err.into_err();
+                    match &inner {
+                        aws_sdk_s3::operation::get_object::GetObjectError::NoSuchKey(_) => {
+                            Err(StorageError::NotFound {
+                                path: key.to_string(),
+                                source: Arc::new(inner),
+                            })
+                        }
+                        aws_sdk_s3::operation::get_object::GetObjectError::InvalidObjectState(
+                            msg,
+                        ) => {
+                            tracing::error!("invalid object state: {}", msg);
+                            Err(StorageError::Generic {
+                                source: Arc::new(inner),
+                            })
+                        }
+                        _ => {
+                            if inner.code() == Some("SlowDown") {
+                                Err(StorageError::Backoff)
+                            } else {
                                 tracing::error!("error: {}", inner.to_string());
                                 Err(StorageError::Generic {
                                     source: Arc::new(inner),
                                 })
                             }
                         }
                     }
-                    _ => Err(StorageError::Generic {
-                        source: Arc::new(e),
-                    }),
                 }
-            }
+                _ => Err(StorageError::Generic {
+                    source: Arc::new(e),
+                }),
+            },
         }
     }
 
@@ -350,9 +354,13 @@ impl S3Storage {
                                 })
                             }
                             _ => {
-                                Err(StorageError::Generic {
-                                    source: Arc::new(inner),
-                                })
+                                if inner.code() == Some("SlowDown") {
+                                    Err(StorageError::Backoff)
+                                } else {
+                                    Err(StorageError::Generic {
+                                        source: Arc::new(inner),
+                                    })
+                                }
                             }
                         }
                     }
@@ -602,6 +610,8 @@ impl S3Storage {
                     path: key.to_string(),
                     source: Arc::new(err),
                 }
+            } else if err.meta().code() == Some("SlowDown") {
+                StorageError::Backoff
             } else {
                 StorageError::Generic {
                     source: Arc::new(err),
@@ -793,6 +803,7 @@ impl S3Storage {
                             path: key.to_string(),
                             source: Arc::new(inner),
                         }),
+                        Some("SlowDown") => Err(StorageError::Backoff),
                         _ => {
                             tracing::error!(error = %inner, key = %key, "Failed to delete object from S3");
                             Err(StorageError::Generic {
@@ -864,9 +875,15 @@ impl S3Storage {
                 tracing::trace!("Successfully deleted objects from S3");
                 Ok(out)
             }
-            Err(e) => Err(StorageError::Generic {
-                source: Arc::new(e),
-            }),
+            Err(e) => {
+                if e.code() == Some("SlowDown") {
+                    Err(StorageError::Backoff)
+                } else {
+                    Err(StorageError::Generic {
+                        source: Arc::new(e),
+                    })
+                }
+            }
         }
     }