diff --git a/crates/manager-core/src/apps_api.rs b/crates/manager-core/src/apps_api.rs index 617e12c..3b0c24c 100644 --- a/crates/manager-core/src/apps_api.rs +++ b/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"); + } } } diff --git a/crates/manager-core/src/group_docs_service.rs b/crates/manager-core/src/group_docs_service.rs index 2649b6f..9ea6607 100644 --- a/crates/manager-core/src/group_docs_service.rs +++ b/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()) diff --git a/crates/manager-core/src/group_kv_service.rs b/crates/manager-core/src/group_kv_service.rs index 6347fee..4221457 100644 --- a/crates/manager-core/src/group_kv_service.rs +++ b/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()) diff --git a/crates/manager-core/src/groups_api.rs b/crates/manager-core/src/groups_api.rs index f6141e8..0a294b8 100644 --- a/crates/manager-core/src/groups_api.rs +++ b/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)) } diff --git a/crates/manager-core/src/materialize.rs b/crates/manager-core/src/materialize.rs index 088f5da..05bb180 100644 --- a/crates/manager-core/src/materialize.rs +++ b/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 Result 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", diff --git a/crates/orchestrator-core/src/realtime.rs b/crates/orchestrator-core/src/realtime.rs index bd30799..aabd5c9 100644 --- a/crates/orchestrator-core/src/realtime.rs +++ b/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>>, 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::() { + 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); diff --git a/crates/orchestrator-core/src/realtime_api.rs b/crates/orchestrator-core/src/realtime_api.rs index 367be30..1595c72 100644 --- a/crates/orchestrator-core/src/realtime_api.rs +++ b/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::*;