From 636106e9ea2f94bebb49bcb2c81fb72e24b58f46 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 11 Jul 2026 15:09:01 +0200 Subject: [PATCH] feat(quota): per-group total-byte quotas for shared KV + docs (Track A M4) M3 quotas capped a group's shared KV/docs by ROW count only; a group could still blow past storage with few-but-huge values. Add the symmetric total-bytes ceiling (files already had one). - group_quota: PICLOUD_GROUP_{KV,DOCS}_MAX_TOTAL_BYTES env vars (default 256 MiB) + accessors. - GroupKvRepo/GroupDocsRepo: total_bytes(group) = SUM(octet_length(value::text)) (default Ok(0) so non-Postgres impls skip the check). - GroupKv/DocsServiceImpl: check the PROJECTED total (old value's bytes subtracted, new added) on every set/create/update, so a same-or-smaller update near the cap is still allowed (unlike a naive used+new check). New TotalBytesQuotaExceeded errors; +with_max_total_bytes for tests. - KV set switched has->get to obtain the old value's size for the delta. Pinned by group_kv_service + group_docs_service per_group_byte_quota_uses_the_ projected_total unit tests (reject over-cap, allow same/smaller update near cap). No migration. CLAUDE.md env-var table updated. Co-Authored-By: Claude Opus 4.8 (1M context) --- CLAUDE.md | 2 + crates/manager-core/src/group_docs_repo.rs | 18 +++ crates/manager-core/src/group_docs_service.rs | 104 ++++++++++++++++++ crates/manager-core/src/group_kv_repo.rs | 19 ++++ crates/manager-core/src/group_kv_service.rs | 98 ++++++++++++++++- crates/manager-core/src/group_quota.rs | 29 ++++- crates/shared/src/group_docs.rs | 5 + crates/shared/src/group_kv.rs | 6 + 8 files changed, 274 insertions(+), 7 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 1cd367e..bca9144 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -134,6 +134,8 @@ Environment variables consumed by the `picloud` binary: | `PICLOUD_QUEUE_MAX_PAYLOAD_BYTES` | `262144` (256 KB) | Per-message JSON-encoded payload cap for `queue::enqueue`. | | `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. | +| `PICLOUD_GROUP_DOCS_MAX_TOTAL_BYTES` | `268435456` (256 MiB) | §11.6 M4 per-group quota: max total stored JSON bytes across a group's shared-docs collections (projected-total check, as KV). | | `PICLOUD_GROUP_FILES_MAX_TOTAL_BYTES` | `10737418240` (10 GiB) | §11.6 M3 per-group quota: max total stored bytes across a group's shared-files collections (`files::shared_collection().create`). | ## Out of MVP diff --git a/crates/manager-core/src/group_docs_repo.rs b/crates/manager-core/src/group_docs_repo.rs index c9ab2d4..1f13143 100644 --- a/crates/manager-core/src/group_docs_repo.rs +++ b/crates/manager-core/src/group_docs_repo.rs @@ -88,6 +88,13 @@ pub trait GroupDocsRepo: Send + Sync { let _ = group_id; Ok(0) } + + /// §11.6 M4 quota: total stored JSON bytes across the group's shared-docs + /// collections (`octet_length(data::text)`). Default `Ok(0)`. + async fn total_bytes(&self, group_id: GroupId) -> Result { + let _ = group_id; + Ok(0) + } } pub struct PostgresGroupDocsRepo { @@ -281,6 +288,17 @@ impl GroupDocsRepo for PostgresGroupDocsRepo { .await?; Ok(u64::try_from(n).unwrap_or(0)) } + + async fn total_bytes(&self, group_id: GroupId) -> Result { + let (n,): (i64,) = sqlx::query_as( + "SELECT COALESCE(SUM(octet_length(data::text)), 0)::BIGINT \ + FROM group_docs WHERE group_id = $1", + ) + .bind(group_id.into_inner()) + .fetch_one(&self.pool) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) + } } fn encode_cursor(last_id: &Uuid) -> String { diff --git a/crates/manager-core/src/group_docs_service.rs b/crates/manager-core/src/group_docs_service.rs index 83697f8..2649b6f 100644 --- a/crates/manager-core/src/group_docs_service.rs +++ b/crates/manager-core/src/group_docs_service.rs @@ -32,6 +32,10 @@ pub struct GroupDocsServiceImpl { /// §11.6 per-group quota: max total docs across the group's shared-docs /// collections (`PICLOUD_GROUP_DOCS_MAX_ROWS`). max_rows: u64, + /// §11.6 M4 per-group quota: max total stored bytes across the group's + /// shared-docs collections (`PICLOUD_GROUP_DOCS_MAX_TOTAL_BYTES`). Checked on + /// the projected total so a same/smaller update near the cap is still allowed. + max_total_bytes: u64, /// §11.6: fires `shared = true` docs triggers on a shared-collection write. events: Arc, } @@ -63,6 +67,7 @@ impl GroupDocsServiceImpl { Self { events: 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(), repo, resolver, authz, @@ -70,6 +75,34 @@ impl GroupDocsServiceImpl { } } + /// Override the per-group total-bytes quota (§11.6 M4). Mainly for tests. + #[must_use] + pub fn with_max_total_bytes(mut self, max_total_bytes: u64) -> Self { + self.max_total_bytes = max_total_bytes; + self + } + + /// §11.6 M4: reject a write whose PROJECTED total (old doc's bytes subtracted, + /// new doc's added — `old_len = 0` for a create) would exceed the group's + /// total-bytes ceiling. `new_len` is the JSON-encoded size of the incoming + /// data (already validated ≤ per-doc cap). + async fn check_total_bytes( + &self, + group_id: GroupId, + new_len: u64, + old_len: u64, + ) -> Result<(), GroupDocsError> { + let used = self.repo.total_bytes(group_id).await?; + let projected = used.saturating_sub(old_len) + new_len; + if projected > self.max_total_bytes { + return Err(GroupDocsError::TotalBytesQuotaExceeded { + limit: self.max_total_bytes, + actual: projected, + }); + } + Ok(()) + } + /// §11.6: fire `shared = true` docs triggers on the owning group. /// Best-effort; an emit failure is logged, never surfaced. #[allow(clippy::too_many_arguments)] // op/collection/id + new + old payloads @@ -201,6 +234,9 @@ impl GroupDocsService for GroupDocsServiceImpl { actual: usize::try_from(count).unwrap_or(usize::MAX), }); } + // §11.6 M4 byte quota: a create only adds bytes (old_len = 0). + let new_len = serde_json::to_vec(&data).map_or(0u64, |b| b.len() as u64); + self.check_total_bytes(group_id, new_len, 0).await?; let created = self.repo.create(group_id, collection, data.clone()).await?; self.emit_shared( cx, @@ -265,6 +301,17 @@ impl GroupDocsService for GroupDocsServiceImpl { validate_data(&data)?; self.check_data_size(&data)?; self.check_write(cx, group_id).await?; + // §11.6 M4 byte quota: check the projected total before writing. Fetch the + // current doc for its old byte size (a missing doc → old_len 0, and the + // update below no-ops to `None`). + let old_len = self + .repo + .get(group_id, collection, id) + .await? + .and_then(|d| serde_json::to_vec(&d.data).ok()) + .map_or(0u64, |b| b.len() as u64); + let new_len = serde_json::to_vec(&data).map_or(0u64, |b| b.len() as u64); + self.check_total_bytes(group_id, new_len, old_len).await?; match self .repo .update(group_id, collection, id, data.clone()) @@ -447,6 +494,16 @@ mod tests { next_cursor: None, }) } + + async fn total_bytes(&self, group_id: GroupId) -> Result { + let data = self.data.lock().await; + Ok(data + .iter() + .filter(|((g, _), _)| *g == group_id) + .flat_map(|(_, docs)| docs.values()) + .map(|v| serde_json::to_vec(v).map_or(0, |b| b.len() as u64)) + .sum()) + } } #[derive(Default)] @@ -578,4 +635,51 @@ mod tests { .unwrap_err(); assert!(matches!(err, GroupDocsError::InvalidData)); } + + #[tokio::test] + async fn per_group_byte_quota_uses_the_projected_total() { + // §11.6 M4: the shared-docs byte ceiling is checked on the PROJECTED + // total — a create over the cap is rejected, and a shrinking update near + // the cap is allowed (the delta rule). + let app = AppId::new(); + let group = GroupId::new(); + let mut resolver = FakeResolver::default(); + resolver.map.insert((app, "articles".into()), group); + let docs = GroupDocsServiceImpl::new( + Arc::new(InMemoryGroupDocsRepo::default()), + Arc::new(resolver), + Arc::new(DenyingAuthzRepo), + ) + .with_max_total_bytes(40); + let cx = cx_with(app, Some(owner())); + + // {"t":"xxxxx"} encodes to 13 bytes. + let id = docs + .create(&cx, "articles", serde_json::json!({"t": "xxxxx"})) + .await + .unwrap(); // used = 13 + docs.create(&cx, "articles", serde_json::json!({"t": "xxxxx"})) + .await + .unwrap(); // used = 26 + // A third 13-byte doc → projected 39 ≤ 40 → ok. + docs.create(&cx, "articles", serde_json::json!({"t": "xxxxx"})) + .await + .unwrap(); // used = 39 + // A fourth → projected 52 > 40 → rejected. + let err = docs + .create(&cx, "articles", serde_json::json!({"t": "xxxxx"})) + .await + .unwrap_err(); + assert!( + matches!( + err, + GroupDocsError::TotalBytesQuotaExceeded { limit: 40, .. } + ), + "got {err:?}" + ); + // Shrinking an existing doc frees budget (projected 39 - 13 + 9 = 35). + docs.update(&cx, "articles", id, serde_json::json!({"t": "y"})) + .await + .expect("a shrinking update near the cap must be allowed"); + } } diff --git a/crates/manager-core/src/group_kv_repo.rs b/crates/manager-core/src/group_kv_repo.rs index 106fbd0..e53ecdc 100644 --- a/crates/manager-core/src/group_kv_repo.rs +++ b/crates/manager-core/src/group_kv_repo.rs @@ -68,6 +68,14 @@ pub trait GroupKvRepo: Send + Sync { let _ = group_id; Ok(0) } + + /// §11.6 M4 quota: total stored JSON bytes across all of the group's shared-KV + /// collections (`octet_length(value::text)`). Default `Ok(0)` so non-Postgres + /// impls skip the byte-quota check. + async fn total_bytes(&self, group_id: GroupId) -> Result { + let _ = group_id; + Ok(0) + } } pub struct PostgresGroupKvRepo { @@ -224,6 +232,17 @@ impl GroupKvRepo for PostgresGroupKvRepo { .await?; Ok(u64::try_from(n).unwrap_or(0)) } + + async fn total_bytes(&self, group_id: GroupId) -> Result { + let (n,): (i64,) = sqlx::query_as( + "SELECT COALESCE(SUM(octet_length(value::text)), 0)::BIGINT \ + FROM group_kv_entries WHERE group_id = $1", + ) + .bind(group_id.into_inner()) + .fetch_one(&self.pool) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) + } } fn encode_cursor(last_key: &str) -> String { diff --git a/crates/manager-core/src/group_kv_service.rs b/crates/manager-core/src/group_kv_service.rs index 8bb96e3..f42a461 100644 --- a/crates/manager-core/src/group_kv_service.rs +++ b/crates/manager-core/src/group_kv_service.rs @@ -43,6 +43,11 @@ pub struct GroupKvServiceImpl { /// collections (`PICLOUD_GROUP_KV_MAX_ROWS`). Checked only when inserting a /// NEW key. max_rows: u64, + /// §11.6 M4 per-group quota: max total stored bytes across the group's + /// shared-KV collections (`PICLOUD_GROUP_KV_MAX_TOTAL_BYTES`). Checked on the + /// projected total (old value subtracted, new added) so a same/smaller update + /// near the cap is still allowed. + max_total_bytes: u64, /// §11.6: fires `shared = true` triggers on a shared-collection write. /// Defaults to the noop emitter; the host wires the outbox emitter via /// [`Self::with_events`]. @@ -85,6 +90,7 @@ impl GroupKvServiceImpl { Self { events: 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(), repo, resolver, authz, @@ -92,6 +98,13 @@ impl GroupKvServiceImpl { } } + /// Override the per-group total-bytes quota (§11.6 M4). Mainly for tests. + #[must_use] + pub fn with_max_total_bytes(mut self, max_total_bytes: u64) -> Self { + self.max_total_bytes = max_total_bytes; + self + } + /// The structural boundary: resolve the collection to its owning group on /// the calling app's chain. `CollectionNotShared` when no ancestor group /// declares it — the same outcome a foreign app gets, by construction. @@ -204,11 +217,12 @@ impl GroupKvService for GroupKvServiceImpl { }); } self.check_write(cx, group_id).await?; - // Determine insert vs update for the event op (best-effort, like the - // per-app path — the has+set is non-transactional but triggers are - // fire-and-forget). - let existed = self.repo.has(group_id, collection, key).await?; - // §11.6 quota: a NEW key must fit under the group's row ceiling; an + // Determine insert vs update for the event op + quota deltas (best-effort, + // like the per-app path — the get+set is non-transactional but triggers + // are fire-and-forget and quotas are safety rails, not exact accounting). + let existing = self.repo.get(group_id, collection, key).await?; + let existed = existing.is_some(); + // §11.6 row quota: a NEW key must fit under the group's row ceiling; an // update of an existing key is net-zero and exempt. if !existed { let count = self.repo.count_rows(group_id).await?; @@ -219,6 +233,21 @@ impl GroupKvService for GroupKvServiceImpl { }); } } + // §11.6 M4 byte quota: the PROJECTED total (old value's bytes subtracted, + // new value's added) must fit under the group's total-bytes ceiling, so a + // same-or-smaller update near the cap is still allowed. + let old_len = existing + .as_ref() + .and_then(|v| serde_json::to_vec(v).ok()) + .map_or(0u64, |b| b.len() as u64); + let used = self.repo.total_bytes(group_id).await?; + let projected = used.saturating_sub(old_len) + encoded_len as u64; + if projected > self.max_total_bytes { + return Err(GroupKvError::TotalBytesQuotaExceeded { + limit: self.max_total_bytes, + actual: projected, + }); + } self.repo .set(group_id, collection, key, value.clone()) .await?; @@ -370,6 +399,15 @@ mod tests { let data = self.data.lock().await; Ok(data.keys().filter(|(g, _, _)| *g == group_id).count() as u64) } + + async fn total_bytes(&self, group_id: GroupId) -> Result { + let data = self.data.lock().await; + Ok(data + .iter() + .filter(|((g, _, _), _)| *g == group_id) + .map(|(_, v)| serde_json::to_vec(v).map_or(0, |b| b.len() as u64)) + .sum()) + } } /// Maps `(app_id, lowercased name) -> owning group`. Anything absent is @@ -488,6 +526,56 @@ mod tests { .expect("update of an existing key must be allowed at the cap"); } + #[tokio::test] + async fn per_group_byte_quota_uses_the_projected_total() { + // §11.6 M4: the byte ceiling is checked on the PROJECTED total, so a + // write that would push the group over is rejected, but a same/smaller + // update near the cap is still allowed (unlike a naive used+new check). + let app = AppId::new(); + let group = GroupId::new(); + let mut resolver = FakeResolver::default(); + resolver.map.insert((app, "catalog".into()), group); + // A 20-byte ceiling; each `"xxxxx"` value encodes to 7 bytes ("\"xxxxx\""). + let kv = GroupKvServiceImpl::new( + Arc::new(InMemoryGroupKvRepo::default()), + Arc::new(resolver), + Arc::new(DenyingAuthzRepo), + ) + .with_max_rows(1000) + .with_max_total_bytes(20); + let cx = cx_with(app, Some(owner())); + + kv.set(&cx, "catalog", "a", serde_json::json!("xxxxx")) + .await + .unwrap(); // used = 7 + kv.set(&cx, "catalog", "b", serde_json::json!("xxxxx")) + .await + .unwrap(); // used = 14 + // A third 7-byte value → projected 21 > 20 → rejected. + let err = kv + .set(&cx, "catalog", "c", serde_json::json!("xxxxx")) + .await + .unwrap_err(); + assert!( + matches!(err, GroupKvError::TotalBytesQuotaExceeded { limit: 20, .. }), + "got {err:?}" + ); + // A same-size update of an existing key stays within budget (projected + // = 14 - 7 + 7 = 14 ≤ 20) — the delta rule, not used+new. + kv.set(&cx, "catalog", "a", serde_json::json!("yyyyy")) + .await + .expect("a same-size update near the cap must be allowed"); + // Growing an existing value past the cap is still rejected. + let err = kv + .set(&cx, "catalog", "a", serde_json::json!("zzzzzzzzzzzzzzz")) + .await + .unwrap_err(); + assert!( + matches!(err, GroupKvError::TotalBytesQuotaExceeded { .. }), + "got {err:?}" + ); + } + #[tokio::test] async fn unrelated_app_gets_collection_not_shared() { // app_a's chain declares "catalog"; app_b's does not. diff --git a/crates/manager-core/src/group_quota.rs b/crates/manager-core/src/group_quota.rs index 74db1b7..b4763e9 100644 --- a/crates/manager-core/src/group_quota.rs +++ b/crates/manager-core/src/group_quota.rs @@ -3,14 +3,23 @@ //! `PICLOUD_*_MAX_*_BYTES` caps). One default applies to every group; per-group //! configurable limits are deferred. //! -//! - KV / docs: a per-group ROW-count ceiling (across the group's collections of -//! that kind). An update of an existing key/doc is net-zero and exempt. +//! - KV / docs: a per-group ROW-count ceiling AND a per-group TOTAL-BYTES ceiling +//! (across the group's collections of that kind). Row count exempts an update +//! of an existing key/doc (net-zero rows); the byte ceiling is checked on the +//! projected total (old value's bytes subtracted, new value's added) so a +//! same-or-smaller update near the cap is still allowed. //! - files: a per-group TOTAL-BYTES ceiling (sum of stored blob sizes). /// Default per-group shared-KV row ceiling. Override `PICLOUD_GROUP_KV_MAX_ROWS`. pub const DEFAULT_GROUP_KV_MAX_ROWS: u64 = 100_000; /// Default per-group shared-docs row ceiling. Override `PICLOUD_GROUP_DOCS_MAX_ROWS`. pub const DEFAULT_GROUP_DOCS_MAX_ROWS: u64 = 100_000; +/// Default per-group shared-KV total-bytes ceiling (256 MiB). Override +/// `PICLOUD_GROUP_KV_MAX_TOTAL_BYTES`. +pub const DEFAULT_GROUP_KV_MAX_TOTAL_BYTES: u64 = 256 * 1024 * 1024; +/// Default per-group shared-docs total-bytes ceiling (256 MiB). Override +/// `PICLOUD_GROUP_DOCS_MAX_TOTAL_BYTES`. +pub const DEFAULT_GROUP_DOCS_MAX_TOTAL_BYTES: u64 = 256 * 1024 * 1024; /// Default per-group shared-files total-bytes ceiling (10 GiB). Override /// `PICLOUD_GROUP_FILES_MAX_TOTAL_BYTES`. pub const DEFAULT_GROUP_FILES_MAX_TOTAL_BYTES: u64 = 10 * 1024 * 1024 * 1024; @@ -35,6 +44,22 @@ pub fn group_docs_max_rows_from_env() -> u64 { u64_from_env("PICLOUD_GROUP_DOCS_MAX_ROWS", DEFAULT_GROUP_DOCS_MAX_ROWS) } +#[must_use] +pub fn group_kv_max_total_bytes_from_env() -> u64 { + u64_from_env( + "PICLOUD_GROUP_KV_MAX_TOTAL_BYTES", + DEFAULT_GROUP_KV_MAX_TOTAL_BYTES, + ) +} + +#[must_use] +pub fn group_docs_max_total_bytes_from_env() -> u64 { + u64_from_env( + "PICLOUD_GROUP_DOCS_MAX_TOTAL_BYTES", + DEFAULT_GROUP_DOCS_MAX_TOTAL_BYTES, + ) +} + #[must_use] pub fn group_files_max_total_bytes_from_env() -> u64 { u64_from_env( diff --git a/crates/shared/src/group_docs.rs b/crates/shared/src/group_docs.rs index 591942b..ff8d830 100644 --- a/crates/shared/src/group_docs.rs +++ b/crates/shared/src/group_docs.rs @@ -181,6 +181,11 @@ pub enum GroupDocsError { #[error("group docs: quota exceeded ({actual} rows; limit {limit})")] QuotaExceeded { limit: usize, actual: usize }, + /// The owning group's shared-docs store would exceed its total-bytes quota + /// (`PICLOUD_GROUP_DOCS_MAX_TOTAL_BYTES`). `actual` is the projected total. + #[error("group docs: total-bytes quota exceeded ({actual} bytes; limit {limit})")] + TotalBytesQuotaExceeded { limit: u64, actual: u64 }, + #[error("group docs backend error: {0}")] Backend(String), } diff --git a/crates/shared/src/group_kv.rs b/crates/shared/src/group_kv.rs index 253e191..99b550a 100644 --- a/crates/shared/src/group_kv.rs +++ b/crates/shared/src/group_kv.rs @@ -141,6 +141,12 @@ pub enum GroupKvError { #[error("group kv: quota exceeded ({actual} rows; limit {limit})")] QuotaExceeded { limit: usize, actual: usize }, + /// The owning group's shared-KV store would exceed its total-bytes quota + /// (`PICLOUD_GROUP_KV_MAX_TOTAL_BYTES`). `actual` is the projected total + /// (old value's bytes subtracted, new value's added). + #[error("group kv: total-bytes quota exceeded ({actual} bytes; limit {limit})")] + TotalBytesQuotaExceeded { limit: u64, actual: u64 }, + /// Anything else — Postgres unavailable, serialization failure, etc. #[error("group kv backend error: {0}")] Backend(String),