fix(quota): make the per-group ceiling a bound, not a hint

Audit #8. `GroupKvServiceImpl` read the group's usage (`count_rows` /
`projected_total_bytes`) on one pooled connection and wrote on another. N
concurrent writers therefore each observed the same pre-write total, each
concluded it had room, and together sailed past the ceiling. The code
called this out as "best-effort ... quotas are safety rails, not exact
accounting" — but the overshoot is not small. With a ceiling of 5 and 20
concurrent writers, 19 rows land.

Route group-KV mutations through `atomic_write::GroupKvWriter`, whose
Postgres impl runs the quota reads, the row write, and the shared-trigger
fan-out on ONE connection in ONE transaction — and, crucially, takes a
per-group `pg_advisory_xact_lock` first.

The lock is the fix, not the transaction. Putting the check inside the
writing tx is necessary but NOT sufficient: under READ COMMITTED each
transaction's `COUNT(*)`/`SUM(...)` still sees a snapshot without the other
writers' uncommitted rows, so they all still pass. Serializing the
check-then-write per group is what makes a writer see its predecessor's row.
The lock key is namespaced per kind (kv/docs/…) so writes drawing on
different ceilings don't contend. A delete takes no lock — it only frees
space.

The quota POLICY (row ceiling, projected-bytes ceiling, and the
upper-bound fast path that skips the O(n) SUM scan for a group nowhere near
its cap) moves to one place, `group_quota::check_group_write`, over a small
`GroupUsage` trait. Both backends — the transactional one reading through
`&mut *tx`, and the in-memory one the unit tests use — share it, so there is
exactly one copy of the decision.

`set_if` gains a correctness nicety on the way: the row ceiling now keys on
whether the write ADDS a row, so a compare-and-swap against an absent key
with a `Some` precondition (which cannot swap, and so consumes nothing) is
no longer charged for one.

tests/atomic_write.rs pins both ceilings against 20/16 concurrent writers.
Both tests fail without the lock (19 rows stored against a ceiling of 5).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/src/group_kv_service.rs

@@ -22,13 +22,17 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use picloud_shared::{
-    GroupId, GroupKvError, GroupKvService, KvListPage, NoopEventEmitter, SdkCallCx, ServiceEvent,
+    GroupId, GroupKvError, GroupKvService, KvListPage, NoopEventEmitter, SdkCallCx,
     ServiceEventEmitter,
 };
 
+use crate::atomic_write::{
+    BestEffortGroupKvWriter, GroupKvTarget, GroupKvWriter, PostgresGroupKvWriter,
+};
 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;
 
 /// The registry `kind` this service resolves.
@@ -48,10 +52,10 @@ pub struct GroupKvServiceImpl {
     /// 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`].
-    events: Arc<dyn ServiceEventEmitter>,
+    /// Mutations: the quota decision, the row write, and the shared-trigger
+    /// fan-out, which for the host all commit in one advisory-locked
+    /// transaction. Reads stay on `repo`.
+    writer: Arc<dyn GroupKvWriter>,
 }
 
 impl GroupKvServiceImpl {
@@ -64,11 +68,26 @@ impl GroupKvServiceImpl {
         Self::with_max_value_bytes(repo, resolver, authz, kv_max_value_bytes_from_env())
     }
 
-    /// Wire the event emitter (§11.6 shared-collection triggers). Without this
-    /// the service uses the noop emitter and shared writes fire nothing.
+    /// Wire the event emitter (§11.6 shared-collection triggers) on the
+    /// best-effort writer. Without this — or [`Self::with_atomic_writes`], which
+    /// supersedes it — shared writes fire nothing.
     #[must_use]
     pub fn with_events(mut self, events: Arc<dyn ServiceEventEmitter>) -> Self {
-        self.events = events;
+        self.writer = Arc::new(BestEffortGroupKvWriter::new(self.repo.clone(), events));
+        self
+    }
+
+    /// Use the transactional writer: the quota reads, the row write, and the
+    /// shared-trigger fan-out all run on ONE connection under a per-group
+    /// advisory lock. That makes the quota an actual bound (concurrent writers
+    /// to a group serialize, so one sees the other's row) and makes an emit
+    /// failure roll the write back instead of silently losing the event.
+    ///
+    /// Supersedes [`Self::with_events`] — the transactional writer resolves and
+    /// writes the fan-out itself and needs no injected emitter.
+    #[must_use]
+    pub fn with_atomic_writes(mut self, pool: sqlx::PgPool) -> Self {
+        self.writer = Arc::new(PostgresGroupKvWriter::new(pool));
         self
     }
 
@@ -88,7 +107,10 @@ impl GroupKvServiceImpl {
         max_value_bytes: usize,
     ) -> Self {
         Self {
-            events: Arc::new(NoopEventEmitter),
+            writer: Arc::new(BestEffortGroupKvWriter::new(
+                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(),
             repo,
@@ -123,39 +145,19 @@ impl GroupKvServiceImpl {
             .ok_or_else(|| GroupKvError::CollectionNotShared(collection.to_string()))
     }
 
-    /// §11.6: fire `shared = true` kv triggers on the owning group. Best-effort
-    /// — an emit failure is logged, never surfaced (the write already
-    /// committed), matching the per-app `KvServiceImpl` behaviour.
-    async fn emit_shared(
-        &self,
-        cx: &SdkCallCx,
-        group_id: GroupId,
-        op: &'static str,
-        collection: &str,
-        key: &str,
-        payload: Option<serde_json::Value>,
-    ) {
-        crate::group_collection_repo::best_effort_emit_shared(
-            &*self.events,
-            cx,
-            group_id,
-            ServiceEvent {
-                source: "kv",
-                op,
-                collection: Some(collection.to_string()),
-                key: Some(key.to_string()),
-                payload,
-                old_payload: None,
-            },
-        )
-        .await;
+    /// The ceilings a shared-collection write must fit under.
+    fn quota(&self) -> GroupWriteQuota {
+        GroupWriteQuota {
+            max_rows: self.max_rows,
+            max_total_bytes: self.max_total_bytes,
+            max_value_bytes: self.max_value_bytes,
+        }
     }
 
-    /// Encode `value` and enforce the per-value byte cap, returning its encoded
-    /// length. (The per-group byte quota measures canonically in SQL via
-    /// `projected_total_bytes`, so it no longer consumes this length.) Mirrors
-    /// the sibling `group_docs_service::check_data_size`.
-    fn check_value_size(&self, value: &serde_json::Value) -> Result<usize, GroupKvError> {
+    /// Enforce the per-value byte cap. The per-GROUP byte quota is measured
+    /// canonically in SQL inside the writing transaction — see
+    /// `group_quota::check_group_write`.
+    fn check_value_size(&self, value: &serde_json::Value) -> Result<(), GroupKvError> {
         let encoded_len = serde_json::to_vec(value)
             .map(|v| v.len())
             .map_err(|e| GroupKvError::Backend(format!("encode value: {e}")))?;
@@ -165,7 +167,7 @@ impl GroupKvServiceImpl {
                 actual: encoded_len,
             });
         }
-        Ok(encoded_len)
+        Ok(())
     }
 
     async fn check_read(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupKvError> {
@@ -222,61 +224,18 @@ impl GroupKvService for GroupKvServiceImpl {
         let group_id = self.owning_group(cx, collection).await?;
         self.check_value_size(&value)?;
         self.check_write(cx, group_id).await?;
-        // 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();
-        // One cheap COUNT backs BOTH the row quota AND the byte-quota fast path
-        // below (an index-only count, far lighter than the per-value text SUM).
-        let count = self.repo.count_rows(group_id).await?;
-        // §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 && count >= self.max_rows {
-            return Err(GroupKvError::QuotaExceeded {
-                limit: usize::try_from(self.max_rows).unwrap_or(usize::MAX),
-                actual: usize::try_from(count).unwrap_or(usize::MAX),
-            });
-        }
-        // §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.
-        //
-        // Fast path: every stored value is capped at `max_value_bytes`, so the
-        // whole collection can hold at most `rows_after * max_value_bytes`. When
-        // that exact upper bound already fits under the ceiling there is no way to
-        // exceed it — skip the O(collection-size) `SUM(octet_length(...))` scan
-        // entirely. The scan runs only for a group actually near its byte cap.
-        let rows_after = if existed { count } else { count + 1 };
-        let upper_bound = rows_after.saturating_mul(self.max_value_bytes as u64);
-        if upper_bound > self.max_total_bytes {
-            // Consistent-metric projection (canonical `octet_length(value::text)`
-            // for the current SUM, the old row, and the new value) — no
-            // Rust-compact vs DB-canonical drift.
-            let projected = self
-                .repo
-                .projected_total_bytes(group_id, collection, key, &value)
-                .await?;
-            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?;
-        self.emit_shared(
-            cx,
-            group_id,
-            if existed { "update" } else { "insert" },
-            collection,
-            key,
-            Some(value),
-        )
-        .await;
-        Ok(())
+        self.writer
+            .set(
+                cx,
+                GroupKvTarget {
+                    group_id,
+                    collection,
+                    key,
+                },
+                value,
+                self.quota(),
+            )
+            .await
     }
 
     async fn set_if(
@@ -290,43 +249,19 @@ impl GroupKvService for GroupKvServiceImpl {
         let group_id = self.owning_group(cx, collection).await?;
         self.check_value_size(&new)?;
         self.check_write(cx, group_id).await?;
-        // Same quota rules as `set` (best-effort pre-check; the swap itself is
-        // atomic in the repo). A swap that inserts a NEW key (expected absent)
-        // must fit the row ceiling; the byte ceiling uses the projected total.
-        let existing = self.repo.get(group_id, collection, key).await?;
-        if expected.is_none() && existing.is_none() {
-            let count = self.repo.count_rows(group_id).await?;
-            if count >= self.max_rows {
-                return Err(GroupKvError::QuotaExceeded {
-                    limit: usize::try_from(self.max_rows).unwrap_or(usize::MAX),
-                    actual: usize::try_from(count).unwrap_or(usize::MAX),
-                });
-            }
-        }
-        let projected = self
-            .repo
-            .projected_total_bytes(group_id, collection, key, &new)
-            .await?;
-        if projected > self.max_total_bytes {
-            return Err(GroupKvError::TotalBytesQuotaExceeded {
-                limit: self.max_total_bytes,
-                actual: projected,
-            });
-        }
-        let op = if expected.is_some() {
-            "update"
-        } else {
-            "insert"
-        };
-        let swapped = self
-            .repo
-            .set_if(group_id, collection, key, expected, new.clone())
-            .await?;
-        if swapped {
-            self.emit_shared(cx, group_id, op, collection, key, Some(new))
-                .await;
-        }
-        Ok(swapped)
+        self.writer
+            .set_if(
+                cx,
+                GroupKvTarget {
+                    group_id,
+                    collection,
+                    key,
+                },
+                expected,
+                new,
+                self.quota(),
+            )
+            .await
     }
 
     async fn delete(
@@ -337,12 +272,16 @@ impl GroupKvService for GroupKvServiceImpl {
     ) -> Result<bool, GroupKvError> {
         let group_id = self.owning_group(cx, collection).await?;
         self.check_write(cx, group_id).await?;
-        let deleted = self.repo.delete(group_id, collection, key).await?.is_some();
-        if deleted {
-            self.emit_shared(cx, group_id, "delete", collection, key, None)
-                .await;
-        }
-        Ok(deleted)
+        self.writer
+            .delete(
+                cx,
+                GroupKvTarget {
+                    group_id,
+                    collection,
+                    key,
+                },
+            )
+            .await
     }
 
     async fn has(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result<bool, GroupKvError> {