test(quota): pin docs ceiling, group docs byte race, and files disk-ordering rollback

Extends the atomic_write suite with three write-path gaps:
- an app cannot exceed PICLOUD_APP_DOCS_MAX_ROWS (create-only, update
  net-zero, delete frees a slot) via a real PostgresDocsWriter;
- concurrent writers cannot push a group past its docs total-bytes quota
  (16 racers against a 450-byte ceiling under pg_advisory_xact_lock);
- a files create with a broken outbox leaves no row AND no blob on disk,
  while a files delete with a broken outbox keeps both the row and the
  readable blob (the disk-ordering invariant).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/tests/atomic_write.rs

@@ -22,15 +22,15 @@
 use std::sync::Arc;
 
 use picloud_manager_core::atomic_write::{
-    FilesWriter, GroupDocsTarget, GroupDocsWriter, GroupFilesQuota, GroupFilesWriter,
-    GroupKvTarget, GroupKvWriter, KvWriter, PostgresFilesWriter, PostgresGroupDocsWriter,
-    PostgresGroupFilesWriter, PostgresGroupKvWriter, PostgresKvWriter,
+    DocsWriter, FilesWriter, GroupDocsTarget, GroupDocsWriter, GroupFilesQuota, GroupFilesWriter,
+    GroupKvTarget, GroupKvWriter, KvWriter, PostgresDocsWriter, 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,
+    AppId, DocsError, ExecutionId, FileUpdate, FilesError, GroupDocsError, GroupFilesError,
+    GroupId, GroupKvError, KvError, NewFile, RequestId, ScriptId, SdkCallCx,
 };
 use sqlx::PgPool;
 use uuid::Uuid;
@@ -159,6 +159,59 @@ async fn kv_count(pool: &PgPool, app: Uuid) -> i64 {
     n
 }
 
+async fn files_count(pool: &PgPool, app: Uuid) -> i64 {
+    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM files WHERE app_id = $1")
+        .bind(app)
+        .fetch_one(pool)
+        .await
+        .expect("count files");
+    n
+}
+
+/// Like `setup`, but the watching trigger is `kind='files'` so a files write into
+/// `collection` produces an outbox fan-out (which `break_outbox_for` can fail).
+async fn setup_files(pool: &PgPool, collection: &str) -> Fixture {
+    let f = setup(pool, "unwatched").await; // app/group/admin, KV trigger on a name we won't use
+    let uniq = Uuid::new_v4().simple().to_string();
+    let (script,): (Uuid,) = sqlx::query_as(
+        "INSERT INTO scripts (app_id, name, source) VALUES ($1, $2, '') RETURNING id",
+    )
+    .bind(f.app)
+    .bind(format!("fh-{uniq}"))
+    .fetch_one(pool)
+    .await
+    .expect("script");
+    let (admin,): (Uuid,) = sqlx::query_as(
+        "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id",
+    )
+    .bind(format!("fa-{uniq}"))
+    .fetch_one(pool)
+    .await
+    .expect("admin");
+    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, 'files', 3, 'exponential', 1000, $3) RETURNING id",
+    )
+    .bind(f.app)
+    .bind(script)
+    .bind(admin)
+    .fetch_one(pool)
+    .await
+    .expect("files trigger");
+    sqlx::query(
+        "INSERT INTO files_trigger_details (trigger_id, collection_glob, ops) \
+         VALUES ($1, $2, '{}')",
+    )
+    .bind(trigger)
+    .bind(collection)
+    .execute(pool)
+    .await
+    .expect("files_trigger_details");
+    f
+}
+
 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)
@@ -484,6 +537,72 @@ async fn concurrent_writers_cannot_push_a_group_past_its_docs_row_quota() {
         .expect("cleanup");
 }
 
+/// Group docs BYTE quota under concurrency — the one group quota that was only
+/// covered single-threaded. The lock is keyed per-(group, KIND), so docs-bytes
+/// could plausibly diverge from docs-rows and KV-bytes; race it directly. Mirrors
+/// the group-KV byte race: ~100-byte docs, a ceiling that fits about 4, and a
+/// `max_value_bytes` low enough that the upper-bound fast path can't short-circuit
+/// the projected-total scan.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn concurrent_writers_cannot_push_a_group_past_its_docs_byte_quota() {
+    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_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);
+        set.spawn(async move {
+            let cx = cx(app);
+            w.create(
+                &cx,
+                GroupDocsTarget {
+                    group_id: GroupId::from(group),
+                    collection: "articles",
+                },
+                serde_json::json!({ "blob": "x".repeat(90), "n": i }),
+                quota,
+            )
+            .await
+        });
+    }
+    while let Some(r) = set.join_next().await {
+        match r.expect("task") {
+            Ok(_) | Err(GroupDocsError::TotalBytesQuotaExceeded { .. }) => {}
+            Err(e) => panic!("unexpected error: {e}"),
+        }
+    }
+
+    let (bytes,): (i64,) = sqlx::query_as(
+        "SELECT COALESCE(SUM(octet_length(data::text)), 0)::BIGINT \
+         FROM group_docs WHERE group_id = $1",
+    )
+    .bind(group)
+    .fetch_one(&pool)
+    .await
+    .expect("bytes");
+    assert!(
+        bytes <= 450,
+        "stored docs bytes ({bytes}) must not exceed the 450-byte ceiling — \
+         concurrent writers must serialize on the docs-bytes lock"
+    );
+    assert!(bytes > 0, "some writers should have succeeded");
+
+    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
@@ -636,6 +755,56 @@ async fn an_app_cannot_exceed_its_key_ceiling_but_updates_stay_free() {
     assert_eq!(kv_count(&pool, f.app).await, 3);
 }
 
+/// The per-app DOCS row ceiling (`PICLOUD_APP_DOCS_MAX_ROWS`). The KV ceiling was
+/// pinned; docs — create-only, and written with a different check (`>=` BEFORE the
+/// insert, vs KV's `>` after) — was not, so an off-by-one or a dropped check would
+/// have been invisible.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn an_app_cannot_exceed_its_docs_ceiling() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let f = setup(&pool, "unwatched").await;
+    let writer = PostgresDocsWriter::new(pool.clone(), 3);
+    let cx = cx(f.app);
+
+    let mut ids = Vec::new();
+    for i in 0..3 {
+        ids.push(
+            writer
+                .create(&cx, "c", serde_json::json!({ "n": i }))
+                .await
+                .expect("under the ceiling"),
+        );
+    }
+    let err = writer
+        .create(&cx, "c", serde_json::json!({ "n": 3 }))
+        .await
+        .expect_err("a fourth doc must be refused");
+    assert!(matches!(err, DocsError::QuotaExceeded { .. }), "got {err}");
+
+    // An UPDATE replaces in place — net-zero rows — so it stays allowed at the
+    // ceiling.
+    writer
+        .update(&cx, "c", ids[0], serde_json::json!({ "n": "updated" }))
+        .await
+        .expect("an update at the ceiling is net-zero rows and must be allowed");
+
+    // Deleting frees a slot.
+    writer.delete(&cx, "c", ids[0]).await.expect("delete");
+    writer
+        .create(&cx, "c", serde_json::json!({ "n": 3 }))
+        .await
+        .expect("room after a delete");
+
+    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM docs WHERE app_id = $1")
+        .bind(f.app)
+        .fetch_one(&pool)
+        .await
+        .expect("count");
+    assert_eq!(n, 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 {
@@ -737,3 +906,106 @@ async fn concurrent_uploads_cannot_push_an_app_past_its_disk_ceiling() {
 
     let _ = std::fs::remove_dir_all(&root);
 }
+
+/// The files disk/transaction ORDERING invariant (CLAUDE.md write-path rules):
+/// create/update write the blob first and unlink it on rollback; delete commits
+/// the metadata removal first and unlinks after — "the reverse would destroy the
+/// bytes of a row a rollback keeps." The disk mechanics were tested but never with
+/// a transaction actually rolling back. This drives both directions with the
+/// outbox fault injection.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn files_rollback_respects_the_disk_ordering() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let f = setup_files(&pool, "assets").await;
+    let root = std::env::temp_dir().join(format!("picloud-fr-{}", Uuid::new_v4().simple()));
+    let writer = PostgresFilesWriter::new(pool.clone(), root.clone(), u64::MAX);
+    let repo = FsFilesRepo::new(
+        pool.clone(),
+        FilesConfig {
+            root: root.clone(),
+            max_file_size_bytes: 1024 * 1024,
+        },
+    );
+    let mk = |n: u8| NewFile {
+        name: format!("f{n}.bin"),
+        content_type: "application/octet-stream".into(),
+        data: vec![b'x'; 100],
+    };
+
+    // --- CREATE rollback: a fan-out failure must leave NO orphan blob. --------
+    let tag = Uuid::new_v4().simple().to_string();
+    break_outbox_for(&pool, f.app, &tag).await;
+    let err = writer
+        .create(&cx(f.app), "assets", mk(0))
+        .await
+        .expect_err("a create whose fan-out fails must error");
+    unbreak_outbox(&pool, &tag).await;
+    assert!(format!("{err}").contains("event emit"), "got {err}");
+    assert_eq!(
+        files_count(&pool, f.app).await,
+        0,
+        "the metadata row must be rolled back"
+    );
+    // No blob may remain on disk — a create writes the bytes inside the tx and
+    // must unlink them when it rolls back, or they leak forever (the sweeper only
+    // reaps *.tmp.*, not finalized blobs).
+    let disk = count_files_on_disk(&root);
+    assert_eq!(
+        disk, 0,
+        "a rolled-back create must leave no orphan blob, found {disk}"
+    );
+
+    // --- DELETE rollback: a fan-out failure must KEEP the blob readable. -------
+    let id = writer
+        .create(&cx(f.app), "assets", mk(1))
+        .await
+        .expect("healthy create lands");
+    assert_eq!(files_count(&pool, f.app).await, 1);
+
+    let tag2 = Uuid::new_v4().simple().to_string();
+    break_outbox_for(&pool, f.app, &tag2).await;
+    let err = writer
+        .delete(&cx(f.app), "assets", id)
+        .await
+        .expect_err("a delete whose fan-out fails must error");
+    unbreak_outbox(&pool, &tag2).await;
+    assert!(format!("{err}").contains("event emit"), "got {err}");
+    assert_eq!(
+        files_count(&pool, f.app).await,
+        1,
+        "the row must survive the rolled-back delete"
+    );
+    // The blob must STILL be readable (checksum-verified). Inverting the delete
+    // order — unlink before commit — would leave this row pointing at bytes that
+    // no longer exist, i.e. silent data loss surfacing later as Corrupted.
+    let survivor = repo
+        .get(AppId::from(f.app), "assets", id)
+        .await
+        .expect("a rolled-back delete must not destroy the blob")
+        .expect("the file must still exist");
+    assert_eq!(survivor, vec![b'x'; 100]);
+
+    let _ = std::fs::remove_dir_all(&root);
+}
+
+/// Count finalized blob files (not `*.tmp.*`) anywhere under a files root.
+fn count_files_on_disk(root: &std::path::Path) -> usize {
+    fn walk(dir: &std::path::Path, acc: &mut usize) {
+        let Ok(entries) = std::fs::read_dir(dir) else {
+            return;
+        };
+        for e in entries.flatten() {
+            let p = e.path();
+            if p.is_dir() {
+                walk(&p, acc);
+            } else if !p.to_string_lossy().contains(".tmp.") {
+                *acc += 1;
+            }
+        }
+    }
+    let mut n = 0;
+    walk(root, &mut n);
+    n
+}