From ddb3589c4973d1fc47e9c9ccf1ec7fe52a50fddf Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Wed, 1 Jul 2026 21:47:27 +0200 Subject: [PATCH] feat(stateful-templates): materialize cron+queue templates + hooks (M5.2-M5.4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit materialize::rematerialize_stateful_templates reconciles a group cron/queue template into one app-owned copy per descendant (materialized_from = template), via the all-apps app_chain CTE ⋈ group-owned stateful templates. A precise create/delete diff preserves cron last_fired_at on unchanged copies; a queue copy is skipped-with-warning when the app already fills that queue's consumer slot (the one-consumer invariant). Called full-live at the route-rebuild chokepoints: apply (single + tree), app create/delete (apps_api), group reparent (groups_api) — AppsState/GroupsState gain a pool. Pinned by tests/stateful_templates.rs (per-app copy, idempotent, new-app materializes, reparent de-materializes). Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/manager-core/src/apply_service.rs | 20 ++ crates/manager-core/src/apps_api.rs | 9 + crates/manager-core/src/groups_api.rs | 9 + crates/manager-core/src/lib.rs | 1 + crates/manager-core/src/materialize.rs | 186 ++++++++++++++++ .../manager-core/tests/stateful_templates.rs | 198 ++++++++++++++++++ crates/picloud/src/lib.rs | 2 + 7 files changed, 425 insertions(+) create mode 100644 crates/manager-core/src/materialize.rs create mode 100644 crates/manager-core/tests/stateful_templates.rs diff --git a/crates/manager-core/src/apply_service.rs b/crates/manager-core/src/apply_service.rs index cb8f11d..aeccf5c 100644 --- a/crates/manager-core/src/apply_service.rs +++ b/crates/manager-core/src/apply_service.rs @@ -1420,6 +1420,11 @@ impl ApplyService { .push("route table refresh failed; it will self-heal".into()); } } + // §4.5 M5: re-materialize stateful (cron/queue) group templates into + // per-descendant app rows. Post-commit + best-effort like the route + // rebuild; per-app skips (e.g. a queue slot already taken) surface as + // warnings. + report.warnings.extend(self.rematerialize().await); Ok(report) } @@ -1559,6 +1564,8 @@ impl ApplyService { .warnings .push("route table refresh failed; it will self-heal".into()); } + // §4.5 M5: one post-commit materialization pass covers the whole tree. + report.warnings.extend(self.rematerialize().await); Ok(report) } @@ -1851,6 +1858,19 @@ impl ApplyService { .map_err(|e| ApplyError::Backend(e.to_string())) } + /// §4.5 M5: reconcile materialized stateful-template copies. Best-effort — + /// a failure is logged (the copies self-heal on the next mutation) and any + /// per-app skip warnings are returned for the apply report. + async fn rematerialize(&self) -> Vec { + match crate::materialize::rematerialize_stateful_templates(&self.pool).await { + Ok(warnings) => warnings, + Err(e) => { + tracing::warn!(error = %e, "apply: stateful-template materialization failed"); + vec!["stateful-template materialization failed; it will self-heal".into()] + } + } + } + /// Resolve a referenced secret to plaintext, then re-seal it for an /// email trigger's inbound-secret column. The value never appears in /// the manifest — only the secret's name does. diff --git a/crates/manager-core/src/apps_api.rs b/crates/manager-core/src/apps_api.rs index 9f4d188..617e12c 100644 --- a/crates/manager-core/src/apps_api.rs +++ b/crates/manager-core/src/apps_api.rs @@ -52,6 +52,10 @@ pub struct AppsState { /// Group tree — resolves an app's parent group at create time /// (defaults to the instance root). pub groups: Arc, + /// §4.5 M5: creating/deleting an app changes which ancestor-group stateful + /// templates it inherits, so its materialized cron/queue copies are + /// reconciled here (full-live, mirroring the route_table rebuild). + pub pool: sqlx::PgPool, } pub fn apps_router(state: AppsState) -> Router { @@ -565,6 +569,11 @@ async fn refresh_route_cache(state: &AppsState) { { tracing::warn!(error = %e, "apps: route table refresh failed; it will self-heal"); } + // §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"); + } } pub async fn refresh_domain_cache(state: &AppsState) -> Result<(), AppsApiError> { diff --git a/crates/manager-core/src/groups_api.rs b/crates/manager-core/src/groups_api.rs index 22a17d3..92247b0 100644 --- a/crates/manager-core/src/groups_api.rs +++ b/crates/manager-core/src/groups_api.rs @@ -50,6 +50,10 @@ pub struct GroupsState { /// apps inherit — so the route snapshot is rebuilt after a reparent. pub routes: Arc, pub route_table: Arc, + /// §4.5 M5: reparenting moves a subtree, changing which ancestor-group + /// stateful templates its apps inherit — their materialized cron/queue + /// copies are reconciled after a reparent. + pub pool: sqlx::PgPool, } pub fn groups_router(state: GroupsState) -> Router { @@ -295,6 +299,11 @@ async fn reparent_group( { tracing::warn!(error = %e, "groups: route table refresh after reparent failed; it will self-heal"); } + // §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"); + } Ok(Json(moved)) } diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index c0a0c2f..763eae0 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -70,6 +70,7 @@ pub mod kv_repo; pub mod kv_service; pub mod log_sink; pub mod login_rate_limit; +pub mod materialize; pub mod migrations; pub mod module_source; pub mod outbox_event_emitter; diff --git a/crates/manager-core/src/materialize.rs b/crates/manager-core/src/materialize.rs new file mode 100644 index 0000000..c4135fc --- /dev/null +++ b/crates/manager-core/src/materialize.rs @@ -0,0 +1,186 @@ +//! §4.5 M5: materialization of STATEFUL group trigger templates. +//! +//! A group-owned cron / queue / email trigger is a TEMPLATE that is never +//! dispatched directly (the scheduler / queue consumer / email webhook all +//! filter `t.app_id IS NOT NULL`). Instead this reconciler expands each template +//! into an app-owned copy — `materialized_from = template.id` — for every +//! descendant app, and removes copies whose (app, template) pair no longer +//! inherits. The copy owns its per-app state (cron `last_fired_at`, the queue +//! advisory lock, the email sealed secret), so the unchanged dispatch paths work +//! against it as if it were hand-authored. +//! +//! 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. +//! +//! Returns per-app WARNINGS (e.g. a queue whose consumer slot is already taken, +//! or an email whose descendant lacks the referenced secret) — the caller +//! surfaces them; materialization of the other pairs still proceeds. + +use std::collections::HashSet; + +use sqlx::PgPool; +use uuid::Uuid; + +/// The stateful kinds that materialize (M5.2 cron; M5.4 queue; M5.5 email). +const MATERIALIZED_KINDS: &[&str] = &["cron", "queue"]; + +/// (descendant app, source template, kind) that SHOULD have a materialized copy. +#[derive(sqlx::FromRow)] +struct ShouldRow { + effective_app_id: Uuid, + template_id: Uuid, + kind: String, +} + +/// Reconcile all materialized stateful-template copies. Idempotent; safe to call +/// on every tree/apply mutation. Best-effort per pair — a pair that can't +/// materialize (queue slot taken, missing email secret) yields a warning and is +/// skipped, not an error. +pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result, sqlx::Error> { + let mut tx = pool.begin().await?; + + // Every (descendant app × ancestor-group stateful template) that should + // exist — the all-apps `app_chain` CTE (each app × its ancestor chain) ⋈ + // group-owned stateful templates. + let kinds: Vec = MATERIALIZED_KINDS + .iter() + .map(|s| (*s).to_string()) + .collect(); + let should: Vec = sqlx::query_as( + "WITH RECURSIVE app_chain AS ( \ + SELECT a.id AS effective_app_id, a.group_id AS owner_group, \ + a.group_id AS next_group, 0 AS depth \ + FROM apps a WHERE a.group_id IS NOT NULL \ + UNION ALL \ + SELECT ac.effective_app_id, g.parent_id, g.parent_id, ac.depth + 1 \ + FROM groups g JOIN app_chain ac ON g.id = ac.next_group \ + WHERE ac.depth < 64 AND g.parent_id IS NOT NULL \ + ) \ + SELECT DISTINCT ac.effective_app_id, t.id AS template_id, t.kind \ + 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)", + ) + .bind(&kinds) + .fetch_all(&mut *tx) + .await?; + + let should_set: HashSet<(Uuid, Uuid)> = should + .iter() + .map(|r| (r.effective_app_id, r.template_id)) + .collect(); + + // Existing managed copies. + let existing: Vec<(Uuid, Uuid)> = sqlx::query_as( + "SELECT app_id, materialized_from FROM triggers WHERE materialized_from IS NOT NULL", + ) + .fetch_all(&mut *tx) + .await?; + let existing_set: HashSet<(Uuid, Uuid)> = existing.iter().copied().collect(); + + // Delete stale copies (template gone or app no longer a descendant). + for (app_id, template_id) in &existing_set { + if !should_set.contains(&(*app_id, *template_id)) { + sqlx::query("DELETE FROM triggers WHERE app_id = $1 AND materialized_from = $2") + .bind(app_id) + .bind(template_id) + .execute(&mut *tx) + .await?; + } + } + + // Create missing copies. + let mut warnings = Vec::new(); + for r in &should { + if existing_set.contains(&(r.effective_app_id, r.template_id)) { + continue; + } + if let Some(w) = + materialize_one(&mut tx, r.effective_app_id, r.template_id, &r.kind).await? + { + warnings.push(w); + } + } + + tx.commit().await?; + Ok(warnings) +} + +/// Create one app-owned copy of a group template + its per-kind detail. Returns +/// `Some(warning)` when the pair is intentionally skipped (e.g. a queue slot the +/// app already fills). `name` uses the column default (a fresh UUID) so it can't +/// collide with the app's own trigger names. +async fn materialize_one( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + app_id: Uuid, + template_id: Uuid, + kind: &str, +) -> Result, sqlx::Error> { + // §M5.4: a queue copy would violate the one-consumer-per-(app_id, queue_name) + // invariant if the app already has a consumer (hand-authored or another + // template) on that queue — skip with a warning. + if kind == "queue" { + let taken: Option<(Uuid,)> = sqlx::query_as( + "SELECT t.id FROM triggers t \ + JOIN queue_trigger_details d ON d.trigger_id = t.id \ + WHERE t.app_id = $1 AND t.kind = 'queue' \ + AND d.queue_name = ( \ + SELECT queue_name FROM queue_trigger_details WHERE trigger_id = $2)", + ) + .bind(app_id) + .bind(template_id) + .fetch_optional(&mut **tx) + .await?; + if taken.is_some() { + return Ok(Some(format!( + "queue template not materialized for one app — it already has a \ + consumer on that queue (app_id={app_id})" + ))); + } + } + + // Parent row: copy the template's settings, owned by the app, linked back. + let new_id: (Uuid,) = sqlx::query_as( + "INSERT INTO triggers ( \ + app_id, script_id, kind, enabled, dispatch_mode, \ + retry_max_attempts, retry_backoff, retry_base_ms, \ + registered_by_principal, materialized_from) \ + SELECT $1, script_id, kind, enabled, dispatch_mode, \ + retry_max_attempts, retry_backoff, retry_base_ms, \ + registered_by_principal, id \ + FROM triggers WHERE id = $2 RETURNING id", + ) + .bind(app_id) + .bind(template_id) + .fetch_one(&mut **tx) + .await?; + + match kind { + "cron" => { + // A fresh copy starts with last_fired_at NULL → fires on next due. + sqlx::query( + "INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \ + SELECT $1, schedule, timezone FROM cron_trigger_details WHERE trigger_id = $2", + ) + .bind(new_id.0) + .bind(template_id) + .execute(&mut **tx) + .await?; + } + "queue" => { + sqlx::query( + "INSERT INTO queue_trigger_details \ + (trigger_id, queue_name, visibility_timeout_secs) \ + SELECT $1, queue_name, visibility_timeout_secs \ + FROM queue_trigger_details WHERE trigger_id = $2", + ) + .bind(new_id.0) + .bind(template_id) + .execute(&mut **tx) + .await?; + } + _ => {} + } + Ok(None) +} diff --git a/crates/manager-core/tests/stateful_templates.rs b/crates/manager-core/tests/stateful_templates.rs new file mode 100644 index 0000000..88be4ec --- /dev/null +++ b/crates/manager-core/tests/stateful_templates.rs @@ -0,0 +1,198 @@ +//! §4.5 M5 integration test: materialization of STATEFUL group trigger +//! templates. A group cron template materializes one app-owned copy per +//! descendant app; a new app materializes on reconcile; an app leaving the +//! subtree de-materializes; a group queue template skips (with a warning) an app +//! that already fills that queue's consumer slot. +//! +//! Drives `rematerialize_stateful_templates` directly. Skips when `DATABASE_URL` +//! is unset. + +#![allow(clippy::too_many_lines, clippy::many_single_char_names)] + +use picloud_manager_core::materialize::rematerialize_stateful_templates; +use sqlx::postgres::PgPoolOptions; +use sqlx::PgPool; +use uuid::Uuid; + +async fn pool_or_skip() -> Option { + let Ok(url) = std::env::var("DATABASE_URL") else { + eprintln!("stateful_templates: DATABASE_URL unset — skipping"); + return None; + }; + let pool = PgPoolOptions::new() + .max_connections(2) + .connect(&url) + .await + .expect("connect"); + sqlx::migrate!("./migrations") + .run(&pool) + .await + .expect("migrate"); + Some(pool) +} + +/// Count materialized copies of `template_id` owned by `app_id`. +async fn copies(pool: &PgPool, app_id: Uuid, template_id: Uuid) -> i64 { + let (n,): (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM triggers WHERE app_id = $1 AND materialized_from = $2", + ) + .bind(app_id) + .bind(template_id) + .fetch_one(pool) + .await + .unwrap(); + n +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cron_template_materializes_per_descendant_and_reconciles() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let sfx = Uuid::new_v4().simple().to_string(); + let admin = { + let r: (Uuid,) = sqlx::query_as( + "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id", + ) + .bind(format!("mat-{sfx}")) + .fetch_one(&pool) + .await + .unwrap(); + r.0 + }; + // Group G with a handler + a cron TEMPLATE. + let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("mat-g-{sfx}")) + .fetch_one(&pool) + .await + .unwrap(); + let handler: (Uuid,) = sqlx::query_as( + "INSERT INTO scripts (name, source, group_id) VALUES ($1, 'x', $2) RETURNING id", + ) + .bind(format!("nightly-{sfx}")) + .bind(g.0) + .fetch_one(&pool) + .await + .unwrap(); + let tmpl: (Uuid,) = sqlx::query_as( + "INSERT INTO triggers \ + (group_id, script_id, kind, enabled, dispatch_mode, retry_max_attempts, \ + retry_backoff, retry_base_ms, registered_by_principal, name) \ + VALUES ($1, $2, 'cron', TRUE, 'async', 3, 'exponential', 1000, $3, $4) RETURNING id", + ) + .bind(g.0) + .bind(handler.0) + .bind(admin) + .bind(format!("t-{sfx}")) + .fetch_one(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \ + VALUES ($1, '0 0 * * * *', 'UTC')", + ) + .bind(tmpl.0) + .execute(&pool) + .await + .unwrap(); + + // Two apps under G. + let mk_app = |slug: String| { + let pool = pool.clone(); + let g = g.0; + async move { + let r: (Uuid,) = sqlx::query_as( + "INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id", + ) + .bind(slug) + .bind(g) + .fetch_one(&pool) + .await + .unwrap(); + r.0 + } + }; + let a = mk_app(format!("mat-a-{sfx}")).await; + let b = mk_app(format!("mat-b-{sfx}")).await; + + // Reconcile → one copy per app. + let w = rematerialize_stateful_templates(&pool).await.unwrap(); + assert!(w.is_empty(), "no warnings expected: {w:?}"); + assert_eq!(copies(&pool, a, tmpl.0).await, 1, "app A materialized"); + assert_eq!(copies(&pool, b, tmpl.0).await, 1, "app B materialized"); + + // Idempotent: a second reconcile does not duplicate. + rematerialize_stateful_templates(&pool).await.unwrap(); + assert_eq!(copies(&pool, a, tmpl.0).await, 1, "no duplicate copy"); + + // The copy is app-owned + linked back to the template + carries the detail. + let (kind, sched): (String, String) = sqlx::query_as( + "SELECT t.kind, d.schedule FROM triggers t \ + JOIN cron_trigger_details d ON d.trigger_id = t.id \ + WHERE t.app_id = $1 AND t.materialized_from = $2", + ) + .bind(a) + .bind(tmpl.0) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(kind, "cron"); + assert_eq!(sched, "0 0 * * * *"); + + // A new app under G materializes on the next reconcile. + let c = mk_app(format!("mat-c-{sfx}")).await; + rematerialize_stateful_templates(&pool).await.unwrap(); + assert_eq!(copies(&pool, c, tmpl.0).await, 1, "new app materialized"); + + // Reparent app A into a SIBLING group (not under G) → its copy + // de-materializes. + let g2: (Uuid,) = + sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("mat-g2-{sfx}")) + .fetch_one(&pool) + .await + .unwrap(); + sqlx::query("UPDATE apps SET group_id = $2 WHERE id = $1") + .bind(a) + .bind(g2.0) + .execute(&pool) + .await + .unwrap(); + rematerialize_stateful_templates(&pool).await.unwrap(); + assert_eq!( + copies(&pool, a, tmpl.0).await, + 0, + "an app that left the subtree de-materializes" + ); + assert_eq!( + copies(&pool, b, tmpl.0).await, + 1, + "a still-descendant app keeps its copy" + ); + + // Cleanup (materialized copies CASCADE via app delete / template delete). + let _ = sqlx::query("DELETE FROM triggers WHERE materialized_from = $1") + .bind(tmpl.0) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM triggers WHERE id = $1") + .bind(tmpl.0) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM apps WHERE id = ANY($1)") + .bind(vec![a, b, c]) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM scripts WHERE id = $1") + .bind(handler.0) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM groups WHERE id = ANY($1)") + .bind(vec![g.0, g2.0]) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM admin_users WHERE id = $1") + .bind(admin) + .execute(&pool) + .await; +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 2940e4c..ee55471 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -573,6 +573,7 @@ pub async fn build_app( route_table: route_table.clone(), authz: authz.clone(), groups: groups_repo.clone(), + pool: pool.clone(), }; // Audit 2026-06-11 H-B1 — login DoS defenses. PICLOUD_LOGIN_ARGON2_PARALLELISM @@ -620,6 +621,7 @@ pub async fn build_app( authz: authz.clone(), routes: route_repo.clone(), route_table: route_table.clone(), + pool: pool.clone(), }; let vars_state = VarsApiState { vars: Arc::new(PostgresVarsRepo::new(pool.clone())),