fix: abort multipart upload on errors

adds a test that proves it works too!

.config/nextest.toml

@@ -53,27 +53,15 @@ heavy-load-tests = { max-threads = 4 }
 io-heavy-sim = { max-threads = 2 }
 
 [[profile.default.overrides]]
-filter = "test(/import_with_verification/) | test(/plc_migration/)"
+filter = "binary(it) and test(/^ripple_cluster::/)"
 test-group = "serial-env-tests"
 
 [[profile.default.overrides]]
-filter = "binary(handle_domains)"
+filter = "package(tranquil-signal) and test(/^tests::/)"
 test-group = "serial-env-tests"
 
 [[profile.default.overrides]]
-filter = "binary(ripple_cluster)"
+filter = "binary(it) and test(/^whole_story::/)"
-test-group = "serial-env-tests"
-
-[[profile.default.overrides]]
-filter = "package(tranquil-signal)"
-test-group = "serial-env-tests"
-
-[[profile.default.overrides]]
-filter = "package(tranquil-config)"
-test-group = "serial-env-tests"
-
-[[profile.default.overrides]]
-filter = "binary(whole_story)"
 test-group = "heavy-load-tests"
 
 [[profile.default.overrides]]
@@ -81,7 +69,7 @@ filter = "test(/two_node_stress_concurrent_load/)"
 test-group = "heavy-load-tests"
 
 [[profile.default.overrides]]
-filter = "binary(repo_lifecycle)"
+filter = "binary(it) and test(/^repo_lifecycle::/)"
 test-group = "heavy-load-tests"
 
 [[profile.default.overrides]]
@@ -103,7 +91,7 @@ filter = "binary(gauntlet_smoke)"
 slow-timeout = { period = "300s", terminate-after = 8 }
 
 [[profile.default.overrides]]
-filter = "binary(compaction_restart) | binary(mst_refcount_integrity) | binary(gc_compaction_restart)"
+filter = "binary(compaction_restart) | binary(mst_refcount_integrity) | binary(it) and test(/^gc_compaction_restart::/)"
 slow-timeout = { period = "120s", terminate-after = 4 }
 
 [[profile.default.overrides]]
@@ -117,27 +105,15 @@ slow-timeout = { period = "300s", terminate-after = 4 }
 test-group = "io-heavy-sim"
 
 [[profile.ci.overrides]]
-filter = "test(/import_with_verification/) | test(/plc_migration/)"
+filter = "binary(it) and test(/^ripple_cluster::/)"
 test-group = "serial-env-tests"
 
 [[profile.ci.overrides]]
-filter = "binary(handle_domains)"
+filter = "package(tranquil-signal) and test(/^tests::/)"
 test-group = "serial-env-tests"
 
 [[profile.ci.overrides]]
-filter = "binary(ripple_cluster)"
+filter = "binary(it) and test(/^whole_story::/)"
-test-group = "serial-env-tests"
-
-[[profile.ci.overrides]]
-filter = "package(tranquil-signal)"
-test-group = "serial-env-tests"
-
-[[profile.ci.overrides]]
-filter = "package(tranquil-config)"
-test-group = "serial-env-tests"
-
-[[profile.ci.overrides]]
-filter = "binary(whole_story)"
 test-group = "heavy-load-tests"
 
 [[profile.ci.overrides]]
@@ -145,7 +121,7 @@ filter = "test(/two_node_stress_concurrent_load/)"
 test-group = "heavy-load-tests"
 
 [[profile.ci.overrides]]
-filter = "binary(repo_lifecycle)"
+filter = "binary(it) and test(/^repo_lifecycle::/)"
 test-group = "heavy-load-tests"
 
 [[profile.ci.overrides]]

.tangled/workflows/lint.yml

@@ -0,0 +1,25 @@
+when:
+  - event: ["push", "pull_request"]
+    branch: ["main"]
+
+engine: microvm
+image: nixos
+
+environment:
+  CARGO_INCREMENTAL: "0"
+
+dependencies:
+  - rustup
+  - gcc
+  - pkg-config
+  - openssl
+  - protobuf
+  - just
+
+steps:
+  - name: fmt
+    command: cargo fmt -- --check
+  - name: check
+    command: cargo check --all-targets
+  - name: clippy
+    command: just clippy

.tangled/workflows/publish-cachix-aarch64.yml

@@ -1,21 +0,0 @@
-when:
-  - event: ["push", "manual"]
-    branch: ["main"]
-
-engine: nixery
-
-dependencies:
-  nixpkgs:
-    - nix
-    - cachix
-    - gnugrep
-
-steps:
-  - name: Authenticate
-    command: |
-      cachix authtoken "$CACHIX_AUTH_TOKEN"
-
-  - name: Build and push aarch64
-    command: |
-      SUBS="--option extra-substituters https://tranquil.cachix.org --option extra-trusted-public-keys tranquil.cachix.org-1:PoO+mGL6a6LcJiPakMDHN4E218/ei/7v2sxeDtNkSRg="
-      nix-store -qR --include-outputs $(nix-store -qd $(nix build .#packages.x86_64-linux.tranquil-pds-aarch64 $SUBS --print-out-paths --no-link)) | grep -v '\.drv$' | cachix push tranquil

.tangled/workflows/publish-cachix-x86_64.yml

@@ -1,28 +0,0 @@
-when:
-  - event: ["push", "manual"]
-    branch: ["main"]
-
-engine: nixery
-
-dependencies:
-  nixpkgs:
-    - nix
-    - cachix
-    - gnugrep
-
-steps:
-  - name: Authenticate
-    command: |
-      cachix authtoken "$CACHIX_AUTH_TOKEN"
-
-  - name: Build and push x86_64
-    command: |
-      SUBS="--option extra-substituters https://tranquil.cachix.org --option extra-trusted-public-keys tranquil.cachix.org-1:PoO+mGL6a6LcJiPakMDHN4E218/ei/7v2sxeDtNkSRg="
-      nix-store -qR --include-outputs $(nix-store -qd $(nix build .#packages.x86_64-linux.tranquil-pds $SUBS --print-out-paths --no-link)) | grep -v '\.drv$' | cachix push tranquil
-      nix-store -qR --include-outputs $(nix-store -qd $(nix build .#packages.x86_64-linux.tranquil-frontend $SUBS --print-out-paths --no-link)) | grep -v '\.drv$' | cachix push tranquil
-
-  - name: Build and push devShell
-    command: |
-      SUBS="--option extra-substituters https://tranquil.cachix.org --option extra-trusted-public-keys tranquil.cachix.org-1:PoO+mGL6a6LcJiPakMDHN4E218/ei/7v2sxeDtNkSRg="
-      nix develop $SUBS --profile dev-profile -c true
-      cachix push tranquil dev-profile

.tangled/workflows/publish-image.yml

@@ -1,36 +0,0 @@
-when:
-  - event: [ "manual" ]
-  - event: [ "push" ]
-    branch: [ "main" ]
-
-engine: nixery
-
-dependencies:
-  nixpkgs:
-    - kaniko
-
-environment:
-  DOCKER_CONFIG: "/kaniko/.docker"
-
-steps:
-  - name: Configure Kaniko
-    command: |
-      mkdir -p /kaniko/.docker/
-      echo "{
-        \"auths\": {
-          \"https://atcr.io/v1\":{
-            \"auth\": \"$ATCR_CREDENTIALS\"
-          }
-        }
-      }" > /kaniko/.docker/config.json
-
-  - name: Build image
-    command: |
-      executor \
-        --context=$(pwd) \
-        --ignore-path=$(pwd) \
-        --dockerfile=$(pwd)/Dockerfile \
-        --destination="atcr.io/tranquil.farm/tranquil-pds:latest" \
-        --destination="atcr.io/tranquil.farm/tranquil-pds:$TANGLED_COMMIT_SHA" \
-        --push-retry=3 \
-        --skip-push-permission-check

.tangled/workflows/test.yml

@@ -0,0 +1,25 @@
+when:
+  - event: ["push", "pull_request"]
+    branch: ["main"]
+
+engine: microvm
+image: nixos
+
+environment:
+  CARGO_INCREMENTAL: "0"
+  CARGO_PROFILE_DEV_DEBUG: "0"
+  CARGO_PROFILE_TEST_DEBUG: "0"
+  TMPDIR: "/dev/shm"
+
+dependencies:
+  - rustup
+  - gcc
+  - pkg-config
+  - openssl
+  - protobuf
+  - just
+  - cargo-nextest
+
+steps:
+  - name: test
+    command: just test

Cargo.toml

@@ -63,8 +63,9 @@ backon = "1"
 bincode = { version = "2", features = ["serde"] }
 anyhow = "1.0"
 async-trait = "0.1"
-aws-config = "1.8"
+aws-config = { version = "1.8", default-features = false, features = ["rt-tokio", "credentials-process", "sso"] }
-aws-sdk-s3 = "1.118"
+aws-sdk-s3 = { version = "1.118", default-features = false, features = ["sigv4a", "rt-tokio"] }
+aws-smithy-http-client = { version = "1.1", features = ["rustls-ring"] }
 axum = { version = "0.8", features = ["ws", "macros"] }
 base32 = "0.5"
 base64 = "0.22"
@@ -80,7 +81,7 @@ ed25519-dalek = { version = "2.1", features = ["pkcs8"] }
 foca = { version = "1", features = ["bincode-codec", "tracing"] }
 futures = "0.3"
 futures-util = "0.3"
-governor = "0.10"
+governor = { version = "0.10", default-features = false, features = ["std", "dashmap", "jitter"] }
 h3 = "0.0.8"
 h3-quinn = "0.0.10"
 hex = "0.4"
@@ -115,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"] }
@@ -170,3 +173,20 @@ lto = "fat"
 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-api/Cargo.toml

@@ -51,3 +51,4 @@ webauthn-rs = { workspace = true }
 [features]
 bsky = ["bsky-support"]
 bsky-support = []
+low-bcrypt-cost = []

crates/tranquil-api/src/common.rs

@@ -1,4 +1,4 @@
-use bcrypt::{DEFAULT_COST, hash};
+use bcrypt::hash;
 use chrono::{DateTime, Utc};
 use std::collections::HashMap;
 use tracing::{error, warn};
@@ -8,6 +8,12 @@ use tranquil_pds::api::error::DbResultExt;
 use tranquil_pds::types::{AtIdentifier, Did, Handle, PasswordHash};
 use tranquil_types::{DiscordUsername, SignalUsername, TelegramUsername};
 
+#[cfg(not(feature = "low-bcrypt-cost"))]
+const PASSWORD_HASH_COST: u32 = bcrypt::DEFAULT_COST;
+// Use a lower bcrypt cost in tests.
+#[cfg(feature = "low-bcrypt-cost")]
+const PASSWORD_HASH_COST: u32 = 4;
+
 pub struct ResolvedRepo {
     pub user_id: uuid::Uuid,
     pub did: Did,
@@ -278,7 +284,7 @@ pub async fn verify_credential(
 }
 
 pub fn hash_or_internal_error(value: &str) -> Result<PasswordHash, ApiError> {
-    bcrypt::hash(value, DEFAULT_COST)
+    bcrypt::hash(value, PASSWORD_HASH_COST)
         .map(PasswordHash::new)
         .map_err(|e| {
             error!("Bcrypt hash error: {:?}", e);
@@ -288,7 +294,7 @@ pub fn hash_or_internal_error(value: &str) -> Result<PasswordHash, ApiError> {
 
 pub async fn hash_password_async(password: &str) -> Result<PasswordHash, ApiError> {
     let password = password.to_string();
-    tokio::task::spawn_blocking(move || hash(password, DEFAULT_COST))
+    tokio::task::spawn_blocking(move || hash(password, PASSWORD_HASH_COST))
         .await
         .map_err(|e| {
             error!("Failed to spawn blocking task: {:?}", e);

crates/tranquil-lexicon/tests/resolve_integration.rs

@@ -395,7 +395,7 @@ async fn test_fetch_schema_error_status_gives_meaningful_error() {
 }
 
 #[tokio::test]
-async fn test_plc_server_timeout() {
+async fn jail_test_plc_server_timeout() {
     let plc_server = MockServer::start().await;
     let did = "did:plc:timeout123";
 

crates/tranquil-pds/Cargo.toml

@@ -90,7 +90,7 @@ tranquil-infra = { workspace = true, features = ["testing"] }
 tempfile = "3"
 ciborium = { workspace = true }
 ctor = { workspace = true }
-testcontainers = { workspace = true }
+testcontainers = { workspace = true, features = ["watchdog"] }
 testcontainers-modules = { workspace = true }
 tranquil-ripple = { workspace = true }
 tranquil-sync = { workspace = true }
@@ -102,4 +102,5 @@ hex = { workspace = true }
 tokio-tungstenite = { workspace = true }
 aws-config = { workspace = true }
 aws-sdk-s3 = { workspace = true }
+aws-smithy-http-client = { workspace = true }
 redis = { workspace = true }

crates/tranquil-pds/src/state.rs

@@ -251,7 +251,9 @@ impl AppState {
 
         let mut state = match cfg.storage.repo_backend() {
             tranquil_config::RepoBackend::TranquilStore => {
-                tracing::info!("tranquil-store repo backend active. Thank you so much for believing in us!");
+                tracing::info!(
+                    "tranquil-store repo backend active. Thank you so much for believing in us!"
+                );
                 Self::from_store(shutdown).await
             }
             tranquil_config::RepoBackend::Postgres => {

crates/tranquil-pds/tests/common/mod.rs

@@ -83,28 +83,41 @@ fn has_external_infra() -> bool {
         || (std::env::var("DATABASE_URL").is_ok()
             && (std::env::var("S3_ENDPOINT").is_ok() || std::env::var("BLOB_STORAGE_PATH").is_ok()))
 }
+
+#[cfg(not(feature = "external-infra"))]
+fn started_container_ids() -> Vec<String> {
+    let db = DB_CONTAINER.get().map(|c| c.id().to_string());
+    #[cfg(feature = "s3")]
+    let s3 = S3_CONTAINER.get().map(|c| c.id().to_string());
+    #[cfg(not(feature = "s3"))]
+    let s3: Option<String> = None;
+    db.into_iter().chain(s3).collect()
+}
+
+#[cfg(feature = "external-infra")]
+fn started_container_ids() -> Vec<String> {
+    Vec::new()
+}
+
 #[cfg(test)]
 #[ctor::dtor]
 fn cleanup() {
     if let Some(temp_dir) = TEST_TEMP_DIR.get() {
         let _ = std::fs::remove_dir_all(temp_dir);
     }
-    if has_external_infra() {
+    let ids = started_container_ids();
+    if ids.is_empty() {
         return;
     }
     if std::env::var("XDG_RUNTIME_DIR").is_ok() {
         let _ = std::process::Command::new("podman")
-            .args(["rm", "-f", "--filter", "label=tranquil_pds_test=true"])
+            .args(["rm", "-f"])
+            .args(&ids)
             .output();
     }
     let _ = std::process::Command::new("docker")
-        .args([
+        .args(["rm", "-f"])
-            "container",
+        .args(&ids)
-            "prune",
-            "-f",
-            "--filter",
-            "label=tranquil_pds_test=true",
-        ])
         .output();
 }
 
@@ -257,6 +270,13 @@ async fn setup_with_testcontainers() -> String {
     }
     let sdk_config = aws_config::defaults(BehaviorVersion::latest())
         .region("us-east-1")
+        .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,
+                ))
+                .build_https(),
+        )
         .endpoint_url(&s3_endpoint)
         .credentials_provider(Credentials::new(
             "minioadmin",
@@ -734,7 +754,40 @@ pub async fn spawn_cluster(pool: Option<sqlx::PgPool>, node_count: usize) -> Vec
     let first = &instances[0];
     APP_PORT.set(first.port).ok();
 
-    tokio::time::sleep(Duration::from_millis(2000)).await;
+    let caches: Vec<Arc<dyn Cache>> = instances
+        .iter()
+        .map(|instance| instance.cache.clone().expect("cluster node has no cache"))
+        .collect();
+    futures::future::join_all(caches.iter().enumerate().map(|(i, cache)| async move {
+        cache
+            .set(
+                &format!("__cluster_ready_{i}"),
+                "1",
+                Duration::from_secs(60),
+            )
+            .await
+            .expect("cluster readiness probe failed");
+    }))
+    .await;
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        let ready = futures::future::join_all(caches.iter().flat_map(|cache| {
+            (0..caches.len()).map(move |i| async move {
+                cache.get(&format!("__cluster_ready_{i}")).await.is_some()
+            })
+        }))
+        .await
+        .into_iter()
+        .all(|seen| seen);
+        if ready {
+            break;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "cluster nodes did not converge within 10s"
+        );
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
 
     instances
 }

crates/tranquil-pds/tests/it/account_lifecycle.rs

@@ -1,6 +1,4 @@
-mod common;
+use crate::common::*;
-mod helpers;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/account_notifications.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::{base_url, client, create_account_and_login, get_test_repos, user_id_of};
-use common::{base_url, client, create_account_and_login, get_test_repos, user_id_of};
 use serde_json::{Value, json};
 use tranquil_db_traits::{CommsChannel, CommsType};
 use tranquil_types::{Did, Recipient};

crates/tranquil-pds/tests/it/actor.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::{base_url, client, create_account_and_login};
-use common::{base_url, client, create_account_and_login};
 use serde_json::{Value, json};
 
 #[tokio::test]

crates/tranquil-pds/tests/it/admin_email.rs

@@ -1,5 +1,3 @@
-mod common;
-
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_db_traits::CommsType;
@@ -7,10 +5,10 @@ use tranquil_types::Did;
 
 #[tokio::test]
 async fn test_send_email_success() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
-    let (access_jwt, did) = common::create_admin_account_and_login(&client).await;
+    let (access_jwt, did) = crate::common::create_admin_account_and_login(&client).await;
     let res = client
         .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
         .bearer_auth(&access_jwt)
@@ -26,7 +24,7 @@ async fn test_send_email_success() {
     assert_eq!(res.status(), StatusCode::OK);
     let body: Value = res.json().await.expect("Invalid JSON");
     assert_eq!(body["sent"], true);
-    let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
+    let user_id = crate::common::user_id_of(repos, &Did::new(did).unwrap()).await;
     let comms = repos
         .infra
         .get_latest_comms_for_user(user_id, CommsType::AdminEmail, 1)
@@ -43,10 +41,10 @@ async fn test_send_email_success() {
 
 #[tokio::test]
 async fn test_send_email_default_subject() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
-    let (access_jwt, did) = common::create_admin_account_and_login(&client).await;
+    let (access_jwt, did) = crate::common::create_admin_account_and_login(&client).await;
     let res = client
         .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
         .bearer_auth(&access_jwt)
@@ -61,7 +59,7 @@ async fn test_send_email_default_subject() {
     assert_eq!(res.status(), StatusCode::OK);
     let body: Value = res.json().await.expect("Invalid JSON");
     assert_eq!(body["sent"], true);
-    let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
+    let user_id = crate::common::user_id_of(repos, &Did::new(did).unwrap()).await;
     let comms = repos
         .infra
         .get_latest_comms_for_user(user_id, CommsType::AdminEmail, 10)
@@ -83,9 +81,9 @@ async fn test_send_email_default_subject() {
 
 #[tokio::test]
 async fn test_send_email_recipient_not_found() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let (access_jwt, _) = common::create_admin_account_and_login(&client).await;
+    let (access_jwt, _) = crate::common::create_admin_account_and_login(&client).await;
     let res = client
         .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
         .bearer_auth(&access_jwt)
@@ -104,9 +102,9 @@ async fn test_send_email_recipient_not_found() {
 
 #[tokio::test]
 async fn test_send_email_missing_content() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let (access_jwt, did) = common::create_admin_account_and_login(&client).await;
+    let (access_jwt, did) = crate::common::create_admin_account_and_login(&client).await;
     let res = client
         .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
         .bearer_auth(&access_jwt)
@@ -125,9 +123,9 @@ async fn test_send_email_missing_content() {
 
 #[tokio::test]
 async fn test_send_email_missing_recipient() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let (access_jwt, _) = common::create_admin_account_and_login(&client).await;
+    let (access_jwt, _) = crate::common::create_admin_account_and_login(&client).await;
     let res = client
         .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
         .bearer_auth(&access_jwt)
@@ -144,8 +142,8 @@ async fn test_send_email_missing_recipient() {
 
 #[tokio::test]
 async fn test_send_email_requires_auth() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
         .json(&json!({
@@ -161,11 +159,11 @@ async fn test_send_email_requires_auth() {
 
 #[tokio::test]
 async fn test_send_email_rejects_garbage_stored_email() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
-    let (access_jwt, did) = common::create_admin_account_and_login(&client).await;
+    let (access_jwt, did) = crate::common::create_admin_account_and_login(&client).await;
-    let user_id = common::user_id_of(repos, &Did::new(did.clone()).unwrap()).await;
+    let user_id = crate::common::user_id_of(repos, &Did::new(did.clone()).unwrap()).await;
     repos
         .user
         .update_email(user_id, "not-an-email")

crates/tranquil-pds/tests/it/admin_invite.rs

@@ -1,6 +1,4 @@
-mod common;
+use crate::common::*;
-
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/admin_moderation.rs

@@ -1,6 +1,4 @@
-mod common;
+use crate::common::*;
-
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/admin_search.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::Value;
 

crates/tranquil-pds/tests/it/admin_stats.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::{base_url, client, create_admin_account_and_login};
-use common::{base_url, client, create_admin_account_and_login};
 use serde_json::Value;
 
 #[tokio::test]

crates/tranquil-pds/tests/it/auth_extractor.rs

@@ -1,10 +1,7 @@
-mod common;
+use crate::common::{base_url, client, create_account_and_login, pds_endpoint};
-mod helpers;
+use crate::helpers::verify_new_account;
-
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use chrono::Utc;
-use common::{base_url, client, create_account_and_login, pds_endpoint};
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};
@@ -215,7 +212,7 @@ async fn test_oauth_admin_extractor_allows_oauth_tokens() {
     let did = account["did"].as_str().unwrap().to_string();
     verify_new_account(&http_client, &did).await;
 
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     repos
         .user
         .set_admin_status(&tranquil_types::Did::new(did.clone()).unwrap(), true)

crates/tranquil-pds/tests/it/banned_words.rs

@@ -13,10 +13,8 @@
  * - we maintain this code precisely because we believe everyone deserves an experience on the web that is free from this kinda language
  */
 
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::json;
 

crates/tranquil-pds/tests/it/change_password.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/delete_account.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
 use chrono::Utc;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/did_web.rs

@@ -1,7 +1,6 @@
-mod common;
+use crate::common::*;
 use base64::Engine;
 use base64::engine::general_purpose::URL_SAFE_NO_PAD;
-use common::*;
 use k256::ecdsa::{SigningKey, signature::Signer};
 use reqwest::StatusCode;
 use serde_json::{Value, json};
@@ -100,7 +99,7 @@ async fn test_external_did_web_no_local_doc() {
     let mock_addr = mock_uri.trim_start_matches("http://");
     let did = format!("did:web:{}", mock_addr.replace(":", "%3A"));
     let handle = format!("xw{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
-    let pds_endpoint = common::pds_endpoint();
+    let pds_endpoint = crate::common::pds_endpoint();
 
     let reserve_res = client
         .post(format!(
@@ -203,7 +202,7 @@ async fn assert_reserved_key_placement_rejected(
     let mock_addr = mock_uri.trim_start_matches("http://");
     let did = format!("did:web:{}", mock_addr.replace(":", "%3A"));
     let handle = format!("wm{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
-    let pds_endpoint = common::pds_endpoint();
+    let pds_endpoint = crate::common::pds_endpoint();
 
     let signing_key = reserve_signing_key(&client, base, &did).await;
     let signing_multibase = signing_key
@@ -525,8 +524,8 @@ async fn test_did_web_byod_flow() {
         unique_id
     );
     let handle = format!("by{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
-    let pds_endpoint = common::pds_endpoint();
+    let pds_endpoint = crate::common::pds_endpoint();
-    let pds_hostname = common::pds_hostname();
+    let pds_hostname = crate::common::pds_hostname();
     let pds_did = format!("did:web:{}", pds_hostname);
 
     let temp_key = SigningKey::random(&mut rand::thread_rng());
@@ -579,7 +578,7 @@ async fn test_did_web_byod_flow() {
         "BYOD accounts should require verification"
     );
 
-    let access_jwt = common::verify_new_account(&client, returned_did).await;
+    let access_jwt = crate::common::verify_new_account(&client, returned_did).await;
 
     let res = client
         .get(format!(

crates/tranquil-pds/tests/it/email_update.rs

@@ -1,11 +1,10 @@
-mod common;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_db_traits::CommsType;
 use tranquil_types::Did;
 
 async fn get_email_update_token(did: &str) -> String {
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let parsed_did = Did::new(did.to_string()).unwrap();
     let user = repos
         .user
@@ -60,14 +59,14 @@ async fn create_verified_account(
     assert_eq!(res.status(), StatusCode::OK);
     let body: Value = res.json().await.expect("Invalid JSON");
     let did = body["did"].as_str().expect("No did").to_string();
-    let jwt = common::verify_new_account(client, &did).await;
+    let jwt = crate::common::verify_new_account(client, &did).await;
     (jwt, did)
 }
 
 #[tokio::test]
 async fn test_request_email_update_returns_token_required() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("er{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
     let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
@@ -88,9 +87,9 @@ async fn test_request_email_update_returns_token_required() {
 
 #[tokio::test]
 async fn test_update_email_flow_success() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("eu{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
     let (access_jwt, did) = create_verified_account(&client, base_url, &handle, &email).await;
@@ -136,8 +135,8 @@ async fn test_update_email_flow_success() {
 
 #[tokio::test]
 async fn test_update_email_requires_token_when_verified() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("ed{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
     let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
@@ -157,8 +156,8 @@ async fn test_update_email_requires_token_when_verified() {
 
 #[tokio::test]
 async fn test_update_email_same_email_noop() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("es{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
     let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
@@ -179,8 +178,8 @@ async fn test_update_email_same_email_noop() {
 
 #[tokio::test]
 async fn test_update_email_invalid_token() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("eb{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
     let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
@@ -214,8 +213,8 @@ async fn test_update_email_invalid_token() {
 
 #[tokio::test]
 async fn test_update_email_no_auth() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
 
     let res = client
         .post(format!("{}/xrpc/com.atproto.server.updateEmail", base_url))
@@ -230,8 +229,8 @@ async fn test_update_email_no_auth() {
 
 #[tokio::test]
 async fn test_update_email_invalid_format() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("ef{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
     let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
@@ -248,9 +247,9 @@ async fn test_update_email_invalid_format() {
 
 #[tokio::test]
 async fn test_confirm_email_confirms_existing_email() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("ec{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
 
@@ -323,9 +322,9 @@ async fn test_confirm_email_confirms_existing_email() {
 
 #[tokio::test]
 async fn test_confirm_email_rejects_wrong_email() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("ew{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@jola.dev", handle);
 
@@ -391,8 +390,8 @@ async fn test_confirm_email_rejects_wrong_email() {
 
 #[tokio::test]
 async fn test_confirm_email_invalid_token() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("ei{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@jola.dev", handle);
 
@@ -433,9 +432,9 @@ async fn test_confirm_email_invalid_token() {
 
 #[tokio::test]
 async fn test_unverified_account_can_update_email_without_token() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("ev{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@nel.pet", handle);
 
@@ -503,9 +502,9 @@ async fn test_unverified_account_can_update_email_without_token() {
 
 #[tokio::test]
 async fn test_update_email_to_same_as_another_user_allowed() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
 
     let handle1 = format!("d1{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email1 = format!("{}@jola.dev", handle1);
@@ -557,8 +556,8 @@ async fn test_update_email_to_same_as_another_user_allowed() {
 
 #[tokio::test]
 async fn test_check_email_in_use_distinguishes_empty_from_invalid() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base = common::base_url().await;
+    let base = crate::common::base_url().await;
 
     let resp = client
         .post(format!("{}/xrpc/_account.checkEmailInUse", base))

crates/tranquil-pds/tests/it/feed_proxy_aud.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::*;
-
 use base64::Engine;
 use base64::engine::general_purpose::URL_SAFE_NO_PAD;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use std::sync::{Arc, Mutex};

crates/tranquil-pds/tests/it/firehose_inline_blocks.rs

@@ -1,11 +1,7 @@
-mod common;
+use crate::common::*;
-mod firehose;
+use crate::firehose::FirehoseConsumer;
-mod helpers;
+use crate::helpers::build_car_with_signature;
-
 use cid::Cid;
-use common::*;
-use firehose::FirehoseConsumer;
-use helpers::build_car_with_signature;
 use iroh_car::CarReader;
 use k256::ecdsa::SigningKey;
 use multihash::Multihash;

crates/tranquil-pds/tests/it/firehose_validation.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-
 use cid::Cid;
-use common::*;
 use futures::{SinkExt, stream::StreamExt};
 use iroh_car::CarReader;
 use reqwest::StatusCode;

crates/tranquil-pds/tests/it/gc_after_delete.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
 use chrono::Utc;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_types::{Did, Nsid, Rkey};

crates/tranquil-pds/tests/it/gc_compaction_restart.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-
 use chrono::Utc;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/identity.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::*;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use wiremock::matchers::{method, path};
@@ -98,7 +97,7 @@ async fn test_create_did_web_account_and_resolve() {
     let did = format!("did:web:{}", mock_addr.replace(":", "%3A"));
     let handle = format!("wu{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let base = base_url().await;
-    let pds_endpoint = common::pds_endpoint();
+    let pds_endpoint = crate::common::pds_endpoint();
 
     let reserve_res = client
         .post(format!(
@@ -220,7 +219,7 @@ async fn test_did_web_lifecycle() {
     let handle = format!("lc{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let did = format!("did:web:{}:u:{}", mock_addr.replace(":", "%3A"), handle);
     let email = format!("{}@test.com", handle);
-    let pds_endpoint = common::pds_endpoint();
+    let pds_endpoint = crate::common::pds_endpoint();
 
     let reserve_res = client
         .post(format!(

crates/tranquil-pds/tests/it/import_verification.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::*;
-use common::*;
 use iroh_car::CarHeader;
 use reqwest::StatusCode;
 use serde_json::json;

crates/tranquil-pds/tests/it/invite.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::*;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/invite_registration.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::*;
-use common::*;
 use reqwest::{Client, StatusCode};
 use serde_json::{Value, json};
 use tranquil_pds::api::error::ApiError;

crates/tranquil-pds/tests/it/jwt_security.rs

@@ -1,8 +1,7 @@
 #![allow(unused_imports)]
-mod common;
+use crate::common::{base_url, client, create_account_and_login, get_test_repos};
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use chrono::{Duration, Utc};
-use common::{base_url, client, create_account_and_login, get_test_repos};
 use k256::SecretKey;
 use k256::ecdsa::{Signature, SigningKey, signature::Signer};
 use rand::rngs::OsRng;

crates/tranquil-pds/tests/it/legacy_2fa.rs

@@ -1,6 +1,4 @@
-mod common;
+use crate::common::{base_url, client, create_account_and_login, get_test_repos, user_id_of};
-
-use common::{base_url, client, create_account_and_login, get_test_repos, user_id_of};
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_db_traits::{CommsChannel, CommsType};

crates/tranquil-pds/tests/it/lifecycle_record.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
 use chrono::Utc;
-use common::*;
-use helpers::*;
 use reqwest::{StatusCode, header};
 use serde_json::{Value, json};
 use std::time::Duration;

crates/tranquil-pds/tests/it/lifecycle_session.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
 use chrono::Utc;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/lifecycle_social.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
 use chrono::Utc;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/main.rs

@@ -0,0 +1,85 @@
+#[path = "../common/mod.rs"]
+mod common;
+#[path = "../firehose/mod.rs"]
+mod firehose;
+#[path = "../helpers/mod.rs"]
+mod helpers;
+#[path = "../mst_verify/mod.rs"]
+mod mst_verify;
+
+mod account_lifecycle;
+mod account_notifications;
+mod actor;
+mod admin_email;
+mod admin_invite;
+mod admin_moderation;
+mod admin_search;
+mod admin_stats;
+mod apple_sso_unit;
+mod auth_extractor;
+mod banned_words;
+mod car_export_detects_corruption;
+mod change_password;
+mod commit_signing;
+mod delete_account;
+mod did_web;
+mod dpop_unit;
+mod email_update;
+mod feed_proxy_aud;
+mod firehose_inline_blocks;
+mod firehose_validation;
+mod gc_after_delete;
+mod gc_compaction_restart;
+mod identity;
+mod image_processing;
+mod import_verification;
+mod invite;
+mod invite_registration;
+mod jwt_security;
+mod legacy_2fa;
+mod lifecycle_record;
+mod lifecycle_session;
+mod lifecycle_social;
+mod moderation;
+mod mst_diff_equivalence;
+mod mst_firehose_e2e;
+mod mst_inductive_firehose;
+mod mst_integrity;
+mod mst_property_fuzz;
+mod mst_repair_audit;
+mod mst_repair_durability;
+mod mst_structural_repair;
+mod notifications;
+mod oauth;
+mod oauth_lifecycle;
+mod oauth_permission_sets;
+mod oauth_scopes;
+mod oauth_security;
+mod oauth_token_eviction;
+mod password_reset;
+mod plc_operations;
+mod plc_validation;
+mod rate_limit;
+mod record_validation;
+mod repair_leaf_loss;
+mod repo_batch;
+mod repo_blob;
+mod repo_conformance;
+mod repo_lifecycle;
+mod ripple_cluster;
+mod scope_edge_cases;
+mod security_fixes;
+mod server;
+mod session_management;
+mod shutdown_unit;
+mod signing_key;
+mod sso;
+mod store_parity;
+mod sync_blob;
+mod sync_conformance;
+mod sync_deprecated;
+mod sync_repo;
+mod user_blocks_reachability;
+mod validation_edge_cases;
+mod verify_live_commit;
+mod whole_story;

crates/tranquil-pds/tests/it/moderation.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/mst_firehose_e2e.rs

@@ -1,16 +1,13 @@
-mod common;
-mod firehose;
-
 use std::collections::BTreeMap;
 use std::io::Cursor;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 
+use crate::common::*;
+use crate::firehose::{FirehoseConsumer, ParsedCommitFrame};
 use bytes::Bytes;
 use cid::Cid;
-use common::*;
-use firehose::{FirehoseConsumer, ParsedCommitFrame};
 use iroh_car::CarReader;
 use jacquard_common::smol_str::SmolStr;
 use jacquard_repo::commit::Commit;
@@ -31,7 +28,7 @@ async fn car_to_blocks(car_bytes: &[u8]) -> BTreeMap<Cid, Bytes> {
     blocks
 }
 
-fn op_to_verified(op: &firehose::ParsedRepoOp) -> Result<VerifiedWriteOp, String> {
+fn op_to_verified(op: &crate::firehose::ParsedRepoOp) -> Result<VerifiedWriteOp, String> {
     let key = SmolStr::new(&op.path);
     match op.action {
         RepoAction::Create => {

crates/tranquil-pds/tests/it/mst_inductive_firehose.rs

@@ -1,17 +1,14 @@
-mod common;
-mod mst_verify;
-
 use std::collections::BTreeMap;
 use std::str::FromStr;
 use std::sync::Arc;
 
+use crate::common::*;
+use crate::mst_verify::{extract_event_blocks, inline_to_store};
 use cid::Cid;
-use common::*;
 use jacquard_common::smol_str::SmolStr;
 use jacquard_repo::commit::Commit;
 use jacquard_repo::mst::{Mst, VerifiedWriteOp};
 use jacquard_repo::storage::{BlockStore, MemoryBlockStore};
-use mst_verify::{extract_event_blocks, inline_to_store};
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_db_traits::{RepoEventType, SequenceNumber, SequencedEvent};

crates/tranquil-pds/tests/it/mst_integrity.rs

@@ -1,17 +1,13 @@
-mod common;
-mod firehose;
-mod helpers;
-
 use std::collections::BTreeMap;
 use std::io::Cursor;
 use std::sync::Arc;
 use std::time::Duration;
 
+use crate::common::*;
+use crate::firehose::FirehoseConsumer;
+use crate::helpers::build_car_with_signature;
 use bytes::Bytes;
 use cid::Cid;
-use common::*;
-use firehose::FirehoseConsumer;
-use helpers::build_car_with_signature;
 use iroh_car::CarReader;
 use jacquard_repo::commit::Commit;
 use jacquard_repo::mst::Mst;

crates/tranquil-pds/tests/it/mst_property_fuzz.rs

@@ -1,16 +1,13 @@
-mod common;
-mod mst_verify;
-
 use std::collections::HashMap;
 use std::str::FromStr;
 
+use crate::common::*;
+use crate::mst_verify::{extract_event_blocks, inline_to_store};
 use cid::Cid;
-use common::*;
 use jacquard_common::smol_str::SmolStr;
 use jacquard_repo::commit::Commit;
 use jacquard_repo::mst::{Mst, VerifiedWriteOp};
 use jacquard_repo::storage::BlockStore;
-use mst_verify::{extract_event_blocks, inline_to_store};
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng};
 use reqwest::StatusCode;

crates/tranquil-pds/tests/it/notifications.rs

@@ -1,13 +1,12 @@
-mod common;
 use tranquil_db_traits::{CommsChannel, CommsStatus, CommsType};
 use tranquil_types::{Did, Recipient};
 
 #[tokio::test]
 async fn test_enqueue_comms() {
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
-    let (_, did) = common::create_account_and_login(&common::client()).await;
+    let (_, did) = crate::common::create_account_and_login(&crate::common::client()).await;
-    let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
+    let user_id = crate::common::user_id_of(repos, &Did::new(did).unwrap()).await;
-    repos
+    let comms_id = repos
         .infra
         .enqueue_comms(
             Some(user_id),
@@ -21,10 +20,13 @@ async fn test_enqueue_comms() {
         .expect("Failed to enqueue comms");
     let comms = repos
         .infra
-        .get_latest_comms_for_user(user_id, CommsType::Welcome, 1)
+        .get_latest_comms_for_user(user_id, CommsType::Welcome, 2)
         .await
         .expect("DB error");
-    let row = comms.first().expect("Comms not found");
+    let row = comms
+        .iter()
+        .find(|c| c.id == comms_id)
+        .expect("Comms not found");
     assert_eq!(row.user_id, Some(user_id));
     assert_eq!(row.recipient, "test@nel.pet");
     assert_eq!(row.subject.as_deref(), Some("Test Subject"));
@@ -36,9 +38,9 @@ async fn test_enqueue_comms() {
 
 #[tokio::test]
 async fn test_comms_queue_status_index() {
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
-    let (_, did) = common::create_account_and_login(&common::client()).await;
+    let (_, did) = crate::common::create_account_and_login(&crate::common::client()).await;
-    let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
+    let user_id = crate::common::user_id_of(repos, &Did::new(did).unwrap()).await;
     let initial_count = repos
         .infra
         .count_comms_by_type(user_id, CommsType::PasswordReset)

crates/tranquil-pds/tests/it/oauth.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::{base_url, client, get_test_repos};
-mod helpers;
+use crate::helpers::verify_new_account;
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
-use common::{base_url, client, get_test_repos};
-use helpers::verify_new_account;
 use reqwest::{StatusCode, redirect};
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};

crates/tranquil-pds/tests/it/oauth_lifecycle.rs

@@ -1,10 +1,7 @@
-mod common;
+use crate::common::{base_url, client};
-mod helpers;
+use crate::helpers::verify_new_account;
-
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use chrono::Utc;
-use common::{base_url, client};
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};

crates/tranquil-pds/tests/it/oauth_permission_sets.rs

@@ -1,10 +1,7 @@
-mod common;
+use crate::common::{base_url, client, create_account_and_login};
-mod helpers;
+use crate::helpers::verify_new_account;
-
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use chrono::Utc;
-use common::{base_url, client, create_account_and_login};
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};
@@ -59,7 +56,7 @@ async fn setup_mock_client_metadata(redirect_uri: &str) -> MockServer {
 }
 
 async fn seed_permission_set(nsid: &str, granular_scope: &str) {
-    let state = common::get_test_app_state().await;
+    let state = crate::common::get_test_app_state().await;
     let key = tranquil_pds::cache_keys::permission_set_key(
         &tranquil_types::Nsid::new(nsid).unwrap(),
         None,
@@ -597,7 +594,7 @@ async fn test_grant_row_keeps_include_jwt_carries_expanded() {
     );
 
     let token_id = token_id_from_jwt(&session.access_token);
-    let token_data = common::get_test_repos()
+    let token_data = crate::common::get_test_repos()
         .await
         .oauth
         .get_token_by_id(&token_id)
@@ -916,7 +913,7 @@ async fn test_consent_post_errors_when_set_unresolvable() {
         "Consent GET should succeed"
     );
 
-    let state = common::get_test_app_state().await;
+    let state = crate::common::get_test_app_state().await;
     let key = tranquil_pds::cache_keys::permission_set_key(
         &tranquil_types::Nsid::new(UNRESOLVABLE_NSID).unwrap(),
         None,
@@ -1211,7 +1208,7 @@ async fn test_consent_remember_persists_set_preference() {
 
     let did: tranquil_types::Did = delegated_did.parse().expect("valid did");
     let client_id_typed = tranquil_types::ClientId::new(client_id.clone());
-    let stored_prefs = common::get_test_repos()
+    let stored_prefs = crate::common::get_test_repos()
         .await
         .oauth
         .get_scope_preferences(&did, &client_id_typed)
@@ -1595,7 +1592,7 @@ async fn test_consent_post_drops_unpresented_scope() {
     let access_token = token_body["access_token"].as_str().unwrap().to_string();
 
     let token_id = token_id_from_jwt(&access_token);
-    let token_data = common::get_test_repos()
+    let token_data = crate::common::get_test_repos()
         .await
         .oauth
         .get_token_by_id(&token_id)

crates/tranquil-pds/tests/it/oauth_scopes.rs

@@ -1,10 +1,7 @@
-mod common;
+use crate::common::{base_url, client, get_test_repos};
-mod helpers;
+use crate::helpers::verify_new_account;
-
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use chrono::Utc;
-use common::{base_url, client, get_test_repos};
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};

crates/tranquil-pds/tests/it/oauth_security.rs

@@ -1,10 +1,8 @@
 #![allow(unused_imports)]
-mod common;
+use crate::common::{base_url, client, create_account_and_login};
-mod helpers;
+use crate::helpers::verify_new_account;
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use chrono::Utc;
-use common::{base_url, client, create_account_and_login};
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};

crates/tranquil-pds/tests/it/oauth_token_eviction.rs

@@ -1,10 +1,7 @@
-mod common;
+use crate::common::{base_url, client, get_test_repos};
-mod helpers;
+use crate::helpers::verify_new_account;
-
 use chrono::{DateTime, Duration, Utc};
-use common::{base_url, client, get_test_repos};
 use futures::StreamExt;
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_oauth::{

crates/tranquil-pds/tests/it/password_reset.rs

@@ -1,15 +1,13 @@
-mod common;
+use crate::helpers::verify_new_account;
-mod helpers;
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_db_traits::CommsType;
 
 #[tokio::test]
 async fn test_request_password_reset_creates_code() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("pr{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@example.com", handle);
     let payload = json!({
@@ -55,8 +53,8 @@ async fn test_request_password_reset_creates_code() {
 
 #[tokio::test]
 async fn test_request_password_reset_unknown_email_returns_ok() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.requestPasswordReset",
@@ -71,9 +69,9 @@ async fn test_request_password_reset_unknown_email_returns_ok() {
 
 #[tokio::test]
 async fn test_reset_password_with_valid_token() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("pr2{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@example.com", handle);
     let old_password = "Oldpass123!";
@@ -167,8 +165,8 @@ async fn test_reset_password_with_valid_token() {
 
 #[tokio::test]
 async fn test_reset_password_with_invalid_token() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.resetPassword",
@@ -188,9 +186,9 @@ async fn test_reset_password_with_invalid_token() {
 
 #[tokio::test]
 async fn test_reset_password_with_expired_token() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("pr3{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@example.com", handle);
     let payload = json!({
@@ -249,9 +247,9 @@ async fn test_reset_password_with_expired_token() {
 
 #[tokio::test]
 async fn test_reset_password_invalidates_sessions() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let handle = format!("pr4{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@example.com", handle);
     let payload = json!({
@@ -320,8 +318,8 @@ async fn test_reset_password_invalidates_sessions() {
 
 #[tokio::test]
 async fn test_request_password_reset_empty_email() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.requestPasswordReset",
@@ -338,9 +336,9 @@ async fn test_request_password_reset_empty_email() {
 
 #[tokio::test]
 async fn test_reset_password_creates_notification() {
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("pr5{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let email = format!("{}@example.com", handle);
     let payload = json!({

crates/tranquil-pds/tests/it/plc_operations.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::*;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::json;
 use tranquil_types::Did;

crates/tranquil-pds/tests/it/rate_limit.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::{base_url, client};
-use common::{base_url, client};
 use reqwest::StatusCode;
 use serde_json::json;
 

crates/tranquil-pds/tests/it/repair_leaf_loss.rs

@@ -1,9 +1,6 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-
 use cid::Cid;
-use common::*;
-use helpers::*;
 use jacquard_repo::commit::Commit;
 use jacquard_repo::storage::BlockStore;
 use serde_json::json;

crates/tranquil-pds/tests/it/repo_batch.rs

@@ -1,6 +1,5 @@
-mod common;
+use crate::common::*;
 use chrono::Utc;
-use common::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_db_traits::{Backlink, BacklinkPath};

crates/tranquil-pds/tests/it/repo_blob.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::*;
-use common::*;
 use reqwest::{StatusCode, header};
 use serde_json::Value;
 

crates/tranquil-pds/tests/it/repo_conformance.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
 use chrono::Utc;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/repo_lifecycle.rs

@@ -1,9 +1,6 @@
-mod common;
+use crate::common::*;
-mod firehose;
+use crate::firehose::{FirehoseConsumer, ParsedCommitFrame};
-
 use cid::Cid;
-use common::*;
-use firehose::{FirehoseConsumer, ParsedCommitFrame};
 use iroh_car::CarReader;
 use jacquard_repo::commit::Commit;
 use reqwest::StatusCode;
@@ -12,8 +9,6 @@ use std::io::Cursor;
 use std::str::FromStr;
 use tranquil_scopes::RepoAction;
 
-mod helpers;
-
 async fn create_post_record(client: &reqwest::Client, token: &str, did: &str, text: &str) -> Value {
     let payload = json!({
         "repo": did,
@@ -384,7 +379,7 @@ async fn test_firehose_commit_signature_verification() {
     let client = client();
     let (token, did) = create_account_and_login(&client).await;
 
-    let key_bytes = helpers::get_user_signing_key(&did)
+    let key_bytes = crate::helpers::get_user_signing_key(&did)
         .await
         .expect("Failed to get signing key");
     let signing_key =

crates/tranquil-pds/tests/it/ripple_cluster.rs

@@ -1,5 +1,3 @@
-mod common;
-
 use reqwest::StatusCode;
 use serde_json::json;
 use std::sync::Arc;
@@ -25,14 +23,14 @@ where
     }
 }
 
-fn cache_for(nodes: &[common::ServerInstance], idx: usize) -> Arc<dyn Cache> {
+fn cache_for(nodes: &[crate::common::ServerInstance], idx: usize) -> Arc<dyn Cache> {
     nodes[idx]
         .cache
         .clone()
         .unwrap_or_else(|| panic!("node {idx} should have a cache"))
 }
 
-fn rl_for(nodes: &[common::ServerInstance], idx: usize) -> Arc<dyn DistributedRateLimiter> {
+fn rl_for(nodes: &[crate::common::ServerInstance], idx: usize) -> Arc<dyn DistributedRateLimiter> {
     nodes[idx]
         .distributed_rate_limiter
         .clone()
@@ -41,10 +39,10 @@ fn rl_for(nodes: &[common::ServerInstance], idx: usize) -> Arc<dyn DistributedRa
 
 #[tokio::test]
 async fn cluster_formation() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
     assert!(nodes.len() >= 3, "expected at least 3 cluster nodes");
 
-    let client = common::client();
+    let client = crate::common::client();
     let results: Vec<_> = futures::future::join_all(nodes.iter().map(|node| {
         let client = client.clone();
         let url = node.url.clone();
@@ -71,8 +69,8 @@ async fn cluster_formation() {
 
 #[tokio::test]
 async fn cluster_any_node_access() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
-    let client = common::client();
+    let client = crate::common::client();
 
     let handle = format!("u{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let payload = serde_json::json!({
@@ -97,7 +95,7 @@ async fn cluster_any_node_access() {
         .expect("no accessJwt")
         .to_string();
 
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let user = repos
         .user
         .get_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
@@ -176,7 +174,7 @@ async fn cluster_any_node_access() {
 
 #[tokio::test]
 async fn cache_convergence() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
 
     let cache_a = nodes[0].cache.as_ref().expect("node 0 should have a cache");
     let cache_b = nodes[1].cache.as_ref().expect("node 1 should have a cache");
@@ -218,7 +216,7 @@ async fn cache_convergence() {
 
 #[tokio::test]
 async fn rate_limit_convergence() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
 
     let rl_a = nodes[0]
         .distributed_rate_limiter
@@ -274,7 +272,7 @@ async fn rate_limit_convergence() {
 
 #[tokio::test]
 async fn delete_convergence() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
     let cache_0 = cache_for(nodes, 0);
     let cache_1 = cache_for(nodes, 1);
 
@@ -308,7 +306,7 @@ async fn delete_convergence() {
 
 #[tokio::test]
 async fn three_node_transitive_convergence() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
     let cache_0 = cache_for(nodes, 0);
     let cache_2 = cache_for(nodes, 2);
 
@@ -331,7 +329,7 @@ async fn three_node_transitive_convergence() {
 
 #[tokio::test]
 async fn cluster_overwrite_conflict_resolution() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
     let cache_0 = cache_for(nodes, 0);
     let cache_1 = cache_for(nodes, 1);
     let cache_2 = cache_for(nodes, 2);
@@ -374,7 +372,7 @@ async fn cluster_overwrite_conflict_resolution() {
 
 #[tokio::test]
 async fn cluster_bulk_key_convergence() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
     let cache_0 = cache_for(nodes, 0);
     let cache_1 = cache_for(nodes, 1);
     let cache_2 = cache_for(nodes, 2);
@@ -433,7 +431,7 @@ async fn cluster_bulk_key_convergence() {
 
 #[tokio::test]
 async fn cluster_concurrent_multi_node_writes() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
     let cache_0 = cache_for(nodes, 0);
     let cache_1 = cache_for(nodes, 1);
     let cache_2 = cache_for(nodes, 2);
@@ -542,7 +540,7 @@ async fn cluster_concurrent_multi_node_writes() {
 
 #[tokio::test]
 async fn cluster_rate_limit_multi_node_convergence() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
     let rl_0 = rl_for(nodes, 0);
     let rl_1 = rl_for(nodes, 1);
     let rl_2 = rl_for(nodes, 2);
@@ -632,7 +630,7 @@ fn create_account_on_node<'a>(
             .expect("no accessJwt")
             .to_string();
 
-        let repos = common::get_test_repos().await;
+        let repos = crate::common::get_test_repos().await;
         let user = repos
             .user
             .get_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
@@ -689,9 +687,9 @@ fn create_account_on_node<'a>(
 }
 
 #[tokio::test]
-async fn cross_node_rate_limit_via_login() {
+async fn jail_cross_node_rate_limit_via_login() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
-    let client = common::client();
+    let client = crate::common::client();
 
     let now_ms = u64::try_from(
         std::time::SystemTime::now()
@@ -772,8 +770,8 @@ async fn cross_node_rate_limit_via_login() {
 
 #[tokio::test]
 async fn cross_node_handle_resolution_from_cache() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
-    let client = common::client();
+    let client = crate::common::client();
     let cache_0 = cache_for(nodes, 0);
 
     let fake_handle = format!("cached-{}.test", uuid::Uuid::new_v4().simple());
@@ -825,8 +823,8 @@ async fn cross_node_handle_resolution_from_cache() {
 
 #[tokio::test]
 async fn cross_node_cache_delete_observable_via_http() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
-    let client = common::client();
+    let client = crate::common::client();
     let cache_0 = cache_for(nodes, 0);
     let cache_1 = cache_for(nodes, 1);
 
@@ -892,8 +890,8 @@ async fn cross_node_cache_delete_observable_via_http() {
 
 #[tokio::test]
 async fn cross_node_email_update_status() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
-    let client = common::client();
+    let client = crate::common::client();
     let cache_0 = cache_for(nodes, 0);
     let cache_1 = cache_for(nodes, 1);
 
@@ -967,8 +965,8 @@ async fn cross_node_email_update_status() {
 
 #[tokio::test]
 async fn cross_node_session_revocation() {
-    let nodes = common::cluster().await;
+    let nodes = crate::common::cluster().await;
-    let client = common::client();
+    let client = crate::common::client();
 
     let (token, _did) = create_account_on_node(&client, &nodes[0].url).await;
 

crates/tranquil-pds/tests/it/security_fixes.rs

@@ -1,4 +1,3 @@
-mod common;
 use tranquil_pds::comms::{SendError, is_valid_phone_number};
 use tranquil_pds::image::{ImageError, ImageProcessor};
 
@@ -84,7 +83,7 @@ fn test_send_error_display() {
 
 #[tokio::test]
 async fn test_signup_queue_authentication() {
-    use common::{base_url, client, create_account_and_login};
+    use crate::common::{base_url, client, create_account_and_login};
     let base = base_url().await;
     let http_client = client();
 

crates/tranquil-pds/tests/it/server.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::verify_new_account;
-use common::*;
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/session_management.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 

crates/tranquil-pds/tests/it/signing_key.rs

@@ -1,13 +1,11 @@
-mod common;
+use crate::helpers::verify_new_account;
-mod helpers;
-use helpers::verify_new_account;
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 
 #[tokio::test]
 async fn test_reserve_signing_key_without_did() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -29,9 +27,9 @@ async fn test_reserve_signing_key_without_did() {
 
 #[tokio::test]
 async fn test_reserve_signing_key_with_did() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let target_did = "did:plc:test123456";
     let res = client
         .post(format!(
@@ -60,9 +58,9 @@ async fn test_reserve_signing_key_with_did() {
 
 #[tokio::test]
 async fn test_reserve_signing_key_stores_private_key() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -100,8 +98,8 @@ async fn test_reserve_signing_key_stores_private_key() {
 
 #[tokio::test]
 async fn test_reserve_signing_key_unique_keys() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res1 = client
         .post(format!(
             "{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -131,8 +129,8 @@ async fn test_reserve_signing_key_unique_keys() {
 
 #[tokio::test]
 async fn test_reserve_signing_key_is_public() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -151,9 +149,9 @@ async fn test_reserve_signing_key_is_public() {
 
 #[tokio::test]
 async fn test_create_account_with_reserved_signing_key() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
-    let repos = common::get_test_repos().await;
+    let repos = crate::common::get_test_repos().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -203,8 +201,8 @@ async fn test_create_account_with_reserved_signing_key() {
 
 #[tokio::test]
 async fn test_create_account_with_invalid_signing_key() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let handle = format!("bk{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
     let res = client
         .post(format!(
@@ -227,8 +225,8 @@ async fn test_create_account_with_invalid_signing_key() {
 
 #[tokio::test]
 async fn test_create_account_cannot_reuse_signing_key() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -280,8 +278,8 @@ async fn test_create_account_cannot_reuse_signing_key() {
 
 #[tokio::test]
 async fn test_reserved_key_tokens_work() {
-    let client = common::client();
+    let client = crate::common::client();
-    let base_url = common::base_url().await;
+    let base_url = crate::common::base_url().await;
     let res = client
         .post(format!(
             "{}/xrpc/com.atproto.server.reserveSigningKey",

crates/tranquil-pds/tests/it/sso.rs

@@ -1,6 +1,4 @@
-mod common;
+use crate::common::{base_url, client, create_account_and_login, get_test_repos};
-
-use common::{base_url, client, create_account_and_login, get_test_repos};
 use reqwest::StatusCode;
 use serde_json::{Value, json};
 use tranquil_db_traits::{CommsChannel, CommsType, SsoAction, SsoProviderType};

crates/tranquil-pds/tests/it/store_parity.rs

@@ -1,6 +1,3 @@
-mod common;
-mod helpers;
-
 use std::sync::Arc;
 use tranquil_db::PostgresRepositories;
 use tranquil_db_traits::{Backlink, BacklinkPath, CommsChannel, CommsType};
@@ -85,7 +82,7 @@ async fn create_store_repos() -> Arc<PostgresRepositories> {
 }
 
 async fn create_pg_repos() -> Arc<PostgresRepositories> {
-    let db_url = common::get_db_connection_string().await;
+    let db_url = crate::common::get_db_connection_string().await;
     let pool = sqlx::postgres::PgPoolOptions::new()
         .max_connections(5)
         .connect(&db_url)
@@ -117,7 +114,7 @@ fn test_handle(suffix: &str) -> Handle {
 }
 
 fn test_cid(seed: u8) -> CidLink {
-    CidLink::from_cid(&helpers::make_cid(&[seed]))
+    CidLink::from_cid(&crate::helpers::make_cid(&[seed]))
 }
 
 fn test_nsid(name: &str) -> Nsid {
@@ -139,7 +136,7 @@ fn test_at_uri(did: &Did, collection: &Nsid, rkey: &Rkey) -> AtUri {
 }
 
 async fn seed_user(repos: &PostgresRepositories, did: &Did, handle: &Handle) -> Uuid {
-    let commit_cid = CidLink::from_cid(&helpers::make_cid(did.as_str().as_bytes()));
+    let commit_cid = CidLink::from_cid(&crate::helpers::make_cid(did.as_str().as_bytes()));
     let input = tranquil_db_traits::CreatePasswordAccountInput {
         handle: handle.clone(),
         email: None,
@@ -1780,7 +1777,7 @@ async fn parity_prune_events_older_than() {
     let baseline = f.pg.repo.get_max_seq().await.unwrap();
     f.pg.repo.insert_commit_event(&event).await.unwrap();
     f.store.repo.insert_commit_event(&event).await.unwrap();
-    let pg_seq = common::sequenced_event_for_did(&f.pg, baseline, &did)
+    let pg_seq = crate::common::sequenced_event_for_did(&f.pg, baseline, &did)
         .await
         .seq;
 

crates/tranquil-pds/tests/it/sync_blob.rs

@@ -1,5 +1,4 @@
-mod common;
+use crate::common::*;
-use common::*;
 use reqwest::StatusCode;
 use reqwest::header;
 use serde_json::Value;

crates/tranquil-pds/tests/it/sync_conformance.rs

@@ -1,8 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::Value;
 

crates/tranquil-pds/tests/it/sync_deprecated.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::Value;
 

crates/tranquil-pds/tests/it/sync_repo.rs

@@ -1,7 +1,5 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use reqwest::header;
 use serde_json::{Value, json};

crates/tranquil-pds/tests/it/user_blocks_reachability.rs

@@ -1,8 +1,6 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
 use chrono::Utc;
-use common::*;
-use helpers::*;
 use reqwest::StatusCode;
 use serde_json::json;
 use std::sync::LazyLock;

crates/tranquil-pds/tests/it/verify_live_commit.rs

@@ -1,7 +1,6 @@
 use bytes::Bytes;
 use cid::Cid;
 use std::collections::HashMap;
-mod common;
 
 #[tokio::test]
 #[ignore = "depends on external live server state; run manually with --ignored"]

crates/tranquil-pds/tests/it/whole_story.rs

@@ -1,10 +1,7 @@
-mod common;
+use crate::common::*;
-mod helpers;
+use crate::helpers::*;
-
 use chrono::Utc;
-use common::*;
 use futures::{StreamExt, future::join_all};
-use helpers::*;
 use k256::ecdsa::SigningKey;
 use reqwest::{StatusCode, header};
 use serde_json::{Value, json};
@@ -1171,7 +1168,7 @@ async fn test_backup_restore_workflow() {
 }
 
 #[tokio::test]
-async fn test_scale_1000_posts_with_pagination() {
+async fn jail_test_scale_1000_posts_with_pagination() {
     let client = client();
     let base = base_url().await;
     let (did, jwt) = setup_new_user("scale-posts").await;
@@ -1282,7 +1279,7 @@ async fn test_scale_1000_posts_with_pagination() {
 }
 
 #[tokio::test]
-async fn test_scale_many_users_social_graph() {
+async fn jail_test_scale_many_users_social_graph() {
     let client = client();
     let base = base_url().await;
 
@@ -1693,7 +1690,7 @@ async fn test_scale_batch_operations() {
 }
 
 #[tokio::test]
-async fn test_scale_reply_thread_depth() {
+async fn jail_test_scale_reply_thread_depth() {
     let client = client();
     let base = base_url().await;
     let (did, jwt) = setup_new_user("deep-thread").await;

crates/tranquil-ripple/src/transport.rs

@@ -1327,7 +1327,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn timed_out_write_resets_stream_instead_of_truncating() {
+    async fn jail_timed_out_write_resets_stream_instead_of_truncating() {
         let shutdown = CancellationToken::new();
         let (sender, _rx_sender) =
             Transport::bind("127.0.0.1:0".parse().unwrap(), None, shutdown.clone())
@@ -1368,7 +1368,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn write_timeout_keeps_connection() {
+    async fn jail_write_timeout_keeps_connection() {
         use futures::StreamExt;
 
         let shutdown = CancellationToken::new();

crates/tranquil-ripple/tests/two_node_convergence.rs

@@ -36,8 +36,22 @@ async fn spawn_pair(
         .await
         .expect("node B failed to start");
 
-    tokio::time::sleep(Duration::from_millis(2000)).await;
+    cache_a
-
+        .set("__ready_a", "1", Duration::from_secs(60))
+        .await
+        .expect("readiness probe on A failed");
+    cache_b
+        .set("__ready_b", "1", Duration::from_secs(60))
+        .await
+        .expect("readiness probe on B failed");
+    poll_until(10_000, 50, || {
+        let cache_a = cache_a.clone();
+        let cache_b = cache_b.clone();
+        async move {
+            cache_b.get("__ready_a").await.is_some() && cache_a.get("__ready_b").await.is_some()
+        }
+    })
+    .await;
     ((cache_a, rl_a), (cache_b, rl_b))
 }
 
@@ -780,7 +794,5 @@ async fn two_node_stress_concurrent_load() {
         r.unwrap_or_else(|e| panic!("task {i} panicked: {e}"));
     });
 
-    tokio::time::sleep(Duration::from_secs(12)).await;
-
     shutdown.cancel();
 }

crates/tranquil-scopes/src/coverage.rs

@@ -92,11 +92,7 @@ fn blob_covers(g: &BlobScope, r: &BlobScope) -> bool {
 }
 
 fn rpc_covers(g: &RpcScope, r: &RpcScope) -> bool {
-    let lxm_ok = match &g.lxm {
+    let lxm_ok = g.lxm == "*" || g.lxm == r.lxm;
-        None => true,
-        Some(gl) if gl == "*" => true,
-        Some(gl) => r.lxm.as_deref() == Some(gl.as_str()),
-    };
     let aud_ok = match &g.aud {
         None => true,
         Some(ga) if ga == "*" => true,

crates/tranquil-scopes/src/parser.rs

@@ -93,7 +93,7 @@ impl BlobScope {
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RpcScope {
-    pub lxm: Option<String>,
+    pub lxm: String,
     pub aud: Option<String>,
 }
 
@@ -245,9 +245,10 @@ pub fn parse_scope(scope: &str) -> ParsedScope {
         let lxm = lxm_positional.or_else(|| params.get("lxm").and_then(|v| v.first().cloned()));
         let aud = params.get("aud").and_then(|v| v.first().cloned());
 
-        let is_lxm_wildcard = lxm.as_deref() == Some("*") || lxm.is_none();
+        let Some(lxm) = lxm else {
-        let is_aud_wildcard = aud.as_deref() == Some("*");
+            return ParsedScope::Unknown(scope.to_string());
-        if is_lxm_wildcard && is_aud_wildcard {
+        };
+        if lxm == "*" && aud.as_deref() == Some("*") {
             return ParsedScope::Unknown(scope.to_string());
         }
 
@@ -400,7 +401,7 @@ mod tests {
         let scope = parse_scope("rpc:app.bsky.feed.getTimeline?aud=did:web:api.bsky.app");
         match scope {
             ParsedScope::Rpc(r) => {
-                assert_eq!(r.lxm, Some("app.bsky.feed.getTimeline".to_string()));
+                assert_eq!(r.lxm, "app.bsky.feed.getTimeline");
                 assert_eq!(r.aud, Some("did:web:api.bsky.app".to_string()));
             }
             _ => panic!("Expected Rpc scope"),
@@ -510,6 +511,12 @@ mod tests {
         assert!(matches!(scope4, ParsedScope::Rpc(_)));
     }
 
+    #[test]
+    fn test_rpc_lxm_required() {
+        let bare = parse_scope("rpc");
+        assert!(matches!(bare, ParsedScope::Unknown(_)));
+    }
+
     #[test]
     fn test_url_encoded_aud_with_fragment() {
         let scope =
@@ -527,10 +534,7 @@ mod tests {
         );
         match scope2 {
             ParsedScope::Rpc(r) => {
-                assert_eq!(
+                assert_eq!(r.lxm, "com.atproto.moderation.createReport");
-                    r.lxm,
-                    Some("com.atproto.moderation.createReport".to_string())
-                );
                 assert_eq!(r.aud, Some("did:web:api.bsky.app#bsky_appview".to_string()));
             }
             _ => panic!("Expected Rpc scope"),

crates/tranquil-scopes/src/permissions.rs

@@ -49,10 +49,7 @@ impl ScopePermissions {
 pub fn superseded_by_transition_generic(scope: &ParsedScope) -> bool {
     match scope {
         ParsedScope::Repo(_) | ParsedScope::Blob(_) => true,
-        ParsedScope::Rpc(rpc) => !rpc
+        ParsedScope::Rpc(rpc) => !(rpc.lxm == "*" || rpc.lxm.starts_with("chat.bsky.")),
-            .lxm
-            .as_deref()
-            .is_some_and(|lxm| lxm == "*" || lxm.starts_with("chat.bsky.")),
         ParsedScope::Account(_)
         | ParsedScope::Identity(_)
         | ParsedScope::TransitionEmail
@@ -193,9 +190,9 @@ impl ScopePermissions {
 
         let has_permission = self.find_rpc_scopes().any(|rpc_scope| {
             let lxm_matches = match &rpc_scope.lxm {
-                None => true,
+                scope_lxm if scope_lxm == "*" => true,
-                Some(scope_lxm) if scope_lxm == lxm => true,
+                scope_lxm if scope_lxm == lxm => true,
-                Some(scope_lxm) if scope_lxm.ends_with(".*") => {
+                scope_lxm if scope_lxm.ends_with(".*") => {
                     let prefix = scope_lxm.strip_suffix(".*").unwrap();
                     lxm.starts_with(prefix) && lxm.chars().nth(prefix.len()) == Some('.')
                 }
@@ -203,7 +200,7 @@ impl ScopePermissions {
             };
 
             let aud_matches = match &rpc_scope.aud {
-                None => true,
+                None => false,
                 Some(scope_aud) if scope_aud == "*" => true,
                 Some(scope_aud) => scope_aud == aud,
             };
@@ -474,6 +471,24 @@ mod tests {
         assert!(!perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getAuthorFeed")));
     }
 
+    #[test]
+    fn test_rpc_wildcard_lxm() {
+        let perms = ScopePermissions::from_scope_string(Some(
+            "rpc:*?aud=did:web:api.bsky.app#bsky_appview",
+        ));
+        let aud = "did:web:api.bsky.app#bsky_appview";
+        let other = "did:web:other.app#bsky_appview";
+        assert!(perms.allows_rpc(aud, &c("app.bsky.feed.getTimeline")));
+        assert!(!perms.allows_rpc(other, &c("app.bsky.feed.getTimeline")));
+    }
+
+    #[test]
+    fn test_rpc_wildcard_lxm_without_aud() {
+        let perms = ScopePermissions::from_scope_string(Some("rpc:*"));
+        let aud = "did:web:api.bsky.app#bsky_appview";
+        assert!(!perms.allows_rpc(aud, &c("app.bsky.feed.getTimeline")));
+    }
+
     #[test]
     fn test_granular_account() {
         let perms = ScopePermissions::from_scope_string(Some("account:email?action=read"));

crates/tranquil-storage/Cargo.toml

@@ -5,7 +5,7 @@ edition.workspace = true
 license.workspace = true
 
 [features]
-s3 = ["dep:aws-config", "dep:aws-sdk-s3"]
+s3 = ["dep:aws-config", "dep:aws-sdk-s3", "dep:aws-smithy-http-client"]
 
 [dependencies]
 tranquil-config = { workspace = true }
@@ -14,9 +14,17 @@ tranquil-infra = { workspace = true }
 async-trait = { workspace = true }
 aws-config = { workspace = true, optional = true }
 aws-sdk-s3 = { workspace = true, optional = true }
+aws-smithy-http-client = { workspace = true, optional = true }
 bytes = { workspace = true }
 futures = { workspace = true }
 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 {
+        pub async fn new(bucket: &str, endpoint: Option<&str>, path: &str) -> Self {
-            let cfg = tranquil_config::get();
+            Self {
-            let bucket = cfg
+                client: create_s3_client(endpoint).await,
-                .storage
+                bucket: bucket.to_string(),
-                .s3_bucket
+                path: path
-                .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();
+                    .to_string(),
-            Self {
-                client,
-                bucket,
-                path,
             }
         }
 
@@ -148,28 +138,28 @@ mod s3 {
         }
     }
 
-    async fn create_s3_client() -> Client {
+    async fn create_s3_client(endpoint: Option<&str>) -> Client {
-        let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
+        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,
+            ))
+            .build_https();
 
         let config = aws_config::defaults(BehaviorVersion::latest())
-            .region(region_provider)
+            .http_client(http_client)
             .load()
             .await;
 
-        tranquil_config::get()
+        let region = config
-            .storage
+            .region()
-            .s3_endpoint
+            .cloned()
-            .as_deref()
+            .unwrap_or_else(|| Region::from_static("us-east-1"));
-            .map_or_else(
+        let builder = aws_sdk_s3::config::Builder::from(&config).region(region);
-                || Client::new(&config),
+        let builder = match endpoint {
-                |endpoint| {
+            Some(endpoint) => builder.endpoint_url(endpoint).force_path_style(true),
-                    let s3_config = aws_sdk_s3::config::Builder::from(&config)
+            None => builder,
-                        .endpoint_url(endpoint)
+        };
-                        .force_path_style(true)
+        Client::from_conf(builder.build())
-                        .build();
-                    Client::from_conf(s3_config)
-                },
-            )
     }
 
     #[async_trait]
@@ -250,7 +240,7 @@ mod s3 {
         async fn put_stream(
             &self,
             key: &str,
-            stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
+            mut stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
         ) -> Result<StreamUploadResult, StorageError> {
             use futures::StreamExt;
 
@@ -339,12 +329,10 @@ mod s3 {
                     .await;
             };
 
-            let result: Result<UploadState, StorageError> = {
+            let result: Result<StreamUploadResult, StorageError> = async {
                 let mut state = initial_state;
 
-                let chunk_results: Vec<Result<Bytes, std::io::Error>> = stream.collect().await;
+                while let Some(chunk_result) = stream.next().await {
-
-                for chunk_result in chunk_results {
                     match chunk_result {
                         Ok(chunk) => {
                             state.hasher.update(&chunk);
@@ -369,17 +357,9 @@ mod s3 {
                                 state.part_number += 1;
                             }
                         }
-                        Err(e) => {
+                        Err(e) => return Err(StorageError::Io(e)),
-                            abort_upload().await;
-                            return Err(StorageError::Io(e));
                     }
                 }
-                }
-
-                Ok(state)
-            };
-
-            let mut state = result?;
 
                 if !state.buffer.is_empty() {
                     let part = upload_part(
@@ -395,7 +375,6 @@ mod s3 {
                 }
 
                 if state.completed_parts.is_empty() {
-                abort_upload().await;
                     return Err(StorageError::Other("Empty upload".to_string()));
                 }
 
@@ -422,6 +401,13 @@ mod s3 {
                     size: state.total_size,
                 })
             }
+            .await;
+
+            if result.is_err() {
+                abort_upload().await;
+            }
+            result
+        }
 
         async fn copy(&self, src_key: &str, dst_key: &str) -> Result<(), StorageError> {
             let src_path = self.resolve_path(src_key);
@@ -597,7 +583,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,170 @@
+#![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::dto::{
+    AbortMultipartUploadInput, AbortMultipartUploadOutput, CreateMultipartUploadInput,
+    CreateMultipartUploadOutput,
+};
+use s3s::service::S3ServiceBuilder;
+use s3s::{S3, S3Request, S3Response, S3Result};
+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) {
+    start_s3_with(|fs| fs).await
+}
+
+async fn start_s3_with<T: S3>(wrap: impl FnOnce(FileSystem) -> T) -> (TempDir, S3BlobStorage) {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::create_dir(root.path().join(BUCKET)).unwrap();
+
+    let mut builder = S3ServiceBuilder::new(wrap(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);
+}
+
+struct FailingUploadPart(FileSystem);
+
+#[async_trait::async_trait]
+impl S3 for FailingUploadPart {
+    async fn create_multipart_upload(
+        &self,
+        req: S3Request<CreateMultipartUploadInput>,
+    ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
+        self.0.create_multipart_upload(req).await
+    }
+
+    async fn abort_multipart_upload(
+        &self,
+        req: S3Request<AbortMultipartUploadInput>,
+    ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
+        self.0.abort_multipart_upload(req).await
+    }
+}
+
+#[tokio::test]
+async fn put_stream_part_failure_aborts_upload() {
+    let (root, storage) = start_s3_with(FailingUploadPart).await;
+    let chunks = [Ok(Bytes::from("hello"))];
+
+    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);
+}

crates/tranquil-store/Cargo.toml

@@ -46,6 +46,7 @@ test-harness = ["dep:tempfile"]
 jemalloc = ["dep:tikv-jemallocator"]
 gauntlet-cli = ["test-harness", "dep:clap", "dep:toml", "dep:tracing-subscriber", "jemalloc"]
 gauntlet-jemalloc-prof = []
+skip-fsync = []
 
 [[bin]]
 name = "tranquil-gauntlet"