feat(quota): add per-app storage ceilings

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) <noreply@anthropic.com>

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. |

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<Option<Value>, 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<DocId, DocsError> {
         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<Uuid>,
+        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,

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<u64, DocsRepoError>
+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::*;

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
     }
 

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<Uuid>,
+    incoming: i64,
+) -> Result<u64, FilesRepoError>
+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::*;

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
     }
 

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,

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),

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,

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<Option<serde_json::Value>, 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<String, KvRepoError> {
         .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<u64, KvRepoError>
+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))
+}

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
     }
 

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;

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::<u64>() {
@@ -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)

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);
+}

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<dyn DeadLetterService> = 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
     // `<root>/files/groups/<group_id>/...`, resolved from cx.app_id's chain.

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).

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")]

crates/shared/src/group_files.rs

@@ -202,6 +202,7 @@ impl From<crate::FilesError> 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),
         }
     }

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`.