Add extstore s3 driver

.github/workflows/per-pr.yml

@@ -106,7 +106,7 @@ jobs:
           - group: rust-1.88
             crates: client common common-wasm macros protos sdk-core workflow
           - group: rust-1.92
-            crates: sdk
+            crates: aws-s3driver sdk
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
       - uses: ./.github/actions/setup

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "crates/client",
     "crates/common",
     "crates/common-wasm",
+    "crates/aws-s3driver",
     "crates/protos",
     "crates/macros",
     "crates/sdk",

changelog/added/solemn-simulacrum.md

@@ -0,0 +1 @@
+Added experimental `temporalio-aws-s3driver` crate with an S3 storage driver for external storage payloads, and an `aws-sdk` feature, enabled by default, with an `aws-sdk-s3` backed client.

crates/aws-s3driver/Cargo.toml

@@ -0,0 +1,44 @@
+[package]
+name = "temporalio-aws-s3driver"
+version = "1.1.0"
+edition = "2024"
+rust-version = "1.92.0"
+authors = ["Temporal Technologies Inc. <sdk@temporal.io>"]
+license-file = { workspace = true }
+description = "Amazon S3 external storage driver for the Temporal Rust SDK"
+homepage = "https://temporal.io/"
+repository = "https://github.com/temporalio/sdk-rust"
+keywords = ["temporal", "workflow", "s3"]
+categories = ["development-tools"]
+readme = "README.md"
+
+[package.metadata.docs.rs]
+all-features = true
+
+[features]
+default = ["aws-sdk"]
+aws-sdk = ["dep:aws-sdk-s3"]
+
+[dependencies]
+# The application configures the S3 client, including its TLS stack and async runtime.
+aws-sdk-s3 = { version = "1", default-features = false, optional = true }
+futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
+prost = { workspace = true }
+sha2 = "0.11"
+thiserror = { workspace = true }
+
+[dependencies.temporalio-common]
+path = "../common"
+version = "~1.1.0"
+default-features = false
+features = ["experimental"]
+
+[dev-dependencies]
+aws-sdk-s3 = { version = "1", default-features = false, features = ["rt-tokio"] }
+aws-smithy-http-client = { version = "1", features = ["test-util"] }
+aws-smithy-types = "1"
+http = "1"
+tokio = { version = "1.47", features = ["macros", "rt", "sync", "time"] }
+
+[lints]
+workspace = true

crates/aws-s3driver/README.md

@@ -0,0 +1,135 @@
+# Amazon S3 External Storage Driver
+
+> [!WARNING]
+> This crate is experimental and may change in the future.
+
+`temporalio-aws-s3driver` stores and retrieves Temporal payloads in Amazon S3 using external
+storage.
+
+`S3StorageDriver` has no AWS dependency. It makes requests through the `S3StorageDriverClient`
+trait. The `aws-sdk` feature, enabled by default, adds `AwsSdkS3StorageDriverClient`, backed by the
+official [`aws-sdk-s3`](https://crates.io/crates/aws-sdk-s3) crate. To bring your own client without
+depending on the AWS SDK, disable default features and implement `S3StorageDriverClient`.
+
+## Quick Start
+
+```rust,ignore
+use std::sync::Arc;
+use temporalio_aws_s3driver::{AwsSdkS3StorageDriverClient, S3StorageDriver};
+
+let config = aws_config::load_from_env().await;
+let driver = S3StorageDriver::new(
+    Arc::new(AwsSdkS3StorageDriverClient::new(aws_sdk_s3::Client::new(&config))),
+    "my-temporal-payloads",
+);
+```
+
+To choose the bucket per payload, use `S3StorageDriver::with_bucket_selector`.
+
+## Required IAM Permissions
+
+Before uploading a payload, the driver checks whether the object already exists with a `HeadObject`
+request. That check needs `s3:GetObject`. It also needs `s3:ListBucket`, because without it S3
+reports a missing object as `403 Forbidden` instead of `404 Not Found`, and the upload fails.
+
+```json
+[
+  {
+    "Effect": "Allow",
+    "Action": ["s3:PutObject", "s3:GetObject"],
+    "Resource": "arn:aws:s3:::my-temporal-payloads/*"
+  },
+  {
+    "Effect": "Allow",
+    "Action": "s3:ListBucket",
+    "Resource": "arn:aws:s3:::my-temporal-payloads"
+  }
+]
+```
+
+Components that store payloads need all three permissions. Components that only retrieve payloads
+need only `s3:GetObject`.
+
+## S3 Storage Key Specification
+
+All Temporal S3 drivers generate S3 keys in a consistent manner.
+
+### Key format
+
+Workflow key:
+
+```text
+v0/ns/{namespace}/wt/{workflow-type}/wi/{workflow-id}/ri/{run-id}/d/{hash-algorithm}/{hex-digest}
+```
+
+Activity key:
+
+```text
+v0/ns/{namespace}/at/{activity-type}/ai/{activity-id}/ri/{run-id}/d/{hash-algorithm}/{hex-digest}
+```
+
+Fallback key (unknown target):
+
+```text
+v0/d/{hash-algorithm}/{hex-digest}
+```
+
+- If no namespace, workflow, or activity information is available, the fallback is used.
+- Dynamic path segments are percent-encoded (rules below).
+- Missing values (including a missing `run-id`) are encoded as `null`.
+- `hex-digest` is lower-case SHA-256 hex (64 characters).
+
+### Percent-encoding rules
+
+The Temporal SDKs escape anything that isn't listed in S3's safe character set: https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html
+
+Safe Characters:
+
+```text
+Alphanumeric characters
+  0-9
+  a-z
+  A-Z
+
+Special characters
+  Exclamation point (!)
+  Hyphen (-)
+  Underscore (_)
+  Period (.)
+  Asterisk (*)
+  Single quotation mark (')
+  Opening parenthesis (()
+  Closing parenthesis ())
+```
+
+### Examples
+
+Workflow key example:
+
+```text
+input:
+  namespace=payments prod
+  workflow-type=ChargeWorkflow
+  workflow-id=order+123=abc
+  run-id=3f1d6c7a-8b2e-4f7a-9d0a-87a6f95e4d31
+  hash-algorithm=sha256
+  hex-digest=9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08
+
+output:
+  v0/ns/payments%20prod/wt/ChargeWorkflow/wi/order%2B123%3Dabc/ri/3f1d6c7a-8b2e-4f7a-9d0a-87a6f95e4d31/d/sha256/9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08
+```
+
+Activity key example:
+
+```text
+input:
+  namespace=payments prod
+  activity-type=Capture/Charge
+  activity-id=activity id+42
+  run-id=9e1d1fd9-2f8a-4c40-93e2-731f31b9268b
+  hash-algorithm=sha256
+  hex-digest=2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824
+
+output:
+  v0/ns/payments%20prod/at/Capture%2FCharge/ai/activity%20id%2B42/ri/9e1d1fd9-2f8a-4c40-93e2-731f31b9268b/d/sha256/2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824
+```

crates/aws-s3driver/src/aws_sdk_client.rs

@@ -0,0 +1,194 @@
+use crate::client::{BoxError, S3StorageDriverClient};
+use aws_sdk_s3::{Client, primitives::ByteStream};
+use futures_util::future::BoxFuture;
+use std::collections::HashMap;
+
+/// [`S3StorageDriverClient`] backed by an [`aws_sdk_s3::Client`] from the official AWS SDK.
+///
+/// This API is experimental and may change in the future.
+#[derive(Clone, Debug)]
+pub struct AwsSdkS3StorageDriverClient {
+    client: Client,
+}
+
+impl AwsSdkS3StorageDriverClient {
+    /// Create a client that makes requests with the given AWS S3 client.
+    pub fn new(client: Client) -> Self {
+        Self { client }
+    }
+}
+
+impl S3StorageDriverClient for AwsSdkS3StorageDriverClient {
+    fn put_object<'a>(
+        &'a self,
+        bucket: &'a str,
+        key: &'a str,
+        data: Vec<u8>,
+    ) -> BoxFuture<'a, Result<(), BoxError>> {
+        Box::pin(async move {
+            self.client
+                .put_object()
+                .bucket(bucket)
+                .key(key)
+                .body(ByteStream::from(data))
+                .send()
+                .await?;
+            Ok(())
+        })
+    }
+
+    fn object_exists<'a>(
+        &'a self,
+        bucket: &'a str,
+        key: &'a str,
+    ) -> BoxFuture<'a, Result<bool, BoxError>> {
+        Box::pin(async move {
+            match self
+                .client
+                .head_object()
+                .bucket(bucket)
+                .key(key)
+                .send()
+                .await
+            {
+                Ok(_) => Ok(true),
+                Err(err) if err.as_service_error().is_some_and(|err| err.is_not_found()) => {
+                    Ok(false)
+                }
+                Err(err) => Err(err.into()),
+            }
+        })
+    }
+
+    fn get_object<'a>(
+        &'a self,
+        bucket: &'a str,
+        key: &'a str,
+    ) -> BoxFuture<'a, Result<Vec<u8>, BoxError>> {
+        Box::pin(async move {
+            let output = self
+                .client
+                .get_object()
+                .bucket(bucket)
+                .key(key)
+                .send()
+                .await?;
+            Ok(output.body.collect().await?.into_bytes().to_vec())
+        })
+    }
+
+    fn describe(&self) -> HashMap<String, String> {
+        self.client
+            .config()
+            .region()
+            .map(|region| HashMap::from([("client_region".to_owned(), region.to_string())]))
+            .unwrap_or_default()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aws_sdk_s3::{
+        Config,
+        config::{BehaviorVersion, Credentials, Region},
+    };
+    use aws_smithy_http_client::test_util::infallible_client_fn;
+    use aws_smithy_types::body::SdkBody;
+    use std::sync::{Arc, Mutex};
+
+    fn client(
+        respond: impl Fn(http::Request<SdkBody>) -> http::Response<SdkBody> + Send + Sync + 'static,
+    ) -> AwsSdkS3StorageDriverClient {
+        let config = Config::builder()
+            .behavior_version(BehaviorVersion::latest())
+            .region(Region::new("us-west-2"))
+            .credentials_provider(Credentials::new("id", "secret", None, None, "test"))
+            .force_path_style(true)
+            .http_client(infallible_client_fn(respond))
+            .build();
+        AwsSdkS3StorageDriverClient::new(Client::from_conf(config))
+    }
+
+    fn status(code: u16) -> http::Response<SdkBody> {
+        http::Response::builder()
+            .status(code)
+            .body(SdkBody::empty())
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn object_exists_returns_true_for_200_and_false_for_404() {
+        assert!(
+            client(|_| status(200))
+                .object_exists("my-bucket", "my-key")
+                .await
+                .unwrap()
+        );
+        assert!(
+            !client(|_| status(404))
+                .object_exists("my-bucket", "my-key")
+                .await
+                .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn object_exists_fails_for_errors_other_than_404() {
+        client(|_| status(403))
+            .object_exists("my-bucket", "my-key")
+            .await
+            .unwrap_err();
+    }
+
+    #[tokio::test]
+    async fn put_object_sends_data_to_bucket_and_key() {
+        let requests = Arc::new(Mutex::new(Vec::new()));
+        let recorded = requests.clone();
+        let client = client(move |request| {
+            recorded.lock().unwrap().push((
+                request.method().clone(),
+                request.uri().path().to_owned(),
+                request.body().bytes().map(<[u8]>::to_vec),
+            ));
+            status(200)
+        });
+
+        client
+            .put_object("my-bucket", "my-key", vec![0, 1])
+            .await
+            .unwrap();
+
+        assert_eq!(
+            *requests.lock().unwrap(),
+            [(
+                http::Method::PUT,
+                "/my-bucket/my-key".to_owned(),
+                Some(vec![0, 1])
+            )]
+        );
+    }
+
+    #[tokio::test]
+    async fn get_object_returns_the_response_body() {
+        let client = client(|_| {
+            http::Response::builder()
+                .status(200)
+                .body(SdkBody::from(vec![0, 1]))
+                .unwrap()
+        });
+
+        assert_eq!(
+            client.get_object("my-bucket", "my-key").await.unwrap(),
+            [0, 1]
+        );
+    }
+
+    #[test]
+    fn describe_returns_the_client_region() {
+        assert_eq!(
+            client(|_| status(200)).describe(),
+            HashMap::from([("client_region".to_owned(), "us-west-2".to_owned())])
+        );
+    }
+}