fix(materialize): take the per-(app,queue) lock so a consumer slot can't double-fill

The one-consumer-per-(app_id, queue_name) invariant has no DB unique index (the
parent's app_id and the detail's queue_name live in different tables), so it is
enforced by a pg_advisory_xact_lock on hashtext(app_id||queue_name) around a
SELECT-then-INSERT. But `materialize` guarded its own check-then-insert only
with the GLOBAL rematerialize lock, not that per-(app,queue) key — so a
tree-apply / app-create rematerialize could interleave with an interactive
`create_queue_trigger` for the same (app, queue): each SELECT sees no consumer
(the other's INSERT uncommitted), both INSERT, and the app ends up with TWO
enabled consumers draining the same queue, splitting messages unpredictably.

Fix: `materialize_one` now resolves the template's queue_name and takes the SAME
`advisory_lock_key(app_id, queue_name)` (exposed `pub(crate)`) before its slot
check + insert, so the two paths mutually exclude. No deadlock: the interactive
path never acquires the global rematerialize lock, so there is no lock cycle,
and the rematerialize lock already serializes reconcilers against each other.
No schema change.

stateful_templates DB + journey coverage stays green.

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

crates/manager-core/src/materialize.rs

@@ -25,6 +25,7 @@
 
 use std::collections::HashSet;
 
+use picloud_shared::AppId;
 use sqlx::PgPool;
 use uuid::Uuid;
 
@@ -175,22 +176,39 @@ async fn materialize_one(
     // as a competing consumer (a different store), so the slot check is skipped
     // — every descendant intentionally gets a consumer.
     if kind == "queue" && !shared {
-        let taken: Option<(Uuid,)> = sqlx::query_as(
+        // Resolve the queue name so we can take the SAME per-(app, queue)
-            "SELECT t.id FROM triggers t \
+        // advisory lock the interactive `create_queue_trigger` uses. Without it,
-             JOIN queue_trigger_details d ON d.trigger_id = t.id \
+        // this check-then-insert (serialized only against other reconcilers by
-             WHERE t.app_id = $1 AND t.kind = 'queue' \
+        // REMATERIALIZE_LOCK) could race an interactive consumer-create and
-               AND d.queue_name = ( \
+        // double-fill the slot — the two paths held different locks.
-                   SELECT queue_name FROM queue_trigger_details WHERE trigger_id = $2)",
+        let queue_name: Option<(String,)> =
-        )
+            sqlx::query_as("SELECT queue_name FROM queue_trigger_details WHERE trigger_id = $1")
-        .bind(app_id)
+                .bind(template_id)
-        .bind(template_id)
+                .fetch_optional(&mut **tx)
-        .fetch_optional(&mut **tx)
+                .await?;
-        .await?;
+        if let Some((queue_name,)) = queue_name {
-        if taken.is_some() {
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
-            return Ok(Some(format!(
+                .bind(crate::trigger_repo::advisory_lock_key(
-                "queue template not materialized for one app — it already has a \
+                    AppId::from(app_id),
-                 consumer on that queue (app_id={app_id})"
+                    &queue_name,
-            )));
+                ))
+                .execute(&mut **tx)
+                .await?;
+            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 = $2",
+            )
+            .bind(app_id)
+            .bind(&queue_name)
+            .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})"
+                )));
+            }
         }
     }
 

crates/manager-core/src/trigger_repo.rs

@@ -1878,7 +1878,13 @@ impl TriggerRepo for PostgresTriggerRepo {
 /// Stable per-(app_id, queue_name) i64 derived from the UUID bytes +
 /// the queue name. Used as the key for `pg_advisory_xact_lock` so
 /// concurrent `create_queue_trigger` calls for the same queue serialize.
-fn advisory_lock_key(app_id: AppId, queue_name: &str) -> i64 {
+/// Stable pg-advisory-lock key for the one-consumer-per-`(app_id, queue_name)`
+/// invariant. `pub(crate)` so `materialize` can take the SAME lock the
+/// interactive `create_queue_trigger` uses — otherwise a tree-apply
+/// rematerialize could race an interactive consumer-create and double-fill the
+/// slot (there is no DB unique index: the parent's `app_id` and the detail's
+/// `queue_name` live in different tables).
+pub(crate) fn advisory_lock_key(app_id: AppId, queue_name: &str) -> i64 {
     use std::collections::hash_map::DefaultHasher;
     use std::hash::{Hash, Hasher};
     let mut h = DefaultHasher::new();