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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
//! `NonDeterminismError` and `DurabilityError` taxonomy.
use aion_core::{SearchAttributeError, WorkflowId};
use aion_store::StoreError;
/// A deterministic replay mismatch between the workflow command stream and recorded history.
#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq)]
#[error(
"non-determinism in workflow {workflow_id} at sequence {seq}: expected {expected}, found {found}"
)]
pub struct NonDeterminismError {
/// Workflow whose recorded history diverged from the replayed command stream.
pub workflow_id: WorkflowId,
/// Sequence position of the recorded event at the cursor mismatch.
pub seq: u64,
/// Shape of the command the workflow issued, including family and correlation key.
pub expected: String,
/// Shape of the recorded event found at the cursor position, including family and key.
pub found: String,
}
/// Errors returned by durability recording, replay, and recovery operations.
#[derive(thiserror::Error, Debug)]
pub enum DurabilityError {
/// The backing event store rejected or failed a durability operation.
#[error("store error: {0}")]
Store(#[from] StoreError),
/// Replay detected that workflow code no longer matches recorded history.
#[error("non-determinism violation: {0}")]
NonDeterminism(#[from] NonDeterminismError),
/// Recorded history is malformed or internally inconsistent.
#[error("history shape error: {reason}")]
HistoryShape {
/// Human-readable description of the malformed recorded history.
reason: String,
},
/// A search attribute update did not satisfy the registered schema.
#[error("search attribute validation error: {0}")]
SearchAttribute(#[from] SearchAttributeError),
/// A durable write was refused because this engine's task epoch has closed.
///
/// 🔴 DELIBERATELY NOT [`crate::EngineError::EngineTaskEpochClosed`]. That
/// variant carries a documented single-construction-site invariant —
/// `aion-client`'s `map_engine_error` gives it no arm on the stated ground
/// that it is unreachable through any transport-exposed operation — and a
/// second construction site would silently take that transport's catch-all.
/// This is a different layer with a different audience: it surfaces as
/// `{error, _}` inside running workflow code, never over the wire.
#[error("engine task epoch closed: {reason}")]
EngineTaskEpochClosed {
/// What was refused, and what it means for the run.
reason: String,
},
/// A durable write was refused because it was offered on behalf of a run
/// whose generation the history has already moved past (aion#213).
///
/// Continue-as-new does not stop the predecessor's process at the instant
/// its terminal lands — the API path never touches the pid, and the NIF
/// path's `cancel_pid` follows the append — so a run that is durably over
/// can still reach its recorder with work it had already begun. Sequencing
/// that append would put a `TimerStarted` (or an activity record, or a
/// signal) for a dead run inside the SUCCESSOR's segment, where replay
/// reads it as the successor's own. Nothing is appended, and the run this
/// names is over: the refusal is the correct outcome, not a fault to
/// retry.
#[error(
"run `{run_id}` of workflow `{workflow_id}` cannot append: its generation has been \
superseded, so the append belongs to no live run"
)]
RunSuperseded {
/// Workflow whose history the append was offered into.
workflow_id: aion_core::WorkflowId,
/// The superseded run the append was offered on behalf of.
run_id: aion_core::RunId,
},
}
#[cfg(test)]
mod tests {
use super::{DurabilityError, NonDeterminismError};
use aion_core::WorkflowId;
use aion_store::StoreError;
fn non_determinism_error() -> NonDeterminismError {
NonDeterminismError {
workflow_id: WorkflowId::new(uuid::Uuid::nil()),
seq: 42,
expected: "activity schedule ordinal 7".to_owned(),
found: "timer fired timer:named:deadline".to_owned(),
}
}
#[test]
fn non_determinism_display_includes_context() {
let error = non_determinism_error();
let message = error.to_string();
assert!(message.contains("00000000-0000-0000-0000-000000000000"));
assert!(message.contains("42"));
assert!(message.contains("activity schedule ordinal 7"));
assert!(message.contains("timer fired timer:named:deadline"));
}
/// The operator-facing `RunSuperseded` sentence: it names both identities,
/// and it is ONE sentence rather than two halves with a hole between them.
///
/// # 🔴 A WRAPPED LITERAL THAT LOSES ITS `\` LEAVES A HOLE NO TOOL CATCHES
///
/// This message shipped once as `"…has been superseded, …"` — the
/// line-continuation backslash was dropped when the literal was re-wrapped.
/// `cargo fmt` does not touch the inside of string literals and clippy has
/// nothing to say about them, so nothing failed. The string is not internal:
/// `aion-server`'s wire mapping sends `source.to_string()` to API callers
/// verbatim, so an operator reads it in an incident. The run of spaces is
/// what the assertion below looks for, because that is the shape the defect
/// takes every time.
#[test]
fn run_superseded_reads_as_one_sentence_naming_both_identities() {
// Two DISTINCT identifiers: with identical fixtures the two
// `contains` assertions below collapse into one assertion twice, and
// dropping either `{workflow_id}` or `{run_id}` from the message
// would go unnoticed.
let workflow_id = WorkflowId::new(uuid::Uuid::from_u128(0xA1));
let run_id = aion_core::RunId::new(uuid::Uuid::from_u128(0xB2));
let error = DurabilityError::RunSuperseded {
workflow_id: workflow_id.clone(),
run_id: run_id.clone(),
};
let message = error.to_string();
assert!(message.contains(&workflow_id.to_string()));
assert!(message.contains(&run_id.to_string()));
assert!(
message.contains("its generation has been superseded"),
"the wrapped literal must read as one sentence: {message}"
);
assert!(
!message.contains(" "),
"a run of spaces means a dropped `\\` line continuation: {message}"
);
}
#[test]
fn durability_error_display_mentions_underlying_cause() {
let store = DurabilityError::Store(StoreError::SequenceConflict {
expected: 10,
found: 11,
});
let non_determinism = DurabilityError::NonDeterminism(non_determinism_error());
let history_shape = DurabilityError::HistoryShape {
reason: "activity result without preceding schedule".to_owned(),
};
let store_message = store.to_string();
let non_determinism_message = non_determinism.to_string();
let history_shape_message = history_shape.to_string();
assert!(!store_message.is_empty());
assert!(store_message.contains("sequence conflict"));
assert!(!non_determinism_message.is_empty());
assert!(non_determinism_message.contains("activity schedule ordinal 7"));
assert!(non_determinism_message.contains("timer fired timer:named:deadline"));
assert!(!history_shape_message.is_empty());
assert!(history_shape_message.contains("activity result without preceding schedule"));
}
}