Introduce SnapshotsRepository backed by object_store

Cargo.toml

@@ -77,6 +77,8 @@ arrow = { version = "53.1.0", default-features = false }
 assert2 = "0.3.11"
 async-channel = "2.1.1"
 async-trait = "0.1.73"
+aws-config = { version = "1.5.4", default-features = false, features = ["rt-tokio", "sso", "rustls"] }
+aws-credential-types = { version = "1.2.0", default-features = false }
 axum = { version = "0.7.5", default-features = false }
 base64 = "0.22"
 bitflags = { version = "2.6.0" }
@@ -134,6 +136,7 @@ metrics-exporter-prometheus = { version = "0.15", default-features = false, feat
     "async-runtime",
 ] }
 moka = "0.12.5"
+object_store = { version = "0.11.1", features = ["aws"] }
 once_cell = "1.18"
 opentelemetry = { version = "0.24.0" }
 opentelemetry-http = { version = "0.13.0" }

crates/core/src/worker_api/partition_processor_manager.rs

@@ -82,6 +82,8 @@ pub enum SnapshotError {
     SnapshotExportError(PartitionId, #[source] anyhow::Error),
     #[error("Snapshot failed for partition {0}: {1}")]
     SnapshotMetadataHeaderError(PartitionId, #[source] io::Error),
+    #[error("Error putting partition id {0} snapshot into repository: {1}")]
+    RepositoryPutError(PartitionId, #[source] anyhow::Error),
     #[error("Internal error creating snapshot for partition {0}: {1}")]
     Internal(PartitionId, String),
 }
@@ -94,6 +96,7 @@ impl SnapshotError {
             SnapshotError::InvalidState(partition_id) => *partition_id,
             SnapshotError::SnapshotExportError(partition_id, _) => *partition_id,
             SnapshotError::SnapshotMetadataHeaderError(partition_id, _) => *partition_id,
+            SnapshotError::RepositoryPutError(partition_id, _) => *partition_id,
             SnapshotError::Internal(partition_id, _) => *partition_id,
         }
     }

crates/service-client/Cargo.toml

@@ -13,8 +13,8 @@ options_schema = ["dep:schemars", "restate-types/schemars"]
 
 [dependencies]
 arc-swap = { workspace = true }
-aws-config = { version = "1.5.4", default-features = false, features = ["rt-tokio", "sso"] }
-aws-credential-types = {version = "1.2.0", default-features = false}
+aws-config = { workspace = true }
+aws-credential-types = { workspace = true }
 aws-sdk-lambda = {version = "1.36.0", default-features = false, features = ["rt-tokio"]}
 aws-sdk-sts = {version = "1.35.0", default-features = false, features = ["rt-tokio"]}
 base64 = { workspace = true }

crates/types/Cargo.toml

@@ -74,6 +74,7 @@ tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 ulid = { workspace = true }
 xxhash-rust = { workspace = true, features = ["xxh3"] }
+url = "2.5.3"
 
 [dev-dependencies]
 restate-test-util = { workspace = true }