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
//! `WorkflowFold` — decodes `EventMeta` + payload, routes on dispatch,
//! and applies the deterministic task-lifecycle state transition.
//!
//! The chain is single-writer (the task-lease holder), so the fold does
//! not arbitrate contention between writers — it replays the writer's
//! cursor advances. It does enforce one structural invariant, though: a
//! terminal task (`Done`/`Failed`) is never moved by a later transition
//! or retry, so a duplicate / replayed / buggy-writer event can't
//! resurrect a settled task. Same chain → same state.
use super::super::super::redex::{RedexError, RedexEvent, RedexFold};
use super::super::meta::{
compute_checksum, compute_checksum_with_meta, EventMeta, EVENT_META_SIZE,
};
use super::dispatch::{
DISPATCH_TASK_ADVANCED, DISPATCH_TASK_CANCEL_REQUESTED, DISPATCH_TASK_DELETED,
DISPATCH_TASK_LINKED, DISPATCH_TASK_RETRIED, DISPATCH_TASK_SUBMITTED,
DISPATCH_TASK_TRANSITIONED,
};
use super::state::WorkflowState;
use super::types::{
AdvancedPayload, CancelRequestedPayload, DeletedPayload, LinkedPayload, RetriedPayload,
SubmittedPayload, TaskState, TaskStatus, TransitionedPayload,
};
/// Fold implementation for the task-lifecycle model.
pub struct WorkflowFold;
impl RedexFold<WorkflowState> for WorkflowFold {
fn apply(&mut self, ev: &RedexEvent, state: &mut WorkflowState) -> Result<(), RedexError> {
// Decode failures use `RedexError::Decode` (recoverable —
// skip-and-continue even under the `Stop` policy) so one
// corrupt event can't wedge the fold task forever; same
// rationale as `TasksFold`.
if ev.payload.len() < EVENT_META_SIZE {
return Err(RedexError::Decode(format!(
"workflow payload too short: {} bytes (need >= {})",
ev.payload.len(),
EVENT_META_SIZE
)));
}
let meta = EventMeta::from_bytes(&ev.payload[..EVENT_META_SIZE])
.ok_or_else(|| RedexError::Decode("bad EventMeta prefix".into()))?;
let tail = &ev.payload[EVENT_META_SIZE..];
// Verify the ingest-time checksum over (header-with-zeroed-
// checksum ++ tail); fall back to the legacy tail-only hash
// for records written by pre-fix adapters.
let v2_expected = compute_checksum_with_meta(&meta, tail);
let valid = meta.checksum == v2_expected || meta.checksum == compute_checksum(tail);
if !valid {
return Err(RedexError::Decode(format!(
"workflow fold: EventMeta checksum mismatch at seq {} (got {:#010x}, v2 expected {:#010x})",
ev.entry.seq, meta.checksum, v2_expected
)));
}
match meta.dispatch {
DISPATCH_TASK_SUBMITTED => {
let p: SubmittedPayload =
postcard::from_bytes(tail).map_err(|e| RedexError::Decode(e.to_string()))?;
// Submit is the baseline; a re-submit of a live id
// resets it to the fresh state (the log is the source
// of truth) and clears any stale cancel signal.
state.tasks.insert(p.id, TaskState::submitted());
state.cancelled.remove(&p.id);
}
DISPATCH_TASK_TRANSITIONED => {
let p: TransitionedPayload =
postcard::from_bytes(tail).map_err(|e| RedexError::Decode(e.to_string()))?;
if let Some(t) = state.tasks.get_mut(&p.id) {
// Terminal is terminal: a `Done`/`Failed` task is
// never moved by a plain transition. The sanctioned
// way out of `Failed` is `retry`; out of `Done`
// there is none (a fresh `submit` resets instead).
// This guards replay / duplicate / buggy-writer
// events from resurrecting a settled task (review #2).
if !t.status.is_terminal() {
t.status = p.status;
}
}
// A transition for an unknown id is a no-op: the submit
// we never observed simply isn't in our view.
}
DISPATCH_TASK_ADVANCED => {
let p: AdvancedPayload =
postcard::from_bytes(tail).map_err(|e| RedexError::Decode(e.to_string()))?;
if let Some(t) = state.tasks.get_mut(&p.id) {
t.step = t.step.saturating_add(1);
// A new step starts with a clean attempt counter.
t.attempts = 0;
}
}
DISPATCH_TASK_RETRIED => {
let p: RetriedPayload =
postcard::from_bytes(tail).map_err(|e| RedexError::Decode(e.to_string()))?;
if let Some(t) = state.tasks.get_mut(&p.id) {
// Retry re-runs the current step — the sanctioned
// `Failed → Running` exit. It must not resurrect a
// `Done` task, which is terminal success (review #2).
if t.status != TaskStatus::Done {
t.attempts = t.attempts.saturating_add(1);
t.status = TaskStatus::Running;
}
}
}
DISPATCH_TASK_DELETED => {
let p: DeletedPayload =
postcard::from_bytes(tail).map_err(|e| RedexError::Decode(e.to_string()))?;
// Cascade: delete reclaims the WHOLE subtree (shards /
// spawned children), not just the named task — an
// orphaned shard would keep running and keep holding its
// claim (corrections #4). The subtree is computed from
// the folded lineage, so it's deterministic / replayable.
let subtree = state.subtree(p.id);
// Detach the root from its parent's child list.
if let Some(parent) = state.parents.get(&p.id).copied() {
if let Some(sibs) = state.children.get_mut(&parent) {
sibs.retain(|c| *c != p.id);
}
}
for t in subtree {
state.tasks.remove(&t);
state.cancelled.remove(&t);
state.children.remove(&t);
state.parents.remove(&t);
}
}
DISPATCH_TASK_CANCEL_REQUESTED => {
let p: CancelRequestedPayload =
postcard::from_bytes(tail).map_err(|e| RedexError::Decode(e.to_string()))?;
// Record the signal for the worker to observe; the
// status transition itself is the worker's to make.
state.cancelled.insert(p.id);
}
DISPATCH_TASK_LINKED => {
let p: LinkedPayload =
postcard::from_bytes(tail).map_err(|e| RedexError::Decode(e.to_string()))?;
// Record the lineage edge (idempotent — a duplicate link
// doesn't double-insert the child).
let kids = state.children.entry(p.parent).or_default();
if !kids.contains(&p.child) {
kids.push(p.child);
}
state.parents.insert(p.child, p.parent);
}
other => {
// Unknown dispatches in the CortEX-internal range are
// forward-compatibility — log and skip.
tracing::debug!(
dispatch = other,
seq = ev.entry.seq,
"workflow fold: ignoring unknown dispatch"
);
}
}
Ok(())
}
}