1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
//! Dead-letter inspection endpoint — the discovery half of outbox redrive.
//!
//! `POST /outbox/dead-letters` lists the dead-lettered rows of one workflow: the durable evidence
//! behind a workflow that is still `Running` while an activity never returned. Each row reports
//! `failure_delivered`, which is the fact an operator needs before acting:
//!
//! - `false` — the workflow was never told the activity failed. It is still waiting, judgment never
//! passed, and `ClusterCommand::RedriveOutboxRow` will re-queue the work.
//! - `true` — the failure reached the workflow, which has already reacted, and that reaction is
//! recorded history. Redrive REFUSES such a row: re-running it would re-execute a possibly
//! non-idempotent activity behind a recorded judgment.
//!
//! Auth is the deployment-wide deploy grant — the same authority that may issue the redrive itself
//! (`/cluster/command`), so what an operator can see here is exactly what they can act on there.
use aion_core::WorkflowId;
use aion_proto::WireError;
use axum::{Json, extract::State};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use super::auth::HttpCaller;
use super::error::HttpWireError;
use crate::ServerState;
/// Request body: the workflow whose dead letters are being inspected.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct DeadLettersRequest {
/// The workflow to enumerate dead-lettered outbox rows for.
pub workflow_id: String,
}
/// One dead-lettered outbox row, as an operator sees it.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct DeadLetterRow {
/// The row's durable idempotency key (`"{workflow_id}:{ordinal}"`).
pub dispatch_key: String,
/// The fan-out ordinal within the workflow.
pub ordinal: u64,
/// The activity type that never got through.
pub activity_type: String,
/// Routing identity the row was staged with.
pub namespace: String,
/// Pool/flavour selector the row was staged with.
pub task_queue: String,
/// Optional node affinity, or `null` for an unpinned row.
pub node: Option<String>,
/// Dispatch attempts spent before the row dead-lettered.
pub attempt: u32,
/// Whether the workflow was TOLD this activity failed. `true` means judgment has passed and
/// redrive will refuse the row.
pub failure_delivered: bool,
/// Whether an ordinary (non-forced) redrive would accept this row.
pub redrivable: bool,
}
/// Response body: the workflow's dead letters, ordered by ordinal.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct DeadLettersResponse {
/// The dead-lettered rows.
pub dead_letters: Vec<DeadLetterRow>,
}
/// `POST /outbox/dead-letters`.
pub(crate) async fn list_dead_letters(
State(state): State<ServerState>,
HttpCaller(caller): HttpCaller,
Json(request): Json<DeadLettersRequest>,
) -> Result<Json<DeadLettersResponse>, HttpWireError> {
// GATE FIRST, exactly like the cluster-command seam: the dead-letter table is a
// deployment-wide operational view, not a per-namespace read.
super::cluster_command::deploy_gate(&caller)?;
let Some(outbox_store) = state.outbox_store() else {
return Err(HttpWireError(WireError::invalid_state(
"the durable outbox is not commissioned on this server, so it has no dead letters",
)));
};
let workflow_id = Uuid::parse_str(&request.workflow_id)
.map(WorkflowId::new)
.map_err(|_error| HttpWireError(WireError::invalid_input("workflow_id must be a UUID")))?;
let rows = crate::worker::list_dead_letters(outbox_store.as_ref(), &workflow_id)
.await
.map_err(|error| HttpWireError(crate::ServerError::from(error).to_wire_error()))?;
Ok(Json(DeadLettersResponse {
dead_letters: rows
.into_iter()
.map(|row| DeadLetterRow {
dispatch_key: row.dispatch_key,
ordinal: row.ordinal,
activity_type: row.activity_type,
namespace: row.namespace,
task_queue: row.task_queue,
node: row.node,
attempt: row.attempt,
failure_delivered: row.failure_delivered,
redrivable: !row.failure_delivered,
})
.collect(),
}))
}