aion_server/worker/
outbox_settle.rs

1//! Terminal-workflow outbox settlement sweep (#253).
2//!
3//! The durable outbox table — not any in-memory cache — is the source of truth
4//! for what may still be delivered, so it must be kept consistent with workflow
5//! terminality. The Recorder settles a workflow's live rows at terminal-record
6//! time (the live window); this module is the BOOT/ADOPTION backstop for the
7//! restart window the incident fell through: a workflow that reached a durable
8//! terminal without its rows being settled (a settle-hook store error, or a
9//! node that died between the terminal append and the settle) must have those
10//! rows retired BEFORE the dispatcher's first claim, or a redialed worker
11//! serves a full zombie round for a dead workflow.
12//!
13//! Liveness is projected from event history with the SAME projection
14//! `list_active` and pause validation use ([`status_from_events`]) — NOT
15//! `list_active` membership, which would wrongly classify `Paused` runs (live,
16//! merely held) as dead. Only the four hard terminals settle; `ContinuedAsNew`
17//! never reaches the settle set because a continued chain's later
18//! `WorkflowStarted` projects the chain `Running`, and a chain whose
19//! replacement run has not started yet is still in flight.
20
21use aion_core::{WorkflowStatus, status_from_events};
22use aion_store::{EventStore, OutboxStore, StoreError};
23use tracing::info;
24
25/// Whether `status` is a settle-eligible workflow terminal: the four hard
26/// terminals, never `ContinuedAsNew` (the workflow continues) and never
27/// `Paused`/`Running` (live).
28#[must_use]
29pub fn is_settle_terminal(status: WorkflowStatus) -> bool {
30    matches!(
31        status,
32        WorkflowStatus::Completed
33            | WorkflowStatus::Failed
34            | WorkflowStatus::Cancelled
35            | WorkflowStatus::TimedOut
36    )
37}
38
39/// Settle every terminal workflow's live outbox rows to `Cancelled`, returning
40/// the settled `dispatch_key`s (#253).
41///
42/// Enumerates the distinct workflow ids owning any `Pending`/`Claimed` row,
43/// projects each workflow's status ONCE from its full recorded history, and
44/// retires the rows of workflows whose projected status is a hard terminal.
45/// Bounded cost: proportional to workflows with live rows; runs once per boot
46/// (before the dispatcher's first claim) and once per shard adoption (after
47/// the fence widened the owned scope).
48///
49/// # Errors
50///
51/// Returns [`StoreError`] when the enumeration, a history read, or a settle
52/// fails; the caller logs and continues (the settle-at-terminal hook and the
53/// stale-claim reconciler gate remain as repair paths).
54pub async fn settle_terminal_outbox_rows(
55    event_store: &dyn EventStore,
56    outbox_store: &dyn OutboxStore,
57) -> Result<Vec<String>, StoreError> {
58    let workflow_ids = outbox_store.list_unsettled_outbox_workflow_ids().await?;
59    let mut settled = Vec::new();
60    for workflow_id in workflow_ids {
61        let history = event_store.read_history(&workflow_id).await?;
62        let status = status_from_events(&history);
63        if !is_settle_terminal(status) {
64            continue;
65        }
66        let keys = outbox_store
67            .cancel_outbox_rows_for_workflow(&workflow_id)
68            .await?;
69        if !keys.is_empty() {
70            info!(
71                workflow_id = %workflow_id,
72                projected_status = ?status,
73                settled = keys.len(),
74                dispatch_keys = ?keys,
75                "settled outbox rows for terminal workflow"
76            );
77            settled.extend(keys);
78        }
79    }
80    Ok(settled)
81}
82
83#[cfg(test)]
84mod tests {
85    use super::is_settle_terminal;
86    use aion_core::WorkflowStatus;
87
88    #[test]
89    fn only_hard_terminals_are_settle_eligible() {
90        assert!(is_settle_terminal(WorkflowStatus::Completed));
91        assert!(is_settle_terminal(WorkflowStatus::Failed));
92        assert!(is_settle_terminal(WorkflowStatus::Cancelled));
93        assert!(is_settle_terminal(WorkflowStatus::TimedOut));
94        // Live states: Running plainly, Paused is live-but-held (#204), and a
95        // ContinuedAsNew projection means the replacement run has not started
96        // yet — the chain is still in flight.
97        assert!(!is_settle_terminal(WorkflowStatus::Running));
98        assert!(!is_settle_terminal(WorkflowStatus::Paused));
99        assert!(!is_settle_terminal(WorkflowStatus::ContinuedAsNew));
100    }
101}
aion_server/worker/outbox_settle.rs

aion_server/worker/
outbox_settle.rs
