fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(stage-3): queue DL fan-out + emit-failure visibility

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-10 21:22:15 +02:00
parent f466b2a15e
commit 649246213e
5 changed files with 212 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
crates/manager-core/src/dispatcher.rs
View File

@@ -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<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`.
@@ -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::<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
@@ -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<Utc>,
) {
// 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::<TriggerEvent>(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
{

6
crates/manager-core/src/docs_service.rs
View File

@@ -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)

2
crates/manager-core/src/files_service.rs
View File

@@ -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");
}
}
}

17
crates/manager-core/src/kv_service.rs
View File

@@ -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)

65
crates/picloud/tests/queue_e2e.rs
View File

@@ -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 {