From fc70f5e224f03d77afc2ccdc20bb86f8291fe4c7 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Mon, 13 Jul 2026 22:31:42 +0200 Subject: [PATCH] fix(materialize): take the per-(app,queue) lock so a consumer slot can't double-fill MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- crates/manager-core/src/materialize.rs | 50 +++++++++++++++++-------- crates/manager-core/src/trigger_repo.rs | 8 +++- 2 files changed, 41 insertions(+), 17 deletions(-) diff --git a/crates/manager-core/src/materialize.rs b/crates/manager-core/src/materialize.rs index 05bb180..d1c828f 100644 --- a/crates/manager-core/src/materialize.rs +++ b/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( - "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})" - ))); + // Resolve the queue name so we can take the SAME per-(app, queue) + // advisory lock the interactive `create_queue_trigger` uses. Without it, + // this check-then-insert (serialized only against other reconcilers by + // REMATERIALIZE_LOCK) could race an interactive consumer-create and + // double-fill the slot — the two paths held different locks. + let queue_name: Option<(String,)> = + sqlx::query_as("SELECT queue_name FROM queue_trigger_details WHERE trigger_id = $1") + .bind(template_id) + .fetch_optional(&mut **tx) + .await?; + if let Some((queue_name,)) = queue_name { + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(crate::trigger_repo::advisory_lock_key( + AppId::from(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})" + ))); + } } } diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index 4fd344d..435da7a 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/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();