977abc25d0
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>
1012 lines
35 KiB
Rust
1012 lines
35 KiB
Rust
//! 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::{
|
|
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, DocsError, 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 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)
|
|
.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 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
|
|
/// 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);
|
|
}
|
|
|
|
/// 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 {
|
|
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);
|
|
}
|
|
|
|
/// 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
|
|
}
|