diff --git a/crates/manager-core/tests/atomic_write.rs b/crates/manager-core/tests/atomic_write.rs index f51301c..c360baf 100644 --- a/crates/manager-core/tests/atomic_write.rs +++ b/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 +}