feat(object_store): add PermissionDenied variant to top-level error (…

…#6194)

* feat(object_store):  add `PermissionDenied` variant to top-level error

* Update object_store/src/lib.rs

Co-authored-by: Raphael Taylor-Davies <1781103+tustvold@users.noreply.github.com>

* refactor: add additional error variant for unauthenticated ops

* fix: include path in unauthenticated error

---------

Co-authored-by: Raphael Taylor-Davies <1781103+tustvold@users.noreply.github.com>

object_store/src/client/retry.rs

@@ -86,6 +86,14 @@ impl Error {
                 path,
                 source: Box::new(self),
             },
+            Some(StatusCode::FORBIDDEN) => crate::Error::PermissionDenied {
+                path,
+                source: Box::new(self),
+            },
+            Some(StatusCode::UNAUTHORIZED) => crate::Error::Unauthenticated {
+                path,
+                source: Box::new(self),
+            },
             _ => crate::Error::Generic {
                 store,
                 source: Box::new(self),
@@ -106,6 +114,10 @@ impl From<Error> for std::io::Error {
                 status: StatusCode::BAD_REQUEST,
                 ..
             } => Self::new(ErrorKind::InvalidInput, err),
+            Error::Client {
+                status: StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN,
+                ..
+            } => Self::new(ErrorKind::PermissionDenied, err),
             Error::Reqwest { source, .. } if source.is_timeout() => {
                 Self::new(ErrorKind::TimedOut, err)
             }

object_store/src/lib.rs

@@ -1274,6 +1274,26 @@ pub enum Error {
     #[snafu(display("Operation not yet implemented."))]
     NotImplemented,
 
+    #[snafu(display(
+        "The operation lacked the necessary privileges to complete for path {}: {}",
+        path,
+        source
+    ))]
+    PermissionDenied {
+        path: String,
+        source: Box<dyn std::error::Error + Send + Sync + 'static>,
+    },
+
+    #[snafu(display(
+        "The operation lacked valid authentication credentials for path {}: {}",
+        path,
+        source
+    ))]
+    Unauthenticated {
+        path: String,
+        source: Box<dyn std::error::Error + Send + Sync + 'static>,
+    },
+
     #[snafu(display("Configuration key: '{}' is not valid for store '{}'.", key, store))]
     UnknownConfigurationKey { store: &'static str, key: String },
 }