diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 25da513..835965b 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -37,9 +37,10 @@ use crate::abandoned_repo::{AbandonedRepo, NewAbandonedExecution}; use crate::dead_letter_repo::{DeadLetterRepo, NewDeadLetter}; use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow, OutboxSourceKind}; use crate::principal_resolver::PrincipalResolver; +use crate::queue_repo::{ClaimedMessage, QueueRepo}; use crate::repo::ScriptRepository; use crate::trigger_config::{BackoffShape, TriggerConfig}; -use crate::trigger_repo::{TriggerKind, TriggerRepo}; +use crate::trigger_repo::{ActiveQueueConsumer, TriggerKind, TriggerRepo}; /// Bundle the dispatcher reads from. Each handle is `Arc` so /// tests can substitute in-memory backings. @@ -53,6 +54,9 @@ pub struct Dispatcher { pub executor: Arc, pub gate: Arc, pub inbox: Arc, + /// 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, pub config: TriggerConfig, /// Stable id for this dispatcher instance — written into /// `outbox.claimed_by` for forensics. In MVP this is the host's @@ -76,10 +80,29 @@ const TICK_INTERVAL: Duration = Duration::from_millis(100); const ASYNC_EXEC_TIMEOUT: Duration = Duration::from_secs(300); impl Dispatcher { - /// Spawn the dispatcher loop as a detached `tokio::task`. The - /// returned `JoinHandle` is dropped — the loop runs for the - /// process lifetime. + /// Spawn the dispatcher loop as a detached `tokio::task`. Also + /// spawns the v1.1.9 queue visibility-timeout reclaim task. Both + /// run for the process lifetime; returned `JoinHandle`s are + /// dropped on purpose. pub fn spawn(self) { + // 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_interval = + Duration::from_millis(u64::from(self.config.queue_reclaim_interval_ms)); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(reclaim_interval); + ticker.tick().await; + loop { + ticker.tick().await; + match reclaim_queue.reclaim_visibility_timeouts().await { + Ok(0) => {} + Ok(n) => tracing::info!(reclaimed = n, "queue visibility-timeout reclaim"), + Err(e) => tracing::warn!(?e, "queue reclaim task errored"), + } + } + }); + tokio::spawn(async move { self.run().await; }); @@ -98,16 +121,12 @@ impl Dispatcher { } async fn tick(&self) -> Result<(), DispatcherError> { - // Cheap gate sample so we don't claim rows we can't dispatch. - // The exact permit budget is reapplied per-row below. + // 1) Outbox arm — KV/docs/pubsub/cron/email/files/dead-letter/http/invoke. let rows = self .outbox .claim_due(&self.instance_id, CLAIM_BATCH) .await .map_err(|e| DispatcherError::Outbox(e.to_string()))?; - if rows.is_empty() { - return Ok(()); - } for row in rows { // Process serially within a tick — the outer ticker is the // pacing mechanism. Concurrent dispatchers are a cluster- @@ -116,9 +135,199 @@ impl Dispatcher { tracing::warn!(?err, "dispatch one errored"); } } + + // 2) Queue arm (v1.1.9). One claim per active consumer per tick; + // bounded by the number of registered consumers so a flood in + // one queue doesn't starve the rest. + if let Err(err) = self.tick_queue_arm().await { + tracing::warn!(?err, "queue arm tick errored"); + } Ok(()) } + /// v1.1.9. Per tick: list every `queue:receive` consumer and + /// attempt one claim per `(app_id, queue_name)`. Each successful + /// claim runs through the same executor path as outbox-backed + /// trigger dispatches, so retry / dead-letter wiring is shared. + async fn tick_queue_arm(&self) -> Result<(), DispatcherError> { + let consumers = self + .triggers + .list_active_queue_consumers() + .await + .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?; + for consumer in consumers { + if let Err(err) = self.dispatch_one_queue(&consumer).await { + tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored"); + } + } + Ok(()) + } + + async fn dispatch_one_queue( + &self, + consumer: &ActiveQueueConsumer, + ) -> Result<(), DispatcherError> { + // Atomic claim — None → nothing pending right now for this queue. + let Some(claimed) = self + .queue + .claim(consumer.app_id, &consumer.queue_name) + .await + .map_err(|e| DispatcherError::Outbox(e.to_string()))? + else { + return Ok(()); + }; + + // Gate admission. If saturated, release the claim by nack- + // immediate (which lets the next tick re-claim). Mirrors the + // outbox arm. + let Ok(permit) = self.gate.try_acquire() else { + let _ = self + .queue + .nack(claimed.id, claimed.claim_token, chrono::Duration::milliseconds(100)) + .await; + return Ok(()); + }; + + // Build the TriggerEvent and ExecRequest. Mirrors build_exec_request + // but uses the consumer + claimed-message data instead of an + // outbox row. + let event = TriggerEvent::Queue { + queue_name: claimed.queue_name.clone(), + message: claimed.payload.clone(), + enqueued_at: claimed.enqueued_at, + attempt: claimed.attempt, + message_id: claimed.id.to_string(), + }; + + let script = match self.scripts.get(consumer.script_id).await { + Ok(Some(s)) => s, + 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, + Some(consumer.trigger_id), + Some(consumer.script_id), + claimed.attempt, + claimed.enqueued_at, + "queue trigger script not found", + ) + .await; + drop(permit); + return Ok(()); + } + Err(e) => return Err(DispatcherError::ResolveTrigger(e.to_string())), + }; + + let principal = self + .principals + .resolve(consumer.registered_by_principal) + .await + .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?; + + let execution_id = ExecutionId::new(); + let exec_req = ExecRequest { + execution_id, + request_id: RequestId::new(), + script_id: consumer.script_id, + script_name: script.name.clone(), + invocation_type: InvocationType::Function, + path: "/trigger/queue".into(), + headers: std::collections::BTreeMap::new(), + body: serde_json::Value::Null, + params: std::collections::BTreeMap::new(), + query: std::collections::BTreeMap::new(), + rest: String::new(), + sandbox_overrides: script.sandbox, + app_id: claimed.app_id, + principal: Some(principal), + trigger_depth: 0, + root_execution_id: execution_id, + is_dead_letter_handler: false, + event: Some(event), + }; + let identity = picloud_orchestrator_core::ScriptIdentity { + script_id: consumer.script_id, + updated_at: script.updated_at, + }; + let outcome = self + .executor + .execute_with_identity(identity, &script.source, exec_req, ASYNC_EXEC_TIMEOUT) + .await; + drop(permit); + + // Best-effort touch on last_fired_at (a failure here doesn't + // change ack/nack behavior). + if let Err(e) = self + .triggers + .touch_queue_trigger_last_fired_at(consumer.trigger_id, Utc::now()) + .await + { + tracing::debug!(?e, "touch last_fired_at failed"); + } + + match outcome { + Ok(_) => { + // Auto-ack on success. + if let Err(e) = self.queue.ack(claimed.id, claimed.claim_token).await { + tracing::warn!(?e, "queue ack failed"); + } + } + Err(err) => { + self.handle_queue_failure(consumer, &claimed, err).await; + } + } + Ok(()) + } + + async fn handle_queue_failure( + &self, + consumer: &ActiveQueueConsumer, + claimed: &ClaimedMessage, + err: ExecError, + ) { + if claimed.attempt < claimed.max_attempts { + // Retry: clear claim + set deliver_after. + let delay_ms = compute_backoff( + claimed.attempt, + consumer.retry_backoff, + consumer.retry_base_ms, + self.config.retry_jitter_pct, + ); + let delay = chrono::Duration::milliseconds(i64::from(delay_ms)); + if let Err(e) = self.queue.nack(claimed.id, claimed.claim_token, delay).await { + tracing::warn!(?e, "queue nack failed"); + } + return; + } + // Exhausted. Dead-letter inline (the helper deletes the queue row + + // writes a dead_letters row in one transaction). The existing + // fan_out_dead_letter path on the dispatcher's outbox arm fires + // registered dead_letter handlers off the new row. + if let Err(e) = self + .queue + .dead_letter( + claimed.id, + claimed.claim_token, + claimed.app_id, + &claimed.queue_name, + Some(consumer.trigger_id), + Some(consumer.script_id), + claimed.attempt, + claimed.enqueued_at, + &err.to_string(), + ) + .await + { + tracing::error!(?e, "queue dead-letter write failed"); + } + // TODO(metrics): bump picloud_queue_dead_letters_total{app_id, queue_name}. + } + async fn dispatch_one(&self, row: OutboxRow) -> Result<(), DispatcherError> { // Depth-limit check — design notes §4: loops aren't DL'd. if row.trigger_depth > self.config.max_trigger_depth { diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index 9dee676..e678d21 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -281,10 +281,11 @@ pub struct CreateQueueTrigger { } /// One consumer the dispatcher's queue arm needs to scan messages for. -/// Pulled by `list_active_queue_consumers(app_id)` per tick. +/// Pulled by `list_active_queue_consumers()` per tick. #[derive(Debug, Clone)] pub struct ActiveQueueConsumer { pub trigger_id: TriggerId, + pub app_id: AppId, pub script_id: ScriptId, pub queue_name: String, pub visibility_timeout_secs: u32, @@ -1290,7 +1291,7 @@ impl TriggerRepo for PostgresTriggerRepo { &self, ) -> Result, TriggerRepoError> { let rows: Vec = sqlx::query_as( - "SELECT t.id AS trigger_id, t.script_id, d.queue_name, \ + "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 \ @@ -1304,6 +1305,7 @@ impl TriggerRepo for PostgresTriggerRepo { .into_iter() .map(|r| ActiveQueueConsumer { trigger_id: r.trigger_id.into(), + app_id: r.app_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), @@ -1561,6 +1563,7 @@ struct QueueDetailRow { #[derive(sqlx::FromRow)] struct QueueConsumerRow { trigger_id: Uuid, + app_id: Uuid, script_id: Uuid, queue_name: String, visibility_timeout_secs: i32, diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 74cbefd..ad701c5 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -344,6 +344,7 @@ pub async fn build_app( executor: executor.clone(), gate, inbox: inbox_resolver, + queue: queue_repo.clone(), config: trigger_config, instance_id: format!("picloud-{}", std::process::id()), }