refactor(object_store): refactor timeout and retry of object store interface

src/common/src/config.rs

@@ -982,6 +982,69 @@ pub struct ObjectStoreConfig {
     #[serde(default = "default::object_store_config::object_store_set_atomic_write_dir")]
     pub object_store_set_atomic_write_dir: bool,
 
+    #[serde(default = "default::object_store_config::object_store_req_retry_interval_ms")]
+    pub object_store_req_retry_interval_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_req_retry_max_delay_ms")]
+    pub object_store_req_retry_max_delay_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_req_retry_max_retry_attempts")]
+    pub object_store_req_retry_max_retry_attempts: usize,
+
+    // upload
+    #[serde(default = "default::object_store_config::object_store_upload_attempt_timeout_ms")]
+    pub object_store_upload_attempt_timeout_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_upload_retry_attempts")]
+    pub object_store_upload_retry_attempts: usize,
+
+    // streaming_upload_init + streaming_upload
+    #[serde(
+        default = "default::object_store_config::object_store_streaming_upload_attempt_timeout_ms"
+    )]
+    pub object_store_streaming_upload_attempt_timeout_ms: u64,
+    #[serde(
+        default = "default::object_store_config::object_store_streaming_upload_retry_attempts"
+    )]
+    pub object_store_streaming_upload_retry_attempts: usize,
+
+    // read
+    #[serde(default = "default::object_store_config::object_store_read_attempt_timeout_ms")]
+    pub object_store_read_attempt_timeout_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_read_retry_attempts")]
+    pub object_store_read_retry_attempts: usize,
+
+    // streaming_read_init + streaming_read
+    #[serde(
+        default = "default::object_store_config::object_store_streaming_read_attempt_timeout_ms"
+    )]
+    pub object_store_streaming_read_attempt_timeout_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_streaming_read_retry_attempts")]
+    pub object_store_streaming_read_retry_attempts: usize,
+
+    // metadata
+    #[serde(default = "default::object_store_config::object_store_metadata_attempt_timeout_ms")]
+    pub object_store_metadata_attempt_timeout_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_metadata_retry_attempts")]
+    pub object_store_metadata_retry_attempts: usize,
+
+    // delete
+    #[serde(default = "default::object_store_config::object_store_delete_attempt_timeout_ms")]
+    pub object_store_delete_attempt_timeout_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_delete_retry_attempts")]
+    pub object_store_delete_retry_attempts: usize,
+
+    // delete_object
+    #[serde(
+        default = "default::object_store_config::object_store_delete_objects_attempt_timeout_ms"
+    )]
+    pub object_store_delete_objects_attempt_timeout_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_delete_objects_retry_attempts")]
+    pub object_store_delete_objects_retry_attempts: usize,
+
+    // list
+    #[serde(default = "default::object_store_config::object_store_list_attempt_timeout_ms")]
+    pub object_store_list_attempt_timeout_ms: u64,
+    #[serde(default = "default::object_store_config::object_store_list_retry_attempts")]
+    pub object_store_list_retry_attempts: usize,
+
     #[serde(default)]
     pub s3: S3ObjectStoreConfig,
 }
@@ -1003,12 +1066,6 @@ pub struct S3ObjectStoreConfig {
     pub object_store_send_buffer_size: Option<usize>,
     #[serde(default = "default::object_store_config::s3::object_store_nodelay")]
     pub object_store_nodelay: Option<bool>,
-    #[serde(default = "default::object_store_config::s3::object_store_req_retry_interval_ms")]
-    pub object_store_req_retry_interval_ms: u64,
-    #[serde(default = "default::object_store_config::s3::object_store_req_retry_max_delay_ms")]
-    pub object_store_req_retry_max_delay_ms: u64,
-    #[serde(default = "default::object_store_config::s3::object_store_req_retry_max_attempts")]
-    pub object_store_req_retry_max_attempts: usize,
     /// For backwards compatibility, users should use `S3ObjectStoreDeveloperConfig` instead.
     #[serde(
         default = "default::object_store_config::s3::developer::object_store_retry_unknown_service_error"
@@ -1670,31 +1727,113 @@ pub mod default {
     }
 
     pub mod object_store_config {
+        const DEFAULT_RETRY_INTERVAL_MS: u64 = 20;
+        const DEFAULT_RETRY_MAX_DELAY_MS: u64 = 10 * 1000;
+        const DEFAULT_RETRY_MAX_RETRY_ATTEMPTS: usize = 3;
+        pub const UNLIMITED_MAX_TIMEOUT: u64 = u64::MAX;
+
+        /// DEPRECATED: This config will be deprecated in the future version, use `object_store_streaming_read_attempt_timeout_ms` instead.
         pub fn object_store_streaming_read_timeout_ms() -> u64 {
-            8 * 60 * 1000
+            UNLIMITED_MAX_TIMEOUT
         }
 
+        /// DEPRECATED: This config will be deprecated in the future version, use `object_store_streaming_upload_attempt_timeout_ms` instead.
         pub fn object_store_streaming_upload_timeout_ms() -> u64 {
-            8 * 60 * 1000
+            UNLIMITED_MAX_TIMEOUT
         }
 
+        /// DEPRECATED: This config will be deprecated in the future version, use `object_store_upload_attempt_timeout_ms` instead.
         pub fn object_store_upload_timeout_ms() -> u64 {
-            8 * 60 * 1000
+            UNLIMITED_MAX_TIMEOUT
         }
 
+        /// DEPRECATED: This config will be deprecated in the future version, use `object_store_read_attempt_timeout_ms` instead.
         pub fn object_store_read_timeout_ms() -> u64 {
-            8 * 60 * 1000
+            UNLIMITED_MAX_TIMEOUT
         }
 
         pub fn object_store_set_atomic_write_dir() -> bool {
             false
         }
 
+        pub fn object_store_req_retry_interval_ms() -> u64 {
+            DEFAULT_RETRY_INTERVAL_MS
+        }
+
+        pub fn object_store_req_retry_max_delay_ms() -> u64 {
+            DEFAULT_RETRY_MAX_DELAY_MS // 10s
+        }
+
+        /// DEPRECATED: This config will be deprecated in the future version
+        pub fn object_store_req_retry_max_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_upload_attempt_timeout_ms() -> u64 {
+            5 * 60 * 1000
+        }
+
+        pub fn object_store_upload_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_streaming_upload_attempt_timeout_ms() -> u64 {
+            60 * 1000
+        }
+
+        pub fn object_store_streaming_upload_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_read_attempt_timeout_ms() -> u64 {
+            60 * 1000
+        }
+
+        pub fn object_store_read_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_streaming_read_attempt_timeout_ms() -> u64 {
+            2 * 60 * 1000
+        }
+
+        pub fn object_store_streaming_read_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_metadata_attempt_timeout_ms() -> u64 {
+            5 * 60 * 1000
+        }
+
+        pub fn object_store_metadata_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_delete_attempt_timeout_ms() -> u64 {
+            2 * 60 * 1000
+        }
+
+        pub fn object_store_delete_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_delete_objects_attempt_timeout_ms() -> u64 {
+            10 * 60 * 1000
+        }
+
+        pub fn object_store_delete_objects_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
+        pub fn object_store_list_attempt_timeout_ms() -> u64 {
+            10 * 60 * 1000
+        }
+
+        pub fn object_store_list_retry_attempts() -> usize {
+            DEFAULT_RETRY_MAX_RETRY_ATTEMPTS
+        }
+
         pub mod s3 {
-            /// Retry config for compute node http timeout error.
-            const DEFAULT_RETRY_INTERVAL_MS: u64 = 20;
-            const DEFAULT_RETRY_MAX_DELAY_MS: u64 = 10 * 1000;
-            const DEFAULT_RETRY_MAX_ATTEMPTS: usize = 8;
             const DEFAULT_IDENTITY_RESOLUTION_TIMEOUT_S: u64 = 5;
 
             const DEFAULT_KEEPALIVE_MS: u64 = 600 * 1000; // 10min
@@ -1715,18 +1854,6 @@ pub mod default {
                 Some(true)
             }
 
-            pub fn object_store_req_retry_interval_ms() -> u64 {
-                DEFAULT_RETRY_INTERVAL_MS
-            }
-
-            pub fn object_store_req_retry_max_delay_ms() -> u64 {
-                DEFAULT_RETRY_MAX_DELAY_MS // 10s
-            }
-
-            pub fn object_store_req_retry_max_attempts() -> usize {
-                DEFAULT_RETRY_MAX_ATTEMPTS
-            }
-
             pub fn identity_resolution_timeout_s() -> u64 {
                 DEFAULT_IDENTITY_RESOLUTION_TIMEOUT_S
             }

src/config/example.toml

@@ -183,19 +183,35 @@ recent_filter_layers = 6
 recent_filter_rotate_interval_ms = 10000
 
 [storage.object_store]
-object_store_streaming_read_timeout_ms = 480000
-object_store_streaming_upload_timeout_ms = 480000
-object_store_upload_timeout_ms = 480000
-object_store_read_timeout_ms = 480000
+object_store_streaming_read_timeout_ms = 0
+object_store_streaming_upload_timeout_ms = 0
+object_store_upload_timeout_ms = 0
+object_store_read_timeout_ms = 0
 object_store_set_atomic_write_dir = false
+object_store_req_retry_interval_ms = 20
+object_store_req_retry_max_delay_ms = 10000
+object_store_req_retry_max_retry_attempts = 3
+object_store_upload_attempt_timeout_ms = 300000
+object_store_upload_retry_attempts = 3
+object_store_streaming_upload_attempt_timeout_ms = 60000
+object_store_streaming_upload_retry_attempts = 3
+object_store_read_attempt_timeout_ms = 60000
+object_store_read_retry_attempts = 3
+object_store_streaming_read_attempt_timeout_ms = 120000
+object_store_streaming_read_retry_attempts = 3
+object_store_metadata_attempt_timeout_ms = 300000
+object_store_metadata_retry_attempts = 3
+object_store_delete_attempt_timeout_ms = 120000
+object_store_delete_retry_attempts = 3
+object_store_delete_objects_attempt_timeout_ms = 600000
+object_store_delete_objects_retry_attempts = 3
+object_store_list_attempt_timeout_ms = 600000
+object_store_list_retry_attempts = 3
 
 [storage.object_store.s3]
 object_store_keepalive_ms = 600000
 object_store_recv_buffer_size = 2097152
 object_store_nodelay = true
-object_store_req_retry_interval_ms = 20
-object_store_req_retry_max_delay_ms = 10000
-object_store_req_retry_max_attempts = 8
 retry_unknown_service_error = false
 identity_resolution_timeout_s = 5
 

src/object_store/src/object/error.rs

@@ -27,8 +27,13 @@ use tokio::sync::oneshot::error::RecvError;
 #[derive(Error, Debug, thiserror_ext::Box, thiserror_ext::Construct)]
 #[thiserror_ext(newtype(name = ObjectError, backtrace, report_debug))]
 pub enum ObjectErrorInner {
-    #[error("s3 error: {0}")]
-    S3(#[source] BoxedError),
+    #[error("s3 error: {inner}")]
+    S3 {
+        // TODO: remove this after switch s3 backend to opendal
+        should_retry: bool,
+        #[source]
+        inner: BoxedError,
+    },
     #[error("disk error: {msg}")]
     Disk {
         msg: String,
@@ -55,17 +60,20 @@ impl ObjectError {
     /// Tells whether the error indicates the target object is not found.
     pub fn is_object_not_found_error(&self) -> bool {
         match self.inner() {
-            ObjectErrorInner::S3(e) => {
-                if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) = e
-                    .downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
+            ObjectErrorInner::S3 {
+                inner,
+                should_retry: _,
+            } => {
+                if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) =
+                    inner.downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
                         GetObjectError,
                         aws_smithy_runtime_api::http::Response<SdkBody>,
                     >>()
                 {
                     return matches!(err.err(), GetObjectError::NoSuchKey(_));
                 }
-                if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) = e
-                    .downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
+                if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) =
+                    inner.downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
                         HeadObjectError,
                         aws_smithy_runtime_api::http::Response<SdkBody>,
                     >>()
@@ -90,6 +98,20 @@ impl ObjectError {
         };
         false
     }
+
+    pub fn should_retry(&self) -> bool {
+        // return true;
+        match self.inner() {
+            ObjectErrorInner::S3 {
+                inner: _,
+                should_retry,
+            } => *should_retry,
+
+            ObjectErrorInner::Opendal(e) => e.is_temporary(),
+
+            _ => false,
+        }
+    }
 }
 
 impl<E, R> From<aws_smithy_runtime_api::client::result::SdkError<E, R>> for ObjectError
@@ -98,7 +120,11 @@ where
     R: Send + Sync + 'static + std::fmt::Debug,
 {
     fn from(e: aws_smithy_runtime_api::client::result::SdkError<E, R>) -> Self {
-        ObjectErrorInner::S3(e.into()).into()
+        ObjectErrorInner::S3 {
+            inner: e.into(),
+            should_retry: false,
+        }
+        .into()
     }
 }
 
@@ -110,7 +136,11 @@ impl From<RecvError> for ObjectError {
 
 impl From<ByteStreamError> for ObjectError {
     fn from(e: ByteStreamError) -> Self {
-        ObjectErrorInner::Internal(e.to_report_string()).into()
+        ObjectErrorInner::S3 {
+            inner: e.into(),
+            should_retry: true,
+        }
+        .into()
     }
 }