From 9ce3c4c7041ed8d32964f3acccd3bf0ad4b29d4d Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 6 Jun 2026 19:08:21 +0200 Subject: [PATCH] feat(v1.1.9): TriggerEvent::Queue + TriggerKind::Queue + types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the data-shape pieces for v1.1.9's queue surface; behavior lands in later commits. Each piece compiles independently. - shared/trigger_event.rs: TriggerEvent::Queue variant (queue_name, message, enqueued_at, attempt, message_id). source() returns "queue". Surfaced to scripts as ctx.event.queue with op = "receive". - manager-core/trigger_repo.rs: TriggerKind::Queue + TriggerDetails::Queue + CreateQueueTrigger + ActiveQueueConsumer + QueueDetailRow + QueueConsumerRow. PostgresTriggerRepo gains create_queue_trigger (advisory-lock-then-SELECT enforces one-consumer-per-queue), list_active_queue_consumers (dispatcher scan), touch_queue_trigger_last_fired_at. hydrate_one hydrates the Queue arm. - manager-core/outbox_repo.rs: OutboxSourceKind::Invoke for invoke_async() outbox rows. - manager-core/dispatcher.rs: placeholder OutboxSourceKind::Invoke arm (logs + drops the row) so the workspace compiles; real arm lands in commit 10. - manager-core/trigger_config.rs: queue_reclaim_interval_ms (30000) + queue_default_visibility_timeout_secs (30) env-overridable knobs. - executor-core/engine.rs: trigger_event_to_dynamic handles Queue → builds ctx.event.queue map. - manager-core/triggers_api.rs: in-memory mock TriggerRepo gains the three new methods (returns Default-ish values for tests). Unit tests: TriggerEvent::Queue serde round-trip, TriggerKind::Queue wire round-trip, advisory_lock_key stability per (app_id, queue_name) pair. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/executor-core/src/engine.rs | 18 ++ crates/manager-core/src/dispatcher.rs | 16 ++ crates/manager-core/src/outbox_repo.rs | 6 + crates/manager-core/src/trigger_config.rs | 22 ++ crates/manager-core/src/trigger_repo.rs | 292 ++++++++++++++++++++++ crates/manager-core/src/triggers_api.rs | 41 +++ crates/shared/src/trigger_event.rs | 58 +++++ 7 files changed, 453 insertions(+) diff --git a/crates/executor-core/src/engine.rs b/crates/executor-core/src/engine.rs index 6a3b448..aa23581 100644 --- a/crates/executor-core/src/engine.rs +++ b/crates/executor-core/src/engine.rs @@ -448,6 +448,24 @@ fn trigger_event_to_dynamic(event: &TriggerEvent) -> Dynamic { ps.insert("published_at".into(), published_at.to_rfc3339().into()); m.insert("pubsub".into(), ps.into()); } + TriggerEvent::Queue { + queue_name, + message, + enqueued_at, + attempt, + message_id, + } => { + // `ctx.event.op` is always "receive" for queue (the only op + // a queue:receive trigger surfaces). + m.insert("op".into(), "receive".into()); + let mut q = Map::new(); + q.insert("queue_name".into(), queue_name.clone().into()); + q.insert("message".into(), json_to_dynamic(message.clone())); + q.insert("enqueued_at".into(), enqueued_at.to_rfc3339().into()); + q.insert("attempt".into(), i64::from(*attempt).into()); + q.insert("message_id".into(), message_id.clone().into()); + m.insert("queue".into(), q.into()); + } TriggerEvent::Email { from, to, diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 535aa0b..25da513 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -163,6 +163,22 @@ impl Dispatcher { return Ok(()); } }, + OutboxSourceKind::Invoke => { + // Wired in commit 10 (invoke_async() — function composition). + // Until then a malformed row is impossible to write (no + // producer), but defensively drop the row so a manual + // INSERT doesn't loop the dispatcher. + tracing::warn!( + outbox_id = %row.id, + "OutboxSourceKind::Invoke not wired yet; dropping row" + ); + self.outbox + .delete(row.id) + .await + .map_err(|e| DispatcherError::Outbox(e.to_string()))?; + drop(permit); + return Ok(()); + } OutboxSourceKind::Kv | OutboxSourceKind::Docs | OutboxSourceKind::DeadLetter diff --git a/crates/manager-core/src/outbox_repo.rs b/crates/manager-core/src/outbox_repo.rs index 30e06a2..1810a45 100644 --- a/crates/manager-core/src/outbox_repo.rs +++ b/crates/manager-core/src/outbox_repo.rs @@ -33,6 +33,10 @@ pub enum OutboxSourceKind { Pubsub, /// v1.1.7. Inbound email POSTed to the webhook receiver. Email, + /// v1.1.9. Fire-and-forget function composition via + /// `invoke_async(target, args)`. The dispatcher fires once per row; + /// callers wrap in `retry::with` if they want retries. + Invoke, } impl OutboxSourceKind { @@ -47,6 +51,7 @@ impl OutboxSourceKind { Self::Files => "files", Self::Pubsub => "pubsub", Self::Email => "email", + Self::Invoke => "invoke", } } @@ -61,6 +66,7 @@ impl OutboxSourceKind { "files" => Some(Self::Files), "pubsub" => Some(Self::Pubsub), "email" => Some(Self::Email), + "invoke" => Some(Self::Invoke), _ => None, } } diff --git a/crates/manager-core/src/trigger_config.rs b/crates/manager-core/src/trigger_config.rs index 67b192a..8859afb 100644 --- a/crates/manager-core/src/trigger_config.rs +++ b/crates/manager-core/src/trigger_config.rs @@ -61,6 +61,18 @@ pub struct TriggerConfig { /// real-world cron precision is per-minute, so a 30s tick is fine. /// Floored at 1s by the scheduler. pub cron_tick_interval_ms: u32, + + /// Queue visibility-timeout reclaim task cadence, in ms (v1.1.9). + /// Default 30 000. The reclaim task clears claims on + /// `queue_messages` whose `claimed_at` exceeds the per-queue + /// `visibility_timeout_secs`, so a crashed consumer doesn't lose + /// the message. + pub queue_reclaim_interval_ms: u32, + + /// Default per-queue visibility-timeout in seconds (v1.1.9). Used + /// at queue-trigger creation when the request omits an explicit + /// value. Default 30. The per-trigger column overrides this. + pub queue_default_visibility_timeout_secs: u32, } impl TriggerConfig { @@ -75,6 +87,8 @@ impl TriggerConfig { dead_letter_retention_days: 30, abandoned_retention_days: 7, cron_tick_interval_ms: 30_000, + queue_reclaim_interval_ms: 30_000, + queue_default_visibility_timeout_secs: 30, } } @@ -101,6 +115,14 @@ impl TriggerConfig { &mut c.cron_tick_interval_ms, "PICLOUD_CRON_TICK_INTERVAL_MS", ); + load_u32( + &mut c.queue_reclaim_interval_ms, + "PICLOUD_QUEUE_RECLAIM_INTERVAL_MS", + ); + load_u32( + &mut c.queue_default_visibility_timeout_secs, + "PICLOUD_QUEUE_DEFAULT_VISIBILITY_TIMEOUT_SECS", + ); c } } diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index 8c75304..9dee676 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -59,6 +59,9 @@ pub enum TriggerKind { Pubsub, /// v1.1.7. Inbound email via the webhook receiver. Email, + /// v1.1.9. Durable queue consumer — exactly one consumer per + /// `(app_id, queue_name)`. + Queue, } impl TriggerKind { @@ -72,6 +75,7 @@ impl TriggerKind { Self::Files => "files", Self::Pubsub => "pubsub", Self::Email => "email", + Self::Queue => "queue", } } @@ -85,6 +89,7 @@ impl TriggerKind { "files" => Some(Self::Files), "pubsub" => Some(Self::Pubsub), "email" => Some(Self::Email), + "queue" => Some(Self::Queue), _ => None, } } @@ -145,6 +150,15 @@ pub enum TriggerDetails { /// surfaced (it's encrypted at rest); we expose only whether one is /// configured so the admin UI can show "signed" vs "unsigned". Email { has_inbound_secret: bool }, + /// v1.1.9. Queue consumer. The queue name is the per-queue grouping + /// key; visibility timeout is the per-trigger override; last_fired_at + /// is updated by the dispatcher when a message is claimed. + Queue { + queue_name: String, + visibility_timeout_secs: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + last_fired_at: Option>, + }, } /// Create payload for a KV trigger. Defaults applied at the admin @@ -251,6 +265,35 @@ pub struct CreateEmailTrigger { pub registered_by_principal: AdminUserId, } +/// Create payload for a queue:receive trigger (v1.1.9). One consumer +/// per `(app_id, queue_name)` is enforced inside `create_queue_trigger` +/// via a pg_advisory_xact_lock + SELECT. +#[derive(Debug, Clone)] +pub struct CreateQueueTrigger { + pub script_id: ScriptId, + pub queue_name: String, + pub visibility_timeout_secs: u32, + pub dispatch_mode: TriggerDispatchMode, + pub retry_max_attempts: u32, + pub retry_backoff: BackoffShape, + pub retry_base_ms: u32, + pub registered_by_principal: AdminUserId, +} + +/// One consumer the dispatcher's queue arm needs to scan messages for. +/// Pulled by `list_active_queue_consumers(app_id)` per tick. +#[derive(Debug, Clone)] +pub struct ActiveQueueConsumer { + pub trigger_id: TriggerId, + pub script_id: ScriptId, + pub queue_name: String, + pub visibility_timeout_secs: u32, + pub retry_max_attempts: u32, + pub retry_backoff: BackoffShape, + pub retry_base_ms: u32, + pub registered_by_principal: AdminUserId, +} + /// What the inbound-email webhook receiver needs to verify + dispatch a /// POST. Returned by `email_inbound_target`; `None` when the trigger /// doesn't exist or isn't `kind = 'email'`. @@ -412,6 +455,34 @@ pub trait TriggerRepo: Send + Sync { trigger_id: Option, script_id: Option, ) -> Result, TriggerRepoError>; + + /// v1.1.9. Create a queue:receive trigger. Enforces exactly one + /// consumer per `(app_id, queue_name)` via `pg_advisory_xact_lock` + /// + SELECT-then-INSERT (a partial unique index across the parent + /// app_id + detail queue_name can't be enforced in a single index). + /// Returns `TriggerRepoError::Invalid` if a consumer already exists. + async fn create_queue_trigger( + &self, + app_id: AppId, + req: CreateQueueTrigger, + ) -> Result; + + /// v1.1.9. Dispatcher's queue arm: list every enabled queue:receive + /// trigger across all apps. The dispatcher walks these per tick and + /// attempts a claim per queue. Bounded by the number of registered + /// consumers, which is small. + async fn list_active_queue_consumers( + &self, + ) -> Result, TriggerRepoError>; + + /// v1.1.9. Update `last_fired_at` on a queue trigger after a + /// successful claim. Best-effort — a failure is logged but doesn't + /// fail the dispatch. + async fn touch_queue_trigger_last_fired_at( + &self, + trigger_id: TriggerId, + at: DateTime, + ) -> Result<(), TriggerRepoError>; } // ---------------------------------------------------------------------------- @@ -1113,6 +1184,167 @@ impl TriggerRepo for PostgresTriggerRepo { }) .collect()) } + + async fn create_queue_trigger( + &self, + app_id: AppId, + req: CreateQueueTrigger, + ) -> Result { + if req.queue_name.is_empty() { + return Err(TriggerRepoError::Invalid( + "queue_name must not be empty".into(), + )); + } + if !(5..=3600).contains(&req.visibility_timeout_secs) { + return Err(TriggerRepoError::Invalid( + "visibility_timeout_secs must be in [5, 3600]".into(), + )); + } + + let mut tx = self.pool.begin().await?; + + // Per-(app_id, queue_name) advisory lock serializes concurrent + // creates so the SELECT-then-INSERT below is atomic. The lock + // is per-transaction (xact_lock) and automatically released on + // commit/rollback. + let lock_key = advisory_lock_key(app_id, &req.queue_name); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(&mut *tx) + .await?; + + // Reject duplicate consumer (one-per-queue invariant). + let existing: 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.into_inner()) + .bind(&req.queue_name) + .fetch_optional(&mut *tx) + .await?; + if existing.is_some() { + return Err(TriggerRepoError::Invalid(format!( + "queue '{}' already has a consumer trigger; remove the existing one first", + req.queue_name + ))); + } + + let parent: TriggerRow = 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 \ + ) VALUES ($1, $2, 'queue', TRUE, $3, $4, $5, $6, $7) \ + RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \ + retry_max_attempts, retry_backoff, retry_base_ms, \ + registered_by_principal, created_at, updated_at", + ) + .bind(app_id.into_inner()) + .bind(req.script_id.into_inner()) + .bind(req.dispatch_mode.as_str()) + .bind(i32::try_from(req.retry_max_attempts).unwrap_or(3)) + .bind(req.retry_backoff.as_str()) + .bind(i32::try_from(req.retry_base_ms).unwrap_or(1000)) + .bind(req.registered_by_principal.into_inner()) + .fetch_one(&mut *tx) + .await?; + + sqlx::query( + "INSERT INTO queue_trigger_details \ + (trigger_id, queue_name, visibility_timeout_secs) \ + VALUES ($1, $2, $3)", + ) + .bind(parent.id) + .bind(&req.queue_name) + .bind(i32::try_from(req.visibility_timeout_secs).unwrap_or(30)) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + + Ok(Trigger { + id: parent.id.into(), + app_id: parent.app_id.into(), + script_id: parent.script_id.into(), + kind: TriggerKind::Queue, + enabled: parent.enabled, + dispatch_mode: dispatch_from_str(&parent.dispatch_mode), + retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3), + retry_backoff: BackoffShape::from_wire(&parent.retry_backoff) + .unwrap_or(BackoffShape::Exponential), + retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000), + registered_by_principal: parent.registered_by_principal.into(), + created_at: parent.created_at, + updated_at: parent.updated_at, + details: TriggerDetails::Queue { + queue_name: req.queue_name, + visibility_timeout_secs: req.visibility_timeout_secs, + last_fired_at: None, + }, + }) + } + + async fn list_active_queue_consumers( + &self, + ) -> Result, TriggerRepoError> { + let rows: Vec = sqlx::query_as( + "SELECT t.id AS trigger_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 \ + FROM triggers t \ + JOIN queue_trigger_details d ON d.trigger_id = t.id \ + WHERE t.kind = 'queue' AND t.enabled = TRUE", + ) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|r| ActiveQueueConsumer { + trigger_id: r.trigger_id.into(), + script_id: r.script_id.into(), + queue_name: r.queue_name, + visibility_timeout_secs: u32::try_from(r.visibility_timeout_secs).unwrap_or(30), + retry_max_attempts: u32::try_from(r.retry_max_attempts).unwrap_or(3), + retry_backoff: BackoffShape::from_wire(&r.retry_backoff) + .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(), + }) + .collect()) + } + + async fn touch_queue_trigger_last_fired_at( + &self, + trigger_id: TriggerId, + at: DateTime, + ) -> Result<(), TriggerRepoError> { + sqlx::query( + "UPDATE queue_trigger_details SET last_fired_at = $2 WHERE trigger_id = $1", + ) + .bind(trigger_id.into_inner()) + .bind(at) + .execute(&self.pool) + .await?; + Ok(()) + } +} + +/// 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 { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + let mut h = DefaultHasher::new(); + app_id.into_inner().hash(&mut h); + queue_name.hash(&mut h); + // Postgres advisory locks key on bigint; reinterpret the u64 as i64. + #[allow(clippy::cast_possible_wrap)] + let key = h.finish() as i64; + key } #[allow(clippy::too_many_lines)] @@ -1223,6 +1455,20 @@ async fn hydrate_one(pool: &PgPool, parent: TriggerRow) -> Result { + let row: QueueDetailRow = sqlx::query_as( + "SELECT queue_name, visibility_timeout_secs, last_fired_at \ + FROM queue_trigger_details WHERE trigger_id = $1", + ) + .bind(parent.id) + .fetch_one(pool) + .await?; + TriggerDetails::Queue { + queue_name: row.queue_name, + visibility_timeout_secs: u32::try_from(row.visibility_timeout_secs).unwrap_or(30), + last_fired_at: row.last_fired_at, + } + } }; Ok(Trigger { @@ -1305,6 +1551,25 @@ struct EmailDetailRow { inbound_secret_encrypted: Option>, } +#[derive(sqlx::FromRow)] +struct QueueDetailRow { + queue_name: String, + visibility_timeout_secs: i32, + last_fired_at: Option>, +} + +#[derive(sqlx::FromRow)] +struct QueueConsumerRow { + trigger_id: Uuid, + script_id: Uuid, + queue_name: String, + visibility_timeout_secs: i32, + retry_max_attempts: i32, + retry_backoff: String, + retry_base_ms: i32, + registered_by_principal: Uuid, +} + #[derive(sqlx::FromRow)] struct EmailInboundRow { app_id: Uuid, @@ -1365,4 +1630,31 @@ mod tests { assert!(collection_matches("widgets", "widgets")); assert!(!collection_matches("widgets", "Widgets")); } + + #[test] + fn trigger_kind_queue_round_trips_through_wire() { + assert_eq!(TriggerKind::Queue.as_str(), "queue"); + assert_eq!(TriggerKind::from_wire("queue"), Some(TriggerKind::Queue)); + } + + #[test] + fn advisory_lock_key_is_stable_per_app_queue_pair() { + let app_id_a: AppId = uuid::Uuid::nil().into(); + let app_id_b: AppId = uuid::Uuid::from_u128(1).into(); + // Same inputs → same key. + assert_eq!( + advisory_lock_key(app_id_a, "jobs"), + advisory_lock_key(app_id_a, "jobs") + ); + // Different app → different key (extremely high probability). + assert_ne!( + advisory_lock_key(app_id_a, "jobs"), + advisory_lock_key(app_id_b, "jobs") + ); + // Different queue → different key. + assert_ne!( + advisory_lock_key(app_id_a, "jobs"), + advisory_lock_key(app_id_a, "other") + ); + } } diff --git a/crates/manager-core/src/triggers_api.rs b/crates/manager-core/src/triggers_api.rs index f580316..ae15c6e 100644 --- a/crates/manager-core/src/triggers_api.rs +++ b/crates/manager-core/src/triggers_api.rs @@ -940,6 +940,47 @@ mod tests { ) -> Result, TriggerRepoError> { Ok(vec![]) } + async fn create_queue_trigger( + &self, + app_id: AppId, + req: crate::trigger_repo::CreateQueueTrigger, + ) -> Result { + let now = Utc::now(); + let id = TriggerId::new(); + let trigger = Trigger { + id, + app_id, + script_id: req.script_id, + kind: TriggerKind::Queue, + enabled: true, + dispatch_mode: req.dispatch_mode, + retry_max_attempts: req.retry_max_attempts, + retry_backoff: req.retry_backoff, + retry_base_ms: req.retry_base_ms, + registered_by_principal: req.registered_by_principal, + created_at: now, + updated_at: now, + details: TriggerDetails::Queue { + queue_name: req.queue_name, + visibility_timeout_secs: req.visibility_timeout_secs, + last_fired_at: None, + }, + }; + self.inner.lock().await.insert(id, trigger.clone()); + Ok(trigger) + } + async fn list_active_queue_consumers( + &self, + ) -> Result, TriggerRepoError> { + Ok(vec![]) + } + async fn touch_queue_trigger_last_fired_at( + &self, + _trigger_id: TriggerId, + _at: chrono::DateTime, + ) -> Result<(), TriggerRepoError> { + Ok(()) + } } struct InMemoryAppRepo { diff --git a/crates/shared/src/trigger_event.rs b/crates/shared/src/trigger_event.rs index 4f0d24e..52e2fa6 100644 --- a/crates/shared/src/trigger_event.rs +++ b/crates/shared/src/trigger_event.rs @@ -207,6 +207,17 @@ pub enum TriggerEvent { message_id: Option, }, + /// A queue:receive trigger fired this handler. v1.1.9. Surfaced to + /// scripts as `ctx.event.queue`. `attempt` is 1-based — it's the + /// current delivery attempt (increments per nack-and-retry). + Queue { + queue_name: String, + message: serde_json::Value, + enqueued_at: DateTime, + attempt: u32, + message_id: String, + }, + /// A dead-letter row fired this handler. The original event is /// nested verbatim plus the dead-letter metadata the design notes /// §4 require. @@ -235,6 +246,7 @@ impl TriggerEvent { Self::Files { .. } => "files", Self::Pubsub { .. } => "pubsub", Self::Email { .. } => "email", + Self::Queue { .. } => "queue", Self::DeadLetter { .. } => "dead_letter", } } @@ -254,3 +266,49 @@ pub struct DeadLetterEventDetail { pub first_attempt_at: DateTime, pub last_attempt_at: DateTime, } + +#[cfg(test)] +mod queue_event_tests { + use super::*; + + #[test] + fn queue_event_round_trip_serde() { + let original = TriggerEvent::Queue { + queue_name: "payments.process".into(), + message: serde_json::json!({ "order_id": 123, "amount": 4999 }), + enqueued_at: DateTime::parse_from_rfc3339("2026-06-06T12:00:00Z") + .unwrap() + .with_timezone(&Utc), + attempt: 2, + message_id: "01234567-89ab-cdef-0123-456789abcdef".into(), + }; + let wire = serde_json::to_value(&original).unwrap(); + assert_eq!(wire["source"], "queue"); + assert_eq!(wire["queue_name"], "payments.process"); + assert_eq!(wire["attempt"], 2); + let back: TriggerEvent = serde_json::from_value(wire).unwrap(); + match back { + TriggerEvent::Queue { + queue_name, + attempt, + .. + } => { + assert_eq!(queue_name, "payments.process"); + assert_eq!(attempt, 2); + } + other => panic!("expected Queue, got {other:?}"), + } + } + + #[test] + fn queue_source_discriminant() { + let event = TriggerEvent::Queue { + queue_name: "x".into(), + message: serde_json::Value::Null, + enqueued_at: Utc::now(), + attempt: 1, + message_id: "id".into(), + }; + assert_eq!(event.source(), "queue"); + } +}