c859c9aab2
Chasing the e2e flakiness turned up a production durability bug, not a test bug. The dispatcher claims an outbox row, executes it, then either deletes it (success) or reschedules it (failure) — both of which clear the claim. If the PROCESS DIES in between, neither runs. And `claim_due` only ever selects `claimed_at IS NULL`. Nothing else in the codebase clears `outbox.claimed_at` — grep it: there are exactly three writers, and those are two of them. So a crash or restart mid-dispatch stranded every in-flight row PERMANENTLY. Its trigger never fired and no retry could notice. The outbox is the universal trigger path, so the loss covered kv / docs / files / cron / pubsub / email / invoke_async / dead-letter alike. This is the same durability class as the audit's #6 (lost trigger event), which was just fixed at the WRITE end — this is the same hole at the READ end. Every other claim-based store already had the safety net: `queue_messages` and `group_queue_messages` have `reclaim_visibility_timeouts`, `workflow_steps` has its own reclaim. The outbox was the one that didn't. `OutboxRepo::reclaim_stale_claims(timeout)` returns rows whose claim is older than `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` (default 600s), run from the dispatcher's existing reclaim ticker. The default is deliberately generous — a script may run for up to 300s (the `scripts.timeout_seconds` CHECK), so a claim held past twice that is abandoned rather than slow; reclaiming a row a LIVE dispatcher is still working on would double-execute it (survivable — dispatch is at-least-once — but not worth courting). A reclaim does NOT bump `attempt_count`: the handler never ran, so it must not consume the row's retry budget, or repeated restarts would dead-letter an event that executed zero times. (Same reasoning as the transient queue `release` fixed earlier in this branch.) This is also the root cause of the flaky e2e suites: each test drops its dispatcher, and `claim_due` is not scoped per app or per dispatcher, so a test's dispatcher could claim another test's row and strand it on teardown. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2790 lines
110 KiB
Rust
2790 lines
110 KiB
Rust
//! The triggers-framework dispatcher.
|
||
//!
|
||
//! Single tokio task that polls the outbox, claims due rows
|
||
//! (`FOR UPDATE SKIP LOCKED`), and routes each to the executor.
|
||
//! Shares the `ExecutionGate` with sync HTTP — they compete for the
|
||
//! same permit budget, matching design notes §2.
|
||
//!
|
||
//! Outcome handling per design notes §3 and §4:
|
||
//! - reply_to.is_some() (sync HTTP): never retry. Deliver to inbox
|
||
//! (or write `abandoned_executions` if the receiver dropped).
|
||
//! - is_dead_letter_handler == true: never retry, never DL. Failure
|
||
//! just annotates the original DL row with `resolution =
|
||
//! 'handler_failed'` and bumps a metric.
|
||
//! - Otherwise on failure: if `attempt_count + 1 < max_attempts`,
|
||
//! reschedule with backoff + jitter. Else, write a `dead_letters`
|
||
//! row and delete from outbox.
|
||
//!
|
||
//! Depth-limit: `trigger_depth > max_trigger_depth` skips execution
|
||
//! entirely (log + metric) and deletes the row — does NOT dead-letter
|
||
//! (design notes §4: depth-exceeded means "you built a loop", and
|
||
//! dead-lettering would just re-fire the same loop).
|
||
|
||
use std::sync::Arc;
|
||
use std::time::Duration;
|
||
|
||
use chrono::{DateTime, Utc};
|
||
use picloud_executor_core::{
|
||
build_execution_log, ExecError, ExecRequest, ExecResponse, InvocationType,
|
||
};
|
||
use picloud_orchestrator_core::{ExecutionGate, ExecutorClient};
|
||
use picloud_shared::{
|
||
AppId, DeadLetterId, ExecResponseSummary, ExecutionId, ExecutionLogSink, ExecutionSource,
|
||
HttpDispatchPayload, InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult,
|
||
QueueMessageId, RequestId, Script, ScriptId, ScriptOwner, ScriptSandbox, TriggerEvent,
|
||
TriggerId,
|
||
};
|
||
use rand::Rng;
|
||
use uuid::Uuid;
|
||
|
||
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::{ActiveQueueConsumer, TriggerKind, TriggerRepo};
|
||
|
||
/// Bundle the dispatcher reads from. Each handle is `Arc<dyn …>` so
|
||
/// tests can substitute in-memory backings.
|
||
pub struct Dispatcher {
|
||
pub outbox: Arc<dyn OutboxRepo>,
|
||
pub triggers: Arc<dyn TriggerRepo>,
|
||
pub scripts: Arc<dyn ScriptRepository>,
|
||
pub dead_letters: Arc<dyn DeadLetterRepo>,
|
||
pub abandoned: Arc<dyn AbandonedRepo>,
|
||
pub principals: Arc<dyn PrincipalResolver>,
|
||
pub executor: Arc<dyn ExecutorClient>,
|
||
pub gate: Arc<ExecutionGate>,
|
||
/// G1: records an `execution_logs` row for every trigger run so
|
||
/// background workers (queue / cron / dead-letter / invoke) show up
|
||
/// in `pic logs`, not just synchronous HTTP. Same sink the
|
||
/// orchestrator's data plane writes through.
|
||
pub log_sink: Arc<dyn ExecutionLogSink>,
|
||
pub inbox: Arc<dyn InboxResolver>,
|
||
/// 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<dyn QueueRepo>,
|
||
/// §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<dyn crate::group_queue_repo::GroupQueueRepo>,
|
||
pub config: TriggerConfig,
|
||
/// Stable id for this dispatcher instance — written into
|
||
/// `outbox.claimed_by` for forensics. In MVP this is the host's
|
||
/// pid; cluster mode (v1.3+) uses node identity.
|
||
pub instance_id: String,
|
||
}
|
||
|
||
/// How many outbox rows the dispatcher tries to claim per tick.
|
||
/// Bounded to keep the working set small even if there's a flood.
|
||
const CLAIM_BATCH: i64 = 8;
|
||
|
||
/// Inputs to `Dispatcher::fan_out_dead_letter`. Pulled out as a struct
|
||
/// so the outbox arm (`handle_failure`) and the queue arm
|
||
/// (`handle_queue_failure`) can both call it — the queue arm doesn't
|
||
/// have an `OutboxRow` to pass.
|
||
struct DeadLetterFanOutCtx {
|
||
app_id: picloud_shared::AppId,
|
||
/// The originating event being failed, verbatim. The DL event
|
||
/// nests this under `original` so handler scripts see what
|
||
/// triggered the chain.
|
||
original: TriggerEvent,
|
||
/// Trigger source the DL row was filed under (`"kv"`, `"queue"`,
|
||
/// `"http"`, …). Drives the `source_filter` match on registered
|
||
/// `dead_letter` triggers.
|
||
source: String,
|
||
dead_letter_id: DeadLetterId,
|
||
attempts: u32,
|
||
last_error: String,
|
||
/// Original trigger id (for trigger_id_filter matching).
|
||
trigger_id: Option<picloud_shared::TriggerId>,
|
||
/// Original script id (for script_id_filter matching).
|
||
script_id: Option<ScriptId>,
|
||
first_attempt_at: DateTime<Utc>,
|
||
last_attempt_at: DateTime<Utc>,
|
||
trigger_depth: u32,
|
||
root_execution_id: Option<ExecutionId>,
|
||
}
|
||
|
||
/// Polling cadence. Short enough that fan-out feels instant; long
|
||
/// enough that an idle dispatcher doesn't burn cycles.
|
||
/// F-Q-009: env-overridable via `PICLOUD_DISPATCHER_TICK_INTERVAL_MS`.
|
||
const DEFAULT_TICK_INTERVAL: Duration = Duration::from_millis(100);
|
||
|
||
/// Hard cap on the wall-clock budget passed to the executor for an
|
||
/// async-dispatched script. Sync HTTP gets a per-script timeout via
|
||
/// the orchestrator path; async rows don't have one, so we apply a
|
||
/// platform-wide ceiling here. Matches `LocalExecutorClient`'s own
|
||
/// 5-minute cap.
|
||
/// F-Q-009: env-overridable via `PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC`.
|
||
///
|
||
/// `pub` so `triggers_api::validate_queue_visibility_timeout` can use
|
||
/// the same default for its warn-threshold — otherwise a change here
|
||
/// would silently leave the visibility-timeout warner out of sync.
|
||
pub const DEFAULT_ASYNC_EXEC_TIMEOUT_SECS: u32 = 300;
|
||
const DEFAULT_ASYNC_EXEC_TIMEOUT: Duration =
|
||
Duration::from_secs(DEFAULT_ASYNC_EXEC_TIMEOUT_SECS as u64);
|
||
|
||
/// Read `PICLOUD_DISPATCHER_TICK_INTERVAL_MS`; invalid values fall back
|
||
/// to the documented default with a tracing-warn.
|
||
fn tick_interval_from_env() -> Duration {
|
||
if let Ok(v) = std::env::var("PICLOUD_DISPATCHER_TICK_INTERVAL_MS") {
|
||
match v.trim().parse::<u64>() {
|
||
Ok(n) if n > 0 => return Duration::from_millis(n),
|
||
_ => tracing::warn!(
|
||
value = %v,
|
||
"ignoring invalid PICLOUD_DISPATCHER_TICK_INTERVAL_MS (want a positive integer)"
|
||
),
|
||
}
|
||
}
|
||
DEFAULT_TICK_INTERVAL
|
||
}
|
||
|
||
/// Read `PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC`; invalid values
|
||
/// fall back to the documented default with a tracing-warn.
|
||
fn async_exec_timeout_from_env() -> Duration {
|
||
if let Ok(v) = std::env::var("PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC") {
|
||
match v.trim().parse::<u64>() {
|
||
Ok(n) if n > 0 => return Duration::from_secs(n),
|
||
_ => tracing::warn!(
|
||
value = %v,
|
||
"ignoring invalid PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC (want a positive integer)"
|
||
),
|
||
}
|
||
}
|
||
DEFAULT_ASYNC_EXEC_TIMEOUT
|
||
}
|
||
|
||
/// F-P-007: per-tick concurrency cap for queue dispatches. Matches the
|
||
/// default execution-gate concurrency so the gate is what bounds real
|
||
/// parallelism, not the dispatcher's serial-await.
|
||
const QUEUE_DISPATCH_PARALLELISM: usize = 32;
|
||
|
||
/// Map an outbox row's source kind to the execution-log `source`. The
|
||
/// wire strings are identical, so this is total in practice; an unknown
|
||
/// kind would default to `Http`.
|
||
fn exec_source(row: &OutboxRow) -> ExecutionSource {
|
||
ExecutionSource::from_wire(row.source_kind.as_str()).unwrap_or_default()
|
||
}
|
||
|
||
/// G1: the slice of an `ExecRequest` needed to write an execution-log
|
||
/// row, captured before the request is moved into the executor.
|
||
struct ExecLogContext {
|
||
app_id: picloud_shared::AppId,
|
||
script_id: ScriptId,
|
||
request_id: RequestId,
|
||
path: String,
|
||
headers: std::collections::BTreeMap<String, String>,
|
||
body: serde_json::Value,
|
||
source: ExecutionSource,
|
||
}
|
||
|
||
impl ExecLogContext {
|
||
fn from_request(req: &ExecRequest, source: ExecutionSource) -> Self {
|
||
Self {
|
||
app_id: req.app_id,
|
||
script_id: req.script_id,
|
||
request_id: req.request_id,
|
||
path: req.path.clone(),
|
||
headers: req.headers.clone(),
|
||
body: req.body.clone(),
|
||
source,
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Dispatcher {
|
||
/// Phase 4 isolation backstop: may `script` run under `app_id`'s context?
|
||
/// True when the script is owned by `app_id` directly, or by a group on
|
||
/// its ancestor chain (an inherited group script). App-owned is the
|
||
/// in-memory fast path — only a group-owned script pays the chain query,
|
||
/// and a query error fails closed (not invocable). This generalizes the
|
||
/// pre-Phase-4 same-app guard (`script.app_id == row.app_id`) to the
|
||
/// hierarchy without changing behavior for app-owned scripts.
|
||
async fn script_invocable(&self, script: &Script, app_id: AppId) -> bool {
|
||
if script.is_owned_by_app(app_id) {
|
||
return true;
|
||
}
|
||
if script.group_id.is_none() {
|
||
return false;
|
||
}
|
||
match self.scripts.is_invocable_by_app(script.id, app_id).await {
|
||
Ok(ok) => ok,
|
||
Err(e) => {
|
||
tracing::error!(
|
||
error = %e, script_id = %script.id, %app_id,
|
||
"chain-membership check failed; treating script as not invocable"
|
||
);
|
||
false
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 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_group_queue = self.group_queue.clone();
|
||
let reclaim_outbox = self.outbox.clone();
|
||
let outbox_claim_timeout = self.config.outbox_claim_timeout_secs;
|
||
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;
|
||
// A dispatcher that died mid-dispatch left its claimed rows
|
||
// stranded — `claim_due` only takes unclaimed rows, so without
|
||
// this they would never fire again.
|
||
match reclaim_outbox
|
||
.reclaim_stale_claims(outbox_claim_timeout)
|
||
.await
|
||
{
|
||
Ok(0) => {}
|
||
Ok(n) => tracing::warn!(
|
||
reclaimed = n,
|
||
timeout_secs = outbox_claim_timeout,
|
||
"reclaimed stale outbox claims — a dispatcher died mid-dispatch"
|
||
),
|
||
Err(e) => tracing::warn!(?e, "outbox reclaim task errored"),
|
||
}
|
||
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"),
|
||
}
|
||
// §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"),
|
||
}
|
||
}
|
||
});
|
||
|
||
tokio::spawn(async move {
|
||
self.run().await;
|
||
});
|
||
}
|
||
|
||
async fn run(self) {
|
||
let mut ticker = tokio::time::interval(tick_interval_from_env());
|
||
// Skip the immediate first fire so we don't race startup.
|
||
ticker.tick().await;
|
||
loop {
|
||
ticker.tick().await;
|
||
if let Err(err) = self.tick().await {
|
||
tracing::warn!(?err, "dispatcher tick errored");
|
||
}
|
||
}
|
||
}
|
||
|
||
async fn tick(&self) -> Result<(), DispatcherError> {
|
||
// 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()))?;
|
||
for row in rows {
|
||
// Process serially within a tick — the outer ticker is the
|
||
// pacing mechanism. Concurrent dispatchers are a cluster-
|
||
// mode concern; v1.1.1 MVP has one.
|
||
if let Err(err) = self.dispatch_one(row).await {
|
||
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.
|
||
///
|
||
/// F-P-007: bound concurrency with `for_each_concurrent` instead of
|
||
/// serial-await. With N consumers and a 100ms tick the worst-case
|
||
/// throughput was N × (claim + executor) / tick; concurrent dispatch
|
||
/// lets one slow handler not block the others. The execution gate
|
||
/// still caps real parallelism — this just removes the
|
||
/// dispatcher-side serialization.
|
||
async fn tick_queue_arm(&self) -> Result<(), DispatcherError> {
|
||
use futures::stream::{self, StreamExt};
|
||
let consumers = self
|
||
.triggers
|
||
.list_active_queue_consumers()
|
||
.await
|
||
.map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?;
|
||
// Concurrency cap matches the execution gate's typical default;
|
||
// overflow attempts return Err(Overloaded) and back off so a
|
||
// queue burst doesn't hammer the executor.
|
||
stream::iter(consumers)
|
||
.for_each_concurrent(QUEUE_DISPATCH_PARALLELISM, |consumer| async move {
|
||
if let Err(err) = self.dispatch_one_queue(&consumer).await {
|
||
tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored");
|
||
}
|
||
})
|
||
.await;
|
||
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<Option<ClaimedMessage>, 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<bool, String> {
|
||
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<bool, String> {
|
||
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()),
|
||
}
|
||
}
|
||
|
||
/// TRANSIENT release — the handler never ran (gate saturated, or the script
|
||
/// was disabled at fire time). Re-queue WITHOUT counting the claim's
|
||
/// pre-increment against `max_attempts`, so overload/disable windows can't
|
||
/// dead-letter a message that executed zero times.
|
||
async fn q_release(
|
||
&self,
|
||
c: &ActiveQueueConsumer,
|
||
id: QueueMessageId,
|
||
token: uuid::Uuid,
|
||
delay: chrono::Duration,
|
||
) -> Result<bool, String> {
|
||
match c.shared_group {
|
||
Some(_) => self
|
||
.group_queue
|
||
.release(id, token, delay)
|
||
.await
|
||
.map_err(|e| e.to_string()),
|
||
None => self
|
||
.queue
|
||
.release(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 into `dead_letters`
|
||
/// (+ the caller fans out `dead_letter` triggers). Group shared queue →
|
||
/// dead-letter into `group_dead_letters` (Track A M2) and return `None` so the
|
||
/// per-app fan-out is skipped (a competing-consumer message has no single app
|
||
/// to fire per-app handlers under). 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<TriggerId>,
|
||
script_id: Option<ScriptId>,
|
||
reason: &str,
|
||
) -> Option<DeadLetterId> {
|
||
match c.shared_group {
|
||
Some(group_id) => {
|
||
// §11.6 D3: persist the exhausted message to the group dead-letter
|
||
// store instead of dropping it. We return None (not the dl id) so
|
||
// the per-app `fan_out_dead_letter` below is SKIPPED — firing the
|
||
// consuming app's per-app dead_letter handlers on a shared-queue
|
||
// message (competing consumers → nondeterministic app) would be
|
||
// wrong. Fan-out to a *shared* dead_letter trigger is deferred; the
|
||
// row is operator-visible via the group dead-letters admin API.
|
||
match self
|
||
.group_queue
|
||
.dead_letter(
|
||
claimed.id,
|
||
claimed.claim_token,
|
||
group_id,
|
||
&claimed.queue_name,
|
||
trigger_id,
|
||
script_id,
|
||
claimed.attempt,
|
||
claimed.enqueued_at,
|
||
reason,
|
||
)
|
||
.await
|
||
{
|
||
Ok(dl_id) => tracing::warn!(
|
||
reason,
|
||
queue = %claimed.queue_name,
|
||
dead_letter_id = %dl_id.into_inner(),
|
||
"shared-queue message dead-lettered"
|
||
),
|
||
Err(e) => tracing::error!(?e, "shared-queue dead-letter write failed"),
|
||
}
|
||
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,
|
||
consumer: &ActiveQueueConsumer,
|
||
) -> Result<(), DispatcherError> {
|
||
// Atomic claim — None → nothing pending right now for this queue.
|
||
let Some(claimed) = self
|
||
.q_claim(consumer)
|
||
.await
|
||
.map_err(DispatcherError::Outbox)?
|
||
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 {
|
||
// Transient: never executed → release without burning the retry
|
||
// budget (else sustained overload could dead-letter a message that
|
||
// ran zero times).
|
||
let _ = self
|
||
.q_release(
|
||
consumer,
|
||
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
|
||
.q_terminal(
|
||
consumer,
|
||
&claimed,
|
||
Some(consumer.trigger_id),
|
||
Some(consumer.script_id),
|
||
"queue trigger script not found",
|
||
)
|
||
.await;
|
||
drop(permit);
|
||
return Ok(());
|
||
}
|
||
Err(e) => return Err(DispatcherError::ResolveTrigger(e.to_string())),
|
||
};
|
||
|
||
// Audit 2026-06-11 H-F1 — same-app guard. validate_trigger_target
|
||
// rejects cross-app rows at write time; this is the runtime
|
||
// backstop for hand-edited triggers, partial backup restores, or
|
||
// a script re-pointed across apps without re-validating the
|
||
// existing consumers. Dead-letter the message so the misfire is
|
||
// observable rather than executing one app's script under
|
||
// another app's SdkCallCx.
|
||
// Phase 4: the target must be invocable in the claimed message's app
|
||
// context — owned by that app or an inherited group script on its
|
||
// chain. App-owned scripts take the in-memory fast path; only a group
|
||
// script pays the chain query, and a non-member fails closed
|
||
// (dead-lettered, not silently run under another app's SdkCallCx).
|
||
if !self.script_invocable(&script, claimed.app_id).await {
|
||
tracing::error!(
|
||
script_app = ?script.app_id,
|
||
claim_app = %claimed.app_id,
|
||
script_id = %consumer.script_id,
|
||
"queue consumer script belongs to a different app; dead-lettering"
|
||
);
|
||
let _ = self
|
||
.q_terminal(
|
||
consumer,
|
||
&claimed,
|
||
Some(consumer.trigger_id),
|
||
Some(consumer.script_id),
|
||
"queue consumer target belongs to a different app",
|
||
)
|
||
.await;
|
||
drop(permit);
|
||
return Ok(());
|
||
}
|
||
|
||
// Fire-time `enabled` re-check (§4.3). The queue arm does not flow
|
||
// through `dispatch_one`'s unified `!active` gate; its only other
|
||
// `enabled` guard is the per-tick `list_active_queue_consumers`
|
||
// snapshot, which is stale for messages already claimed in this
|
||
// tick. A script disabled after the list query but before this
|
||
// claimed message executes must NOT run (`script` here is a fresh
|
||
// read from line ~331, so `enabled` is current). Release the claim
|
||
// (nack) rather than ack/dead-letter: the message stays queued and
|
||
// is processed when the script is re-enabled, and the next tick
|
||
// won't re-claim it because the list filters `s.enabled`. Without
|
||
// this nack the message would sit claimed indefinitely —
|
||
// `reclaim_visibility_timeouts` only reclaims for enabled triggers.
|
||
if !script.enabled {
|
||
tracing::info!(
|
||
script_id = %consumer.script_id,
|
||
trigger_id = %consumer.trigger_id,
|
||
"queue consumer script disabled at fire time; releasing claim"
|
||
);
|
||
if let Err(e) = self
|
||
.q_release(
|
||
consumer,
|
||
claimed.id,
|
||
claimed.claim_token,
|
||
chrono::Duration::seconds(1),
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!(?e, "queue release on disabled consumer failed");
|
||
}
|
||
drop(permit);
|
||
return Ok(());
|
||
}
|
||
|
||
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(),
|
||
// Non-HTTP trigger invocation — no request method.
|
||
method: String::new(),
|
||
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,
|
||
script_owner: script.owner(),
|
||
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,
|
||
};
|
||
// G1: queue consumers dispatch outside the outbox, so they need
|
||
// their own log write. Source is always `queue`; no `reply_to`
|
||
// here, so there's no double-logging concern.
|
||
let log_cx = ExecLogContext::from_request(&exec_req, ExecutionSource::Queue);
|
||
let started = Utc::now();
|
||
let outcome = self
|
||
.executor
|
||
.execute_with_identity(
|
||
identity,
|
||
&script.source,
|
||
exec_req,
|
||
async_exec_timeout_from_env(),
|
||
)
|
||
.await;
|
||
let finished = Utc::now();
|
||
drop(permit);
|
||
|
||
self.record_execution_log(&log_cx, &outcome, started, finished)
|
||
.await;
|
||
|
||
// 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.q_ack(consumer, 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
|
||
.q_nack(consumer, 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 outbox arm
|
||
// calls `fan_out_dead_letter` after writing its DL row; the queue
|
||
// arm previously omitted this, so registered `dead_letter`
|
||
// handlers filtered on `source = "queue"` sat idle. Fan out the
|
||
// same way here.
|
||
let now = Utc::now();
|
||
let last_error = err.to_string();
|
||
// Per-app → dead-letter (+ fan out below). Shared group queue →
|
||
// dead-lettered into `group_dead_letters` inside q_terminal (Track A M2),
|
||
// which returns None so the per-app fan-out is skipped.
|
||
let dl_id = self
|
||
.q_terminal(
|
||
consumer,
|
||
claimed,
|
||
Some(consumer.trigger_id),
|
||
Some(consumer.script_id),
|
||
&last_error,
|
||
)
|
||
.await;
|
||
if let Some(dead_letter_id) = dl_id {
|
||
let original = 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(),
|
||
};
|
||
// dead_letter triggers filter on the originating event's
|
||
// wire source ("queue", "kv", "http", …) — same vocabulary
|
||
// as TriggerEvent::source().
|
||
self.fan_out_dead_letter(DeadLetterFanOutCtx {
|
||
app_id: claimed.app_id,
|
||
original,
|
||
source: "queue".to_string(),
|
||
dead_letter_id,
|
||
attempts: claimed.attempt,
|
||
last_error,
|
||
trigger_id: Some(consumer.trigger_id),
|
||
script_id: Some(consumer.script_id),
|
||
first_attempt_at: claimed.enqueued_at,
|
||
last_attempt_at: now,
|
||
// Queue messages always root a depth-1 chain (the queue
|
||
// itself is depth 0; any subsequent DL handler ticks up).
|
||
trigger_depth: 1,
|
||
root_execution_id: None,
|
||
})
|
||
.await;
|
||
}
|
||
// TODO(metrics): bump picloud_queue_dead_letters_total{app_id, queue_name}.
|
||
}
|
||
|
||
#[allow(clippy::too_many_lines)]
|
||
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 {
|
||
tracing::warn!(
|
||
outbox_id = %row.id,
|
||
app_id = %row.app_id,
|
||
trigger_depth = row.trigger_depth,
|
||
"trigger depth exceeded; dropping row"
|
||
);
|
||
// TODO(metrics): bump `picloud_trigger_depth_exceeded{app_id,trigger_id}`.
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
return Ok(());
|
||
}
|
||
|
||
// Gate admission — non-blocking. If the gate is saturated,
|
||
// release the claim by rescheduling so another tick can pick
|
||
// it up. The row stays "due" essentially immediately.
|
||
let Ok(permit) = self.gate.try_acquire() else {
|
||
let next = Utc::now() + chrono::Duration::milliseconds(100);
|
||
self.outbox
|
||
.reschedule(row.id, row.attempt_count, next)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
return Ok(());
|
||
};
|
||
|
||
// Resolve the trigger config (KV / DL) or pull the HTTP
|
||
// payload directly off the outbox row.
|
||
let (resolved, exec_req) = match row.source_kind {
|
||
OutboxSourceKind::Http => match self.build_http_request(&row).await {
|
||
Ok(pair) => pair,
|
||
Err(err) => {
|
||
tracing::warn!(outbox_id = %row.id, ?err, "http exec build failed; dropping");
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
drop(permit);
|
||
return Ok(());
|
||
}
|
||
},
|
||
OutboxSourceKind::Invoke => match self.build_invoke_request(&row).await {
|
||
Ok(pair) => pair,
|
||
Err(err) => {
|
||
tracing::warn!(outbox_id = %row.id, ?err, "invoke exec build failed; dropping");
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
drop(permit);
|
||
return Ok(());
|
||
}
|
||
},
|
||
OutboxSourceKind::Kv
|
||
| OutboxSourceKind::Docs
|
||
| OutboxSourceKind::DeadLetter
|
||
| OutboxSourceKind::Cron
|
||
| OutboxSourceKind::Files
|
||
| OutboxSourceKind::Pubsub
|
||
| OutboxSourceKind::Email => {
|
||
let resolved = self.resolve_trigger(&row).await?;
|
||
let req = match self.build_exec_request(&row, &resolved).await {
|
||
Ok(req) => req,
|
||
Err(err) => {
|
||
tracing::warn!(outbox_id = %row.id, ?err, "exec request build failed; dropping row");
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
drop(permit);
|
||
return Ok(());
|
||
}
|
||
};
|
||
(resolved, req)
|
||
}
|
||
};
|
||
|
||
// §4.3 fire-time re-check, for EVERY outbox source (trigger, async
|
||
// HTTP/202, and invoke): if the target script (or, for triggers, the
|
||
// trigger) was disabled after this row was enqueued, drop it rather
|
||
// than fire a stale event. The match-time `enabled` check can't see a
|
||
// later toggle, so this is the gate that makes a disabled script
|
||
// genuinely non-invocable on the async paths too.
|
||
if !resolved.active {
|
||
tracing::debug!(
|
||
outbox_id = %row.id,
|
||
app_id = %row.app_id,
|
||
"target script/trigger disabled since enqueue; dropping outbox row"
|
||
);
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
drop(permit);
|
||
return Ok(());
|
||
}
|
||
|
||
// The gate permit auto-releases when this scope ends or when
|
||
// the executor finishes. We hand control to the executor and
|
||
// wait synchronously here — sync HTTP and dispatcher share the
|
||
// semaphore so this is intentional.
|
||
let source = resolved.script_source.clone();
|
||
let identity = picloud_orchestrator_core::ScriptIdentity {
|
||
script_id: resolved.script_id,
|
||
updated_at: resolved.script_updated_at,
|
||
};
|
||
// G1: capture the request context before `exec_req` is consumed so
|
||
// we can write an execution-log row for this run (see below).
|
||
let log_cx = ExecLogContext::from_request(&exec_req, exec_source(&row));
|
||
let started = Utc::now();
|
||
let outcome = self
|
||
.executor
|
||
.execute_with_identity(identity, &source, exec_req, async_exec_timeout_from_env())
|
||
.await;
|
||
let finished = Utc::now();
|
||
drop(permit);
|
||
|
||
// G1: persist an execution-log row so this run shows up in
|
||
// `pic logs`. Skip rows with a synchronous receiver (`reply_to`
|
||
// is set) — those are sync HTTP routes that the orchestrator's
|
||
// inbox path already logs, and logging here too would duplicate
|
||
// them. Trigger rows and fire-and-forget async HTTP (202) have no
|
||
// receiver, so the dispatcher is the only place that can log them.
|
||
if row.reply_to.is_none() {
|
||
self.record_execution_log(&log_cx, &outcome, started, finished)
|
||
.await;
|
||
}
|
||
|
||
match outcome {
|
||
Ok(resp) => self.handle_success(&row, &resolved, resp).await,
|
||
Err(err) => self.handle_failure(&row, &resolved, err).await,
|
||
}
|
||
}
|
||
|
||
/// G1: best-effort persistence of an execution-log row for a
|
||
/// dispatcher-run script. A sink failure is logged but never changes
|
||
/// ack/nack/retry behavior — the audit trail is not on the hot path
|
||
/// of message-delivery correctness.
|
||
async fn record_execution_log(
|
||
&self,
|
||
cx: &ExecLogContext,
|
||
outcome: &Result<ExecResponse, ExecError>,
|
||
started: DateTime<Utc>,
|
||
finished: DateTime<Utc>,
|
||
) {
|
||
let log = build_execution_log(
|
||
cx.app_id,
|
||
cx.script_id,
|
||
cx.request_id,
|
||
cx.path.clone(),
|
||
cx.headers.clone(),
|
||
cx.body.clone(),
|
||
cx.source,
|
||
outcome,
|
||
started,
|
||
finished,
|
||
);
|
||
if let Err(e) = self.log_sink.record(log).await {
|
||
tracing::warn!(
|
||
error = %e,
|
||
script_id = %cx.script_id,
|
||
source = cx.source.as_str(),
|
||
"failed to persist trigger execution log"
|
||
);
|
||
}
|
||
}
|
||
|
||
async fn resolve_trigger(&self, row: &OutboxRow) -> Result<ResolvedTrigger, DispatcherError> {
|
||
// For KV and DL kinds, the outbox carries `trigger_id`. Use it
|
||
// to look up the trigger row, then resolve the script.
|
||
let Some(trigger_id) = row.trigger_id else {
|
||
return Err(DispatcherError::ResolveTrigger(
|
||
"outbox row missing trigger_id".into(),
|
||
));
|
||
};
|
||
let trigger = self
|
||
.triggers
|
||
.get(trigger_id)
|
||
.await
|
||
.map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
|
||
.ok_or_else(|| {
|
||
DispatcherError::ResolveTrigger(format!("trigger {trigger_id} not found"))
|
||
})?;
|
||
|
||
let script = self
|
||
.scripts
|
||
.get(trigger.script_id)
|
||
.await
|
||
.map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
|
||
.ok_or_else(|| {
|
||
DispatcherError::ResolveTrigger(format!("script {} not found", trigger.script_id))
|
||
})?;
|
||
|
||
// Audit 2026-06-11 H-F1 sibling — same-app guard mirroring
|
||
// build_http_request / build_invoke_request / dispatch_one_queue.
|
||
// `build_exec_request` stamps `ExecRequest.app_id = row.app_id`
|
||
// while sourcing the body from `trigger.script_id`; without this
|
||
// check a hand-edited outbox/trigger row (or a partial restore, or
|
||
// a script re-pointed across apps) could run one app's script under
|
||
// another app's `SdkCallCx.app_id` — the cross-app isolation
|
||
// boundary. Not reachable via the trigger-create or `apply` paths
|
||
// (both resolve the script within the app's own scope), so this is
|
||
// the runtime backstop the other arms already carry.
|
||
// Phase 4: invocable in the outbox row's app context (app-owned or an
|
||
// inherited group script on the chain); a non-member fails closed.
|
||
if !self.script_invocable(&script, row.app_id).await {
|
||
return Err(DispatcherError::ResolveTrigger(
|
||
"trigger outbox target belongs to a different app".into(),
|
||
));
|
||
}
|
||
|
||
Ok(ResolvedTrigger {
|
||
trigger_kind: trigger.kind,
|
||
is_dead_letter_handler: matches!(trigger.kind, TriggerKind::DeadLetter),
|
||
active: trigger.enabled && script.enabled,
|
||
script_id: script.id,
|
||
script_owner: script.owner(),
|
||
script_source: script.source,
|
||
script_name: script.name,
|
||
script_updated_at: script.updated_at,
|
||
sandbox_overrides: script.sandbox,
|
||
registered_by_principal: trigger.registered_by_principal,
|
||
retry_max_attempts: trigger.retry_max_attempts,
|
||
retry_backoff: trigger.retry_backoff,
|
||
retry_base_ms: trigger.retry_base_ms,
|
||
})
|
||
}
|
||
|
||
async fn build_exec_request(
|
||
&self,
|
||
row: &OutboxRow,
|
||
resolved: &ResolvedTrigger,
|
||
) -> Result<ExecRequest, DispatcherError> {
|
||
let trigger_event: TriggerEvent = serde_json::from_value(row.payload.clone())
|
||
.map_err(|e| DispatcherError::ResolveTrigger(format!("decode payload: {e}")))?;
|
||
|
||
let principal = self
|
||
.principals
|
||
.resolve(resolved.registered_by_principal)
|
||
.await
|
||
.map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?;
|
||
|
||
let execution_id = ExecutionId::new();
|
||
Ok(ExecRequest {
|
||
execution_id,
|
||
request_id: RequestId::new(),
|
||
script_id: resolved.script_id,
|
||
script_name: resolved.script_name.clone(),
|
||
invocation_type: InvocationType::Function,
|
||
path: format!("/trigger/{}", trigger_event.source()),
|
||
// Non-HTTP trigger invocation — no request method.
|
||
method: String::new(),
|
||
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: resolved.sandbox_overrides,
|
||
app_id: row.app_id,
|
||
script_owner: resolved.script_owner,
|
||
principal: Some(principal),
|
||
trigger_depth: row.trigger_depth,
|
||
root_execution_id: row.root_execution_id.unwrap_or(execution_id),
|
||
is_dead_letter_handler: resolved.is_dead_letter_handler,
|
||
event: Some(trigger_event),
|
||
})
|
||
}
|
||
|
||
/// Build an `(ResolvedTrigger, ExecRequest)` for an HTTP outbox
|
||
/// row. HTTP rows don't have a backing `triggers` row (the
|
||
/// `trigger_id` references `routes.id` instead). We pull the
|
||
/// script id off the outbox row, the request shape off the
|
||
/// payload, and synthesize a `ResolvedTrigger` with retry
|
||
/// settings irrelevant for HTTP (sync HTTP is never retried;
|
||
/// async HTTP uses default policy from `TriggerConfig`).
|
||
async fn build_http_request(
|
||
&self,
|
||
row: &OutboxRow,
|
||
) -> Result<(ResolvedTrigger, ExecRequest), DispatcherError> {
|
||
let Some(script_id) = row.script_id else {
|
||
return Err(DispatcherError::ResolveTrigger(
|
||
"HTTP outbox row missing script_id".into(),
|
||
));
|
||
};
|
||
let script = self
|
||
.scripts
|
||
.get(script_id)
|
||
.await
|
||
.map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
|
||
.ok_or_else(|| {
|
||
DispatcherError::ResolveTrigger(format!("script {script_id} not found"))
|
||
})?;
|
||
|
||
// Audit 2026-06-11 H-F1 sibling — same-app guard mirroring
|
||
// build_invoke_request and dispatch_one_queue. A hand-edited
|
||
// outbox row could otherwise execute one app's script under
|
||
// another app's SdkCallCx.
|
||
// Phase 4: invocable in this app's context (see queue/trigger arms).
|
||
if !self.script_invocable(&script, row.app_id).await {
|
||
return Err(DispatcherError::ResolveTrigger(
|
||
"http outbox target belongs to a different app".into(),
|
||
));
|
||
}
|
||
|
||
let payload: HttpDispatchPayload = serde_json::from_value(row.payload.clone())
|
||
.map_err(|e| DispatcherError::ResolveTrigger(format!("decode http payload: {e}")))?;
|
||
|
||
let execution_id = ExecutionId::new();
|
||
let req = ExecRequest {
|
||
execution_id,
|
||
request_id: RequestId::new(),
|
||
script_id,
|
||
script_name: payload.script_name.clone(),
|
||
invocation_type: InvocationType::Http,
|
||
path: payload.path.clone(),
|
||
method: payload.method.clone(),
|
||
headers: payload.headers,
|
||
body: payload.body,
|
||
params: payload.params,
|
||
query: payload.query,
|
||
rest: payload.rest,
|
||
sandbox_overrides: script.sandbox,
|
||
app_id: row.app_id,
|
||
script_owner: script.owner(),
|
||
// HTTP outbox rows don't run as the trigger registrant —
|
||
// they run with no principal (public ingress) or the
|
||
// attached one (origin_principal forensic field is not
|
||
// promoted to execution principal in this MVP).
|
||
principal: None,
|
||
trigger_depth: row.trigger_depth,
|
||
root_execution_id: row.root_execution_id.unwrap_or(execution_id),
|
||
is_dead_letter_handler: false,
|
||
event: None,
|
||
};
|
||
|
||
let resolved = ResolvedTrigger {
|
||
trigger_kind: TriggerKind::Kv, // placeholder; HTTP doesn't have a kind
|
||
is_dead_letter_handler: false,
|
||
// §4.3: an async-HTTP (202) row whose script was disabled after
|
||
// enqueue is dropped at fire time by the post-match active check.
|
||
active: script.enabled,
|
||
script_id,
|
||
script_owner: script.owner(),
|
||
script_source: script.source,
|
||
script_name: payload.script_name,
|
||
script_updated_at: script.updated_at,
|
||
sandbox_overrides: script.sandbox,
|
||
// HTTP outbox rows don't carry a registered_by_principal
|
||
// — use a sentinel zero UUID since this field isn't used
|
||
// downstream for HTTP (no retries, no inbox principal).
|
||
registered_by_principal: picloud_shared::AdminUserId::from(uuid::Uuid::nil()),
|
||
// Async HTTP uses the platform default retry policy from
|
||
// TriggerConfig. Sync HTTP (reply_to.is_some) never retries
|
||
// regardless.
|
||
retry_max_attempts: self.config.retry_max_attempts,
|
||
retry_backoff: self.config.retry_backoff,
|
||
retry_base_ms: self.config.retry_base_ms,
|
||
};
|
||
Ok((resolved, req))
|
||
}
|
||
|
||
/// v1.1.9. Build an `(ResolvedTrigger, ExecRequest)` from an
|
||
/// invoke_async outbox row. The payload was written by
|
||
/// `InvokeService::enqueue_async`. No retry policy — invoke_async
|
||
/// runs once; on failure the row is deleted and a dead_letters row
|
||
/// is written (so the misfire is observable).
|
||
async fn build_invoke_request(
|
||
&self,
|
||
row: &OutboxRow,
|
||
) -> Result<(ResolvedTrigger, ExecRequest), DispatcherError> {
|
||
// Extract the args + script_id from the payload (InvokeService
|
||
// populates these). Defensive: dead_letter the row if any
|
||
// required field is missing.
|
||
let payload = &row.payload;
|
||
let Some(script_id_str) = payload.get("script_id").and_then(|v| v.as_str()) else {
|
||
return Err(DispatcherError::ResolveTrigger(
|
||
"invoke row missing script_id".into(),
|
||
));
|
||
};
|
||
let script_id = picloud_shared::ScriptId::from(
|
||
uuid::Uuid::parse_str(script_id_str)
|
||
.map_err(|e| DispatcherError::ResolveTrigger(format!("script_id: {e}")))?,
|
||
);
|
||
let script = self
|
||
.scripts
|
||
.get(script_id)
|
||
.await
|
||
.map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
|
||
.ok_or_else(|| {
|
||
DispatcherError::ResolveTrigger(format!("script {script_id} not found"))
|
||
})?;
|
||
// Same-app guard — the script could have been re-assigned or
|
||
// the row could have been hand-edited. Reject mismatches.
|
||
// Phase 4: invocable in this app's context (app-owned or inherited).
|
||
if !self.script_invocable(&script, row.app_id).await {
|
||
return Err(DispatcherError::ResolveTrigger(
|
||
"invoke target belongs to a different app".into(),
|
||
));
|
||
}
|
||
|
||
let args = payload
|
||
.get("args")
|
||
.cloned()
|
||
.unwrap_or(serde_json::Value::Null);
|
||
let trigger_depth = payload
|
||
.get("trigger_depth")
|
||
.and_then(serde_json::Value::as_u64)
|
||
.map_or(row.trigger_depth, |n| u32::try_from(n).unwrap_or(u32::MAX));
|
||
|
||
let execution_id = ExecutionId::new();
|
||
let req = ExecRequest {
|
||
execution_id,
|
||
request_id: RequestId::new(),
|
||
script_id: script.id,
|
||
script_name: script.name.clone(),
|
||
invocation_type: InvocationType::Function,
|
||
path: "/invoke_async".into(),
|
||
method: String::new(),
|
||
headers: std::collections::BTreeMap::new(),
|
||
body: args,
|
||
params: std::collections::BTreeMap::new(),
|
||
query: std::collections::BTreeMap::new(),
|
||
rest: String::new(),
|
||
sandbox_overrides: script.sandbox,
|
||
app_id: row.app_id,
|
||
script_owner: script.owner(),
|
||
// invoke_async runs with no principal — same as HTTP outbox.
|
||
principal: None,
|
||
trigger_depth,
|
||
root_execution_id: row.root_execution_id.unwrap_or(execution_id),
|
||
is_dead_letter_handler: false,
|
||
event: None,
|
||
};
|
||
|
||
// Build a synthetic ResolvedTrigger so the dispatch_one outer
|
||
// loop can carry on through the same code path. Retry knobs
|
||
// bound to 1 attempt: invoke_async runs ONCE (callers wrap in
|
||
// retry::with for retries).
|
||
let resolved = ResolvedTrigger {
|
||
trigger_kind: TriggerKind::Cron, // placeholder; not used downstream
|
||
is_dead_letter_handler: false,
|
||
// §4.3: a queued invoke() whose target script was disabled after
|
||
// enqueue is dropped at fire time by the post-match active check.
|
||
active: script.enabled,
|
||
script_id: script.id,
|
||
script_owner: script.owner(),
|
||
script_source: script.source,
|
||
script_name: script.name,
|
||
script_updated_at: script.updated_at,
|
||
sandbox_overrides: script.sandbox,
|
||
registered_by_principal: picloud_shared::AdminUserId::from(uuid::Uuid::nil()),
|
||
retry_max_attempts: 1,
|
||
retry_backoff: BackoffShape::Constant,
|
||
retry_base_ms: 0,
|
||
};
|
||
Ok((resolved, req))
|
||
}
|
||
|
||
async fn handle_success(
|
||
&self,
|
||
row: &OutboxRow,
|
||
_resolved: &ResolvedTrigger,
|
||
resp: ExecResponse,
|
||
) -> Result<(), DispatcherError> {
|
||
if let Some(inbox_id) = row.reply_to {
|
||
self.deliver_inbox(row, inbox_id, InboxResult::Success(summarize(&resp)))
|
||
.await;
|
||
}
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
Ok(())
|
||
}
|
||
|
||
#[allow(clippy::too_many_lines)]
|
||
async fn handle_failure(
|
||
&self,
|
||
row: &OutboxRow,
|
||
resolved: &ResolvedTrigger,
|
||
err: ExecError,
|
||
) -> Result<(), DispatcherError> {
|
||
// Sync HTTP: always single-attempt. Always deliver outcome
|
||
// (success-or-failure) to the inbox. Never retry, never DL.
|
||
if let Some(inbox_id) = row.reply_to {
|
||
let (kind, message) = classify_exec_error(&err);
|
||
self.deliver_inbox(
|
||
row,
|
||
inbox_id,
|
||
InboxResult::Failure {
|
||
kind,
|
||
message: message.clone(),
|
||
},
|
||
)
|
||
.await;
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
return Ok(());
|
||
}
|
||
|
||
// Dead-letter handler: never retry, never DL. Failure
|
||
// annotates the original DL row + bumps a metric.
|
||
if resolved.is_dead_letter_handler {
|
||
tracing::error!(
|
||
outbox_id = %row.id,
|
||
app_id = %row.app_id,
|
||
?err,
|
||
"dead-letter handler failed; not retrying"
|
||
);
|
||
// TODO(metrics): bump `picloud_dead_letter_handler_failures{app_id}`.
|
||
// Annotate the original DL row (id is `row.payload.dead_letter.id`
|
||
// when the payload is a DeadLetter TriggerEvent). Best-effort:
|
||
// if the payload doesn't decode, just log and move on.
|
||
if let Ok(TriggerEvent::DeadLetter { dead_letter_id, .. }) =
|
||
serde_json::from_value::<TriggerEvent>(row.payload.clone())
|
||
{
|
||
if let Err(e) = self
|
||
.dead_letters
|
||
.resolve(dead_letter_id, "handler_failed")
|
||
.await
|
||
{
|
||
tracing::warn!(?e, "could not annotate DL row as handler_failed");
|
||
}
|
||
}
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
return Ok(());
|
||
}
|
||
|
||
// Async event: retry per policy, then dead-letter.
|
||
let attempt = row.attempt_count + 1;
|
||
if attempt < resolved.retry_max_attempts {
|
||
let delay = compute_backoff(
|
||
attempt,
|
||
resolved.retry_backoff,
|
||
resolved.retry_base_ms,
|
||
self.config.retry_jitter_pct,
|
||
);
|
||
let next = Utc::now() + chrono::Duration::milliseconds(i64::from(delay));
|
||
tracing::info!(
|
||
outbox_id = %row.id,
|
||
attempt,
|
||
max_attempts = resolved.retry_max_attempts,
|
||
retry_in_ms = delay,
|
||
"rescheduling outbox row"
|
||
);
|
||
self.outbox
|
||
.reschedule(row.id, attempt, next)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
return Ok(());
|
||
}
|
||
|
||
// Exhausted retries → dead-letter.
|
||
let (op, source) = describe_event(row.source_kind, &row.payload);
|
||
let now = Utc::now();
|
||
let dl_id = match self
|
||
.dead_letters
|
||
.insert(NewDeadLetter {
|
||
app_id: row.app_id,
|
||
original_event_id: row.id,
|
||
source: source.clone(),
|
||
op,
|
||
trigger_id: row.trigger_id,
|
||
script_id: Some(resolved.script_id),
|
||
payload: row.payload.clone(),
|
||
attempt_count: attempt,
|
||
first_attempt_at: row.created_at,
|
||
last_attempt_at: now,
|
||
last_error: err.to_string(),
|
||
})
|
||
.await
|
||
{
|
||
Ok(id) => Some(id),
|
||
Err(e) => {
|
||
tracing::error!(?e, "failed to write dead-letter row");
|
||
None
|
||
}
|
||
};
|
||
|
||
// v1.1.7 fix: fan the dead-letter out to matching handler triggers.
|
||
// This was missing since v1.1.1 — the row was written but
|
||
// `list_matching_dead_letter` had no production caller, so
|
||
// registered dead_letter handlers never fired. The recursion-stop
|
||
// (a dead-letter handler's own failure is not re-dead-lettered)
|
||
// is upheld by the `is_dead_letter_handler` short-circuit at the
|
||
// top of this function, so this fan-out is only reached for
|
||
// non-handler executions.
|
||
if let Some(dl_id) = dl_id {
|
||
// The DL event nests the original verbatim; if the payload
|
||
// can't be decoded back into a TriggerEvent we can't build
|
||
// the nested `original`, so skip the fan-out (the DL row is
|
||
// still written).
|
||
match serde_json::from_value::<TriggerEvent>(row.payload.clone()) {
|
||
Ok(original) => {
|
||
self.fan_out_dead_letter(DeadLetterFanOutCtx {
|
||
app_id: row.app_id,
|
||
original,
|
||
source,
|
||
dead_letter_id: dl_id,
|
||
attempts: attempt,
|
||
last_error: err.to_string(),
|
||
trigger_id: row.trigger_id,
|
||
script_id: Some(resolved.script_id),
|
||
first_attempt_at: row.created_at,
|
||
last_attempt_at: now,
|
||
trigger_depth: row.trigger_depth,
|
||
root_execution_id: row.root_execution_id,
|
||
})
|
||
.await;
|
||
}
|
||
Err(_) => {
|
||
tracing::warn!(
|
||
outbox_id = %row.id,
|
||
"dead-letter payload is not a TriggerEvent; skipping handler fan-out"
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
self.outbox
|
||
.delete(row.id)
|
||
.await
|
||
.map_err(|e| DispatcherError::Outbox(e.to_string()))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Enqueue one outbox row per matching `dead_letter` trigger so its
|
||
/// handler script runs with the dead-letter event as `ctx.event`.
|
||
/// Best-effort: a lookup/insert failure is logged, not propagated
|
||
/// (the dead-letter row itself is already durably written).
|
||
///
|
||
/// Called from both the outbox arm (`handle_failure`) and the queue
|
||
/// arm (`handle_queue_failure`). The queue arm previously omitted
|
||
/// this call, so registered `dead_letter` handlers filtered on
|
||
/// `source = "queue"` sat idle even after queue retry-exhaust.
|
||
async fn fan_out_dead_letter(&self, ctx: DeadLetterFanOutCtx) {
|
||
let DeadLetterFanOutCtx {
|
||
app_id,
|
||
original,
|
||
source,
|
||
dead_letter_id,
|
||
attempts,
|
||
last_error,
|
||
trigger_id,
|
||
script_id,
|
||
first_attempt_at,
|
||
last_attempt_at,
|
||
trigger_depth,
|
||
root_execution_id,
|
||
} = ctx;
|
||
|
||
let matches = match self
|
||
.triggers
|
||
.list_matching_dead_letter(app_id, &source, trigger_id, script_id)
|
||
.await
|
||
{
|
||
Ok(m) => m,
|
||
Err(e) => {
|
||
tracing::error!(?e, "dead-letter trigger lookup failed");
|
||
return;
|
||
}
|
||
};
|
||
|
||
for m in matches {
|
||
let event = TriggerEvent::DeadLetter {
|
||
dead_letter_id,
|
||
original: Box::new(original.clone()),
|
||
attempts,
|
||
last_error: last_error.clone(),
|
||
trigger_id,
|
||
script_id,
|
||
first_attempt_at,
|
||
last_attempt_at,
|
||
};
|
||
let payload = match serde_json::to_value(&event) {
|
||
Ok(p) => p,
|
||
Err(e) => {
|
||
tracing::error!(?e, "failed to serialize dead-letter event");
|
||
continue;
|
||
}
|
||
};
|
||
if let Err(e) = self
|
||
.outbox
|
||
.insert(NewOutboxRow {
|
||
app_id,
|
||
source_kind: OutboxSourceKind::DeadLetter,
|
||
trigger_id: Some(m.trigger_id),
|
||
script_id: Some(m.script_id),
|
||
reply_to: None,
|
||
payload,
|
||
origin_principal: Some(m.registered_by_principal),
|
||
trigger_depth: trigger_depth.saturating_add(1),
|
||
root_execution_id,
|
||
})
|
||
.await
|
||
{
|
||
tracing::error!(?e, "failed to enqueue dead-letter handler delivery");
|
||
}
|
||
}
|
||
}
|
||
|
||
async fn deliver_inbox(&self, row: &OutboxRow, inbox_id: Uuid, result: InboxResult) {
|
||
match self.inbox.deliver(inbox_id, result.clone()).await {
|
||
InboxDeliveryOutcome::Delivered => {}
|
||
InboxDeliveryOutcome::Abandoned => {
|
||
// Receiver was dropped — record forensic row + bump
|
||
// metric.
|
||
let (status_code, summary) = match &result {
|
||
InboxResult::Success(s) => (s.status_code, None),
|
||
InboxResult::Failure { kind, message } => {
|
||
(failure_kind_to_status(*kind), Some(message.clone()))
|
||
}
|
||
};
|
||
if let Err(e) = self
|
||
.abandoned
|
||
.insert(NewAbandonedExecution {
|
||
app_id: row.app_id,
|
||
outbox_id: row.id,
|
||
script_id: row.script_id,
|
||
inbox_id,
|
||
status_code,
|
||
result_summary: summary,
|
||
})
|
||
.await
|
||
{
|
||
tracing::warn!(?e, "abandoned_executions insert failed");
|
||
}
|
||
// TODO(metrics): bump `picloud_abandoned_executions_total{app_id}`.
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
#[derive(Debug)]
|
||
pub struct ResolvedTrigger {
|
||
pub trigger_kind: TriggerKind,
|
||
pub is_dead_letter_handler: bool,
|
||
/// §4.3 fire-time gate: `trigger.enabled && script.enabled` at resolve
|
||
/// time. A row enqueued before either was disabled is dropped, not fired.
|
||
pub active: bool,
|
||
pub script_id: ScriptId,
|
||
pub script_source: String,
|
||
pub script_name: String,
|
||
/// Phase 4b: the resolved script's defining node — the lexical origin
|
||
/// for its `import`s (§5.5). `None` only for a corrupt owner row; the
|
||
/// executor then falls back to `App(app_id)`.
|
||
pub script_owner: Option<ScriptOwner>,
|
||
/// v1.1.3: freshness comparator for the orchestrator's top-level
|
||
/// script cache. The dispatcher hands `(script_id, updated_at)`
|
||
/// in alongside the source so cached ASTs can be reused across
|
||
/// triggered invocations.
|
||
pub script_updated_at: chrono::DateTime<chrono::Utc>,
|
||
pub sandbox_overrides: ScriptSandbox,
|
||
pub registered_by_principal: picloud_shared::AdminUserId,
|
||
pub retry_max_attempts: u32,
|
||
pub retry_backoff: BackoffShape,
|
||
pub retry_base_ms: u32,
|
||
}
|
||
|
||
#[derive(Debug, thiserror::Error)]
|
||
pub enum DispatcherError {
|
||
#[error("outbox: {0}")]
|
||
Outbox(String),
|
||
#[error("resolve trigger: {0}")]
|
||
ResolveTrigger(String),
|
||
}
|
||
|
||
fn summarize(resp: &ExecResponse) -> ExecResponseSummary {
|
||
ExecResponseSummary {
|
||
status_code: resp.status_code,
|
||
headers: resp.headers.clone(),
|
||
body: resp.body.clone(),
|
||
}
|
||
}
|
||
|
||
/// Map `ExecError` onto the design-notes §3 status-code table.
|
||
fn classify_exec_error(err: &ExecError) -> (InboxFailureKind, String) {
|
||
match err {
|
||
ExecError::Parse(s) | ExecError::InvalidResponse(s) => {
|
||
(InboxFailureKind::Validation, s.clone())
|
||
}
|
||
ExecError::Timeout(_) => (InboxFailureKind::Timeout, err.to_string()),
|
||
ExecError::OperationBudgetExceeded => (InboxFailureKind::OperationBudget, err.to_string()),
|
||
ExecError::Overloaded { .. } => (InboxFailureKind::Overloaded, err.to_string()),
|
||
ExecError::Runtime(s) => (InboxFailureKind::Runtime, s.clone()),
|
||
}
|
||
}
|
||
|
||
fn failure_kind_to_status(k: InboxFailureKind) -> u16 {
|
||
match k {
|
||
InboxFailureKind::Validation => 422,
|
||
InboxFailureKind::Runtime => 502,
|
||
InboxFailureKind::Overloaded => 503,
|
||
InboxFailureKind::Timeout => 504,
|
||
InboxFailureKind::OperationBudget => 507,
|
||
InboxFailureKind::Platform => 500,
|
||
}
|
||
}
|
||
|
||
/// `(op, source)` extracted from the outbox payload. Used to seed the
|
||
/// `dead_letters` row when retries exhaust.
|
||
///
|
||
/// The shape of the payload depends on `source_kind`:
|
||
/// * `Http` rows carry an `HttpDispatchPayload` (`method`/`path`).
|
||
/// * `Invoke` rows carry an ad-hoc invoke payload (`script_id`/`args`).
|
||
/// * Every other kind carries a serialized [`TriggerEvent`], which has
|
||
/// `source` and `op` fields at the JSON top level.
|
||
///
|
||
/// F-S-009 (audit fix): previously this function unconditionally read
|
||
/// `source`/`op` from the JSON, which silently produced empty strings
|
||
/// for HTTP rows. The DL row then got written with `source = ""` and
|
||
/// `from_wire("")` returned `None` on replay, defaulting to
|
||
/// `OutboxSourceKind::Kv` — replay rerouted through `resolve_trigger`,
|
||
/// which failed because HTTP rows carry no `trigger_id`. Net result:
|
||
/// HTTP-async DL replays no-op'd silently.
|
||
fn describe_event(source_kind: OutboxSourceKind, payload: &serde_json::Value) -> (String, String) {
|
||
match source_kind {
|
||
OutboxSourceKind::Http => {
|
||
let method = payload
|
||
.get("method")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("?");
|
||
let path = payload.get("path").and_then(|v| v.as_str()).unwrap_or("?");
|
||
(
|
||
format!("{method} {path}"),
|
||
OutboxSourceKind::Http.as_str().to_string(),
|
||
)
|
||
}
|
||
OutboxSourceKind::Invoke => (
|
||
"invoke_async".to_string(),
|
||
OutboxSourceKind::Invoke.as_str().to_string(),
|
||
),
|
||
OutboxSourceKind::Kv
|
||
| OutboxSourceKind::Docs
|
||
| OutboxSourceKind::DeadLetter
|
||
| OutboxSourceKind::Cron
|
||
| OutboxSourceKind::Files
|
||
| OutboxSourceKind::Pubsub
|
||
| OutboxSourceKind::Email => {
|
||
let source = payload
|
||
.get("source")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("")
|
||
.to_string();
|
||
let op = payload
|
||
.get("op")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("")
|
||
.to_string();
|
||
(op, source)
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Compute backoff (ms) for the given attempt + policy + jitter.
|
||
/// Attempt is 1-indexed (first retry = attempt 1).
|
||
#[must_use]
|
||
pub fn compute_backoff(attempt: u32, backoff: BackoffShape, base_ms: u32, jitter_pct: u32) -> u32 {
|
||
let base_ms = u64::from(base_ms);
|
||
let attempt = u64::from(attempt.saturating_sub(1));
|
||
let raw = match backoff {
|
||
BackoffShape::Constant => base_ms,
|
||
BackoffShape::Linear => base_ms * (attempt + 1),
|
||
// 1x base, 2x base, 4x base, … (saturating).
|
||
BackoffShape::Exponential => base_ms.saturating_mul(1u64 << attempt.min(20)),
|
||
};
|
||
let raw = u32::try_from(raw.min(u64::from(u32::MAX))).unwrap_or(u32::MAX);
|
||
apply_jitter(raw, jitter_pct)
|
||
}
|
||
|
||
fn apply_jitter(raw: u32, pct: u32) -> u32 {
|
||
if pct == 0 {
|
||
return raw;
|
||
}
|
||
let pct = pct.min(100);
|
||
// ±span% — bounded by raw itself so we can't underflow when
|
||
// raw + offset goes below zero.
|
||
let span = u64::from(raw) * u64::from(pct) / 100;
|
||
if span == 0 {
|
||
return raw;
|
||
}
|
||
let span_i64 = i64::try_from(span).unwrap_or(i64::MAX);
|
||
let mut rng = rand::thread_rng();
|
||
let offset = rng.gen_range(-span_i64..=span_i64);
|
||
let signed = i64::from(raw).saturating_add(offset).max(0);
|
||
u32::try_from(signed.min(i64::from(u32::MAX))).unwrap_or(u32::MAX)
|
||
}
|
||
|
||
#[cfg(test)]
|
||
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<QueueMessageId, crate::group_queue_repo::GroupQueueRepoError> {
|
||
unreachable!("shared queue not exercised")
|
||
}
|
||
async fn claim(
|
||
&self,
|
||
_group_id: picloud_shared::GroupId,
|
||
_collection: &str,
|
||
) -> Result<
|
||
Option<crate::group_queue_repo::ClaimedGroupMessage>,
|
||
crate::group_queue_repo::GroupQueueRepoError,
|
||
> {
|
||
Ok(None)
|
||
}
|
||
async fn ack(
|
||
&self,
|
||
_id: QueueMessageId,
|
||
_token: Uuid,
|
||
) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
|
||
Ok(false)
|
||
}
|
||
async fn nack(
|
||
&self,
|
||
_id: QueueMessageId,
|
||
_token: Uuid,
|
||
_delay: chrono::Duration,
|
||
) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
|
||
Ok(false)
|
||
}
|
||
async fn release(
|
||
&self,
|
||
_id: QueueMessageId,
|
||
_token: Uuid,
|
||
_delay: chrono::Duration,
|
||
) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
|
||
Ok(false)
|
||
}
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn dead_letter(
|
||
&self,
|
||
_id: QueueMessageId,
|
||
_token: Uuid,
|
||
_group_id: picloud_shared::GroupId,
|
||
_collection: &str,
|
||
_trigger_id: Option<picloud_shared::TriggerId>,
|
||
_script_id: Option<ScriptId>,
|
||
_attempt: u32,
|
||
_first_attempt_at: chrono::DateTime<chrono::Utc>,
|
||
_last_error: &str,
|
||
) -> Result<picloud_shared::DeadLetterId, crate::group_queue_repo::GroupQueueRepoError>
|
||
{
|
||
unreachable!("shared queue not exercised")
|
||
}
|
||
async fn reclaim_visibility_timeouts(
|
||
&self,
|
||
) -> Result<u64, crate::group_queue_repo::GroupQueueRepoError> {
|
||
Ok(0)
|
||
}
|
||
async fn depth(
|
||
&self,
|
||
_group_id: picloud_shared::GroupId,
|
||
_collection: &str,
|
||
) -> Result<u64, crate::group_queue_repo::GroupQueueRepoError> {
|
||
Ok(0)
|
||
}
|
||
async fn depth_pending(
|
||
&self,
|
||
_group_id: picloud_shared::GroupId,
|
||
_collection: &str,
|
||
) -> Result<u64, crate::group_queue_repo::GroupQueueRepoError> {
|
||
Ok(0)
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn exponential_backoff_doubles_per_attempt() {
|
||
// No jitter (pct=0) for a deterministic check.
|
||
assert_eq!(compute_backoff(1, BackoffShape::Exponential, 1000, 0), 1000);
|
||
assert_eq!(compute_backoff(2, BackoffShape::Exponential, 1000, 0), 2000);
|
||
assert_eq!(compute_backoff(3, BackoffShape::Exponential, 1000, 0), 4000);
|
||
assert_eq!(compute_backoff(4, BackoffShape::Exponential, 1000, 0), 8000);
|
||
}
|
||
|
||
#[test]
|
||
fn linear_backoff_scales_with_attempt() {
|
||
assert_eq!(compute_backoff(1, BackoffShape::Linear, 100, 0), 100);
|
||
assert_eq!(compute_backoff(2, BackoffShape::Linear, 100, 0), 200);
|
||
assert_eq!(compute_backoff(5, BackoffShape::Linear, 100, 0), 500);
|
||
}
|
||
|
||
#[test]
|
||
fn constant_backoff_returns_base() {
|
||
for attempt in 1..=5 {
|
||
assert_eq!(
|
||
compute_backoff(attempt, BackoffShape::Constant, 750, 0),
|
||
750
|
||
);
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn jitter_within_pct_of_base() {
|
||
for _ in 0..100 {
|
||
let v = compute_backoff(1, BackoffShape::Constant, 1000, 20);
|
||
// ±20% of 1000 = 800..=1200.
|
||
assert!((800..=1200).contains(&v), "jitter out of range: {v}");
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn describe_event_http_uses_method_and_path() {
|
||
let payload = serde_json::json!({
|
||
"method": "POST",
|
||
"path": "/webhook",
|
||
"headers": {},
|
||
"body": null,
|
||
"params": {},
|
||
"query": {},
|
||
"rest": "",
|
||
"script_name": "hook",
|
||
"timeout_seconds": 30,
|
||
});
|
||
let (op, source) = describe_event(OutboxSourceKind::Http, &payload);
|
||
assert_eq!(op, "POST /webhook");
|
||
assert_eq!(source, "http");
|
||
// Round-trip via from_wire so the DL replay path resolves to the
|
||
// HTTP branch instead of falling back to Kv.
|
||
assert_eq!(
|
||
OutboxSourceKind::from_wire(&source),
|
||
Some(OutboxSourceKind::Http)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn describe_event_invoke_uses_constant_op() {
|
||
let payload = serde_json::json!({
|
||
"script_id": "00000000-0000-0000-0000-000000000000",
|
||
"args": {},
|
||
});
|
||
let (op, source) = describe_event(OutboxSourceKind::Invoke, &payload);
|
||
assert_eq!(op, "invoke_async");
|
||
assert_eq!(source, "invoke");
|
||
assert_eq!(
|
||
OutboxSourceKind::from_wire(&source),
|
||
Some(OutboxSourceKind::Invoke)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn describe_event_trigger_event_reads_top_level_fields() {
|
||
let payload = serde_json::json!({
|
||
"source": "kv",
|
||
"op": "set",
|
||
"collection": "users",
|
||
"key": "alice",
|
||
"value": {"name": "Alice"},
|
||
});
|
||
let (op, source) = describe_event(OutboxSourceKind::Kv, &payload);
|
||
assert_eq!(op, "set");
|
||
assert_eq!(source, "kv");
|
||
}
|
||
|
||
#[test]
|
||
fn classify_exec_error_covers_every_variant() {
|
||
let parse = classify_exec_error(&ExecError::Parse("nope".into()));
|
||
assert!(matches!(parse.0, InboxFailureKind::Validation));
|
||
let invalid = classify_exec_error(&ExecError::InvalidResponse("bad".into()));
|
||
assert!(matches!(invalid.0, InboxFailureKind::Validation));
|
||
let timeout = classify_exec_error(&ExecError::Timeout(30));
|
||
assert!(matches!(timeout.0, InboxFailureKind::Timeout));
|
||
let budget = classify_exec_error(&ExecError::OperationBudgetExceeded);
|
||
assert!(matches!(budget.0, InboxFailureKind::OperationBudget));
|
||
let runtime = classify_exec_error(&ExecError::Runtime("threw".into()));
|
||
assert!(matches!(runtime.0, InboxFailureKind::Runtime));
|
||
let overload = classify_exec_error(&ExecError::Overloaded {
|
||
retry_after_secs: 1,
|
||
});
|
||
assert!(matches!(overload.0, InboxFailureKind::Overloaded));
|
||
}
|
||
|
||
#[test]
|
||
fn failure_kind_status_codes_match_design_notes() {
|
||
assert_eq!(failure_kind_to_status(InboxFailureKind::Validation), 422);
|
||
assert_eq!(failure_kind_to_status(InboxFailureKind::Runtime), 502);
|
||
assert_eq!(failure_kind_to_status(InboxFailureKind::Overloaded), 503);
|
||
assert_eq!(failure_kind_to_status(InboxFailureKind::Timeout), 504);
|
||
assert_eq!(
|
||
failure_kind_to_status(InboxFailureKind::OperationBudget),
|
||
507
|
||
);
|
||
assert_eq!(failure_kind_to_status(InboxFailureKind::Platform), 500);
|
||
}
|
||
|
||
// ----------------------------------------------------------------
|
||
// Queue-arm fire-time `enabled` gate (§4.3).
|
||
//
|
||
// `dispatch_one_queue` re-reads the script fresh after claiming a
|
||
// message and, if it has been disabled since the per-tick
|
||
// `list_active_queue_consumers` snapshot, releases the claim (nack)
|
||
// without resolving a principal or executing. This test proves that
|
||
// gate end-to-end with in-memory stubs.
|
||
//
|
||
// Regression property: the principal resolver returns a valid
|
||
// `Principal` and the executor records `executed = true` before
|
||
// erroring, so DELETING the `if !script.enabled` gate makes the flow
|
||
// fall through to resolve → execute, flipping `executed` and failing
|
||
// `assert!(!executed)`. (Verified by temporarily removing the gate.)
|
||
// ----------------------------------------------------------------
|
||
mod queue_enabled_gate {
|
||
use super::*;
|
||
use std::sync::atomic::{AtomicBool, Ordering};
|
||
use std::sync::Arc;
|
||
|
||
use async_trait::async_trait;
|
||
use chrono::Utc;
|
||
use picloud_executor_core::{ExecError, ExecRequest, ExecResponse};
|
||
use picloud_orchestrator_core::{ExecutionGate, ExecutorClient};
|
||
use picloud_shared::{
|
||
AdminUserId, AppId, InstanceRole, Principal, Script, ScriptId, ScriptSandbox,
|
||
};
|
||
use uuid::Uuid;
|
||
|
||
use crate::abandoned_repo::{AbandonedRepo, NewAbandonedExecution};
|
||
use crate::dead_letter_repo::{DeadLetterRepo, NewDeadLetter};
|
||
use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow};
|
||
use crate::principal_resolver::{PrincipalResolver, PrincipalResolverError};
|
||
use crate::queue_repo::{ClaimedMessage, NewQueueMessage, QueueRepo, QueueStats};
|
||
use crate::repo::{NewScript, ScriptPatch, ScriptRepository, ScriptRepositoryError};
|
||
use crate::trigger_config::BackoffShape;
|
||
use crate::trigger_repo::{ActiveQueueConsumer, TriggerRepo};
|
||
|
||
// ---- ScriptRepository: only `get` is exercised. ----
|
||
pub(super) struct DisabledScriptRepo {
|
||
pub(super) script: Script,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl ScriptRepository for DisabledScriptRepo {
|
||
async fn get(&self, _id: ScriptId) -> Result<Option<Script>, ScriptRepositoryError> {
|
||
Ok(Some(self.script.clone()))
|
||
}
|
||
async fn get_by_name(
|
||
&self,
|
||
_app_id: AppId,
|
||
_name: &str,
|
||
) -> Result<Option<Script>, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn get_by_name_inherited(
|
||
&self,
|
||
_app_id: AppId,
|
||
_name: &str,
|
||
) -> Result<Option<Script>, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn is_invocable_by_app(
|
||
&self,
|
||
_script_id: ScriptId,
|
||
app_id: AppId,
|
||
) -> Result<bool, ScriptRepositoryError> {
|
||
Ok(self.script.is_owned_by_app(app_id))
|
||
}
|
||
async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_for_app(
|
||
&self,
|
||
_app_id: AppId,
|
||
) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_for_group(
|
||
&self,
|
||
_group_id: picloud_shared::GroupId,
|
||
) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_for_user(
|
||
&self,
|
||
_user_id: AdminUserId,
|
||
) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create(&self, _input: NewScript) -> Result<Script, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn update(
|
||
&self,
|
||
_id: ScriptId,
|
||
_patch: ScriptPatch,
|
||
) -> Result<Script, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn delete(&self, _id: ScriptId) -> Result<(), ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn count_routes_for_script(
|
||
&self,
|
||
_script_id: ScriptId,
|
||
) -> Result<i64, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn count_triggers_for_script(
|
||
&self,
|
||
_script_id: ScriptId,
|
||
) -> Result<i64, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_imports(
|
||
&self,
|
||
_script_id: ScriptId,
|
||
) -> Result<Vec<Script>, ScriptRepositoryError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
// ---- QueueRepo: `claim` returns the message, `nack` records. ----
|
||
struct ClaimNackQueue {
|
||
claimed: ClaimedMessage,
|
||
nacked: Arc<AtomicBool>,
|
||
released: Arc<AtomicBool>,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl QueueRepo for ClaimNackQueue {
|
||
async fn enqueue(
|
||
&self,
|
||
_msg: NewQueueMessage,
|
||
) -> Result<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn claim(
|
||
&self,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
) -> Result<Option<ClaimedMessage>, crate::queue_repo::QueueRepoError> {
|
||
Ok(Some(self.claimed.clone()))
|
||
}
|
||
async fn ack(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn nack(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
_retry_delay: chrono::Duration,
|
||
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||
self.nacked.store(true, Ordering::SeqCst);
|
||
Ok(true)
|
||
}
|
||
async fn release(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
_retry_delay: chrono::Duration,
|
||
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||
self.released.store(true, Ordering::SeqCst);
|
||
Ok(true)
|
||
}
|
||
async fn reclaim_visibility_timeouts(
|
||
&self,
|
||
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn depth(
|
||
&self,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn depth_pending(
|
||
&self,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_for_app(
|
||
&self,
|
||
_app_id: AppId,
|
||
) -> Result<Vec<(String, QueueStats)>, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn dead_letter(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
_trigger_id: Option<picloud_shared::TriggerId>,
|
||
_script_id: Option<ScriptId>,
|
||
_attempt: u32,
|
||
_first_attempt_at: chrono::DateTime<Utc>,
|
||
_last_error: &str,
|
||
) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
// ---- PrincipalResolver: returns a valid Principal (regression). ----
|
||
pub(super) struct OkPrincipals;
|
||
|
||
#[async_trait]
|
||
impl PrincipalResolver for OkPrincipals {
|
||
async fn resolve(
|
||
&self,
|
||
user_id: AdminUserId,
|
||
) -> Result<Principal, PrincipalResolverError> {
|
||
Ok(Principal {
|
||
user_id,
|
||
instance_role: InstanceRole::Owner,
|
||
scopes: None,
|
||
app_binding: None,
|
||
})
|
||
}
|
||
}
|
||
|
||
// ---- ExecutorClient: records execution then errors (regression). ----
|
||
pub(super) struct RecordingExecutor {
|
||
pub(super) executed: Arc<AtomicBool>,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl ExecutorClient for RecordingExecutor {
|
||
async fn execute(
|
||
&self,
|
||
_source: &str,
|
||
_req: ExecRequest,
|
||
_timeout: std::time::Duration,
|
||
) -> Result<ExecResponse, ExecError> {
|
||
self.executed.store(true, Ordering::SeqCst);
|
||
Err(ExecError::Runtime("stub".into()))
|
||
}
|
||
// Intentionally NOT overriding `execute_with_identity`: the
|
||
// default impl forwards to `execute`, which is what gate
|
||
// removal would reach.
|
||
}
|
||
|
||
// ---- Remaining deps: never exercised by the disabled path. ----
|
||
pub(super) struct UnusedTriggers;
|
||
|
||
#[async_trait]
|
||
impl TriggerRepo for UnusedTriggers {
|
||
async fn create_kv_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreateKvTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create_docs_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreateDocsTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create_dead_letter_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreateDeadLetterTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create_cron_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreateCronTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create_files_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreateFilesTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create_pubsub_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreatePubsubTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create_email_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreateEmailTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn email_inbound_target(
|
||
&self,
|
||
_trigger_id: picloud_shared::TriggerId,
|
||
) -> Result<
|
||
Option<crate::trigger_repo::EmailInboundTarget>,
|
||
crate::trigger_repo::TriggerRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_for_app(
|
||
&self,
|
||
_app_id: AppId,
|
||
) -> Result<Vec<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn get(
|
||
&self,
|
||
_id: picloud_shared::TriggerId,
|
||
) -> Result<Option<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn delete(
|
||
&self,
|
||
_id: picloud_shared::TriggerId,
|
||
) -> Result<bool, crate::trigger_repo::TriggerRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_matching_kv(
|
||
&self,
|
||
_app_id: AppId,
|
||
_collection: &str,
|
||
_op: picloud_shared::KvEventOp,
|
||
) -> Result<
|
||
Vec<crate::trigger_repo::KvTriggerMatch>,
|
||
crate::trigger_repo::TriggerRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_matching_docs(
|
||
&self,
|
||
_app_id: AppId,
|
||
_collection: &str,
|
||
_op: picloud_shared::DocsEventOp,
|
||
) -> Result<
|
||
Vec<crate::trigger_repo::DocsTriggerMatch>,
|
||
crate::trigger_repo::TriggerRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_matching_files(
|
||
&self,
|
||
_app_id: AppId,
|
||
_collection: &str,
|
||
_op: picloud_shared::FilesEventOp,
|
||
) -> Result<
|
||
Vec<crate::trigger_repo::FilesTriggerMatch>,
|
||
crate::trigger_repo::TriggerRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_matching_dead_letter(
|
||
&self,
|
||
_app_id: AppId,
|
||
_source: &str,
|
||
_trigger_id: Option<picloud_shared::TriggerId>,
|
||
_script_id: Option<ScriptId>,
|
||
) -> Result<
|
||
Vec<crate::trigger_repo::DeadLetterTriggerMatch>,
|
||
crate::trigger_repo::TriggerRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn create_queue_trigger(
|
||
&self,
|
||
_app_id: AppId,
|
||
_req: crate::trigger_repo::CreateQueueTrigger,
|
||
) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_active_queue_consumers(
|
||
&self,
|
||
) -> Result<Vec<ActiveQueueConsumer>, crate::trigger_repo::TriggerRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn touch_queue_trigger_last_fired_at(
|
||
&self,
|
||
_trigger_id: picloud_shared::TriggerId,
|
||
_at: chrono::DateTime<Utc>,
|
||
) -> Result<(), crate::trigger_repo::TriggerRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
pub(super) struct UnusedDeadLetters;
|
||
|
||
#[async_trait]
|
||
impl DeadLetterRepo for UnusedDeadLetters {
|
||
async fn insert(
|
||
&self,
|
||
_row: NewDeadLetter,
|
||
) -> Result<picloud_shared::DeadLetterId, crate::dead_letter_repo::DeadLetterRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn get(
|
||
&self,
|
||
_id: picloud_shared::DeadLetterId,
|
||
) -> Result<
|
||
Option<crate::dead_letter_repo::DeadLetterRow>,
|
||
crate::dead_letter_repo::DeadLetterRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_for_app(
|
||
&self,
|
||
_app_id: AppId,
|
||
_unresolved_only: bool,
|
||
_limit: i64,
|
||
_offset: i64,
|
||
) -> Result<
|
||
Vec<crate::dead_letter_repo::DeadLetterRow>,
|
||
crate::dead_letter_repo::DeadLetterRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn unresolved_count(
|
||
&self,
|
||
_app_id: AppId,
|
||
) -> Result<i64, crate::dead_letter_repo::DeadLetterRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn resolve(
|
||
&self,
|
||
_id: picloud_shared::DeadLetterId,
|
||
_reason: &str,
|
||
) -> Result<(), crate::dead_letter_repo::DeadLetterRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn gc(
|
||
&self,
|
||
_older_than: chrono::DateTime<Utc>,
|
||
_limit: i64,
|
||
) -> Result<u64, crate::dead_letter_repo::DeadLetterRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
struct UnusedOutbox;
|
||
|
||
#[async_trait]
|
||
impl OutboxRepo for UnusedOutbox {
|
||
async fn reclaim_stale_claims(
|
||
&self,
|
||
_timeout_secs: u32,
|
||
) -> Result<u64, crate::outbox_repo::OutboxRepoError> {
|
||
Ok(0)
|
||
}
|
||
async fn insert(
|
||
&self,
|
||
_row: NewOutboxRow,
|
||
) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn claim_due(
|
||
&self,
|
||
_claimed_by: &str,
|
||
_limit: i64,
|
||
) -> Result<Vec<OutboxRow>, crate::outbox_repo::OutboxRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn delete(&self, _id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn reschedule(
|
||
&self,
|
||
_id: Uuid,
|
||
_attempt_count: u32,
|
||
_next_attempt_at: chrono::DateTime<Utc>,
|
||
) -> Result<(), crate::outbox_repo::OutboxRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
pub(super) struct UnusedAbandoned;
|
||
|
||
#[async_trait]
|
||
impl AbandonedRepo for UnusedAbandoned {
|
||
async fn insert(
|
||
&self,
|
||
_row: NewAbandonedExecution,
|
||
) -> Result<Uuid, crate::abandoned_repo::AbandonedRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn gc(
|
||
&self,
|
||
_older_than: chrono::DateTime<Utc>,
|
||
_limit: i64,
|
||
) -> Result<u64, crate::abandoned_repo::AbandonedRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
pub(super) struct UnusedLogSink;
|
||
|
||
#[async_trait]
|
||
impl ExecutionLogSink for UnusedLogSink {
|
||
async fn record(
|
||
&self,
|
||
_log: picloud_shared::ExecutionLog,
|
||
) -> Result<(), picloud_shared::LogSinkError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
pub(super) struct UnusedInbox;
|
||
|
||
#[async_trait]
|
||
impl InboxResolver for UnusedInbox {
|
||
async fn deliver(
|
||
&self,
|
||
_inbox_id: Uuid,
|
||
_result: picloud_shared::InboxResult,
|
||
) -> picloud_shared::InboxDeliveryOutcome {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
pub(super) fn disabled_script(app_id: AppId) -> Script {
|
||
Script {
|
||
id: ScriptId::new(),
|
||
app_id: Some(app_id),
|
||
group_id: None,
|
||
name: "worker".into(),
|
||
description: None,
|
||
version: 1,
|
||
source: "0".into(),
|
||
kind: picloud_shared::ScriptKind::Endpoint,
|
||
timeout_seconds: 30,
|
||
memory_limit_mb: 64,
|
||
sandbox: ScriptSandbox::default(),
|
||
enabled: false,
|
||
created_at: Utc::now(),
|
||
updated_at: Utc::now(),
|
||
}
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn disabled_queue_consumer_releases_claim_without_executing() {
|
||
let app_id = AppId::new();
|
||
let script = disabled_script(app_id);
|
||
|
||
let claimed = ClaimedMessage {
|
||
id: picloud_shared::QueueMessageId::new(),
|
||
app_id,
|
||
queue_name: "jobs".into(),
|
||
payload: serde_json::Value::Null,
|
||
enqueued_at: Utc::now(),
|
||
attempt: 1,
|
||
max_attempts: 5,
|
||
claim_token: Uuid::new_v4(),
|
||
};
|
||
|
||
let consumer = ActiveQueueConsumer {
|
||
trigger_id: picloud_shared::TriggerId::new(),
|
||
app_id,
|
||
script_id: script.id,
|
||
queue_name: "jobs".into(),
|
||
visibility_timeout_secs: 30,
|
||
retry_max_attempts: 5,
|
||
retry_backoff: BackoffShape::Exponential,
|
||
retry_base_ms: 1000,
|
||
registered_by_principal: AdminUserId::new(),
|
||
shared_group: None,
|
||
};
|
||
|
||
let nacked = Arc::new(AtomicBool::new(false));
|
||
let released = Arc::new(AtomicBool::new(false));
|
||
let executed = Arc::new(AtomicBool::new(false));
|
||
|
||
let dispatcher = Dispatcher {
|
||
outbox: Arc::new(UnusedOutbox),
|
||
triggers: Arc::new(UnusedTriggers),
|
||
scripts: Arc::new(DisabledScriptRepo {
|
||
script: script.clone(),
|
||
}),
|
||
dead_letters: Arc::new(UnusedDeadLetters),
|
||
abandoned: Arc::new(UnusedAbandoned),
|
||
principals: Arc::new(OkPrincipals),
|
||
executor: Arc::new(RecordingExecutor {
|
||
executed: executed.clone(),
|
||
}),
|
||
gate: Arc::new(ExecutionGate::new(1)),
|
||
log_sink: Arc::new(UnusedLogSink),
|
||
inbox: Arc::new(UnusedInbox),
|
||
queue: Arc::new(ClaimNackQueue {
|
||
claimed,
|
||
nacked: nacked.clone(),
|
||
released: released.clone(),
|
||
}),
|
||
group_queue: Arc::new(NoopGroupQueue),
|
||
config: TriggerConfig::from_env(),
|
||
instance_id: "test-instance".into(),
|
||
};
|
||
|
||
let result = dispatcher.dispatch_one_queue(&consumer).await;
|
||
|
||
// 1. The disabled path returns Ok(()).
|
||
assert!(result.is_ok(), "dispatch_one_queue returned {result:?}");
|
||
// 2. The claim was RELEASED (transient — the handler never ran), not
|
||
// nacked: a disabled-at-fire release must not burn the retry
|
||
// budget (audit fix #4).
|
||
assert!(
|
||
released.load(Ordering::SeqCst),
|
||
"expected a transient release for a disabled consumer"
|
||
);
|
||
assert!(
|
||
!nacked.load(Ordering::SeqCst),
|
||
"a disabled-at-fire release must NOT count as a nack (retry budget)"
|
||
);
|
||
// 3. The executor was never reached. If the `if !script.enabled`
|
||
// gate is deleted, the flow falls through to resolve →
|
||
// execute and this flips to true.
|
||
assert!(
|
||
!executed.load(Ordering::SeqCst),
|
||
"executor ran for a disabled queue consumer; fire-time gate missing"
|
||
);
|
||
}
|
||
}
|
||
|
||
// ----------------------------------------------------------------
|
||
// OUTBOX (async-HTTP) arm fire-time `enabled` gate (§4.3).
|
||
//
|
||
// `dispatch_one` builds an `ExecRequest` from an HTTP outbox row via
|
||
// `build_http_request`, which sets `resolved.active = script.enabled`
|
||
// but does NOT itself reject a disabled script. The unified gate at
|
||
// `if !resolved.active` then drops the row (delete) without executing.
|
||
// This test proves that gate end-to-end for an HTTP-source row whose
|
||
// target script was disabled after the row was enqueued.
|
||
//
|
||
// Regression property (mirrors the queue test): `RecordingExecutor`
|
||
// flips `executed = true` before erroring, so DELETING the
|
||
// `if !resolved.active` gate makes the flow fall through to execute,
|
||
// flipping `executed` and failing `assert!(!executed)`.
|
||
// ----------------------------------------------------------------
|
||
mod outbox_enabled_gate {
|
||
use super::*;
|
||
use std::sync::atomic::{AtomicBool, Ordering};
|
||
use std::sync::{Arc, Mutex};
|
||
|
||
use async_trait::async_trait;
|
||
use chrono::Utc;
|
||
use picloud_orchestrator_core::ExecutionGate;
|
||
use picloud_shared::AppId;
|
||
use uuid::Uuid;
|
||
|
||
use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow, OutboxSourceKind};
|
||
|
||
// Reuse the stub trait impls + helpers from the queue test so the
|
||
// two gate tests share one set of in-memory deps.
|
||
use super::queue_enabled_gate::{
|
||
disabled_script, DisabledScriptRepo, OkPrincipals, RecordingExecutor, UnusedAbandoned,
|
||
UnusedDeadLetters, UnusedInbox, UnusedLogSink, UnusedTriggers,
|
||
};
|
||
|
||
// QueueRepo is never reached on the outbox arm — a panic stub keeps
|
||
// the surface honest without pulling the queue test's claim stub.
|
||
struct UnusedQueue;
|
||
|
||
#[async_trait]
|
||
impl crate::queue_repo::QueueRepo for UnusedQueue {
|
||
async fn enqueue(
|
||
&self,
|
||
_msg: crate::queue_repo::NewQueueMessage,
|
||
) -> Result<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn claim(
|
||
&self,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
) -> Result<Option<crate::queue_repo::ClaimedMessage>, crate::queue_repo::QueueRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn ack(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn nack(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
_retry_delay: chrono::Duration,
|
||
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn release(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
_retry_delay: chrono::Duration,
|
||
) -> Result<bool, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn reclaim_visibility_timeouts(
|
||
&self,
|
||
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn depth(
|
||
&self,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn depth_pending(
|
||
&self,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
) -> Result<u64, crate::queue_repo::QueueRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn list_for_app(
|
||
&self,
|
||
_app_id: AppId,
|
||
) -> Result<
|
||
Vec<(String, crate::queue_repo::QueueStats)>,
|
||
crate::queue_repo::QueueRepoError,
|
||
> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn dead_letter(
|
||
&self,
|
||
_message_id: QueueMessageId,
|
||
_claim_token: Uuid,
|
||
_app_id: AppId,
|
||
_queue_name: &str,
|
||
_trigger_id: Option<picloud_shared::TriggerId>,
|
||
_script_id: Option<picloud_shared::ScriptId>,
|
||
_attempt: u32,
|
||
_first_attempt_at: chrono::DateTime<Utc>,
|
||
_last_error: &str,
|
||
) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
|
||
{
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
// OutboxRepo whose `delete` records the id it was called with; the
|
||
// other methods are never reached on the disabled-drop path.
|
||
struct RecordingOutbox {
|
||
deleted: Arc<Mutex<Option<Uuid>>>,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl OutboxRepo for RecordingOutbox {
|
||
async fn reclaim_stale_claims(
|
||
&self,
|
||
_timeout_secs: u32,
|
||
) -> Result<u64, crate::outbox_repo::OutboxRepoError> {
|
||
Ok(0)
|
||
}
|
||
async fn insert(
|
||
&self,
|
||
_row: NewOutboxRow,
|
||
) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn claim_due(
|
||
&self,
|
||
_claimed_by: &str,
|
||
_limit: i64,
|
||
) -> Result<Vec<OutboxRow>, crate::outbox_repo::OutboxRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
async fn delete(&self, id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> {
|
||
*self.deleted.lock().unwrap() = Some(id);
|
||
Ok(())
|
||
}
|
||
async fn reschedule(
|
||
&self,
|
||
_id: Uuid,
|
||
_attempt_count: u32,
|
||
_next_attempt_at: chrono::DateTime<Utc>,
|
||
) -> Result<(), crate::outbox_repo::OutboxRepoError> {
|
||
unimplemented!("not used by this test")
|
||
}
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn disabled_http_outbox_row_is_dropped_without_executing() {
|
||
let app_id = AppId::new();
|
||
let script = disabled_script(app_id);
|
||
|
||
// Minimal valid HttpDispatchPayload (see shared::outbox_writer).
|
||
let payload = serde_json::json!({
|
||
"script_name": "worker",
|
||
"path": "/hook",
|
||
"method": "POST",
|
||
"headers": {},
|
||
"body": null,
|
||
"params": {},
|
||
"query": {},
|
||
"rest": "",
|
||
"timeout_seconds": 30,
|
||
});
|
||
|
||
let row_id = Uuid::new_v4();
|
||
let row = OutboxRow {
|
||
id: row_id,
|
||
// Same app_id as the script so the same-app guard passes.
|
||
app_id,
|
||
source_kind: OutboxSourceKind::Http,
|
||
trigger_id: None,
|
||
script_id: Some(script.id),
|
||
reply_to: None,
|
||
payload,
|
||
origin_principal: None,
|
||
trigger_depth: 0,
|
||
root_execution_id: None,
|
||
attempt_count: 0,
|
||
next_attempt_at: Utc::now(),
|
||
created_at: Utc::now(),
|
||
};
|
||
|
||
let deleted = Arc::new(Mutex::new(None));
|
||
let executed = Arc::new(AtomicBool::new(false));
|
||
|
||
let dispatcher = Dispatcher {
|
||
outbox: Arc::new(RecordingOutbox {
|
||
deleted: deleted.clone(),
|
||
}),
|
||
triggers: Arc::new(UnusedTriggers),
|
||
scripts: Arc::new(DisabledScriptRepo {
|
||
script: script.clone(),
|
||
}),
|
||
dead_letters: Arc::new(UnusedDeadLetters),
|
||
abandoned: Arc::new(UnusedAbandoned),
|
||
principals: Arc::new(OkPrincipals),
|
||
executor: Arc::new(RecordingExecutor {
|
||
executed: executed.clone(),
|
||
}),
|
||
gate: Arc::new(ExecutionGate::new(1)),
|
||
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(),
|
||
};
|
||
|
||
let result = dispatcher.dispatch_one(row).await;
|
||
|
||
// 1. The disabled-drop path returns Ok(()).
|
||
assert!(result.is_ok(), "dispatch_one returned {result:?}");
|
||
// 2. The outbox row was deleted with its own id.
|
||
assert_eq!(
|
||
*deleted.lock().unwrap(),
|
||
Some(row_id),
|
||
"expected the disabled HTTP outbox row to be deleted"
|
||
);
|
||
// 3. The executor was never reached. If the `if !resolved.active`
|
||
// gate at dispatch_one is deleted, the flow falls through to
|
||
// execute and this flips to true.
|
||
assert!(
|
||
!executed.load(Ordering::SeqCst),
|
||
"executor ran for a disabled HTTP outbox row; fire-time gate missing"
|
||
);
|
||
}
|
||
}
|
||
}
|