From 649246213e14a00862c0b01a31481cd280295673 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Wed, 10 Jun 2026 21:22:15 +0200 Subject: [PATCH] fix(stage-3): queue DL fan-out + emit-failure visibility MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the queue DL fan-out gap (audit Medium) and surfaces the KV/docs/files non-transactional emit gap to operators. - handle_queue_failure previously called queue.dead_letter to write the DL row but never invoked fan_out_dead_letter. The comment claimed "the outbox arm fires registered dead_letter handlers off the new row" — but queue DL rows are written via a separate path that bypasses the outbox entirely, so handlers filtered on source="queue" sat idle forever. Refactor fan_out_dead_letter to take a DeadLetterFanOutCtx struct so both the outbox arm and the queue arm can call it; the queue arm constructs a TriggerEvent::Queue from the claimed message and passes it through. - New integration test queue_dead_letter_fans_out_to_dead_letter_handler registers a dead_letter trigger filtered on "queue", forces queue exhaustion, asserts the handler fires with the correctly-shaped event. - KV/docs/files services committed the data write then ran events.emit as a separate operation, logging-and-swallowing on failure. Bump the six call sites from tracing::warn to tracing::error with an event_emit_failure=true marker so operators can grep them. The full single-tx repo refactor (extending ServiceEventEmitter with emit_in_tx + tx-aware *_repo methods) is documented in kv_service.rs as a v1.2 follow-up — it's a meaningful redesign that deserves its own pass (pubsub_service::fan_out_publish is the reference shape). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/manager-core/src/dispatcher.rs | 169 +++++++++++++++++------ crates/manager-core/src/docs_service.rs | 6 +- crates/manager-core/src/files_service.rs | 2 +- crates/manager-core/src/kv_service.rs | 17 ++- crates/picloud/tests/queue_e2e.rs | 65 +++++++++ 5 files changed, 212 insertions(+), 47 deletions(-) diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 467f71e..355c55c 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -68,6 +68,33 @@ pub struct Dispatcher { /// 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, + /// Original script id (for script_id_filter matching). + script_id: Option, + first_attempt_at: DateTime, + last_attempt_at: DateTime, + trigger_depth: u32, + root_execution_id: Option, +} + /// 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`. @@ -369,10 +396,14 @@ impl Dispatcher { return; } // Exhausted. Dead-letter inline (the helper deletes the queue row + - // writes a dead_letters row in one transaction). The existing - // fan_out_dead_letter path on the dispatcher's outbox arm fires - // registered dead_letter handlers off the new row. - if let Err(e) = self + // 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(); + let dl_id = match self .queue .dead_letter( claimed.id, @@ -383,11 +414,44 @@ impl Dispatcher { Some(consumer.script_id), claimed.attempt, claimed.enqueued_at, - &err.to_string(), + &last_error, ) .await { - tracing::error!(?e, "queue dead-letter write failed"); + Ok(dl_id) => Some(dl_id), + Err(e) => { + tracing::error!(?e, "queue dead-letter write failed"); + None + } + }; + 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}. } @@ -754,6 +818,7 @@ impl Dispatcher { Ok(()) } + #[allow(clippy::too_many_lines)] async fn handle_failure( &self, row: &OutboxRow, @@ -871,8 +936,35 @@ impl Dispatcher { // top of this function, so this fan-out is only reached for // non-handler executions. if let Some(dl_id) = dl_id { - self.fan_out_dead_letter(row, resolved, dl_id, &source, attempt, &err, now) - .await; + // 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::(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 @@ -886,31 +978,30 @@ impl Dispatcher { /// 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). - #[allow(clippy::too_many_arguments)] - async fn fan_out_dead_letter( - &self, - row: &OutboxRow, - resolved: &ResolvedTrigger, - dead_letter_id: DeadLetterId, - source: &str, - attempt: u32, - err: &ExecError, - now: DateTime, - ) { - // 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). - let Ok(original) = serde_json::from_value::(row.payload.clone()) else { - tracing::warn!( - outbox_id = %row.id, - "dead-letter payload is not a TriggerEvent; skipping handler fan-out" - ); - return; - }; + /// + /// 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(row.app_id, source, row.trigger_id, Some(resolved.script_id)) + .list_matching_dead_letter(app_id, &source, trigger_id, script_id) .await { Ok(m) => m, @@ -924,12 +1015,12 @@ impl Dispatcher { let event = TriggerEvent::DeadLetter { dead_letter_id, original: Box::new(original.clone()), - 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, + 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, @@ -941,15 +1032,15 @@ impl Dispatcher { if let Err(e) = self .outbox .insert(NewOutboxRow { - app_id: row.app_id, + 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: row.trigger_depth.saturating_add(1), - root_execution_id: row.root_execution_id, + trigger_depth: trigger_depth.saturating_add(1), + root_execution_id, }) .await { diff --git a/crates/manager-core/src/docs_service.rs b/crates/manager-core/src/docs_service.rs index 30bb17a..42308e2 100644 --- a/crates/manager-core/src/docs_service.rs +++ b/crates/manager-core/src/docs_service.rs @@ -184,7 +184,7 @@ impl DocsService for DocsServiceImpl { ) .await { - tracing::warn!(error = %e, source = "docs", op = "create", "event emit failed"); + tracing::error!(error = %e, source = "docs", op = "create", event_emit_failure = true, "event emit failed"); } Ok(row.id) } @@ -262,7 +262,7 @@ impl DocsService for DocsServiceImpl { ) .await { - tracing::warn!(error = %e, source = "docs", op = "update", "event emit failed"); + tracing::error!(error = %e, source = "docs", op = "update", event_emit_failure = true, "event emit failed"); } Ok(()) } @@ -291,7 +291,7 @@ impl DocsService for DocsServiceImpl { ) .await { - tracing::warn!(error = %e, source = "docs", op = "delete", "event emit failed"); + tracing::error!(error = %e, source = "docs", op = "delete", event_emit_failure = true, "event emit failed"); } } Ok(was_present) diff --git a/crates/manager-core/src/files_service.rs b/crates/manager-core/src/files_service.rs index ada7220..8464db0 100644 --- a/crates/manager-core/src/files_service.rs +++ b/crates/manager-core/src/files_service.rs @@ -97,7 +97,7 @@ impl FilesServiceImpl { ) .await { - tracing::warn!(error = %e, source = "files", op, "event emit failed"); + tracing::error!(error = %e, source = "files", op, event_emit_failure = true, "event emit failed"); } } } diff --git a/crates/manager-core/src/kv_service.rs b/crates/manager-core/src/kv_service.rs index d6134d2..71b07e8 100644 --- a/crates/manager-core/src/kv_service.rs +++ b/crates/manager-core/src/kv_service.rs @@ -155,8 +155,17 @@ impl KvService for KvServiceImpl { "insert" }; // Emit unconditionally; the noop emitter drops it, the outbox - // emitter persists it. Best-effort: a failed emit is logged - // but does not roll back the write. + // emitter persists it. + // + // Audit finding (Medium): this is non-transactional with the + // data write — the row above has already committed by the time + // emit() runs, so an emit failure means triggers silently + // don't fire. Logged at `error` level (with + // `event_emit_failure = true` for grepability); operators see + // the gap. The full transactional refactor — extending the + // ServiceEventEmitter trait with `emit_in_tx` and the KvRepo + // surface with `set_with_tx` — is deferred to a v1.2 design + // pass (pubsub_repo::fan_out_publish is the reference shape). if let Err(e) = self .events .emit( @@ -172,7 +181,7 @@ impl KvService for KvServiceImpl { ) .await { - tracing::warn!(error = %e, source = "kv", op, "event emit failed"); + tracing::error!(error = %e, source = "kv", op, event_emit_failure = true, "event emit failed"); } Ok(()) } @@ -198,7 +207,7 @@ impl KvService for KvServiceImpl { ) .await { - tracing::warn!(error = %e, source = "kv", op = "delete", "event emit failed"); + tracing::error!(error = %e, source = "kv", op = "delete", event_emit_failure = true, "event emit failed"); } } Ok(was_present) diff --git a/crates/picloud/tests/queue_e2e.rs b/crates/picloud/tests/queue_e2e.rs index 019302d..f57837b 100644 --- a/crates/picloud/tests/queue_e2e.rs +++ b/crates/picloud/tests/queue_e2e.rs @@ -235,6 +235,71 @@ async fn queue_receive_dead_letters_after_max_attempts() { ); } +/// Audit fix: dead_letter triggers filtered on `source = "queue"` MUST +/// fire when a queue message exhausts its retries. Before the queue arm +/// called `fan_out_dead_letter`, the DL row was written but no handler +/// delivery was enqueued, so registered `dead_letter` handlers sat idle. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn queue_dead_letter_fans_out_to_dead_letter_handler() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let (server, app_id) = server_for(pool.clone(), "qdlfan").await; + + // DL handler writes to a different KV key so we don't race the + // failing handler's own marker write. + let dl_handler_source = r#" + kv::collection("e2e_markers").set("dl_marker", ctx.event); + #{ ok: true } + "#; + let dl_handler = create_script(&server, &app_id, "dl-handler", dl_handler_source).await; + server + .post(&format!("/api/v1/admin/apps/{app_id}/triggers/dead_letter")) + .json(&json!({ "script_id": dl_handler, "source_filter": "queue" })) + .await + .assert_status(axum::http::StatusCode::CREATED); + + // Failing consumer with max_attempts=1 so we exhaust quickly. + let failing = create_script(&server, &app_id, "failing", THROW_HANDLER).await; + server + .post(&format!("/api/v1/admin/apps/{app_id}/triggers/queue")) + .json(&json!({ + "script_id": failing, + "queue_name": "failing-fanout", + "visibility_timeout_secs": 30, + "max_attempts": 1 + })) + .await + .assert_status(axum::http::StatusCode::CREATED); + + enqueue_directly(&pool, &app_id, "failing-fanout", json!({ "x": 1 })).await; + + // Poll the DL handler's marker. + for _ in 0..200 { + let row: Option<(Value,)> = sqlx::query_as( + "SELECT value FROM kv_entries WHERE app_id = $1 \ + AND collection = 'e2e_markers' AND key = 'dl_marker'", + ) + .bind(Uuid::parse_str(&app_id).expect("uuid")) + .fetch_optional(&pool) + .await + .expect("kv read"); + if let Some((event,)) = row { + // Sanity: the dead-letter event names the queue source and + // nests the original Queue event verbatim. + assert_eq!(event["source"], "dead_letter"); + assert_eq!(event["dead_letter"]["original"]["source"], "queue"); + assert_eq!( + event["dead_letter"]["original"]["queue"]["queue_name"], + "failing-fanout" + ); + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("dead_letter handler never fired for queue-exhausted message"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn queue_one_consumer_per_queue_rejected() { let Some(pool) = pool_or_skip().await else {