orbit_runtime/
model.rs

1//! Execution history events and the state derived from them.
2//!
3//! The event history is authoritative (RFC-0001 §9.8): [`ExecutionState`] is
4//! always the left fold of [`apply`] over an execution's events, so a runtime
5//! restart reconstructs exactly the state that was committed.
6
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use std::collections::BTreeMap;
10
11/// Structured task error, as reported by a worker or produced by the runtime.
12#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
13pub struct TaskError {
14    pub code: String,
15    pub message: String,
16    #[serde(default = "yes")]
17    pub retryable: bool,
18    #[serde(default, skip_serializing_if = "Option::is_none")]
19    pub details: Option<Value>,
20}
21
22fn yes() -> bool {
23    true
24}
25
26impl TaskError {
27    pub fn new(code: &str, message: impl Into<String>, retryable: bool) -> Self {
28        TaskError {
29            code: code.into(),
30            message: message.into(),
31            retryable,
32            details: None,
33        }
34    }
35}
36
37/// Runtime-generated error codes.
38pub mod codes {
39    pub const TIMEOUT: &str = "TIMEOUT";
40    pub const QUEUE_TIMEOUT: &str = "QUEUE_TIMEOUT";
41    pub const EXPRESSION_ERROR: &str = "EXPRESSION_ERROR";
42    pub const INPUT_CONTRACT_VIOLATION: &str = "INPUT_CONTRACT_VIOLATION";
43    pub const OUTPUT_CONTRACT_VIOLATION: &str = "OUTPUT_CONTRACT_VIOLATION";
44    pub const CANCELLED: &str = "CANCELLED";
45    pub const WORKFLOW_TIMEOUT: &str = "WORKFLOW_TIMEOUT";
46    pub const ADAPTER_DISABLED: &str = "ADAPTER_DISABLED";
47    pub const HTTP_ERROR: &str = "HTTP_ERROR";
48}
49
50#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
51#[serde(tag = "type")]
52pub enum Event {
53    WorkflowCreated {
54        workflow: String,
55        version: String,
56        definition_hash: String,
57        input: Value,
58        /// Set for child workflows started by a `workflow "Name"` step.
59        #[serde(default, skip_serializing_if = "Option::is_none")]
60        parent: Option<ParentRef>,
61    },
62    /// A `for` step expanded into one unit per element (`step[0]`, `step[1]`, ...).
63    StepFanOut {
64        step: String,
65        items: Vec<Value>,
66    },
67    WorkflowStarted {},
68    /// The unit is ready and queued for a worker. `input` is the fully evaluated payload.
69    StepScheduled {
70        step: String,
71        attempt: u32,
72        input: Value,
73        idempotency_key: String,
74        redelivery: bool,
75    },
76    StepSkipped {
77        step: String,
78        reason: String,
79    },
80    StepDispatched {
81        step: String,
82        attempt: u32,
83        token: String,
84        worker_id: String,
85        worker_session: String,
86        deadline_ms: Option<u64>,
87    },
88    StepSucceeded {
89        step: String,
90        attempt: u32,
91        output: Value,
92    },
93    StepAttemptFailed {
94        step: String,
95        attempt: u32,
96        error: TaskError,
97    },
98    /// The outcome of a dispatched attempt is unknown (worker lost, runtime restarted).
99    /// The same attempt is redelivered with the same idempotency key.
100    StepAttemptUncertain {
101        step: String,
102        attempt: u32,
103        reason: String,
104    },
105    RetryScheduled {
106        step: String,
107        next_attempt: u32,
108        delay_ms: u64,
109        fire_at_ms: u64,
110    },
111    StepFailed {
112        step: String,
113        error: TaskError,
114    },
115    /// A queued unit was withdrawn because the workflow is being aborted.
116    StepCancelled {
117        step: String,
118    },
119    TimerStarted {
120        step: String,
121        fire_at_ms: u64,
122    },
123    TimerFired {
124        step: String,
125    },
126    /// A result arrived for an attempt that no longer matters (duplicate or late).
127    StaleResultIgnored {
128        step: String,
129        token: String,
130        outcome: String,
131    },
132    CancelRequested {
133        reason: String,
134    },
135    WorkflowTimeoutReached {},
136    CompensationStarted {
137        reason: String,
138        steps: Vec<String>,
139    },
140    CompensationCompleted {},
141    CompensationFailed {
142        step: String,
143        error: TaskError,
144    },
145    WorkflowCompleted {
146        output: Value,
147    },
148    WorkflowFailed {
149        error: TaskError,
150        compensation: CompensationOutcome,
151    },
152    WorkflowCancelled {
153        compensation: CompensationOutcome,
154    },
155    WorkflowTimedOut {
156        compensation: CompensationOutcome,
157    },
158}
159
160impl Event {
161    pub fn name(&self) -> &'static str {
162        match self {
163            Event::WorkflowCreated { .. } => "WorkflowCreated",
164            Event::WorkflowStarted { .. } => "WorkflowStarted",
165            Event::StepFanOut { .. } => "StepFanOut",
166            Event::StepScheduled { .. } => "StepScheduled",
167            Event::StepSkipped { .. } => "StepSkipped",
168            Event::StepDispatched { .. } => "StepDispatched",
169            Event::StepSucceeded { .. } => "StepSucceeded",
170            Event::StepAttemptFailed { .. } => "StepAttemptFailed",
171            Event::StepAttemptUncertain { .. } => "StepAttemptUncertain",
172            Event::RetryScheduled { .. } => "RetryScheduled",
173            Event::StepFailed { .. } => "StepFailed",
174            Event::StepCancelled { .. } => "StepCancelled",
175            Event::TimerStarted { .. } => "TimerStarted",
176            Event::TimerFired { .. } => "TimerFired",
177            Event::StaleResultIgnored { .. } => "StaleResultIgnored",
178            Event::CancelRequested { .. } => "CancelRequested",
179            Event::WorkflowTimeoutReached { .. } => "WorkflowTimeoutReached",
180            Event::CompensationStarted { .. } => "CompensationStarted",
181            Event::CompensationCompleted { .. } => "CompensationCompleted",
182            Event::CompensationFailed { .. } => "CompensationFailed",
183            Event::WorkflowCompleted { .. } => "WorkflowCompleted",
184            Event::WorkflowFailed { .. } => "WorkflowFailed",
185            Event::WorkflowCancelled { .. } => "WorkflowCancelled",
186            Event::WorkflowTimedOut { .. } => "WorkflowTimedOut",
187        }
188    }
189
190    pub fn step(&self) -> Option<&str> {
191        match self {
192            Event::StepScheduled { step, .. }
193            | Event::StepFanOut { step, .. }
194            | Event::StepSkipped { step, .. }
195            | Event::StepDispatched { step, .. }
196            | Event::StepSucceeded { step, .. }
197            | Event::StepAttemptFailed { step, .. }
198            | Event::StepAttemptUncertain { step, .. }
199            | Event::RetryScheduled { step, .. }
200            | Event::StepFailed { step, .. }
201            | Event::StepCancelled { step }
202            | Event::TimerStarted { step, .. }
203            | Event::TimerFired { step }
204            | Event::StaleResultIgnored { step, .. }
205            | Event::CompensationFailed { step, .. } => Some(step),
206            _ => None,
207        }
208    }
209}
210
211#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
212#[serde(rename_all = "snake_case")]
213pub enum CompensationOutcome {
214    /// Nothing needed compensating.
215    None,
216    Completed,
217    Failed,
218}
219
220/// A recorded event with its position and commit time.
221#[derive(Debug, Clone, Serialize, Deserialize)]
222pub struct HistoryEvent {
223    pub seq: u64,
224    pub timestamp_ms: u64,
225    #[serde(flatten)]
226    pub event: Event,
227}
228
229#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
230#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
231pub enum WorkflowStatus {
232    Running,
233    Compensating,
234    Completed,
235    Failed,
236    Cancelled,
237    TimedOut,
238}
239
240impl WorkflowStatus {
241    pub fn is_terminal(self) -> bool {
242        matches!(
243            self,
244            WorkflowStatus::Completed
245                | WorkflowStatus::Failed
246                | WorkflowStatus::Cancelled
247                | WorkflowStatus::TimedOut
248        )
249    }
250
251    pub fn as_str(self) -> &'static str {
252        match self {
253            WorkflowStatus::Running => "RUNNING",
254            WorkflowStatus::Compensating => "COMPENSATING",
255            WorkflowStatus::Completed => "COMPLETED",
256            WorkflowStatus::Failed => "FAILED",
257            WorkflowStatus::Cancelled => "CANCELLED",
258            WorkflowStatus::TimedOut => "TIMED_OUT",
259        }
260    }
261}
262
263#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
264#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
265pub enum UnitStatus {
266    Pending,
267    /// A `for` step whose elements are running.
268    Running,
269    /// Queued, waiting for a worker.
270    Scheduled,
271    Dispatched,
272    /// Transient: an attempt failed and the decider will choose retry or failure.
273    AttemptFailed,
274    /// Transient: the outcome of the attempt is unknown and it will be redelivered.
275    Uncertain,
276    RetryWait,
277    /// Durable timer running (`sleep`).
278    Waiting,
279    Succeeded,
280    Skipped,
281    Failed,
282    Cancelled,
283}
284
285impl UnitStatus {
286    pub fn is_terminal(self) -> bool {
287        matches!(
288            self,
289            UnitStatus::Succeeded
290                | UnitStatus::Skipped
291                | UnitStatus::Failed
292                | UnitStatus::Cancelled
293        )
294    }
295}
296
297/// Why a running workflow is being wound down.
298#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
299#[serde(tag = "kind", rename_all = "snake_case")]
300pub enum AbortReason {
301    Failed { error: TaskError },
302    Cancelled { reason: String },
303    TimedOut,
304}
305
306/// One schedulable unit: a workflow step, or the compensation of a step
307/// (key `compensate:<step>`).
308#[derive(Debug, Clone, Default, Serialize, Deserialize)]
309pub struct UnitState {
310    pub status: Option<UnitStatus>,
311    pub attempt: u32,
312    #[serde(skip_serializing_if = "Option::is_none")]
313    pub input: Option<Value>,
314    #[serde(skip_serializing_if = "Option::is_none")]
315    pub idempotency_key: Option<String>,
316    #[serde(skip_serializing_if = "Option::is_none")]
317    pub output: Option<Value>,
318    #[serde(skip_serializing_if = "Option::is_none")]
319    pub error: Option<TaskError>,
320    #[serde(skip_serializing_if = "Option::is_none")]
321    pub last_error: Option<TaskError>,
322    pub redelivery: bool,
323    #[serde(skip_serializing_if = "Option::is_none")]
324    pub scheduled_at_ms: Option<u64>,
325    #[serde(skip_serializing_if = "Option::is_none")]
326    pub dispatched_at_ms: Option<u64>,
327    #[serde(skip_serializing_if = "Option::is_none")]
328    pub deadline_ms: Option<u64>,
329    #[serde(skip_serializing_if = "Option::is_none")]
330    pub wake_at_ms: Option<u64>,
331    #[serde(skip_serializing_if = "Option::is_none")]
332    pub finished_at_ms: Option<u64>,
333    #[serde(skip_serializing_if = "Option::is_none")]
334    pub worker_id: Option<String>,
335    #[serde(skip_serializing_if = "Option::is_none")]
336    pub worker_session: Option<String>,
337    /// Token of the in-flight dispatch.
338    #[serde(skip_serializing_if = "Option::is_none")]
339    pub token: Option<String>,
340    /// Every token issued for this unit, with its attempt number. Results for
341    /// any of them are accepted while the unit is unresolved (reconciliation).
342    #[serde(default, skip_serializing_if = "Vec::is_empty")]
343    pub tokens: Vec<(String, u32)>,
344    /// Commit order of success, used to compensate in reverse.
345    #[serde(skip_serializing_if = "Option::is_none")]
346    pub succeeded_seq: Option<u64>,
347    pub dispatch_count: u32,
348}
349
350impl UnitState {
351    pub fn status(&self) -> UnitStatus {
352        self.status.unwrap_or(UnitStatus::Pending)
353    }
354}
355
356#[derive(Debug, Clone, Serialize, Deserialize)]
357pub struct ExecutionState {
358    pub id: String,
359    #[serde(default, skip_serializing_if = "Option::is_none")]
360    pub parent: Option<ParentRef>,
361    pub workflow: String,
362    pub version: String,
363    pub definition_hash: String,
364    pub input: Value,
365    pub status: WorkflowStatus,
366    pub created_at_ms: u64,
367    pub updated_at_ms: u64,
368    #[serde(skip_serializing_if = "Option::is_none")]
369    pub finished_at_ms: Option<u64>,
370    pub units: BTreeMap<String, UnitState>,
371    #[serde(skip_serializing_if = "Option::is_none")]
372    pub abort: Option<AbortReason>,
373    #[serde(default, skip_serializing_if = "Vec::is_empty")]
374    pub compensation_order: Vec<String>,
375    #[serde(skip_serializing_if = "Option::is_none")]
376    pub compensation: Option<CompensationOutcome>,
377    #[serde(skip_serializing_if = "Option::is_none")]
378    pub output: Option<Value>,
379    #[serde(skip_serializing_if = "Option::is_none")]
380    pub error: Option<TaskError>,
381    pub last_seq: u64,
382}
383
384/// Link from a child execution to the parent unit that started it.
385#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
386pub struct ParentRef {
387    pub execution_id: String,
388    pub step: String,
389}
390
391/// Unit key of element `i` of a `for` step.
392pub fn element_key(step: &str, i: usize) -> String {
393    format!("{step}[{i}]")
394}
395
396/// A unit key decomposed: `[compensate:]step[[index]]`.
397#[derive(Debug, Clone, Copy, PartialEq, Eq)]
398pub struct UnitKey<'a> {
399    pub compensation: bool,
400    pub step: &'a str,
401    pub index: Option<usize>,
402}
403
404pub fn parse_key(key: &str) -> UnitKey<'_> {
405    let (compensation, rest) = match key.strip_prefix("compensate:") {
406        Some(r) => (true, r),
407        None => (false, key),
408    };
409    if let Some(open) = rest.find('[') {
410        if let Some(index) = rest[open + 1..]
411            .strip_suffix(']')
412            .and_then(|n| n.parse().ok())
413        {
414            return UnitKey {
415                compensation,
416                step: &rest[..open],
417                index: Some(index),
418            };
419        }
420    }
421    UnitKey {
422        compensation,
423        step: rest,
424        index: None,
425    }
426}
427
428pub fn compensation_key(step: &str) -> String {
429    format!("compensate:{step}")
430}
431
432impl ExecutionState {
433    pub fn new(id: &str) -> Self {
434        ExecutionState {
435            id: id.into(),
436            parent: None,
437            workflow: String::new(),
438            version: String::new(),
439            definition_hash: String::new(),
440            input: Value::Null,
441            status: WorkflowStatus::Running,
442            created_at_ms: 0,
443            updated_at_ms: 0,
444            finished_at_ms: None,
445            units: BTreeMap::new(),
446            abort: None,
447            compensation_order: vec![],
448            compensation: None,
449            output: None,
450            error: None,
451            last_seq: 0,
452        }
453    }
454
455    /// Rebuild from a full history.
456    pub fn replay(id: &str, history: &[HistoryEvent]) -> Self {
457        let mut s = ExecutionState::new(id);
458        for h in history {
459            apply(&mut s, h);
460        }
461        s
462    }
463
464    pub fn unit(&self, key: &str) -> Option<&UnitState> {
465        self.units.get(key)
466    }
467}
468
469/// Apply one event. Pure and total: unknown combinations are ignored so that
470/// replaying an older history never panics.
471pub fn apply(s: &mut ExecutionState, h: &HistoryEvent) {
472    let ts = h.timestamp_ms;
473    s.last_seq = h.seq;
474    s.updated_at_ms = ts;
475    match &h.event {
476        Event::WorkflowCreated {
477            workflow,
478            version,
479            definition_hash,
480            input,
481            parent,
482        } => {
483            s.parent = parent.clone();
484            s.workflow = workflow.clone();
485            s.version = version.clone();
486            s.definition_hash = definition_hash.clone();
487            s.input = input.clone();
488            s.created_at_ms = ts;
489            s.status = WorkflowStatus::Running;
490        }
491        Event::WorkflowStarted {} => {}
492        Event::StepScheduled {
493            step,
494            attempt,
495            input,
496            idempotency_key,
497            redelivery,
498        } => {
499            let u = s.units.entry(step.clone()).or_default();
500            u.status = Some(UnitStatus::Scheduled);
501            u.attempt = *attempt;
502            u.input = Some(input.clone());
503            u.idempotency_key = Some(idempotency_key.clone());
504            u.redelivery = *redelivery;
505            u.scheduled_at_ms = Some(ts);
506            u.token = None;
507            u.deadline_ms = None;
508            u.wake_at_ms = None;
509        }
510        Event::StepFanOut { step, items } => {
511            let u = s.units.entry(step.clone()).or_default();
512            u.status = Some(UnitStatus::Running);
513            u.input = Some(Value::Array(items.clone()));
514            u.scheduled_at_ms = Some(ts);
515            u.attempt = 1;
516            for i in 0..items.len() {
517                s.units.entry(element_key(step, i)).or_default();
518            }
519        }
520        Event::StepSkipped { step, .. } => {
521            let u = s.units.entry(step.clone()).or_default();
522            u.status = Some(UnitStatus::Skipped);
523            u.output = Some(Value::Null);
524            u.finished_at_ms = Some(ts);
525        }
526        Event::StepDispatched {
527            step,
528            attempt,
529            token,
530            worker_id,
531            worker_session,
532            deadline_ms,
533        } => {
534            let u = s.units.entry(step.clone()).or_default();
535            u.status = Some(UnitStatus::Dispatched);
536            u.attempt = *attempt;
537            u.token = Some(token.clone());
538            u.tokens.push((token.clone(), *attempt));
539            u.worker_id = Some(worker_id.clone());
540            u.worker_session = Some(worker_session.clone());
541            u.dispatched_at_ms = Some(ts);
542            u.deadline_ms = *deadline_ms;
543            u.dispatch_count += 1;
544        }
545        Event::StepSucceeded {
546            step,
547            attempt,
548            output,
549        } => {
550            let u = s.units.entry(step.clone()).or_default();
551            u.status = Some(UnitStatus::Succeeded);
552            u.attempt = *attempt;
553            u.output = Some(output.clone());
554            u.finished_at_ms = Some(ts);
555            u.succeeded_seq = Some(h.seq);
556            u.token = None;
557            u.wake_at_ms = None;
558        }
559        Event::StepAttemptFailed { step, error, .. } => {
560            let u = s.units.entry(step.clone()).or_default();
561            u.status = Some(UnitStatus::AttemptFailed);
562            u.last_error = Some(error.clone());
563            u.token = None;
564        }
565        Event::StepAttemptUncertain { step, .. } => {
566            let u = s.units.entry(step.clone()).or_default();
567            u.status = Some(UnitStatus::Uncertain);
568            u.token = None;
569        }
570        Event::RetryScheduled {
571            step,
572            next_attempt,
573            fire_at_ms,
574            ..
575        } => {
576            let u = s.units.entry(step.clone()).or_default();
577            u.status = Some(UnitStatus::RetryWait);
578            u.attempt = *next_attempt;
579            u.wake_at_ms = Some(*fire_at_ms);
580        }
581        Event::StepFailed { step, error } => {
582            let u = s.units.entry(step.clone()).or_default();
583            u.status = Some(UnitStatus::Failed);
584            u.error = Some(error.clone());
585            u.finished_at_ms = Some(ts);
586            u.token = None;
587            if s.abort.is_none() && !step.starts_with("compensate:") {
588                s.abort = Some(AbortReason::Failed {
589                    error: TaskError {
590                        message: format!("step `{step}` failed: {}", error.message),
591                        ..error.clone()
592                    },
593                });
594            }
595        }
596        Event::StepCancelled { step } => {
597            let u = s.units.entry(step.clone()).or_default();
598            u.status = Some(UnitStatus::Cancelled);
599            u.finished_at_ms = Some(ts);
600            u.token = None;
601        }
602        Event::TimerStarted { step, fire_at_ms } => {
603            let u = s.units.entry(step.clone()).or_default();
604            u.status = Some(UnitStatus::Waiting);
605            u.wake_at_ms = Some(*fire_at_ms);
606            u.scheduled_at_ms = Some(ts);
607            u.attempt = 1;
608        }
609        Event::TimerFired { step } => {
610            let u = s.units.entry(step.clone()).or_default();
611            u.status = Some(UnitStatus::Succeeded);
612            u.output = Some(Value::Object(Default::default()));
613            u.finished_at_ms = Some(ts);
614            u.succeeded_seq = Some(h.seq);
615            u.wake_at_ms = None;
616        }
617        Event::StaleResultIgnored { .. } => {}
618        Event::CancelRequested { reason } => {
619            if s.abort.is_none() {
620                s.abort = Some(AbortReason::Cancelled {
621                    reason: reason.clone(),
622                });
623            }
624        }
625        Event::WorkflowTimeoutReached {} => {
626            if s.abort.is_none() {
627                s.abort = Some(AbortReason::TimedOut);
628            }
629        }
630        Event::CompensationStarted { steps, .. } => {
631            s.status = WorkflowStatus::Compensating;
632            s.compensation_order = steps.clone();
633            for st in steps {
634                s.units.entry(compensation_key(st)).or_default();
635            }
636        }
637        Event::CompensationCompleted {} => s.compensation = Some(CompensationOutcome::Completed),
638        Event::CompensationFailed { .. } => s.compensation = Some(CompensationOutcome::Failed),
639        Event::WorkflowCompleted { output } => {
640            s.status = WorkflowStatus::Completed;
641            s.output = Some(output.clone());
642            s.finished_at_ms = Some(ts);
643        }
644        Event::WorkflowFailed {
645            error,
646            compensation,
647        } => {
648            s.status = WorkflowStatus::Failed;
649            s.error = Some(error.clone());
650            s.compensation = Some(*compensation);
651            s.finished_at_ms = Some(ts);
652        }
653        Event::WorkflowCancelled { compensation } => {
654            s.status = WorkflowStatus::Cancelled;
655            s.compensation = Some(*compensation);
656            s.finished_at_ms = Some(ts);
657        }
658        Event::WorkflowTimedOut { compensation } => {
659            s.status = WorkflowStatus::TimedOut;
660            s.compensation = Some(*compensation);
661            s.finished_at_ms = Some(ts);
662        }
663    }
664}
orbit_runtime/model.rs

orbit_runtime/
model.rs
