diff --git a/crates/manager-core/src/apply_service.rs b/crates/manager-core/src/apply_service.rs index b803c4b..bef098a 100644 --- a/crates/manager-core/src/apply_service.rs +++ b/crates/manager-core/src/apply_service.rs @@ -260,6 +260,10 @@ pub enum BundleTrigger { dispatch_mode: Option, #[serde(default)] retry_max_attempts: Option, + /// §11.6 D3: `true` for a shared-QUEUE group consumer — materializes a + /// competing consumer per descendant app, all claiming the group store. + #[serde(default)] + shared: bool, }, } @@ -313,8 +317,9 @@ impl BundleTrigger { Self::Kv { shared, .. } | Self::Docs { shared, .. } | Self::Files { shared, .. } - | Self::Pubsub { shared, .. } => *shared, - Self::Cron { .. } | Self::Email { .. } | Self::Queue { .. } => false, + | Self::Pubsub { shared, .. } + | Self::Queue { shared, .. } => *shared, + Self::Cron { .. } | Self::Email { .. } => false, } } @@ -390,7 +395,9 @@ impl BundleTrigger { .. } => format!("pubsub|{script}|{topic_pattern}|{sealed}|{shared}"), Self::Email { script, .. } => format!("email|{script}"), - Self::Queue { queue_name, .. } => format!("queue|{queue_name}"), + Self::Queue { + queue_name, shared, .. + } => format!("queue|{queue_name}|{shared}"), } } @@ -787,9 +794,10 @@ impl ApplyService { BundleTrigger::Pubsub { topic_pattern, .. } => { (topic_pattern.split('.').next().unwrap_or(""), "topic") } + BundleTrigger::Queue { queue_name, .. } => (queue_name.as_str(), "queue"), _ => { return Err(ApplyError::Invalid(format!( - "a `shared` trigger must be a kv/docs/files/pubsub kind; \ + "a `shared` trigger must be a kv/docs/files/pubsub/queue kind; \ `{}` has no shared collection store", t.kind_str() ))); @@ -3416,7 +3424,7 @@ fn current_trigger_identity(t: &Trigger, name_by_id: &HashMap) Some(format!("pubsub|{script}|{topic_pattern}|{sealed}|{shared}")) } TriggerDetails::Email { .. } => Some(format!("email|{script}")), - TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}")), + TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}|{shared}")), TriggerDetails::DeadLetter { .. } => None, } } @@ -4254,6 +4262,7 @@ mod tests { visibility_timeout_secs: vis, dispatch_mode: None, retry_max_attempts: None, + shared: false, }; assert!(validate_trigger_shape(&queue(Some(10))).is_err()); assert!(validate_trigger_shape(&queue(Some( @@ -4715,6 +4724,7 @@ mod tests { visibility_timeout_secs: None, dispatch_mode: None, retry_max_attempts: None, + shared: false, }]; let p = compute_diff(¤t, &b); assert!( diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index d1d3412..558008e 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -31,7 +31,8 @@ use picloud_orchestrator_core::{ExecutionGate, ExecutorClient}; use picloud_shared::{ AppId, DeadLetterId, ExecResponseSummary, ExecutionId, ExecutionLogSink, ExecutionSource, HttpDispatchPayload, InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult, - RequestId, Script, ScriptId, ScriptOwner, ScriptSandbox, TriggerEvent, + QueueMessageId, RequestId, Script, ScriptId, ScriptOwner, ScriptSandbox, TriggerEvent, + TriggerId, }; use rand::Rng; use uuid::Uuid; @@ -65,6 +66,10 @@ pub struct Dispatcher { /// v1.1.9. Reads `queue_messages` for the queue arm + the reclaim /// task. None in tests / harnesses that don't exercise queues. pub queue: Arc, + /// §11.6 D3. The group shared-queue store. A materialized consumer of a + /// SHARED group queue template (`consumer.shared_group.is_some()`) claims + /// from here instead of the per-app `queue`. + pub group_queue: Arc, pub config: TriggerConfig, /// Stable id for this dispatcher instance — written into /// `outbox.claimed_by` for forensics. In MVP this is the host's @@ -225,6 +230,7 @@ impl Dispatcher { // Reclaim task: independent cadence (default 30s) so it doesn't // contend with the per-100ms dispatcher tick. let reclaim_queue = self.queue.clone(); + let reclaim_group_queue = self.group_queue.clone(); let reclaim_interval = Duration::from_millis(u64::from(self.config.queue_reclaim_interval_ms)); tokio::spawn(async move { @@ -237,6 +243,14 @@ impl Dispatcher { Ok(n) => tracing::info!(reclaimed = n, "queue visibility-timeout reclaim"), Err(e) => tracing::warn!(?e, "queue reclaim task errored"), } + // §11.6 D3: the group shared-queue store has the same reclaim. + match reclaim_group_queue.reclaim_visibility_timeouts().await { + Ok(0) => {} + Ok(n) => { + tracing::info!(reclaimed = n, "group-queue visibility-timeout reclaim"); + } + Err(e) => tracing::warn!(?e, "group-queue reclaim task errored"), + } } }); @@ -313,6 +327,126 @@ impl Dispatcher { Ok(()) } + // §11.6 D3: route the four queue store operations to the per-app store or + // the group shared store, based on whether this consumer was materialized + // from a SHARED group queue template (`consumer.shared_group`). A group + // claim is normalized to a `ClaimedMessage` under the CONSUMING app so the + // rest of the queue arm (handler dispatch, event, logging) is unchanged. + async fn q_claim(&self, c: &ActiveQueueConsumer) -> Result, String> { + match c.shared_group { + Some(g) => Ok(self + .group_queue + .claim(g, &c.queue_name) + .await + .map_err(|e| e.to_string())? + .map(|m| ClaimedMessage { + id: m.id, + app_id: c.app_id, + queue_name: m.collection, + payload: m.payload, + enqueued_at: m.enqueued_at, + attempt: m.attempt, + max_attempts: m.max_attempts, + claim_token: m.claim_token, + })), + None => self + .queue + .claim(c.app_id, &c.queue_name) + .await + .map_err(|e| e.to_string()), + } + } + + async fn q_ack( + &self, + c: &ActiveQueueConsumer, + id: QueueMessageId, + token: uuid::Uuid, + ) -> Result { + match c.shared_group { + Some(_) => self + .group_queue + .ack(id, token) + .await + .map_err(|e| e.to_string()), + None => self.queue.ack(id, token).await.map_err(|e| e.to_string()), + } + } + + async fn q_nack( + &self, + c: &ActiveQueueConsumer, + id: QueueMessageId, + token: uuid::Uuid, + delay: chrono::Duration, + ) -> Result { + match c.shared_group { + Some(_) => self + .group_queue + .nack(id, token, delay) + .await + .map_err(|e| e.to_string()), + None => self + .queue + .nack(id, token, delay) + .await + .map_err(|e| e.to_string()), + } + } + + /// Terminal disposition of a message that can't be processed (script + /// missing / cross-app / exhausted). Per-app → dead-letter (+ the caller + /// fans out `dead_letter` triggers). Group shared queue → drop the row + /// (no group dead-letter store yet — documented D3 deferral) + warn. + /// Returns the dead-letter id only for the per-app path (drives fan-out). + async fn q_terminal( + &self, + c: &ActiveQueueConsumer, + claimed: &ClaimedMessage, + trigger_id: Option, + script_id: Option, + reason: &str, + ) -> Option { + match c.shared_group { + Some(_) => { + if let Err(e) = self + .group_queue + .drop_exhausted(claimed.id, claimed.claim_token) + .await + { + tracing::warn!(?e, "shared-queue drop failed"); + } + tracing::warn!( + reason, + queue = %claimed.queue_name, + "shared-queue message dropped (no group dead-letter store yet)" + ); + None + } + None => match self + .queue + .dead_letter( + claimed.id, + claimed.claim_token, + claimed.app_id, + &claimed.queue_name, + trigger_id, + script_id, + claimed.attempt, + claimed.enqueued_at, + reason, + ) + .await + { + Ok(dl_id) => Some(dl_id), + Err(e) => { + tracing::error!(?e, "queue dead-letter write failed"); + None + } + }, + } + } + #[allow(clippy::too_many_lines)] async fn dispatch_one_queue( &self, @@ -320,10 +454,9 @@ impl Dispatcher { ) -> Result<(), DispatcherError> { // Atomic claim — None → nothing pending right now for this queue. let Some(claimed) = self - .queue - .claim(consumer.app_id, &consumer.queue_name) + .q_claim(consumer) .await - .map_err(|e| DispatcherError::Outbox(e.to_string()))? + .map_err(DispatcherError::Outbox)? else { return Ok(()); }; @@ -333,8 +466,8 @@ impl Dispatcher { // outbox arm. let Ok(permit) = self.gate.try_acquire() else { let _ = self - .queue - .nack( + .q_nack( + consumer, claimed.id, claimed.claim_token, chrono::Duration::milliseconds(100), @@ -359,16 +492,11 @@ impl Dispatcher { Ok(None) => { tracing::warn!(script_id = %consumer.script_id, "queue trigger script missing; dead-lettering"); let _ = self - .queue - .dead_letter( - claimed.id, - claimed.claim_token, - claimed.app_id, - &claimed.queue_name, + .q_terminal( + consumer, + &claimed, Some(consumer.trigger_id), Some(consumer.script_id), - claimed.attempt, - claimed.enqueued_at, "queue trigger script not found", ) .await; @@ -398,16 +526,11 @@ impl Dispatcher { "queue consumer script belongs to a different app; dead-lettering" ); let _ = self - .queue - .dead_letter( - claimed.id, - claimed.claim_token, - claimed.app_id, - &claimed.queue_name, + .q_terminal( + consumer, + &claimed, Some(consumer.trigger_id), Some(consumer.script_id), - claimed.attempt, - claimed.enqueued_at, "queue consumer target belongs to a different app", ) .await; @@ -434,8 +557,8 @@ impl Dispatcher { "queue consumer script disabled at fire time; releasing claim" ); if let Err(e) = self - .queue - .nack( + .q_nack( + consumer, claimed.id, claimed.claim_token, chrono::Duration::seconds(1), @@ -515,7 +638,7 @@ impl Dispatcher { match outcome { Ok(_) => { // Auto-ack on success. - if let Err(e) = self.queue.ack(claimed.id, claimed.claim_token).await { + if let Err(e) = self.q_ack(consumer, claimed.id, claimed.claim_token).await { tracing::warn!(?e, "queue ack failed"); } } @@ -542,8 +665,7 @@ impl Dispatcher { ); let delay = chrono::Duration::milliseconds(i64::from(delay_ms)); if let Err(e) = self - .queue - .nack(claimed.id, claimed.claim_token, delay) + .q_nack(consumer, claimed.id, claimed.claim_token, delay) .await { tracing::warn!(?e, "queue nack failed"); @@ -558,27 +680,18 @@ impl Dispatcher { // same way here. let now = Utc::now(); let last_error = err.to_string(); - let dl_id = match self - .queue - .dead_letter( - claimed.id, - claimed.claim_token, - claimed.app_id, - &claimed.queue_name, + // Per-app → dead-letter (+ fan out below). Shared group queue → dropped + // inside q_terminal (no group dead-letter store yet), returns None so + // the fan-out is skipped. + let dl_id = self + .q_terminal( + consumer, + claimed, Some(consumer.trigger_id), Some(consumer.script_id), - claimed.attempt, - claimed.enqueued_at, &last_error, ) - .await - { - Ok(dl_id) => Some(dl_id), - Err(e) => { - tracing::error!(?e, "queue dead-letter write failed"); - None - } - }; + .await; if let Some(dead_letter_id) = dl_id { let original = TriggerEvent::Queue { queue_name: claimed.queue_name.clone(), @@ -1512,6 +1625,72 @@ fn apply_jitter(raw: u32, pct: u32) -> u32 { mod tests { use super::*; + /// §11.6 D3: a do-nothing group-queue store for dispatcher tests that don't + /// exercise SHARED queues (every consumer has `shared_group: None`, so these + /// methods are never reached — they just satisfy the struct field). + struct NoopGroupQueue; + + #[async_trait::async_trait] + impl crate::group_queue_repo::GroupQueueRepo for NoopGroupQueue { + async fn enqueue( + &self, + _msg: crate::group_queue_repo::NewGroupQueueMessage, + ) -> Result { + unreachable!("shared queue not exercised") + } + async fn claim( + &self, + _group_id: picloud_shared::GroupId, + _collection: &str, + ) -> Result< + Option, + crate::group_queue_repo::GroupQueueRepoError, + > { + Ok(None) + } + async fn ack( + &self, + _id: QueueMessageId, + _token: Uuid, + ) -> Result { + Ok(false) + } + async fn nack( + &self, + _id: QueueMessageId, + _token: Uuid, + _delay: chrono::Duration, + ) -> Result { + Ok(false) + } + async fn drop_exhausted( + &self, + _id: QueueMessageId, + _token: Uuid, + ) -> Result { + Ok(false) + } + async fn reclaim_visibility_timeouts( + &self, + ) -> Result { + Ok(0) + } + async fn depth( + &self, + _group_id: picloud_shared::GroupId, + _collection: &str, + ) -> Result { + Ok(0) + } + async fn depth_pending( + &self, + _group_id: picloud_shared::GroupId, + _collection: &str, + ) -> Result { + Ok(0) + } + } + #[test] fn exponential_backoff_doubles_per_attempt() { // No jitter (pct=0) for a deterministic check. @@ -1778,14 +1957,14 @@ mod tests { } async fn ack( &self, - _message_id: picloud_shared::QueueMessageId, + _message_id: QueueMessageId, _claim_token: Uuid, ) -> Result { unimplemented!("not used by this test") } async fn nack( &self, - _message_id: picloud_shared::QueueMessageId, + _message_id: QueueMessageId, _claim_token: Uuid, _retry_delay: chrono::Duration, ) -> Result { @@ -1820,7 +1999,7 @@ mod tests { #[allow(clippy::too_many_arguments)] async fn dead_letter( &self, - _message_id: picloud_shared::QueueMessageId, + _message_id: QueueMessageId, _claim_token: Uuid, _app_id: AppId, _queue_name: &str, @@ -2205,6 +2384,7 @@ mod tests { retry_backoff: BackoffShape::Exponential, retry_base_ms: 1000, registered_by_principal: AdminUserId::new(), + shared_group: None, }; let nacked = Arc::new(AtomicBool::new(false)); @@ -2229,6 +2409,7 @@ mod tests { claimed, nacked: nacked.clone(), }), + group_queue: Arc::new(NoopGroupQueue), config: TriggerConfig::from_env(), instance_id: "test-instance".into(), }; @@ -2310,14 +2491,14 @@ mod tests { } async fn ack( &self, - _message_id: picloud_shared::QueueMessageId, + _message_id: QueueMessageId, _claim_token: Uuid, ) -> Result { unimplemented!("not used by this test") } async fn nack( &self, - _message_id: picloud_shared::QueueMessageId, + _message_id: QueueMessageId, _claim_token: Uuid, _retry_delay: chrono::Duration, ) -> Result { @@ -2354,7 +2535,7 @@ mod tests { #[allow(clippy::too_many_arguments)] async fn dead_letter( &self, - _message_id: picloud_shared::QueueMessageId, + _message_id: QueueMessageId, _claim_token: Uuid, _app_id: AppId, _queue_name: &str, @@ -2461,6 +2642,7 @@ mod tests { log_sink: Arc::new(UnusedLogSink), inbox: Arc::new(UnusedInbox), queue: Arc::new(UnusedQueue), + group_queue: Arc::new(NoopGroupQueue), config: TriggerConfig::from_env(), instance_id: "test-instance".into(), }; diff --git a/crates/manager-core/src/materialize.rs b/crates/manager-core/src/materialize.rs index eba00bf..9f43339 100644 --- a/crates/manager-core/src/materialize.rs +++ b/crates/manager-core/src/materialize.rs @@ -44,6 +44,10 @@ struct ShouldRow { effective_app_id: Uuid, template_id: Uuid, kind: String, + /// §11.6 D3: a `shared` queue template's copies drain the GROUP store as + /// COMPETING consumers, so the one-consumer-per-(app, queue) slot check is + /// skipped for them (each descendant intentionally gets a consumer). + shared: bool, } /// Reconcile all materialized stateful-template copies. Idempotent; safe to call @@ -79,7 +83,7 @@ pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result Result 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" { + // §M5.4: a per-app queue copy would violate the one-consumer-per-(app_id, + // queue_name) invariant if the app already has a consumer on that queue — + // skip with a warning. §11.6 D3: a SHARED queue copy drains the group store + // 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 \ diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index a9f01c5..32448c3 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -337,6 +337,11 @@ pub struct ActiveQueueConsumer { pub retry_backoff: BackoffShape, pub retry_base_ms: u32, pub registered_by_principal: AdminUserId, + /// §11.6 D3: `Some(group)` when this consumer is a materialized copy of a + /// SHARED group queue template — it drains `group_queue_messages(group, + /// queue_name)` (competing consumers) instead of the per-app store. `None` + /// for an ordinary per-app or non-shared-template consumer. + pub shared_group: Option, } /// What the inbound-email webhook receiver needs to verify + dispatch a @@ -1792,13 +1797,19 @@ impl TriggerRepo for PostgresTriggerRepo { &self, ) -> Result, TriggerRepoError> { let rows: Vec = sqlx::query_as( + // §11.6 D3: LEFT JOIN the SOURCE template of a materialized copy; if + // that template is a SHARED group queue, `shared_group` is its + // owning group and this consumer drains the group store instead of + // the per-app one. "SELECT t.id AS trigger_id, t.app_id, t.script_id, d.queue_name, \ d.visibility_timeout_secs, \ t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \ - t.registered_by_principal \ + t.registered_by_principal, tmpl.group_id AS shared_group \ FROM triggers t \ JOIN queue_trigger_details d ON d.trigger_id = t.id \ JOIN scripts s ON s.id = t.script_id \ + LEFT JOIN triggers tmpl \ + ON tmpl.id = t.materialized_from AND tmpl.shared = TRUE \ WHERE t.kind = 'queue' AND t.enabled = TRUE AND s.enabled = TRUE \ AND t.app_id IS NOT NULL", ) @@ -1817,6 +1828,7 @@ impl TriggerRepo for PostgresTriggerRepo { .unwrap_or(BackoffShape::Exponential), retry_base_ms: u32::try_from(r.retry_base_ms).unwrap_or(1000), registered_by_principal: r.registered_by_principal.into(), + shared_group: r.shared_group.map(Into::into), }) .collect()) } @@ -2096,6 +2108,9 @@ struct QueueConsumerRow { retry_backoff: String, retry_base_ms: i32, registered_by_principal: Uuid, + // §11.6 D3: the owning group of a SHARED queue template this consumer was + // materialized from; NULL for a per-app / non-shared consumer. + shared_group: Option, } #[derive(sqlx::FromRow)] diff --git a/crates/manager-core/tests/group_queue.rs b/crates/manager-core/tests/group_queue.rs index 8820075..63cf35e 100644 --- a/crates/manager-core/tests/group_queue.rs +++ b/crates/manager-core/tests/group_queue.rs @@ -8,7 +8,7 @@ //! Deterministic: drives `GroupQueueRepo` directly (no dispatcher). Skips when //! `DATABASE_URL` is unset. -#![allow(clippy::too_many_lines)] +#![allow(clippy::too_many_lines, clippy::many_single_char_names)] use std::collections::HashSet; @@ -52,8 +52,8 @@ async fn competing_consumers_claim_each_message_exactly_once() { let repo = PostgresGroupQueueRepo::new(pool.clone()); // Enqueue 50 distinct messages into the shared `tasks` queue. - const N: usize = 50; - for i in 0..N { + let n_msgs: usize = 50; + for i in 0..n_msgs { repo.enqueue(NewGroupQueueMessage { group_id: group, collection: "tasks".into(), @@ -70,14 +70,9 @@ async fn competing_consumers_claim_each_message_exactly_once() { // Every claimed payload id must be unique — no message delivered twice. let claim_loop = |repo: PostgresGroupQueueRepo, group: GroupId| async move { let mut got: Vec = Vec::new(); - loop { - match repo.claim(group, "tasks").await.unwrap() { - Some(msg) => { - got.push(msg.payload["i"].as_i64().unwrap()); - assert!(repo.ack(msg.id, msg.claim_token).await.unwrap(), "ack ok"); - } - None => break, - } + while let Some(msg) = repo.claim(group, "tasks").await.unwrap() { + got.push(msg.payload["i"].as_i64().unwrap()); + assert!(repo.ack(msg.id, msg.claim_token).await.unwrap(), "ack ok"); } got }; @@ -96,10 +91,10 @@ async fn competing_consumers_claim_each_message_exactly_once() { let unique: HashSet = all.iter().copied().collect(); assert_eq!( all.len(), - N, + n_msgs, "every message delivered exactly once (no dupes)" ); - assert_eq!(unique.len(), N, "all N distinct ids covered"); + assert_eq!(unique.len(), n_msgs, "all distinct ids covered"); assert_eq!( repo.depth(group, "tasks").await.unwrap(), 0, diff --git a/crates/picloud-cli/src/cmds/pull.rs b/crates/picloud-cli/src/cmds/pull.rs index 2d84436..8bbd4ff 100644 --- a/crates/picloud-cli/src/cmds/pull.rs +++ b/crates/picloud-cli/src/cmds/pull.rs @@ -220,6 +220,8 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) -> visibility_timeout_secs: Some(d.visibility_timeout_secs), dispatch_mode, retry_max_attempts, + // app-owned triggers are never shared (group-only). + shared: false, }); } // `email` is skipped: the server stores the sealed secret value, diff --git a/crates/picloud-cli/src/manifest.rs b/crates/picloud-cli/src/manifest.rs index 9a23e34..1ccad2f 100644 --- a/crates/picloud-cli/src/manifest.rs +++ b/crates/picloud-cli/src/manifest.rs @@ -516,6 +516,11 @@ pub struct QueueTriggerSpec { pub dispatch_mode: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub retry_max_attempts: Option, + /// §11.6 D3: `true` for a shared-QUEUE group consumer over a declared + /// `kind = "queue"` collection — competing per-descendant consumers drain + /// one group-owned store. Group-only. + #[serde(default, skip_serializing_if = "is_false")] + pub shared: bool, } /// `[secrets] names = [...]` — declares which secrets the app expects. diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 6161e0f..08269ce 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -461,6 +461,7 @@ pub async fn build_app( log_sink: log_sink.clone(), inbox: inbox_resolver, queue: queue_repo.clone(), + group_queue: group_queue_repo.clone(), config: trigger_config, instance_id: format!("picloud-{}", std::process::id()), }