feat(transaction): Add retry logic to transaction (#1484)

Author: CTTY

Cargo.lock

@@ -55,6 +55,7 @@ async-std = "1.12"
 async-trait = "0.1.88"
 aws-config = "1.6.1"
 aws-sdk-glue = "1.39"
+backon = "1.5.1"
 base64 = "0.22.1"
 bimap = "0.6"
 bytes = "1.10"
@@ -82,6 +83,7 @@ itertools = "0.13"
 linkedbytes = "0.1.8"
 metainfo = "0.7.14"
 mimalloc = "0.1.46"
+mockall = "0.13.1"
 mockito = "1"
 motore-macros = "0.4.3"
 murmur3 = "0.5.2"

Cargo.toml

@@ -57,6 +57,7 @@ arrow-string = { workspace = true }
 as-any = { workspace = true }
 async-std = { workspace = true, optional = true, features = ["attributes"] }
 async-trait = { workspace = true }
+backon = { workspace = true }
 base64 = { workspace = true }
 bimap = { workspace = true }
 bytes = { workspace = true }
@@ -66,6 +67,7 @@ expect-test = { workspace = true }
 fnv = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }
+mockall = { workspace = true }
 moka = { version = "0.12.10", features = ["future"] }
 murmur3 = { workspace = true }
 num-bigint = { workspace = true }

crates/iceberg/Cargo.toml

@@ -29,6 +29,7 @@ use std::sync::Arc;
 use _serde::deserialize_snapshot;
 use async_trait::async_trait;
 pub use memory::MemoryCatalog;
+use mockall::automock;
 use serde_derive::{Deserialize, Serialize};
 use typed_builder::TypedBuilder;
 use uuid::Uuid;
@@ -43,6 +44,7 @@ use crate::{Error, ErrorKind, Result};
 
 /// The catalog API for Iceberg Rust.
 #[async_trait]
+#[automock]
 pub trait Catalog: Debug + Sync + Send {
     /// List namespaces inside the catalog.
     async fn list_namespaces(&self, parent: Option<&NamespaceIdent>)

crates/iceberg/src/catalog/mod.rs

@@ -320,6 +320,11 @@ impl Error {
         self.kind
     }
 
+    /// Return error's retryable status
+    pub fn retryable(&self) -> bool {
+        self.retryable
+    }
+
     /// Return error's message.
     #[inline]
     pub fn message(&self) -> &str {

crates/iceberg/src/error.rs

@@ -98,6 +98,26 @@ pub const RESERVED_PROPERTIES: [&str; 9] = [
     PROPERTY_DEFAULT_SORT_ORDER,
 ];
 
+/// Property key for number of commit retries.
+pub const PROPERTY_COMMIT_NUM_RETRIES: &str = "commit.retry.num-retries";
+/// Default value for number of commit retries.
+pub const PROPERTY_COMMIT_NUM_RETRIES_DEFAULT: usize = 4;
+
+/// Property key for minimum wait time (ms) between retries.
+pub const PROPERTY_COMMIT_MIN_RETRY_WAIT_MS: &str = "commit.retry.min-wait-ms";
+/// Default value for minimum wait time (ms) between retries.
+pub const PROPERTY_COMMIT_MIN_RETRY_WAIT_MS_DEFAULT: u64 = 100;
+
+/// Property key for maximum wait time (ms) between retries.
+pub const PROPERTY_COMMIT_MAX_RETRY_WAIT_MS: &str = "commit.retry.max-wait-ms";
+/// Default value for maximum wait time (ms) between retries.
+pub const PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT: u64 = 60 * 1000; // 1 minute
+
+/// Property key for total maximum retry time (ms).
+pub const PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS: &str = "commit.retry.total-timeout-ms";
+/// Default value for total maximum retry time (ms).
+pub const PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT: u64 = 30 * 60 * 1000; // 30 minutes
+
 /// Reference to [`TableMetadata`].
 pub type TableMetadataRef = Arc<TableMetadata>;