test(storage): use s3s to test S3BlobStorage

Introduces s3s and s3s-fs to run an in-memory s3 server in tests. With this we can promote s3 tests to run in `just test`, they won't need containers, and they're super fast. Adds `just test-s3` to run them more easily.

```
 Nextest run ID db9a9530-738e-4c92-ad3a-23289252cac2 with nextest profile: default                                                                                                              Starting 6 tests across 1 binary                                                                                                                                                                PASS [   0.356s] tranquil-storage::s3 put_stream_error_aborts_upload                                                                                                                        PASS [   0.356s] tranquil-storage::s3 put_stream_empty_aborts_upload                                                                                                                        PASS [   0.358s] tranquil-storage::s3 copy                                                                                                                                                  PASS [   0.360s] tranquil-storage::s3 put_get_head_delete                                                                                                                                   PASS [   0.360s] tranquil-storage::s3 put_stream                                                                                                                                            PASS [   0.447s] tranquil-storage::s3 put_stream_multipart                                                                                                                          ────────────                                                                                                                                                                                     Summary [   0.448s] 6 tests run: 6 passed, 0 skipped
```

Also optimizes more hashing functions because it matters in tests.

Cargo.toml

@@ -116,6 +116,8 @@ regex = "1"
 rsa = "0.9"
 rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] }
 rustls-pemfile = "2"
+s3s = "0.17"
+s3s-fs = "0.17"
 secrecy = { version = "0.10", features = ["serde"] }
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls-webpki-roots", "http2", "charset", "macos-system-configuration"] }
 serde = { version = "1.0", features = ["derive"] }
@@ -172,12 +174,19 @@ codegen-units = 1
 debug = 1
 strip = false
 
+# Optimize all the hashing so it runs faster in tests.
 [profile.dev.package.bcrypt]
 opt-level = 3
 
 [profile.dev.package.blowfish]
 opt-level = 3
 
+[profile.dev.package.sha2]
+opt-level = 3
+
+[profile.dev.package.md-5]
+opt-level = 3
+
 # Set the default to optimized so fortify works in CI.
 [profile.dev.package.tikv-jemalloc-sys]
 opt-level = 1

crates/tranquil-storage/Cargo.toml

@@ -21,3 +21,10 @@ sha2 = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 uuid = { workspace = true }
+
+[dev-dependencies]
+hyper-util = { workspace = true }
+s3s = { workspace = true }
+s3s-fs = { workspace = true }
+tempfile = "3"
+tokio = { workspace = true, features = ["net"] }

crates/tranquil-storage/src/lib.rs

@@ -103,8 +103,8 @@ fn map_io_not_found(key: &str) -> impl FnOnce(std::io::Error) -> StorageError +
 mod s3 {
     use super::*;
     use aws_config::BehaviorVersion;
-    use aws_config::meta::region::RegionProviderChain;
     use aws_sdk_s3::Client;
+    use aws_sdk_s3::config::Region;
     use aws_sdk_s3::primitives::ByteStream;
     use aws_sdk_s3::types::CompletedMultipartUpload;
     use aws_sdk_s3::types::CompletedPart;
@@ -118,24 +118,14 @@ mod s3 {
     }
 
     impl S3BlobStorage {
-        pub async fn new() -> Self {
-            let cfg = tranquil_config::get();
-            let bucket = cfg
-                .storage
-                .s3_bucket
-                .clone()
-                .expect("storage.s3_bucket (S3_BUCKET) must be set");
-            let client = create_s3_client().await;
-            let path = cfg
-                .storage
-                .s3_path
-                .trim_start_matches("/")
-                .trim_end_matches("/")
-                .to_string();
+        pub async fn new(bucket: &str, endpoint: Option<&str>, path: &str) -> Self {
             Self {
-                client,
-                bucket,
-                path,
+                client: create_s3_client(endpoint).await,
+                bucket: bucket.to_string(),
+                path: path
+                    .trim_start_matches("/")
+                    .trim_end_matches("/")
+                    .to_string(),
             }
         }
 
@@ -148,9 +138,7 @@ mod s3 {
         }
     }
 
-    async fn create_s3_client() -> Client {
-        let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
-
+    async fn create_s3_client(endpoint: Option<&str>) -> Client {
         let http_client = aws_smithy_http_client::Builder::new()
             .tls_provider(aws_smithy_http_client::tls::Provider::Rustls(
                 aws_smithy_http_client::tls::rustls_provider::CryptoMode::Ring,
@@ -158,25 +146,20 @@ mod s3 {
             .build_https();
 
         let config = aws_config::defaults(BehaviorVersion::latest())
-            .region(region_provider)
             .http_client(http_client)
             .load()
             .await;
 
-        tranquil_config::get()
-            .storage
-            .s3_endpoint
-            .as_deref()
-            .map_or_else(
-                || Client::new(&config),
-                |endpoint| {
-                    let s3_config = aws_sdk_s3::config::Builder::from(&config)
-                        .endpoint_url(endpoint)
-                        .force_path_style(true)
-                        .build();
-                    Client::from_conf(s3_config)
-                },
-            )
+        let region = config
+            .region()
+            .cloned()
+            .unwrap_or_else(|| Region::from_static("us-east-1"));
+        let builder = aws_sdk_s3::config::Builder::from(&config).region(region);
+        let builder = match endpoint {
+            Some(endpoint) => builder.endpoint_url(endpoint).force_path_style(true),
+            None => builder,
+        };
+        Client::from_conf(builder.build())
     }
 
     #[async_trait]
@@ -604,7 +587,14 @@ pub async fn create_blob_storage() -> Arc<dyn BlobStorage> {
         #[cfg(feature = "s3")]
         "s3" => {
             tracing::info!("Initializing S3 blob storage");
-            Arc::new(S3BlobStorage::new().await)
+            let storage = &cfg.storage;
+            let bucket = storage
+                .s3_bucket
+                .as_deref()
+                .expect("storage.s3_bucket (S3_BUCKET) must be set");
+            Arc::new(
+                S3BlobStorage::new(bucket, storage.s3_endpoint.as_deref(), &storage.s3_path).await,
+            )
         }
         #[cfg(not(feature = "s3"))]
         "s3" => {

crates/tranquil-storage/tests/s3.rs

@@ -0,0 +1,129 @@
+#![cfg(feature = "s3")]
+use bytes::Bytes;
+use hyper_util::rt::{TokioExecutor, TokioIo};
+use hyper_util::server::conn::auto::Builder as ConnBuilder;
+use s3s::auth::SimpleAuth;
+use s3s::service::S3ServiceBuilder;
+use s3s_fs::FileSystem;
+use sha2::{Digest, Sha256};
+use tempfile::TempDir;
+use tokio::net::TcpListener;
+use tranquil_storage::{BlobStorage, S3BlobStorage};
+
+const BUCKET: &str = "bucket";
+const PREFIX: &str = "prefix";
+
+async fn start_s3() -> (TempDir, S3BlobStorage) {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::create_dir(root.path().join(BUCKET)).unwrap();
+
+    let mut builder = S3ServiceBuilder::new(FileSystem::new(root.path()).unwrap());
+    builder.set_auth(SimpleAuth::from_single("test", "test"));
+    let service = builder.build();
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let endpoint = format!("http://{}", listener.local_addr().unwrap());
+
+    tokio::spawn(async move {
+        loop {
+            let (socket, _) = listener.accept().await.unwrap();
+            let conn = ConnBuilder::new(TokioExecutor::new())
+                .serve_connection(TokioIo::new(socket), service.clone())
+                .into_owned();
+            tokio::spawn(conn);
+        }
+    });
+
+    unsafe {
+        std::env::set_var("AWS_ACCESS_KEY_ID", "test");
+        std::env::set_var("AWS_SECRET_ACCESS_KEY", "test");
+        std::env::set_var("AWS_REGION", "us-east-1");
+    }
+
+    let storage = S3BlobStorage::new(BUCKET, Some(&endpoint), PREFIX).await;
+    (root, storage)
+}
+
+#[tokio::test]
+async fn put_get_head_delete() {
+    let (root, storage) = start_s3().await;
+
+    storage
+        .put_bytes("key", "hello world".into())
+        .await
+        .unwrap();
+
+    assert_eq!(storage.get_bytes("key").await.unwrap(), "hello world");
+    assert_eq!(storage.get_head("key", 5).await.unwrap(), "hello");
+    assert!(root.path().join(BUCKET).join(PREFIX).join("key").is_file());
+
+    storage.delete("key").await.unwrap();
+    assert!(storage.get_bytes("key").await.is_err());
+}
+
+#[tokio::test]
+async fn copy() {
+    let (_root, storage) = start_s3().await;
+
+    storage.put_bytes("src", "hello".into()).await.unwrap();
+    storage.copy("src", "dst").await.unwrap();
+
+    assert_eq!(storage.get_bytes("dst").await.unwrap(), "hello");
+}
+
+#[tokio::test]
+async fn put_stream() {
+    let (_root, storage) = start_s3().await;
+    let chunks = ["hello", " ", "world"].map(|c| Ok(Bytes::from(c)));
+
+    let result = storage
+        .put_stream("key", Box::pin(futures::stream::iter(chunks)))
+        .await
+        .unwrap();
+
+    assert_eq!(result.size, 11);
+    assert_eq!(result.sha256_hash[..], Sha256::digest("hello world")[..]);
+    assert_eq!(storage.get_bytes("key").await.unwrap(), "hello world");
+}
+
+#[tokio::test]
+async fn put_stream_error_aborts_upload() {
+    let (root, storage) = start_s3().await;
+    let chunks = [Ok(Bytes::from("hello")), Err(std::io::Error::other("boom"))];
+
+    let result = storage
+        .put_stream("key", Box::pin(futures::stream::iter(chunks)))
+        .await;
+
+    assert!(result.is_err());
+    assert_eq!(std::fs::read_dir(root.path()).unwrap().count(), 1);
+}
+
+#[tokio::test]
+async fn put_stream_empty_aborts_upload() {
+    let (root, storage) = start_s3().await;
+
+    let result = storage
+        .put_stream("key", Box::pin(futures::stream::empty()))
+        .await;
+
+    assert!(result.is_err());
+    assert_eq!(std::fs::read_dir(root.path()).unwrap().count(), 1);
+}
+
+#[tokio::test]
+async fn put_stream_multipart() {
+    let (_root, storage) = start_s3().await;
+    let chunk = Bytes::from(vec![7u8; 1024 * 1024]);
+    let chunks = std::iter::repeat_n(chunk, 6).map(Ok);
+
+    let result = storage
+        .put_stream("key", Box::pin(futures::stream::iter(chunks)))
+        .await
+        .unwrap();
+
+    let expected = vec![7u8; 6 * 1024 * 1024];
+    assert_eq!(result.size, expected.len() as u64);
+    assert_eq!(result.sha256_hash[..], Sha256::digest(&expected)[..]);
+    assert_eq!(storage.get_bytes("key").await.unwrap(), expected);
+}

justfile

@@ -128,6 +128,9 @@ test-import:
 test-misc:
     {{store_run}} --test it -E 'test(/^(actor|commit_signing|image_processing|lifecycle_social|notifications|server|signing_key|verify_live_commit)::/)'
 
+test-s3:
+    {{store_run}} --test s3
+
 test *args:
     {{store_test}} {{args}}