PiCloud/crates/manager-core/tests/atomic_write.rs

//! Audit fix #6: a data write and the trigger fan-out it produces must commit
//! ATOMICALLY.
//!
//! Before this, a service wrote the row (commit #1), then asked the emitter to
//! resolve matching triggers and insert outbox rows (commit #2). If commit #2
//! failed, the row was permanently in the store with its trigger having never
//! fired — unobservable to the caller, unrecoverable by any retry, and only
//! logged. `PostgresKvWriter` runs both on ONE connection in ONE transaction, so
//! an emit failure rolls the write back and surfaces as an error.
//!
//! Fault injection: a Postgres BEFORE-INSERT trigger on `outbox` that raises,
//! scoped to this test's freshly-minted `app_id`.
//!
//! That scoping bounds which ROWS the trigger rejects, but installing it is still
//! `CREATE TRIGGER ... ON outbox` — DDL, taking an ACCESS EXCLUSIVE lock on a
//! table every other suite is concurrently inserting into. On the shared dev
//! database that made this suite fail intermittently under load (and briefly
//! imposed our trigger on everyone else's inserts). So each test here now runs
//! against its OWN database, which is what makes the DDL safe; it also means no
//! test needs to clean up after itself. Skips when `DATABASE_URL` is unset.

use std::sync::Arc;

use picloud_manager_core::atomic_write::{
    FilesWriter, GroupDocsTarget, GroupDocsWriter, GroupFilesQuota, GroupFilesWriter,
    GroupKvTarget, GroupKvWriter, KvWriter, PostgresFilesWriter, PostgresGroupDocsWriter,
    PostgresGroupFilesWriter, PostgresGroupKvWriter, PostgresKvWriter,
};
use picloud_manager_core::files_repo::{FilesConfig, FilesRepo, FsFilesRepo};
use picloud_manager_core::quota::GroupWriteQuota;
use picloud_shared::{
    AppId, ExecutionId, FileUpdate, FilesError, GroupDocsError, GroupFilesError, GroupId,
    GroupKvError, KvError, NewFile, RequestId, ScriptId, SdkCallCx,
};
use sqlx::PgPool;
use uuid::Uuid;

async fn pool_or_skip() -> Option<PgPool> {
    // 24 connections: the quota races below spawn enough concurrent writers to
    // actually contend for the per-group advisory lock.
    picloud_test_support::test_pool_sized("atomic_write", 24).await
}

fn cx(app_id: Uuid) -> SdkCallCx {
    SdkCallCx {
        app_id: AppId::from(app_id),
        script_id: ScriptId::new(),
        principal: None,
        execution_id: ExecutionId::new(),
        request_id: RequestId::new(),
        trigger_depth: 0,
        root_execution_id: ExecutionId::new(),
        is_dead_letter_handler: false,
        event: None,
    }
}

/// An app with one script and one enabled KV trigger watching `collection`.
struct Fixture {
    app: Uuid,
}

async fn setup(pool: &PgPool, collection: &str) -> Fixture {
    let uniq = Uuid::new_v4().simple().to_string();
    let (group,): (Uuid,) =
        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
            .bind(format!("aw-grp-{uniq}"))
            .fetch_one(pool)
            .await
            .expect("group");
    let (app,): (Uuid,) =
        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
            .bind(format!("aw-app-{uniq}"))
            .bind(group)
            .fetch_one(pool)
            .await
            .expect("app");
    let (admin,): (Uuid,) = sqlx::query_as(
        "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id",
    )
    .bind(format!("aw-admin-{uniq}"))
    .fetch_one(pool)
    .await
    .expect("admin");
    let (script,): (Uuid,) = sqlx::query_as(
        "INSERT INTO scripts (app_id, name, source) VALUES ($1, 'handler', '') RETURNING id",
    )
    .bind(app)
    .fetch_one(pool)
    .await
    .expect("script");
    let (trigger,): (Uuid,) = sqlx::query_as(
        "INSERT INTO triggers ( \
            app_id, script_id, kind, retry_max_attempts, retry_backoff, \
            retry_base_ms, registered_by_principal \
         ) VALUES ($1, $2, 'kv', 3, 'exponential', 1000, $3) RETURNING id",
    )
    .bind(app)
    .bind(script)
    .bind(admin)
    .fetch_one(pool)
    .await
    .expect("trigger");
    sqlx::query(
        "INSERT INTO kv_trigger_details (trigger_id, collection_glob, ops) \
         VALUES ($1, $2, '{}')",
    )
    .bind(trigger)
    .bind(collection)
    .execute(pool)
    .await
    .expect("kv_trigger_details");
    Fixture { app }
}

/// Make every outbox insert for `app` fail. Scoped by app_id so parallel tests
/// (which use their own fresh apps) are untouched.
async fn break_outbox_for(pool: &PgPool, app: Uuid, tag: &str) {
    sqlx::query(&format!(
        "CREATE FUNCTION break_outbox_{tag}() RETURNS TRIGGER AS $$ \
         BEGIN \
           IF NEW.app_id = '{app}'::uuid THEN \
             RAISE EXCEPTION 'injected outbox failure'; \
           END IF; \
           RETURN NEW; \
         END; $$ LANGUAGE plpgsql"
    ))
    .execute(pool)
    .await
    .expect("create fn");
    sqlx::query(&format!(
        "CREATE TRIGGER break_outbox_{tag} BEFORE INSERT ON outbox \
         FOR EACH ROW EXECUTE FUNCTION break_outbox_{tag}()"
    ))
    .execute(pool)
    .await
    .expect("create trigger");
}

async fn unbreak_outbox(pool: &PgPool, tag: &str) {
    sqlx::query(&format!(
        "DROP TRIGGER IF EXISTS break_outbox_{tag} ON outbox"
    ))
    .execute(pool)
    .await
    .expect("drop trigger");
    sqlx::query(&format!("DROP FUNCTION IF EXISTS break_outbox_{tag}()"))
        .execute(pool)
        .await
        .expect("drop fn");
}

async fn kv_count(pool: &PgPool, app: Uuid) -> i64 {
    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM kv_entries WHERE app_id = $1")
        .bind(app)
        .fetch_one(pool)
        .await
        .expect("count kv");
    n
}

async fn outbox_count(pool: &PgPool, app: Uuid) -> i64 {
    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM outbox WHERE app_id = $1")
        .bind(app)
        .fetch_one(pool)
        .await
        .expect("count outbox");
    n
}

/// `scripts` is RESTRICT on app delete (code is not data), so drop it first.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_failed_fan_out_rolls_the_kv_write_back() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let f = setup(&pool, "watched").await;
    let writer = PostgresKvWriter::new(pool.clone(), u64::MAX);
    let cx = cx(f.app);
    // A distinct, valid SQL identifier for this run's trigger/function names.
    let tag = Uuid::new_v4().simple().to_string();

    // --- Control: with the outbox healthy, the write lands AND fans out. -----
    writer
        .set(&cx, "watched", "k", serde_json::json!(1))
        .await
        .expect("healthy set succeeds");
    assert_eq!(kv_count(&pool, f.app).await, 1, "the row is written");
    assert_eq!(
        outbox_count(&pool, f.app).await,
        1,
        "the matching trigger fanned out"
    );

    // --- The fix: a fan-out failure must roll the write back. ---------------
    break_outbox_for(&pool, f.app, &tag).await;
    let err = writer
        .set(&cx, "watched", "k2", serde_json::json!(2))
        .await
        .expect_err("an outbox failure must surface as an error, not be swallowed");
    unbreak_outbox(&pool, &tag).await;

    assert!(
        format!("{err}").contains("event emit"),
        "the error should name the emit failure, got: {err}"
    );
    assert_eq!(
        kv_count(&pool, f.app).await,
        1,
        "k2 must NOT be in the store — the write is rolled back with its fan-out. \
         A committed row whose trigger never fired is exactly the durability hole \
         the transactional outbox closes."
    );
    assert_eq!(
        outbox_count(&pool, f.app).await,
        1,
        "and no outbox row for the rolled-back write"
    );

    // --- An unwatched collection still writes (no trigger → no fan-out). ----
    writer
        .set(&cx, "unwatched", "k", serde_json::json!(3))
        .await
        .expect("a write with no matching trigger needs no outbox row");
    assert_eq!(kv_count(&pool, f.app).await, 2);
    assert_eq!(outbox_count(&pool, f.app).await, 1, "still no new fan-out");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_failed_fan_out_rolls_a_delete_back() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let f = setup(&pool, "watched").await;
    let writer = PostgresKvWriter::new(pool.clone(), u64::MAX);
    let cx = cx(f.app);
    let tag = Uuid::new_v4().simple().to_string();

    writer
        .set(&cx, "watched", "doomed", serde_json::json!(1))
        .await
        .expect("seed");
    assert_eq!(kv_count(&pool, f.app).await, 1);

    // A delete that can't fan out must leave the row in place — otherwise the
    // key is gone and nothing downstream ever learns it was removed.
    break_outbox_for(&pool, f.app, &tag).await;
    let err = writer
        .delete(&cx, "watched", "doomed")
        .await
        .expect_err("outbox failure surfaces");
    unbreak_outbox(&pool, &tag).await;
    assert!(format!("{err}").contains("event emit"));
    assert_eq!(
        kv_count(&pool, f.app).await,
        1,
        "the delete rolled back — the key is still there"
    );
}

// ----------------------------------------------------------------------------
// Audit fix #8: the per-group quota must be a BOUND, not a hint.
// ----------------------------------------------------------------------------

async fn mk_group(pool: &PgPool) -> Uuid {
    let (g,): (Uuid,) =
        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
            .bind(format!("q-grp-{}", Uuid::new_v4().simple()))
            .fetch_one(pool)
            .await
            .expect("group");
    g
}

async fn group_kv_count(pool: &PgPool, group: Uuid) -> i64 {
    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM group_kv_entries WHERE group_id = $1")
        .bind(group)
        .fetch_one(pool)
        .await
        .expect("count");
    n
}

/// The service read the group's usage on one pooled connection and wrote on
/// another, so N concurrent writers each saw the same pre-write count, each
/// decided it had room, and together blew past the ceiling. The writer now holds
/// a per-group advisory lock across the check AND the write, so they serialize
/// and each sees its predecessor's row.
///
/// Note this is NOT fixed by merely putting the check in the same transaction:
/// under READ COMMITTED each transaction's `COUNT(*)` still sees a snapshot
/// without the others' uncommitted rows. The lock is what does the work.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_writers_cannot_push_a_group_past_its_row_quota() {
    const MAX_ROWS: u64 = 5;

    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let group = mk_group(&pool).await;
    let writer = Arc::new(PostgresGroupKvWriter::new(pool.clone()));
    let quota = GroupWriteQuota {
        max_rows: MAX_ROWS,
        max_total_bytes: u64::MAX,
        max_value_bytes: 256 * 1024,
    };

    // 20 writers race to insert 20 DISTINCT new keys into an empty group whose
    // ceiling is 5. Exactly 5 may win.
    let app = Uuid::new_v4(); // cx.app_id is not used by the group writer's SQL
    let mut set = tokio::task::JoinSet::new();
    for i in 0..20 {
        let w = Arc::clone(&writer);
        set.spawn(async move {
            let cx = cx(app);
            w.set(
                &cx,
                GroupKvTarget {
                    group_id: GroupId::from(group),
                    collection: "shared",
                    key: &format!("k{i}"),
                },
                serde_json::json!(i),
                quota,
            )
            .await
        });
    }
    let mut ok = 0;
    let mut denied = 0;
    while let Some(r) = set.join_next().await {
        match r.expect("task") {
            Ok(()) => ok += 1,
            Err(GroupKvError::QuotaExceeded { .. }) => denied += 1,
            Err(e) => panic!("unexpected error: {e}"),
        }
    }

    assert_eq!(
        group_kv_count(&pool, group).await,
        i64::try_from(MAX_ROWS).unwrap(),
        "the group must hold EXACTLY its ceiling — a check-then-write race would overshoot"
    );
    let max_rows = usize::try_from(MAX_ROWS).unwrap();
    assert_eq!(ok, max_rows, "exactly {MAX_ROWS} writers may win");
    assert_eq!(denied, 20 - max_rows, "the rest are refused");

    sqlx::query("DELETE FROM groups WHERE id = $1")
        .bind(group)
        .execute(&pool)
        .await
        .expect("cleanup");
}

/// The byte ceiling has the same race, and the same fix.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_writers_cannot_push_a_group_past_its_byte_quota() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let group = mk_group(&pool).await;
    let writer = Arc::new(PostgresGroupKvWriter::new(pool.clone()));

    // A ~100-byte value, and a ceiling that fits about 4 of them. `max_value_bytes`
    // is set low enough that the upper-bound fast path can't short-circuit the
    // scan (rows_after * max_value_bytes must exceed the ceiling immediately).
    let value = serde_json::json!("x".repeat(100));
    let quota = GroupWriteQuota {
        max_total_bytes: 450,
        max_rows: u64::MAX,
        max_value_bytes: 200,
    };

    let app = Uuid::new_v4();
    let mut set = tokio::task::JoinSet::new();
    for i in 0..16 {
        let w = Arc::clone(&writer);
        let v = value.clone();
        set.spawn(async move {
            let cx = cx(app);
            w.set(
                &cx,
                GroupKvTarget {
                    group_id: GroupId::from(group),
                    collection: "shared",
                    key: &format!("k{i}"),
                },
                v,
                quota,
            )
            .await
        });
    }
    while let Some(r) = set.join_next().await {
        match r.expect("task") {
            Ok(()) | Err(GroupKvError::TotalBytesQuotaExceeded { .. }) => {}
            Err(e) => panic!("unexpected error: {e}"),
        }
    }

    let (bytes,): (i64,) = sqlx::query_as(
        "SELECT COALESCE(SUM(octet_length(value::text)), 0)::BIGINT \
         FROM group_kv_entries WHERE group_id = $1",
    )
    .bind(group)
    .fetch_one(&pool)
    .await
    .expect("bytes");
    assert!(
        bytes <= 450,
        "stored bytes ({bytes}) must not exceed the 450-byte ceiling — \
         concurrent writers must not each see the same pre-write total"
    );
    assert!(bytes > 0, "some writers should have succeeded");

    sqlx::query("DELETE FROM groups WHERE id = $1")
        .bind(group)
        .execute(&pool)
        .await
        .expect("cleanup");
}

/// Group DOCS carried the identical check-then-write race as group KV, and the
/// identical fix (a per-group advisory lock, on its own `docs`-namespaced key so
/// it doesn't contend with KV writes).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_writers_cannot_push_a_group_past_its_docs_row_quota() {
    const MAX_DOCS: u64 = 4;

    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let group = mk_group(&pool).await;
    let writer = Arc::new(PostgresGroupDocsWriter::new(pool.clone()));
    let quota = GroupWriteQuota {
        max_rows: MAX_DOCS,
        max_total_bytes: u64::MAX,
        max_value_bytes: 256 * 1024,
    };

    let app = Uuid::new_v4();
    let mut set = tokio::task::JoinSet::new();
    for i in 0..16 {
        let w = Arc::clone(&writer);
        set.spawn(async move {
            let cx = cx(app);
            w.create(
                &cx,
                GroupDocsTarget {
                    group_id: GroupId::from(group),
                    collection: "articles",
                },
                serde_json::json!({ "n": i }),
                quota,
            )
            .await
        });
    }
    let mut ok = 0;
    while let Some(r) = set.join_next().await {
        match r.expect("task") {
            Ok(_) => ok += 1,
            Err(GroupDocsError::QuotaExceeded { .. }) => {}
            Err(e) => panic!("unexpected error: {e}"),
        }
    }

    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM group_docs WHERE group_id = $1")
        .bind(group)
        .fetch_one(&pool)
        .await
        .expect("count");
    assert_eq!(
        n,
        i64::try_from(MAX_DOCS).unwrap(),
        "the group must hold EXACTLY its doc ceiling"
    );
    assert_eq!(ok, usize::try_from(MAX_DOCS).unwrap());

    sqlx::query("DELETE FROM groups WHERE id = $1")
        .bind(group)
        .execute(&pool)
        .await
        .expect("cleanup");
}

/// Group FILES had the worst version of the race: the ceiling is disk (10 GiB by
/// default) and a single file may be 100 MB, so a fleet of concurrent uploads
/// each seeing the same pre-write total could overshoot by GIGABYTES of real
/// disk. Same fix — a per-group advisory lock across the check and the write.
///
/// It also pins the second half of the bug: `update` checked NO quota at all, so
/// a 1-byte file could be grown past the ceiling unchallenged.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_uploads_cannot_push_a_group_past_its_files_byte_quota() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let group = mk_group(&pool).await;
    let root = std::env::temp_dir().join(format!("picloud-aw-{}", Uuid::new_v4().simple()));
    let writer = Arc::new(PostgresGroupFilesWriter::new(pool.clone(), root.clone()));

    // Ten 100-byte uploads race for a 450-byte ceiling: at most 4 may land.
    let quota = GroupFilesQuota {
        max_total_bytes: 450,
    };
    let app = Uuid::new_v4();
    let mut set = tokio::task::JoinSet::new();
    for i in 0..10 {
        let w = Arc::clone(&writer);
        set.spawn(async move {
            let cx = cx(app);
            w.create(
                &cx,
                GroupId::from(group),
                "assets",
                NewFile {
                    name: format!("f{i}.bin"),
                    content_type: "application/octet-stream".into(),
                    data: vec![b'x'; 100],
                },
                quota,
            )
            .await
        });
    }
    let mut created = Vec::new();
    while let Some(r) = set.join_next().await {
        match r.expect("task") {
            Ok(id) => created.push(id),
            Err(GroupFilesError::QuotaExceeded { .. }) => {}
            Err(e) => panic!("unexpected error: {e}"),
        }
    }

    let stored = |pool: PgPool| async move {
        let (n,): (i64,) = sqlx::query_as(
            "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT FROM group_files WHERE group_id = $1",
        )
        .bind(group)
        .fetch_one(&pool)
        .await
        .expect("bytes");
        n
    };
    let bytes = stored(pool.clone()).await;
    assert!(
        bytes <= 450,
        "stored bytes ({bytes}) must not exceed the 450-byte ceiling — concurrent \
         uploads must not each see the same pre-write total"
    );
    assert!(!created.is_empty(), "some uploads should have succeeded");

    // An UPDATE that would blow the ceiling must now be refused too. Previously
    // update consulted no quota at all, so this was a free bypass.
    let victim = created[0];
    let err = writer
        .update(
            &cx(app),
            GroupId::from(group),
            "assets",
            victim,
            FileUpdate {
                name: None,
                content_type: None,
                data: vec![b'y'; 10_000],
            },
            quota,
        )
        .await
        .expect_err("growing a file past the group ceiling must be refused");
    assert!(
        matches!(err, GroupFilesError::QuotaExceeded { .. }),
        "got {err}"
    );
    assert_eq!(
        stored(pool.clone()).await,
        bytes,
        "the refused update must not have changed the stored total"
    );

    sqlx::query("DELETE FROM groups WHERE id = $1")
        .bind(group)
        .execute(&pool)
        .await
        .expect("cleanup");
    let _ = std::fs::remove_dir_all(&root);
}

// ----------------------------------------------------------------------------
// Per-app ceilings.
//
// Before these, an app's OWN store had no ceiling at all — only the group-SHARED
// collections did. And `script_gate` returns Ok for an anonymous principal, so a
// public unauthenticated route could write files/keys/docs in a loop until the
// disk was full. The per-file cap bounds ONE file; nothing bounded the count.
// ----------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_app_cannot_exceed_its_key_ceiling_but_updates_stay_free() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let f = setup(&pool, "unwatched").await;
    // Ceiling of 3 keys.
    let writer = PostgresKvWriter::new(pool.clone(), 3);
    let cx = cx(f.app);

    for i in 0..3 {
        writer
            .set(&cx, "c", &format!("k{i}"), serde_json::json!(i))
            .await
            .expect("under the ceiling");
    }
    let err = writer
        .set(&cx, "c", "k3", serde_json::json!(3))
        .await
        .expect_err("a fourth key must be refused");
    assert!(matches!(err, KvError::QuotaExceeded { .. }), "got {err}");

    // An UPDATE adds no row, so it stays allowed at the ceiling — and pays no
    // COUNT at all. (Rejecting updates here would brick an app the moment it
    // filled up, which is worse than the DoS the ceiling exists to stop.)
    writer
        .set(&cx, "c", "k0", serde_json::json!("updated"))
        .await
        .expect("an update at the ceiling is net-zero rows and must be allowed");

    // Freeing a key makes room again.
    writer.delete(&cx, "c", "k0").await.expect("delete");
    writer
        .set(&cx, "c", "k3", serde_json::json!(3))
        .await
        .expect("room after a delete");

    assert_eq!(kv_count(&pool, f.app).await, 3);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_uploads_cannot_push_an_app_past_its_disk_ceiling() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let f = setup(&pool, "unwatched").await;
    let root = std::env::temp_dir().join(format!("picloud-app-{}", Uuid::new_v4().simple()));
    // 450-byte ceiling; ten 100-byte uploads race for it.
    let writer = Arc::new(PostgresFilesWriter::new(pool.clone(), root.clone(), 450));

    let mut set = tokio::task::JoinSet::new();
    for i in 0..10 {
        let w = Arc::clone(&writer);
        let app = f.app;
        set.spawn(async move {
            w.create(
                &cx(app),
                "assets",
                NewFile {
                    name: format!("f{i}.bin"),
                    content_type: "application/octet-stream".into(),
                    data: vec![b'x'; 100],
                },
            )
            .await
        });
    }
    let mut created = Vec::new();
    while let Some(r) = set.join_next().await {
        match r.expect("task") {
            Ok(id) => created.push(id),
            Err(FilesError::QuotaExceeded { .. }) => {}
            Err(e) => panic!("unexpected error: {e}"),
        }
    }

    let bytes = |pool: PgPool, app: Uuid| async move {
        let (n,): (i64,) = sqlx::query_as(
            "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT FROM files WHERE app_id = $1",
        )
        .bind(app)
        .fetch_one(&pool)
        .await
        .expect("bytes");
        n
    };
    let stored = bytes(pool.clone(), f.app).await;
    assert!(
        stored <= 450,
        "stored bytes ({stored}) must not exceed the ceiling — concurrent uploads \
         must not each see the same pre-write total"
    );
    assert!(!created.is_empty(), "some uploads should have succeeded");

    // Growing a file past the ceiling must be refused too — an update that
    // skipped the check would be a free bypass.
    let err = writer
        .update(
            &cx(f.app),
            "assets",
            created[0],
            FileUpdate {
                name: None,
                content_type: None,
                data: vec![b'y'; 10_000],
            },
        )
        .await
        .expect_err("growing a file past the app ceiling must be refused");
    assert!(matches!(err, FilesError::QuotaExceeded { .. }), "got {err}");
    assert_eq!(
        bytes(pool.clone(), f.app).await,
        stored,
        "the refused update must not have changed the stored total"
    );
    // …and, crucially, the file must SURVIVE. `write_atomic_at` renames over the
    // final path, so a quota check that ran AFTER the blob write would have
    // destroyed the original bytes — the user would permanently lose a file
    // merely by exceeding a quota. Read it back through the repo, which verifies
    // the stored checksum, so both "bytes gone" and "bytes replaced but metadata
    // still has the old checksum" fail here.
    let repo = FsFilesRepo::new(
        pool.clone(),
        FilesConfig {
            root: root.clone(),
            max_file_size_bytes: 1024 * 1024,
        },
    );
    let survivor = repo
        .get(AppId::from(f.app), "assets", created[0])
        .await
        .expect("the refused update must not corrupt or destroy the existing file")
        .expect("the file must still exist");
    assert_eq!(
        survivor,
        vec![b'x'; 100],
        "the original bytes must be intact — a refused update must not touch the blob"
    );

    let _ = std::fs::remove_dir_all(&root);
}