From 966b209d2e3f68c184e88283ed4baf50b94abc59 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 14 Jul 2026 20:32:32 +0200 Subject: [PATCH] feat(quota): add per-app storage ceilings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An app's OWN store had no ceiling of any kind — only the group-SHARED collections did, which is the inversion of what you'd expect. And `authz::script_gate` returns `Ok(())` when there is no principal, so a PUBLIC UNAUTHENTICATED route could reach `files::collection(..).create(..)` and write blobs in a loop until the disk filled. `PICLOUD_FILES_MAX_FILE_SIZE_BYTES` caps ONE file at 100 MB; nothing capped the count. Same for KV keys and docs against Postgres. Ceilings: `PICLOUD_APP_KV_MAX_ROWS` / `PICLOUD_APP_DOCS_MAX_ROWS` (100k) and `PICLOUD_APP_FILES_MAX_TOTAL_BYTES` (10 GiB). They are enforced DIFFERENTLY from the group ones, on purpose — porting the group design as-is would have been a serious regression: * KV/docs check a ROW count, on INSERT only, with NO lock. `kv::set` is the hottest write path in the system; taking a per-app advisory lock on every set would serialize an app's entire data plane, and a `SUM(...)` scan would make write cost grow with the app's stored size. What the lock buys is small — unlocked overshoot is bounded by write concurrency (32) against a ceiling of 100k, ~0.03%. These are anti-DoS rails, not billing. An update adds no row and is bounded by the per-value cap, so it pays nothing at all. * No per-app BYTE ceiling for KV/docs: every value is already capped at `PICLOUD_KV_MAX_VALUE_BYTES`, so `max_rows x max_value_bytes` is ALREADY a finite bound. A second scan would buy nothing. * FILES are the exception and DO lock + sum: one blob may be 100 MB, so 32 racing uploads could overshoot by gigabytes of real disk, and a file write is heavy enough that the lock and the SUM are lost in the noise. Checked on create AND update — an update that skipped the ceiling would be a free bypass (the same bug just fixed for group files: grow a 1-byte file to 100 MB, repeat). `group_quota` is renamed `quota`: it now serves both owners, and the module doc is where the two enforcement strategies are contrasted. Pinned by tests/atomic_write.rs: the key ceiling refuses a new key while still allowing an update at the ceiling (rejecting updates would brick an app the moment it filled up — worse than the DoS the rail exists to stop), and 10 concurrent uploads cannot exceed the disk ceiling, nor can an update grow past it. Co-Authored-By: Claude Opus 4.8 (1M context) --- CLAUDE.md | 3 + crates/manager-core/src/atomic_write.rs | 131 ++++++++++++++-- crates/manager-core/src/docs_repo.rs | 13 ++ crates/manager-core/src/docs_service.rs | 4 +- crates/manager-core/src/files_repo.rs | 28 ++++ crates/manager-core/src/files_service.rs | 9 +- crates/manager-core/src/group_docs_service.rs | 6 +- .../manager-core/src/group_files_service.rs | 2 +- crates/manager-core/src/group_kv_service.rs | 6 +- crates/manager-core/src/kv_repo.rs | 34 ++++ crates/manager-core/src/kv_service.rs | 4 +- crates/manager-core/src/lib.rs | 2 +- .../src/{group_quota.rs => quota.rs} | 56 +++++++ crates/manager-core/tests/atomic_write.rs | 146 +++++++++++++++++- crates/picloud/src/lib.rs | 17 +- crates/shared/src/docs.rs | 5 + crates/shared/src/files.rs | 6 + crates/shared/src/group_files.rs | 1 + crates/shared/src/kv.rs | 5 + 19 files changed, 442 insertions(+), 36 deletions(-) rename crates/manager-core/src/{group_quota.rs => quota.rs} (70%) diff --git a/CLAUDE.md b/CLAUDE.md index 6abdef3..c46befd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -143,6 +143,9 @@ Environment variables consumed by the `picloud` binary: | `PICLOUD_DOCS_MAX_VALUE_BYTES` | `262144` (256 KB) | Per-document JSON-encoded data cap for `docs::create`/`update`. | | `PICLOUD_PUBSUB_MAX_MESSAGE_BYTES` | `262144` (256 KB) | Per-message JSON-encoded payload cap for `pubsub::publish_durable`. Prevents one publish from amplifying into N outbox rows × M MB. | | `PICLOUD_QUEUE_MAX_PAYLOAD_BYTES` | `262144` (256 KB) | Per-message JSON-encoded payload cap for `queue::enqueue`. | +| `PICLOUD_APP_KV_MAX_ROWS` | `100000` | Per-app ceiling on total KV keys. Checked **only when a write ADDS a key** (`kv::set` of a new key, or a `set_if` insert) — an update is net-zero rows and pays nothing. Deliberately **not** advisory-locked: `kv::set` is the hottest write path, and serializing an app's data plane (or scanning `SUM(...)` on every set) is a far worse trade than the ~32-row overshoot (bounded by `PICLOUD_MAX_CONCURRENT_EXECUTIONS`) that the lock would prevent. These are anti-DoS rails, not billing. Together with `PICLOUD_KV_MAX_VALUE_BYTES` this **also bounds stored bytes** (rows x value cap), so there is no separate per-app KV byte ceiling. | +| `PICLOUD_APP_DOCS_MAX_ROWS` | `100000` | Per-app ceiling on total docs (`docs::create`). Same rationale as KV — create-only, unlocked, and bounded in bytes by `PICLOUD_DOCS_MAX_VALUE_BYTES`. | +| `PICLOUD_APP_FILES_MAX_TOTAL_BYTES` | `10737418240` (10 GiB) | Per-app ceiling on total stored blob bytes. **This one IS advisory-locked** and does sum bytes, unlike KV/docs: one blob may be 100 MB, so racing uploads could overshoot by gigabytes of real disk, and a file write is heavy enough that the lock + `SUM` are lost in the noise. Checked on the projected total (the replaced blob subtracted) on **create and update** — an update that skipped it would be a free bypass. Closes the anonymous disk-exhaustion path: `script_gate` passes for an unauthenticated principal, so a public route could previously write blobs unbounded. | | `PICLOUD_GROUP_KV_MAX_ROWS` | `100000` | §11.6 M3 per-group quota: max total keys across a group's shared-KV collections. Checked only on a NEW key (`kv::shared_collection().set`); an update is exempt. | | `PICLOUD_GROUP_DOCS_MAX_ROWS` | `100000` | §11.6 M3 per-group quota: max total docs across a group's shared-docs collections (`docs::shared_collection().create`). | | `PICLOUD_GROUP_KV_MAX_TOTAL_BYTES` | `268435456` (256 MiB) | §11.6 M4 per-group quota: max total stored JSON bytes across a group's shared-KV collections. Checked on the projected total (old value subtracted, new added), so a same/smaller update near the cap is still allowed. | diff --git a/crates/manager-core/src/atomic_write.rs b/crates/manager-core/src/atomic_write.rs index cc978af..087a45f 100644 --- a/crates/manager-core/src/atomic_write.rs +++ b/crates/manager-core/src/atomic_write.rs @@ -54,9 +54,9 @@ use crate::files_repo::{self, FileUpdated, FilesRepo, MetaFields, MetaPatch}; use crate::group_docs_repo::{self, GroupDocsRepo}; use crate::group_files_repo::{self, GroupFilesRepo, GroupFilesRepoError}; use crate::group_kv_repo::{self, GroupKvRepo}; -use crate::group_quota::{check_group_write, GroupUsage, GroupWriteQuota, QuotaOutcome}; use crate::kv_repo::{self, KvRepo}; use crate::outbox_event_emitter::{emit_on, emit_shared_on}; +use crate::quota::{check_group_write, GroupUsage, GroupWriteQuota, QuotaOutcome}; /// Map a `sqlx` failure from the transaction plumbing itself (begin/commit). fn tx_err(e: &sqlx::Error) -> KvError { @@ -123,12 +123,33 @@ pub trait KvWriter: Send + Sync { pub struct PostgresKvWriter { pool: PgPool, + /// Per-app key ceiling. Checked ONLY when a write adds a key — an update is + /// net-zero rows and already bounded by the per-value cap, so it pays + /// nothing. No lock: see the `quota` module for why. + max_rows: u64, } impl PostgresKvWriter { #[must_use] - pub fn new(pool: PgPool) -> Self { - Self { pool } + pub fn new(pool: PgPool, max_rows: u64) -> Self { + Self { pool, max_rows } + } + + /// Refuse a key-adding write once the app is at its ceiling. Runs on the + /// writing transaction's connection. + async fn check_rows( + &self, + conn: &mut PgConnection, + app_id: picloud_shared::AppId, + ) -> Result<(), KvError> { + let count = kv_repo::count_rows_on(&mut *conn, app_id).await?; + if count >= self.max_rows { + return Err(KvError::QuotaExceeded { + limit: self.max_rows, + actual: count, + }); + } + Ok(()) } } @@ -142,6 +163,13 @@ impl KvWriter for PostgresKvWriter { value: Value, ) -> Result, KvError> { let mut tx = self.pool.begin().await.map_err(|e| tx_err(&e))?; + // A NEW key consumes a row against the app's ceiling; an update does not. + if kv_repo::get_on(&mut *tx, cx.app_id, collection, key) + .await? + .is_none() + { + self.check_rows(&mut tx, cx.app_id).await?; + } let previous = kv_repo::set_on(&mut *tx, cx.app_id, collection, key, value.clone()).await?; let op = if previous.is_some() { "update" @@ -172,6 +200,16 @@ impl KvWriter for PostgresKvWriter { "insert" }; let mut tx = self.pool.begin().await.map_err(|e| tx_err(&e))?; + // A swap only ADDS a key when the precondition was "absent" AND the key + // really is absent — with a `Some` precondition against an absent key the + // swap cannot happen at all, so it consumes nothing. + if expected.is_none() + && kv_repo::get_on(&mut *tx, cx.app_id, collection, key) + .await? + .is_none() + { + self.check_rows(&mut tx, cx.app_id).await?; + } let swapped = kv_repo::set_if_on(&mut *tx, cx.app_id, collection, key, expected, new.clone()).await?; if swapped { @@ -381,13 +419,13 @@ fn quota_gate(outcome: QuotaOutcome) -> Result<(), GroupKvError> { /// The key is namespaced per KIND (`kv` / `docs` / …): a KV write and a docs /// write draw on separate ceilings, so serializing them against each other would /// be pure contention for no safety. -fn group_quota_lock_key(group_id: GroupId, kind: &str) -> i64 { +fn quota_lock_key(owner: uuid::Uuid, scope: &str, kind: &str) -> i64 { use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; let mut h = DefaultHasher::new(); - "group_quota".hash(&mut h); + scope.hash(&mut h); kind.hash(&mut h); - group_id.into_inner().hash(&mut h); + owner.hash(&mut h); // Postgres advisory locks key on bigint; reinterpret the u64 as i64. #[allow(clippy::cast_possible_wrap)] let key = h.finish() as i64; @@ -400,7 +438,21 @@ async fn lock_group_quota( kind: &str, ) -> Result<(), sqlx::Error> { sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(group_quota_lock_key(group_id, kind)) + .bind(quota_lock_key(group_id.into_inner(), "group_quota", kind)) + .execute(conn) + .await?; + Ok(()) +} + +/// Per-app FILES only — see the `quota` module for why the per-app KV/docs +/// ceilings deliberately do NOT lock. +async fn lock_app_quota( + conn: &mut PgConnection, + app_id: picloud_shared::AppId, + kind: &str, +) -> Result<(), sqlx::Error> { + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(quota_lock_key(app_id.into_inner(), "app_quota", kind)) .execute(conn) .await?; Ok(()) @@ -778,12 +830,15 @@ pub trait DocsWriter: Send + Sync { pub struct PostgresDocsWriter { pool: PgPool, + /// Per-app doc ceiling. Checked on `create` only — an update is net-zero + /// rows and bounded by the per-doc value cap. No lock: see `quota`. + max_rows: u64, } impl PostgresDocsWriter { #[must_use] - pub fn new(pool: PgPool) -> Self { - Self { pool } + pub fn new(pool: PgPool, max_rows: u64) -> Self { + Self { pool, max_rows } } } @@ -796,6 +851,13 @@ impl DocsWriter for PostgresDocsWriter { data: Value, ) -> Result { let mut tx = self.pool.begin().await.map_err(|e| docs_tx_err(&e))?; + let count = docs_repo::count_rows_on(&mut *tx, cx.app_id).await?; + if count >= self.max_rows { + return Err(DocsError::QuotaExceeded { + limit: self.max_rows, + actual: count, + }); + } let row = docs_repo::create_on(&mut *tx, cx.app_id, collection, data.clone()).await?; emit_on( &mut tx, @@ -1335,12 +1397,42 @@ pub trait FilesWriter: Send + Sync { pub struct PostgresFilesWriter { pool: PgPool, root: PathBuf, + /// Per-app stored-blob ceiling. Unlike KV/docs this one DOES take a lock and + /// DOES sum bytes: a blob may be 100 MB, so racing uploads could overshoot by + /// gigabytes of real disk, and a file write is heavy enough that the lock and + /// the SUM are lost in the noise. + max_total_bytes: u64, } impl PostgresFilesWriter { #[must_use] - pub fn new(pool: PgPool, root: PathBuf) -> Self { - Self { pool, root } + pub fn new(pool: PgPool, root: PathBuf, max_total_bytes: u64) -> Self { + Self { + pool, + root, + max_total_bytes, + } + } + + /// The projected total (the replaced blob's bytes subtracted) must fit under + /// the ceiling. Runs on the writing transaction's connection, under the + /// per-app advisory lock the caller has taken. + async fn check_bytes( + &self, + conn: &mut PgConnection, + app_id: picloud_shared::AppId, + replacing: Option, + incoming: i64, + ) -> Result<(), FilesError> { + let projected = + files_repo::projected_total_bytes_on(&mut *conn, app_id, replacing, incoming).await?; + if projected > self.max_total_bytes { + return Err(FilesError::QuotaExceeded { + limit: self.max_total_bytes, + actual: projected, + }); + } + Ok(()) } } @@ -1360,6 +1452,10 @@ impl FilesWriter for PostgresFilesWriter { let committed = async { let mut tx = self.pool.begin().await.map_err(|e| files_tx_err(&e))?; + lock_app_quota(&mut tx, cx.app_id, "files") + .await + .map_err(|e| files_tx_err(&e))?; + self.check_bytes(&mut tx, cx.app_id, None, size).await?; let meta = files_repo::insert_meta_on( &mut *tx, cx.app_id, @@ -1409,6 +1505,19 @@ impl FilesWriter for PostgresFilesWriter { let checksum = files_repo::write_atomic_at(&self.root, &owner, collection, id, &upd.data)?; let mut tx = self.pool.begin().await.map_err(|e| files_tx_err(&e))?; + lock_app_quota(&mut tx, cx.app_id, "files") + .await + .map_err(|e| files_tx_err(&e))?; + // Growing an existing blob consumes disk too — an update that skipped the + // ceiling would be a free bypass (exactly the group-files bug). The + // projection subtracts the blob being replaced, so a same-size-or-smaller + // update near the cap still goes through. + if let Err(e) = self.check_bytes(&mut tx, cx.app_id, Some(id), size).await { + // The new bytes are already on disk but no row will reference them. + drop(tx); + files_repo::unlink_blob(&self.root, &owner, collection, id); + return Err(e); + } let updated = files_repo::update_meta_on( &mut *tx, cx.app_id, diff --git a/crates/manager-core/src/docs_repo.rs b/crates/manager-core/src/docs_repo.rs index 80efb5b..39bbad6 100644 --- a/crates/manager-core/src/docs_repo.rs +++ b/crates/manager-core/src/docs_repo.rs @@ -488,6 +488,19 @@ where Ok(row.map(|(v,)| v)) } +/// Total doc count for the app, across all its collections. Backs the per-app +/// row ceiling; run only on `create`. +pub(crate) async fn count_rows_on<'c, E>(exec: E, app_id: AppId) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM docs WHERE app_id = $1") + .bind(app_id.into_inner()) + .fetch_one(exec) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) +} + #[cfg(test)] mod sql_shape_tests { use super::*; diff --git a/crates/manager-core/src/docs_service.rs b/crates/manager-core/src/docs_service.rs index 7e30223..4ab4211 100644 --- a/crates/manager-core/src/docs_service.rs +++ b/crates/manager-core/src/docs_service.rs @@ -93,8 +93,8 @@ impl DocsServiceImpl { /// write back instead of silently losing the event. The host always calls /// this; the in-memory unit tests do not. #[must_use] - pub fn with_atomic_writes(mut self, pool: sqlx::PgPool) -> Self { - self.writer = Arc::new(PostgresDocsWriter::new(pool)); + pub fn with_atomic_writes(mut self, pool: sqlx::PgPool, max_rows: u64) -> Self { + self.writer = Arc::new(PostgresDocsWriter::new(pool, max_rows)); self } diff --git a/crates/manager-core/src/files_repo.rs b/crates/manager-core/src/files_repo.rs index e04236a..421b7b0 100644 --- a/crates/manager-core/src/files_repo.rs +++ b/crates/manager-core/src/files_repo.rs @@ -785,6 +785,34 @@ pub(crate) fn unlink_blob(root: &Path, owner_rel: &Path, collection: &str, id: U } } +/// The PROJECTED total stored bytes for the app AFTER this write — the current +/// SUM, minus the bytes of the file being replaced (`replacing = Some(id)` on +/// update, `None` on create), plus the incoming blob's. Backs the per-app disk +/// ceiling; the subtraction is what lets an update near the cap still go through. +pub(crate) async fn projected_total_bytes_on<'c, E>( + exec: E, + app_id: AppId, + replacing: Option, + incoming: i64, +) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let (n,): (i64,) = sqlx::query_as( + "SELECT ( \ + COALESCE((SELECT SUM(size_bytes) FROM files WHERE app_id = $1), 0) \ + - COALESCE((SELECT size_bytes FROM files WHERE app_id = $1 AND id = $2), 0) \ + + $3 \ + )::BIGINT", + ) + .bind(app_id.into_inner()) + .bind(replacing) + .bind(incoming) + .fetch_one(exec) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/manager-core/src/files_service.rs b/crates/manager-core/src/files_service.rs index deab64c..ec07381 100644 --- a/crates/manager-core/src/files_service.rs +++ b/crates/manager-core/src/files_service.rs @@ -54,8 +54,13 @@ impl FilesServiceImpl { /// the metadata back (and unlinks the blob) instead of silently losing the /// event. The host always calls this; the in-memory unit tests do not. #[must_use] - pub fn with_atomic_writes(mut self, pool: sqlx::PgPool, root: std::path::PathBuf) -> Self { - self.writer = Arc::new(PostgresFilesWriter::new(pool, root)); + pub fn with_atomic_writes( + mut self, + pool: sqlx::PgPool, + root: std::path::PathBuf, + max_total_bytes: u64, + ) -> Self { + self.writer = Arc::new(PostgresFilesWriter::new(pool, root, max_total_bytes)); self } diff --git a/crates/manager-core/src/group_docs_service.rs b/crates/manager-core/src/group_docs_service.rs index 2b3c557..2ab4d6e 100644 --- a/crates/manager-core/src/group_docs_service.rs +++ b/crates/manager-core/src/group_docs_service.rs @@ -23,7 +23,7 @@ use crate::docs_filter::{parse_filter, FilterParseError}; use crate::docs_service::docs_max_value_bytes_from_env; use crate::group_collection_repo::GroupCollectionResolver; use crate::group_docs_repo::{GroupDocsRepo, GroupDocsRepoError}; -use crate::group_quota::GroupWriteQuota; +use crate::quota::GroupWriteQuota; /// The registry `kind` this service resolves. const KIND_DOCS: &str = "docs"; @@ -86,8 +86,8 @@ impl GroupDocsServiceImpl { repo.clone(), Arc::new(NoopEventEmitter), )), - max_rows: crate::group_quota::group_docs_max_rows_from_env(), - max_total_bytes: crate::group_quota::group_docs_max_total_bytes_from_env(), + max_rows: crate::quota::group_docs_max_rows_from_env(), + max_total_bytes: crate::quota::group_docs_max_total_bytes_from_env(), repo, resolver, authz, diff --git a/crates/manager-core/src/group_files_service.rs b/crates/manager-core/src/group_files_service.rs index 92e1957..34784ac 100644 --- a/crates/manager-core/src/group_files_service.rs +++ b/crates/manager-core/src/group_files_service.rs @@ -51,7 +51,7 @@ impl GroupFilesServiceImpl { max_file_size_bytes: usize, ) -> Self { Self { - max_total_bytes: crate::group_quota::group_files_max_total_bytes_from_env(), + max_total_bytes: crate::quota::group_files_max_total_bytes_from_env(), writer: Arc::new(BestEffortGroupFilesWriter::new( repo.clone(), Arc::new(NoopEventEmitter), diff --git a/crates/manager-core/src/group_kv_service.rs b/crates/manager-core/src/group_kv_service.rs index 46290ae..ecac694 100644 --- a/crates/manager-core/src/group_kv_service.rs +++ b/crates/manager-core/src/group_kv_service.rs @@ -32,8 +32,8 @@ use crate::atomic_write::{ use crate::authz::{self, AuthzRepo, Capability}; use crate::group_collection_repo::GroupCollectionResolver; use crate::group_kv_repo::{GroupKvRepo, GroupKvRepoError}; -use crate::group_quota::GroupWriteQuota; use crate::kv_service::kv_max_value_bytes_from_env; +use crate::quota::GroupWriteQuota; /// The registry `kind` this service resolves. const KIND_KV: &str = "kv"; @@ -111,8 +111,8 @@ impl GroupKvServiceImpl { repo.clone(), Arc::new(NoopEventEmitter), )), - max_rows: crate::group_quota::group_kv_max_rows_from_env(), - max_total_bytes: crate::group_quota::group_kv_max_total_bytes_from_env(), + max_rows: crate::quota::group_kv_max_rows_from_env(), + max_total_bytes: crate::quota::group_kv_max_total_bytes_from_env(), repo, resolver, authz, diff --git a/crates/manager-core/src/kv_repo.rs b/crates/manager-core/src/kv_repo.rs index a4f0f64..ca16acd 100644 --- a/crates/manager-core/src/kv_repo.rs +++ b/crates/manager-core/src/kv_repo.rs @@ -99,6 +99,27 @@ const KV_LIST_DEFAULT_LIMIT: u32 = 100; // trigger fan-out on ONE connection inside ONE transaction. // ---------------------------------------------------------------------------- +pub(crate) async fn get_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + key: &str, +) -> Result, KvRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option<(serde_json::Value,)> = sqlx::query_as( + "SELECT value FROM kv_entries \ + WHERE app_id = $1 AND collection = $2 AND key = $3", + ) + .bind(app_id.into_inner()) + .bind(collection) + .bind(key) + .fetch_optional(exec) + .await?; + Ok(row.map(|(v,)| v)) +} + /// Upsert. Returns the previous value, so the caller knows whether this was an /// `insert` or an `update` for the emitted `ServiceEvent`. pub(crate) async fn set_on<'c, E>( @@ -324,3 +345,16 @@ fn decode_cursor(cursor: &str) -> Result { .map_err(|_| KvRepoError::InvalidCursor)?; String::from_utf8(bytes).map_err(|_| KvRepoError::InvalidCursor) } + +/// Total key count for the app, across all its collections. Backs the per-app +/// row ceiling; an index-only count, and only run when a write ADDS a key. +pub(crate) async fn count_rows_on<'c, E>(exec: E, app_id: AppId) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM kv_entries WHERE app_id = $1") + .bind(app_id.into_inner()) + .fetch_one(exec) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) +} diff --git a/crates/manager-core/src/kv_service.rs b/crates/manager-core/src/kv_service.rs index 61caab2..672a291 100644 --- a/crates/manager-core/src/kv_service.rs +++ b/crates/manager-core/src/kv_service.rs @@ -86,8 +86,8 @@ impl KvServiceImpl { /// write back instead of silently losing the event. The host always calls /// this; the in-memory unit tests do not (they have no Postgres). #[must_use] - pub fn with_atomic_writes(mut self, pool: PgPool) -> Self { - self.writer = Arc::new(PostgresKvWriter::new(pool)); + pub fn with_atomic_writes(mut self, pool: PgPool, max_rows: u64) -> Self { + self.writer = Arc::new(PostgresKvWriter::new(pool, max_rows)); self } diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index c870ddf..61aaa75 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -64,7 +64,6 @@ pub mod group_members_repo; pub mod group_pubsub_service; pub mod group_queue_repo; pub mod group_queue_service; -pub mod group_quota; pub mod group_repo; pub mod group_scripts_api; pub mod groups_api; @@ -90,6 +89,7 @@ pub mod pubsub_service; pub mod queue_repo; pub mod queue_service; pub mod queues_api; +pub mod quota; pub mod realtime_authority; pub mod repo; pub mod route_admin; diff --git a/crates/manager-core/src/group_quota.rs b/crates/manager-core/src/quota.rs similarity index 70% rename from crates/manager-core/src/group_quota.rs rename to crates/manager-core/src/quota.rs index 1280e68..d1575ea 100644 --- a/crates/manager-core/src/group_quota.rs +++ b/crates/manager-core/src/quota.rs @@ -1,3 +1,33 @@ +//! Storage ceilings — per-app and §11.6 per-group. +//! +//! # Why the two are enforced DIFFERENTLY +//! +//! The group ceilings serialize (a `pg_advisory_xact_lock` across the check and +//! the write) because they must be exact: a group's shared collections are the +//! cross-app blast radius, and with a small ceiling the check-then-write race +//! overshoots wildly (measured: 19 rows stored against a ceiling of 5). Group +//! writes are comparatively rare, so the lock costs little. +//! +//! The per-app KV/docs ceilings deliberately do **not** lock, and check only a +//! ROW count, on INSERT only: +//! +//! * `kv::set` is the hottest write path in the system. Taking a per-app lock +//! on every set would serialize an app's entire data plane; a `SUM(...)` scan +//! on every set would make write cost grow with the app's stored size. Both +//! are far worse trades than what the lock buys here. +//! * What it buys is small: without the lock, overshoot is bounded by write +//! concurrency (`PICLOUD_MAX_CONCURRENT_EXECUTIONS`, 32 by default) against a +//! ceiling of 100k — roughly 0.03%. These are anti-DoS rails, not billing. +//! * A byte ceiling is unnecessary: every value is already capped at +//! `PICLOUD_KV_MAX_VALUE_BYTES`, so `max_rows × max_value_bytes` is ALREADY a +//! finite bound on an app's stored bytes. Add a second scan for nothing. +//! * An update adds no row and cannot exceed the per-value cap, so it is exempt +//! and pays nothing at all — the COUNT runs only when a key/doc is created. +//! +//! Per-app FILES are the exception and DO lock: a single blob may be 100 MB, so +//! 32 racing uploads could overshoot by gigabytes of real disk, and a file write +//! is heavy enough that a lock and a `SUM(size_bytes)` are lost in the noise. +//! //! §11.6 per-group quotas — global env-var ceilings on a group's shared //! collections, enforced in the group write path (mirrors the per-value //! `PICLOUD_*_MAX_*_BYTES` caps). One default applies to every group; per-group @@ -24,6 +54,14 @@ pub const DEFAULT_GROUP_DOCS_MAX_TOTAL_BYTES: u64 = 256 * 1024 * 1024; /// `PICLOUD_GROUP_FILES_MAX_TOTAL_BYTES`. pub const DEFAULT_GROUP_FILES_MAX_TOTAL_BYTES: u64 = 10 * 1024 * 1024 * 1024; +/// Default per-app KV key ceiling. Override `PICLOUD_APP_KV_MAX_ROWS`. +pub const DEFAULT_APP_KV_MAX_ROWS: u64 = 100_000; +/// Default per-app doc ceiling. Override `PICLOUD_APP_DOCS_MAX_ROWS`. +pub const DEFAULT_APP_DOCS_MAX_ROWS: u64 = 100_000; +/// Default per-app stored-blob ceiling (10 GiB). Override +/// `PICLOUD_APP_FILES_MAX_TOTAL_BYTES`. +pub const DEFAULT_APP_FILES_MAX_TOTAL_BYTES: u64 = 10 * 1024 * 1024 * 1024; + fn u64_from_env(var: &str, default: u64) -> u64 { if let Ok(v) = std::env::var(var) { match v.trim().parse::() { @@ -34,6 +72,24 @@ fn u64_from_env(var: &str, default: u64) -> u64 { default } +#[must_use] +pub fn app_kv_max_rows_from_env() -> u64 { + u64_from_env("PICLOUD_APP_KV_MAX_ROWS", DEFAULT_APP_KV_MAX_ROWS) +} + +#[must_use] +pub fn app_docs_max_rows_from_env() -> u64 { + u64_from_env("PICLOUD_APP_DOCS_MAX_ROWS", DEFAULT_APP_DOCS_MAX_ROWS) +} + +#[must_use] +pub fn app_files_max_total_bytes_from_env() -> u64 { + u64_from_env( + "PICLOUD_APP_FILES_MAX_TOTAL_BYTES", + DEFAULT_APP_FILES_MAX_TOTAL_BYTES, + ) +} + #[must_use] pub fn group_kv_max_rows_from_env() -> u64 { u64_from_env("PICLOUD_GROUP_KV_MAX_ROWS", DEFAULT_GROUP_KV_MAX_ROWS) diff --git a/crates/manager-core/tests/atomic_write.rs b/crates/manager-core/tests/atomic_write.rs index bae6d95..0ff22a0 100644 --- a/crates/manager-core/tests/atomic_write.rs +++ b/crates/manager-core/tests/atomic_write.rs @@ -16,14 +16,14 @@ use std::sync::Arc; use picloud_manager_core::atomic_write::{ - GroupDocsTarget, GroupDocsWriter, GroupFilesQuota, GroupFilesWriter, GroupKvTarget, - GroupKvWriter, KvWriter, PostgresGroupDocsWriter, PostgresGroupFilesWriter, - PostgresGroupKvWriter, PostgresKvWriter, + FilesWriter, GroupDocsTarget, GroupDocsWriter, GroupFilesQuota, GroupFilesWriter, + GroupKvTarget, GroupKvWriter, KvWriter, PostgresFilesWriter, PostgresGroupDocsWriter, + PostgresGroupFilesWriter, PostgresGroupKvWriter, PostgresKvWriter, }; -use picloud_manager_core::group_quota::GroupWriteQuota; +use picloud_manager_core::quota::GroupWriteQuota; use picloud_shared::{ - AppId, ExecutionId, FileUpdate, GroupDocsError, GroupFilesError, GroupId, GroupKvError, - NewFile, RequestId, ScriptId, SdkCallCx, + AppId, ExecutionId, FileUpdate, FilesError, GroupDocsError, GroupFilesError, GroupId, + GroupKvError, KvError, NewFile, RequestId, ScriptId, SdkCallCx, }; use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; @@ -193,7 +193,7 @@ async fn a_failed_fan_out_rolls_the_kv_write_back() { return; }; let f = setup(&pool, "watched").await; - let writer = PostgresKvWriter::new(pool.clone()); + 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(); @@ -252,7 +252,7 @@ async fn a_failed_fan_out_rolls_a_delete_back() { return; }; let f = setup(&pool, "watched").await; - let writer = PostgresKvWriter::new(pool.clone()); + let writer = PostgresKvWriter::new(pool.clone(), u64::MAX); let cx = cx(f.app); let tag = Uuid::new_v4().simple().to_string(); @@ -608,3 +608,133 @@ async fn concurrent_uploads_cannot_push_a_group_past_its_files_byte_quota() { .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); + cleanup(&pool, f.app).await; +} + +#[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" + ); + + cleanup(&pool, f.app).await; + let _ = std::fs::remove_dir_all(&root); +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index c1b18dd..8c6f865 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -180,7 +180,10 @@ pub async fn build_app( events.clone(), picloud_manager_core::kv_service::kv_max_value_bytes_from_env(), ) - .with_atomic_writes(pool.clone()), + .with_atomic_writes( + pool.clone(), + picloud_manager_core::quota::app_kv_max_rows_from_env(), + ), ); // §11.6 shared group collections (KV): a separate store keyed by the owning // group, resolved from cx.app_id's chain. Reuses the KV value-size cap. @@ -219,7 +222,10 @@ pub async fn build_app( events.clone(), picloud_manager_core::docs_service::docs_max_value_bytes_from_env(), ) - .with_atomic_writes(pool.clone()), + .with_atomic_writes( + pool.clone(), + picloud_manager_core::quota::app_docs_max_rows_from_env(), + ), ); let dl_service: Arc = Arc::new(PostgresDeadLetterService::new( dl_repo.clone(), @@ -255,7 +261,12 @@ pub async fn build_app( files_max_size, ) // Metadata row + trigger fan-out in one tx; a rollback unlinks the blob. - .with_atomic_writes(pool.clone(), files_root.clone()), + // The per-app disk ceiling is enforced here too (advisory-locked). + .with_atomic_writes( + pool.clone(), + files_root.clone(), + picloud_manager_core::quota::app_files_max_total_bytes_from_env(), + ), ); // §11.6 shared group collections (files): group-keyed blobs under // `/files/groups//...`, resolved from cx.app_id's chain. diff --git a/crates/shared/src/docs.rs b/crates/shared/src/docs.rs index bee921c..bec4a28 100644 --- a/crates/shared/src/docs.rs +++ b/crates/shared/src/docs.rs @@ -220,6 +220,11 @@ pub enum DocsError { #[error("collection name must not be empty")] InvalidCollection, + /// The app is at its per-app doc ceiling (`PICLOUD_APP_DOCS_MAX_ROWS`) and + /// this write would add a doc. An update is exempt (net-zero rows). + #[error("docs: app doc quota exceeded ({actual} docs; limit {limit})")] + QuotaExceeded { limit: u64, actual: u64 }, + /// `create`/`update` was handed a non-object JSON value (data must /// be a JSON object so it can be navigated by field paths in /// queries). diff --git a/crates/shared/src/files.rs b/crates/shared/src/files.rs index 9f51c39..c130cb2 100644 --- a/crates/shared/src/files.rs +++ b/crates/shared/src/files.rs @@ -139,6 +139,12 @@ pub enum FilesError { #[error("missing required field: {0}")] MissingField(&'static str), + /// The app's stored blobs would exceed its total-bytes ceiling + /// (`PICLOUD_APP_FILES_MAX_TOTAL_BYTES`). Checked on the projected total, so + /// a same-size-or-smaller update near the cap still goes through. + #[error("files: app storage quota exceeded ({actual} bytes; limit {limit})")] + QuotaExceeded { limit: u64, actual: u64 }, + /// Blob exceeds the per-file size cap (default 100 MB, /// `PICLOUD_FILES_MAX_FILE_SIZE_BYTES`). #[error("file too large: {size} bytes exceeds limit of {limit} bytes")] diff --git a/crates/shared/src/group_files.rs b/crates/shared/src/group_files.rs index 2bdeb2c..3805322 100644 --- a/crates/shared/src/group_files.rs +++ b/crates/shared/src/group_files.rs @@ -202,6 +202,7 @@ impl From for GroupFilesError { F::NotFound => Self::NotFound, F::Corrupted => Self::Corrupted, F::Forbidden => Self::Forbidden, + F::QuotaExceeded { limit, actual } => Self::QuotaExceeded { limit, actual }, F::Backend(s) => Self::Backend(s), } } diff --git a/crates/shared/src/kv.rs b/crates/shared/src/kv.rs index 456ec98..cb1c4ba 100644 --- a/crates/shared/src/kv.rs +++ b/crates/shared/src/kv.rs @@ -152,6 +152,11 @@ pub enum KvError { #[error("forbidden")] Forbidden, + /// The app is at its per-app row ceiling (`PICLOUD_APP_KV_MAX_ROWS`) and + /// this write would add a key. An update to an existing key is exempt. + #[error("kv: app key quota exceeded ({actual} keys; limit {limit})")] + QuotaExceeded { limit: u64, actual: u64 }, + /// JSON-encoded value exceeded the per-app `max_value_bytes` cap. /// Defends Postgres JSONB columns from anonymous-public-script DoS; /// the limit is configurable via `PICLOUD_KV_MAX_VALUE_BYTES`.