fix(core): cron double-fire, materialize warnings, group quota fast path, SSE channel cap

Remediate the MEDIUM backend correctness/perf findings from the 2026-07-11 audit.

B1 — a group cron template toggled disabled→enabled no longer re-fires every
descendant's cron. Materialization now KEEPS a disabled cron template's copies
(syncing `enabled` in place) instead of delete+recreate, preserving
`last_fired_at`; the scheduler already skips disabled copies, so keeping them is
inert. Queue/email arms still delete-on-disable (freeing the one-consumer slot).

B2 — `rematerialize_stateful_templates` warnings are now logged at the
app-create/delete and group-reparent chokepoints (were silently dropped); only
the `Err` arm was handled before.

B3 — the group KV/docs byte-quota check skips the O(rows) `SUM(octet_length(..))`
scan when a cheap upper bound (`rows_after * max_value_bytes`) is already under
the cap, so the full aggregate only runs near-cap. Every row is validated
≤ max_value_bytes, so the bound is sound (never lets an over-quota write through).

B6 — the in-process SSE broadcaster caps live channels per map
(`PICLOUD_REALTIME_MAX_CHANNELS`, default 100k); a subscribe that would open a
NEW channel past the cap is refused with 503 + `Retry-After: 1` rather than
growing the (app,topic)/(group,topic) maps unboundedly. Check+insert is atomic
under the map mutex (no TOCTOU); an existing-channel subscribe is exempt.

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

crates/manager-core/src/apps_api.rs

@@ -571,8 +571,15 @@ async fn refresh_route_cache(state: &AppsState) {
     }
     // §4.5 M5: reconcile materialized stateful-template copies for the changed
     // app set (best-effort; self-heals on the next mutation).
-    if let Err(e) = crate::materialize::rematerialize_stateful_templates(&state.pool).await {
-        tracing::warn!(error = %e, "apps: stateful-template materialization failed; it will self-heal");
+    match crate::materialize::rematerialize_stateful_templates(&state.pool).await {
+        Ok(warnings) => {
+            for w in warnings {
+                tracing::warn!(warning = %w, "apps: stateful-template materialization warning");
+            }
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "apps: stateful-template materialization failed; it will self-heal");
+        }
     }
 }
 

crates/manager-core/src/group_docs_service.rs

@@ -85,13 +85,24 @@ impl GroupDocsServiceImpl {
     /// §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).
+    /// data (already validated ≤ per-doc cap). `rows_after` is the doc count the
+    /// collection would hold post-write.
+    ///
+    /// Fast path (matches the KV service): every doc is ≤ `max_value_bytes`, so
+    /// the collection holds at most `rows_after * max_value_bytes`. When that
+    /// exact upper bound fits under the ceiling the write can't exceed it — skip
+    /// the O(collection-size) `SUM(octet_length(...))` scan. It runs only near cap.
     async fn check_total_bytes(
         &self,
         group_id: GroupId,
         new_len: u64,
         old_len: u64,
+        rows_after: u64,
     ) -> Result<(), GroupDocsError> {
+        let upper_bound = rows_after.saturating_mul(self.max_value_bytes as u64);
+        if upper_bound <= self.max_total_bytes {
+            return Ok(());
+        }
         let used = self.repo.total_bytes(group_id).await?;
         let projected = used.saturating_sub(old_len) + new_len;
         if projected > self.max_total_bytes {
@@ -234,9 +245,10 @@ 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).
+        // §11.6 M4 byte quota: a create only adds bytes (old_len = 0), one row.
         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?;
+        self.check_total_bytes(group_id, new_len, 0, count + 1)
+            .await?;
         let created = self.repo.create(group_id, collection, data.clone()).await?;
         self.emit_shared(
             cx,
@@ -311,7 +323,11 @@ impl GroupDocsService for GroupDocsServiceImpl {
             .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?;
+        // An update leaves the row count unchanged; the cheap COUNT lets the byte
+        // check skip the text SUM when the collection is comfortably under cap.
+        let count = self.repo.count_rows(group_id).await?;
+        self.check_total_bytes(group_id, new_len, old_len, count)
+            .await?;
         match self
             .repo
             .update(group_id, collection, id, data.clone())

crates/manager-core/src/group_kv_service.rs

@@ -222,31 +222,41 @@ impl GroupKvService for GroupKvServiceImpl {
         // 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 {
-            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),
-                });
-            }
+        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.
-        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,
-            });
+        //
+        // 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 {
+            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())

crates/manager-core/src/groups_api.rs

@@ -318,8 +318,15 @@ async fn reparent_group(
     }
     // §4.5 M5: the moved subtree inherits a different set of ancestor-group
     // stateful templates — reconcile its materialized copies.
-    if let Err(e) = crate::materialize::rematerialize_stateful_templates(&s.pool).await {
-        tracing::warn!(error = %e, "groups: stateful-template materialization after reparent failed; it will self-heal");
+    match crate::materialize::rematerialize_stateful_templates(&s.pool).await {
+        Ok(warnings) => {
+            for w in warnings {
+                tracing::warn!(warning = %w, "groups: stateful-template materialization warning after reparent");
+            }
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "groups: stateful-template materialization after reparent failed; it will self-heal");
+        }
     }
     Ok(Json(moved))
 }

crates/manager-core/src/materialize.rs

@@ -14,7 +14,10 @@
 //!
 //! Full-live like `rebuild_route_table`: called at the same chokepoints (apply,
 //! app create/delete, group reparent). A precise create/delete diff (not
-//! delete-all-then-recreate) preserves `last_fired_at` on unchanged cron copies.
+//! delete-all-then-recreate) preserves `last_fired_at` on unchanged cron copies;
+//! a DISABLED cron template keeps its copies (their `enabled` is synced off)
+//! rather than deleting them, so a later re-enable resumes from the preserved
+//! `last_fired_at` instead of firing from a fresh reference.
 //!
 //! Returns per-app WARNINGS (e.g. a queue whose consumer slot is already taken)
 //! — the caller surfaces them; materialization of the other pairs still
@@ -86,7 +89,8 @@ pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result<Vec<Strin
          SELECT DISTINCT ac.effective_app_id, t.id AS template_id, t.kind, t.shared \
          FROM app_chain ac \
          JOIN triggers t ON t.group_id = ac.owner_group \
-         WHERE t.group_id IS NOT NULL AND t.enabled = TRUE AND t.kind = ANY($1)",
+         WHERE t.group_id IS NOT NULL AND t.kind = ANY($1) \
+           AND (t.enabled = TRUE OR t.kind = 'cron')",
     )
     .bind(&kinds)
     .fetch_all(&mut *tx)
@@ -135,6 +139,21 @@ pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result<Vec<Strin
         }
     }
 
+    // Sync each CRON copy's `enabled` to its template's, in place. A cron
+    // template's copies are kept across a disable (the `should` query includes
+    // disabled cron templates), so a disable→re-enable toggles this flag rather
+    // than delete+recreate — preserving each copy's per-app `last_fired_at` (and
+    // its `created_at` reference). Without this, a re-enabled cron would reset to
+    // a fresh reference and lose/shift its schedule state. Cron-only: queue/email
+    // keep delete-on-disable so the queue one-consumer slot is freed on disable.
+    sqlx::query(
+        "UPDATE triggers c SET enabled = t.enabled \
+         FROM triggers t \
+         WHERE c.materialized_from = t.id AND t.kind = 'cron' AND c.enabled <> t.enabled",
+    )
+    .execute(&mut *tx)
+    .await?;
+
     tx.commit().await?;
     Ok(warnings)
 }
@@ -201,7 +220,11 @@ async fn materialize_one(
 
     match kind {
         "cron" => {
-            // A fresh copy starts with last_fired_at NULL → fires on next due.
+            // A first-ever copy starts with last_fired_at NULL → its reference
+            // is created_at (now), so it fires at the next scheduled slot, not
+            // immediately. Re-enabling a previously-materialized template does
+            // NOT hit this path — that copy is kept and its `enabled` synced, so
+            // last_fired_at survives (see the enabled-sync UPDATE above).
             sqlx::query(
                 "INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \
                  SELECT $1, schedule, timezone FROM cron_trigger_details WHERE trigger_id = $2",

crates/orchestrator-core/src/realtime.rs

@@ -32,6 +32,12 @@ use tokio::sync::broadcast;
 pub const DEFAULT_BROADCAST_CAPACITY: usize = 64;
 const ENV_CAPACITY: &str = "PICLOUD_REALTIME_BROADCAST_CAPACITY";
 
+/// Default cap on the number of live channels PER map (per-app and per-group).
+/// A subscribe that would create a NEW channel past this is refused, so an
+/// attacker naming unbounded distinct topics can't grow the maps until OOM.
+pub const DEFAULT_MAX_CHANNELS: usize = 100_000;
+const ENV_MAX_CHANNELS: &str = "PICLOUD_REALTIME_MAX_CHANNELS";
+
 /// Default GC sweep interval for empty channels.
 pub const DEFAULT_GC_INTERVAL_SECS: u64 = 60;
 
@@ -41,6 +47,8 @@ pub struct InProcessBroadcaster {
     /// subtree app subscribing to the same shared topic shares one channel.
     groups: Mutex<HashMap<(GroupId, String), broadcast::Sender<RealtimeEvent>>>,
     capacity: usize,
+    /// Max live channels per map (see [`DEFAULT_MAX_CHANNELS`]).
+    max_channels: usize,
 }
 
 impl InProcessBroadcaster {
@@ -50,10 +58,19 @@ impl InProcessBroadcaster {
             inner: Mutex::new(HashMap::new()),
             groups: Mutex::new(HashMap::new()),
             capacity: capacity.max(1),
+            max_channels: DEFAULT_MAX_CHANNELS,
         }
     }
 
-    /// Build from `PICLOUD_REALTIME_BROADCAST_CAPACITY` (default 64).
+    /// Override the per-map channel cap (0 → treated as 1; never unbounded).
+    #[must_use]
+    pub fn with_max_channels(mut self, max_channels: usize) -> Self {
+        self.max_channels = max_channels.max(1);
+        self
+    }
+
+    /// Build from `PICLOUD_REALTIME_BROADCAST_CAPACITY` (default 64) and
+    /// `PICLOUD_REALTIME_MAX_CHANNELS` (default 100_000).
     #[must_use]
     pub fn from_env() -> Self {
         let capacity = match std::env::var(ENV_CAPACITY) {
@@ -70,7 +87,21 @@ impl InProcessBroadcaster {
                 }
             },
         };
-        Self::new(capacity)
+        let max_channels = match std::env::var(ENV_MAX_CHANNELS) {
+            Err(_) => DEFAULT_MAX_CHANNELS,
+            Ok(v) => match v.parse::<usize>() {
+                Ok(n) if n > 0 => n,
+                Ok(_) => {
+                    tracing::warn!(env = ENV_MAX_CHANNELS, value = %v, "must be > 0; using default");
+                    DEFAULT_MAX_CHANNELS
+                }
+                Err(e) => {
+                    tracing::warn!(env = ENV_MAX_CHANNELS, value = %v, error = %e, "invalid; using default");
+                    DEFAULT_MAX_CHANNELS
+                }
+            },
+        };
+        Self::new(capacity).with_max_channels(max_channels)
     }
 
     /// Number of live channels in both maps (test/observability helper).
@@ -110,8 +141,16 @@ impl RealtimeBroadcaster for InProcessBroadcaster {
             .inner
             .lock()
             .map_err(|_| BroadcasterError::Unavailable("broadcaster map poisoned".into()))?;
+        let key = (app_id, topic.to_string());
+        // Refuse to open a NEW channel past the cap; subscribing to an existing
+        // one (another receiver) is always allowed.
+        if !g.contains_key(&key) && g.len() >= self.max_channels {
+            return Err(BroadcasterError::Unavailable(
+                "realtime channel limit reached".into(),
+            ));
+        }
         let tx = g
-            .entry((app_id, topic.to_string()))
+            .entry(key)
             .or_insert_with(|| broadcast::channel(self.capacity).0);
         Ok(tx.subscribe())
     }
@@ -145,8 +184,14 @@ impl RealtimeBroadcaster for InProcessBroadcaster {
             .groups
             .lock()
             .map_err(|_| BroadcasterError::Unavailable("group broadcaster map poisoned".into()))?;
+        let key = (group_id, topic.to_string());
+        if !g.contains_key(&key) && g.len() >= self.max_channels {
+            return Err(BroadcasterError::Unavailable(
+                "realtime channel limit reached".into(),
+            ));
+        }
         let tx = g
-            .entry((group_id, topic.to_string()))
+            .entry(key)
             .or_insert_with(|| broadcast::channel(self.capacity).0);
         Ok(tx.subscribe())
     }
@@ -308,6 +353,23 @@ mod tests {
         assert_eq!(b.channel_count(), 0);
     }
 
+    #[tokio::test]
+    async fn subscribe_refuses_new_channel_past_the_cap() {
+        // Bound the maps: a subscribe that would open a NEW channel past the cap
+        // is refused, but subscribing to an already-open channel still works.
+        let b = InProcessBroadcaster::new(16).with_max_channels(2);
+        let app = AppId::new();
+        let _rx1 = b.subscribe(app, "a").await.unwrap();
+        let _rx2 = b.subscribe(app, "b").await.unwrap();
+        // Third distinct topic → over cap → refused.
+        assert!(matches!(
+            b.subscribe(app, "c").await,
+            Err(BroadcasterError::Unavailable(_))
+        ));
+        // An additional receiver on an existing channel is still allowed.
+        assert!(b.subscribe(app, "a").await.is_ok());
+    }
+
     #[tokio::test]
     async fn drop_group_topic_disconnects_subscribers() {
         let b = InProcessBroadcaster::new(16);

crates/orchestrator-core/src/realtime_api.rs

@@ -134,14 +134,7 @@ async fn sse_topic(
     // 4. Subscribe + stream.
     let rx = match state.broadcaster.subscribe(app_id, &topic).await {
         Ok(rx) => rx,
-        Err(e) => {
-            tracing::error!(error = %e, "failed to acquire realtime subscription");
-            return (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                axum::Json(serde_json::json!({ "error": "internal error" })),
-            )
-                .into_response();
-        }
+        Err(e) => return subscribe_unavailable(&e),
     };
 
     let stream = event_stream(rx);
@@ -199,14 +192,7 @@ async fn sse_shared_topic(
 
     let rx = match state.broadcaster.subscribe_group(group_id, &topic).await {
         Ok(rx) => rx,
-        Err(e) => {
-            tracing::error!(error = %e, "failed to acquire shared-topic subscription");
-            return (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                axum::Json(serde_json::json!({ "error": "internal error" })),
-            )
-                .into_response();
-        }
+        Err(e) => return subscribe_unavailable(&e),
     };
 
     let sse = Sse::new(event_stream(rx))
@@ -264,6 +250,20 @@ fn unauthorized() -> Response {
         .into_response()
 }
 
+/// A `subscribe`/`subscribe_group` failure. `BroadcasterError` is transient by
+/// construction (the channel cap is reached, or a rare poisoned lock / cluster
+/// register failure), so it maps to 503 + `Retry-After: 1` — a "come back
+/// later", not a 500. Aligns with the `PICLOUD_REALTIME_MAX_CHANNELS` doc.
+fn subscribe_unavailable(e: &picloud_shared::BroadcasterError) -> Response {
+    tracing::warn!(error = %e, "realtime subscription refused (broadcaster unavailable)");
+    (
+        StatusCode::SERVICE_UNAVAILABLE,
+        [(axum::http::header::RETRY_AFTER, "1")],
+        axum::Json(serde_json::json!({ "error": "realtime broadcaster unavailable" })),
+    )
+        .into_response()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;