fix(stateful-templates): serialize the materialization reconciler (review)

The reconciler read `should` and `existing` in separate statements under READ
COMMITTED, and its queue one-consumer check wasn't locked — so two concurrent
reconcilers (e.g. two parallel app-creates) could (a) spuriously DELETE a valid
materialized cron copy when one committed a copy between the other's two reads
(a lost copy self-healing only on the next mutation), or (b) both pass the
queue slot check and double-fill a consumer slot. Take a fixed xact-scoped
advisory lock (REMATERIALIZE_LOCK_KEY) at the top of the reconcile so each pass
is atomic w.r.t. the others. Reconciles are infrequent + fast, so serializing
is cheap; the ON CONFLICT guard stays as defense in depth.

Found by the M5 review. Pinned by the existing stateful_templates test + the
full journey suite (which previously flaked on the race).

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

crates/manager-core/src/materialize.rs

@@ -25,6 +25,16 @@ use uuid::Uuid;
 /// The stateful kinds that materialize (M5.2 cron; M5.4 queue; M5.5 email).
 const MATERIALIZED_KINDS: &[&str] = &["cron", "queue"];
 
+/// Fixed advisory-lock key serializing all materialization reconcilers (§4.5
+/// M5). Two reconcilers running concurrently (e.g. two parallel app-creates)
+/// would otherwise (a) read `should` and `existing` at different READ-COMMITTED
+/// snapshots — a copy created by one between the two reads could be spuriously
+/// DELETEd by the other (a lost cron copy that only self-heals on the next
+/// mutation), and (b) both pass the queue one-consumer check and double-fill a
+/// slot. An xact-scoped advisory lock makes each reconcile atomic w.r.t. the
+/// others; reconciles are infrequent + fast, so serializing is cheap.
+const REMATERIALIZE_LOCK_KEY: i64 = 0x0004_0005_0041_054d; // "M5" materialize marker
+
 /// (descendant app, source template, kind) that SHOULD have a materialized copy.
 #[derive(sqlx::FromRow)]
 struct ShouldRow {
@@ -40,6 +50,15 @@ struct ShouldRow {
 pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result<Vec<String>, sqlx::Error> {
     let mut tx = pool.begin().await?;
 
+    // Serialize reconcilers (see REMATERIALIZE_LOCK_KEY): the lock is held until
+    // this transaction commits, so the `should`/`existing` reads below are
+    // snapshot-consistent relative to any other reconcile and the queue slot
+    // check can't race.
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(REMATERIALIZE_LOCK_KEY)
+        .execute(&mut *tx)
+        .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.