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    pub const SIGNAL_TIMEOUT: &str = "SIGNAL_TIMEOUT";
49    pub const LIMIT_EXCEEDED: &str = "LIMIT_EXCEEDED";
50}
51
52#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
53#[serde(tag = "type")]
54pub enum Event {
55    WorkflowCreated {
56        workflow: String,
57        version: String,
58        definition_hash: String,
59        input: Value,
60        /// Set for child workflows started by a `workflow "Name"` step.
61        #[serde(default, skip_serializing_if = "Option::is_none")]
62        parent: Option<ParentRef>,
63        /// Set for executions started by a schedule (`/v1/schedules`).
64        #[serde(default, skip_serializing_if = "Option::is_none")]
65        schedule_id: Option<String>,
66        /// The run number, for runs after the first one (continue-as-new).
67        /// Absent means run 1.
68        #[serde(default, skip_serializing_if = "Option::is_none")]
69        run: Option<u32>,
70    },
71    /// A `for` step expanded into one unit per element (`step[0]`, `step[1]`, ...).
72    StepFanOut {
73        step: String,
74        items: Vec<Value>,
75    },
76    WorkflowStarted {},
77    /// The unit is ready and queued for a worker. `input` is the fully evaluated payload.
78    StepScheduled {
79        step: String,
80        attempt: u32,
81        input: Value,
82        idempotency_key: String,
83        redelivery: bool,
84    },
85    StepSkipped {
86        step: String,
87        reason: String,
88    },
89    StepDispatched {
90        step: String,
91        attempt: u32,
92        token: String,
93        worker_id: String,
94        worker_session: String,
95        deadline_ms: Option<u64>,
96    },
97    StepSucceeded {
98        step: String,
99        attempt: u32,
100        output: Value,
101    },
102    StepAttemptFailed {
103        step: String,
104        attempt: u32,
105        error: TaskError,
106    },
107    /// The outcome of a dispatched attempt is unknown (worker lost, runtime restarted).
108    /// The same attempt is redelivered with the same idempotency key.
109    StepAttemptUncertain {
110        step: String,
111        attempt: u32,
112        reason: String,
113    },
114    RetryScheduled {
115        step: String,
116        next_attempt: u32,
117        delay_ms: u64,
118        fire_at_ms: u64,
119    },
120    StepFailed {
121        step: String,
122        error: TaskError,
123    },
124    /// A queued unit was withdrawn because the workflow is being aborted.
125    StepCancelled {
126        step: String,
127    },
128    TimerStarted {
129        step: String,
130        fire_at_ms: u64,
131    },
132    TimerFired {
133        step: String,
134    },
135    /// A result arrived for an attempt that no longer matters (duplicate or late).
136    StaleResultIgnored {
137        step: String,
138        token: String,
139        outcome: String,
140    },
141    CancelRequested {
142        reason: String,
143    },
144    WorkflowTimeoutReached {},
145    CompensationStarted {
146        reason: String,
147        steps: Vec<String>,
148    },
149    CompensationCompleted {},
150    CompensationFailed {
151        step: String,
152        error: TaskError,
153    },
154    WorkflowCompleted {
155        output: Value,
156    },
157    WorkflowFailed {
158        error: TaskError,
159        compensation: CompensationOutcome,
160    },
161    WorkflowCancelled {
162        compensation: CompensationOutcome,
163    },
164    WorkflowTimedOut {
165        compensation: CompensationOutcome,
166    },
167    /// A typed signal sent to the execution (`POST .../signals/<name>`).
168    /// Buffered until an `await` step takes it.
169    SignalReceived {
170        name: String,
171        payload: Value,
172    },
173    /// An `await` step started waiting. `deadline_ms` comes from its `timeout`.
174    SignalAwaitStarted {
175        step: String,
176        signal: String,
177        #[serde(default, skip_serializing_if = "Option::is_none")]
178        deadline_ms: Option<u64>,
179    },
180    /// A server resource limit was reached; the execution is failed with
181    /// `LIMIT_EXCEEDED` (compensation and `on failure` still run).
182    LimitExceeded {
183        limit: String,
184        message: String,
185    },
186    /// Compensation (if any) is done and the `on <kind>` handler steps start.
187    /// `compensation` is the outcome recorded on the terminal event afterwards.
188    HandlersStarted {
189        kind: String,
190        compensation: CompensationOutcome,
191    },
192    /// An `await` step took the signal received at history sequence `signal_seq`;
193    /// its payload is the step output.
194    SignalDelivered {
195        step: String,
196        signal_seq: u64,
197    },
198    /// `continue_as_new`: this run ended and run `next_run` of the same
199    /// execution starts with `input` and an empty history. The runtime
200    /// archives this run's history and writes the new run's first events in
201    /// the same transaction.
202    WorkflowContinuedAsNew {
203        next_run: u32,
204        input: Value,
205    },
206}
207
208impl Event {
209    pub fn name(&self) -> &'static str {
210        match self {
211            Event::WorkflowCreated { .. } => "WorkflowCreated",
212            Event::WorkflowStarted { .. } => "WorkflowStarted",
213            Event::StepFanOut { .. } => "StepFanOut",
214            Event::StepScheduled { .. } => "StepScheduled",
215            Event::StepSkipped { .. } => "StepSkipped",
216            Event::StepDispatched { .. } => "StepDispatched",
217            Event::StepSucceeded { .. } => "StepSucceeded",
218            Event::StepAttemptFailed { .. } => "StepAttemptFailed",
219            Event::StepAttemptUncertain { .. } => "StepAttemptUncertain",
220            Event::RetryScheduled { .. } => "RetryScheduled",
221            Event::StepFailed { .. } => "StepFailed",
222            Event::StepCancelled { .. } => "StepCancelled",
223            Event::TimerStarted { .. } => "TimerStarted",
224            Event::TimerFired { .. } => "TimerFired",
225            Event::StaleResultIgnored { .. } => "StaleResultIgnored",
226            Event::CancelRequested { .. } => "CancelRequested",
227            Event::WorkflowTimeoutReached { .. } => "WorkflowTimeoutReached",
228            Event::CompensationStarted { .. } => "CompensationStarted",
229            Event::CompensationCompleted { .. } => "CompensationCompleted",
230            Event::CompensationFailed { .. } => "CompensationFailed",
231            Event::WorkflowCompleted { .. } => "WorkflowCompleted",
232            Event::WorkflowFailed { .. } => "WorkflowFailed",
233            Event::WorkflowCancelled { .. } => "WorkflowCancelled",
234            Event::WorkflowTimedOut { .. } => "WorkflowTimedOut",
235            Event::SignalReceived { .. } => "SignalReceived",
236            Event::HandlersStarted { .. } => "HandlersStarted",
237            Event::LimitExceeded { .. } => "LimitExceeded",
238            Event::SignalAwaitStarted { .. } => "SignalAwaitStarted",
239            Event::SignalDelivered { .. } => "SignalDelivered",
240            Event::WorkflowContinuedAsNew { .. } => "WorkflowContinuedAsNew",
241        }
242    }
243
244    pub fn step(&self) -> Option<&str> {
245        match self {
246            Event::StepScheduled { step, .. }
247            | Event::StepFanOut { step, .. }
248            | Event::StepSkipped { step, .. }
249            | Event::StepDispatched { step, .. }
250            | Event::StepSucceeded { step, .. }
251            | Event::StepAttemptFailed { step, .. }
252            | Event::StepAttemptUncertain { step, .. }
253            | Event::RetryScheduled { step, .. }
254            | Event::StepFailed { step, .. }
255            | Event::StepCancelled { step }
256            | Event::TimerStarted { step, .. }
257            | Event::TimerFired { step }
258            | Event::StaleResultIgnored { step, .. }
259            | Event::SignalAwaitStarted { step, .. }
260            | Event::SignalDelivered { step, .. }
261            | Event::CompensationFailed { step, .. } => Some(step),
262            _ => None,
263        }
264    }
265}
266
267#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
268#[serde(rename_all = "snake_case")]
269pub enum CompensationOutcome {
270    /// Nothing needed compensating.
271    None,
272    Completed,
273    Failed,
274}
275
276/// A recorded event with its position and commit time.
277#[derive(Debug, Clone, Serialize, Deserialize)]
278pub struct HistoryEvent {
279    pub seq: u64,
280    pub timestamp_ms: u64,
281    #[serde(flatten)]
282    pub event: Event,
283}
284
285#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
286#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
287pub enum WorkflowStatus {
288    Running,
289    Compensating,
290    Completed,
291    Failed,
292    Cancelled,
293    TimedOut,
294    /// The run ended with `continue_as_new`; a newer run of the execution
295    /// carries on. Only archived runs have this status.
296    ContinuedAsNew,
297}
298
299impl WorkflowStatus {
300    pub fn is_terminal(self) -> bool {
301        matches!(
302            self,
303            WorkflowStatus::Completed
304                | WorkflowStatus::Failed
305                | WorkflowStatus::Cancelled
306                | WorkflowStatus::TimedOut
307                | WorkflowStatus::ContinuedAsNew
308        )
309    }
310
311    pub fn as_str(self) -> &'static str {
312        match self {
313            WorkflowStatus::Running => "RUNNING",
314            WorkflowStatus::Compensating => "COMPENSATING",
315            WorkflowStatus::Completed => "COMPLETED",
316            WorkflowStatus::Failed => "FAILED",
317            WorkflowStatus::Cancelled => "CANCELLED",
318            WorkflowStatus::TimedOut => "TIMED_OUT",
319            WorkflowStatus::ContinuedAsNew => "CONTINUED_AS_NEW",
320        }
321    }
322}
323
324#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
325#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
326pub enum UnitStatus {
327    Pending,
328    /// A `for` step whose elements are running.
329    Running,
330    /// Queued, waiting for a worker.
331    Scheduled,
332    Dispatched,
333    /// Transient: an attempt failed and the decider will choose retry or failure.
334    AttemptFailed,
335    /// Transient: the outcome of the attempt is unknown and it will be redelivered.
336    Uncertain,
337    RetryWait,
338    /// Durable timer running (`sleep`).
339    Waiting,
340    Succeeded,
341    Skipped,
342    Failed,
343    Cancelled,
344}
345
346impl UnitStatus {
347    pub fn is_terminal(self) -> bool {
348        matches!(
349            self,
350            UnitStatus::Succeeded
351                | UnitStatus::Skipped
352                | UnitStatus::Failed
353                | UnitStatus::Cancelled
354        )
355    }
356}
357
358/// Why a running workflow is being wound down.
359#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
360#[serde(tag = "kind", rename_all = "snake_case")]
361pub enum AbortReason {
362    Failed { error: TaskError },
363    Cancelled { reason: String },
364    TimedOut,
365}
366
367/// One schedulable unit: a workflow step, or the compensation of a step
368/// (key `compensate:<step>`).
369#[derive(Debug, Clone, Default, Serialize, Deserialize)]
370pub struct UnitState {
371    pub status: Option<UnitStatus>,
372    pub attempt: u32,
373    #[serde(skip_serializing_if = "Option::is_none")]
374    pub input: Option<Value>,
375    #[serde(skip_serializing_if = "Option::is_none")]
376    pub idempotency_key: Option<String>,
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    #[serde(skip_serializing_if = "Option::is_none")]
382    pub last_error: Option<TaskError>,
383    pub redelivery: bool,
384    #[serde(skip_serializing_if = "Option::is_none")]
385    pub scheduled_at_ms: Option<u64>,
386    #[serde(skip_serializing_if = "Option::is_none")]
387    pub dispatched_at_ms: Option<u64>,
388    #[serde(skip_serializing_if = "Option::is_none")]
389    pub deadline_ms: Option<u64>,
390    #[serde(skip_serializing_if = "Option::is_none")]
391    pub wake_at_ms: Option<u64>,
392    #[serde(skip_serializing_if = "Option::is_none")]
393    pub finished_at_ms: Option<u64>,
394    #[serde(skip_serializing_if = "Option::is_none")]
395    pub worker_id: Option<String>,
396    #[serde(skip_serializing_if = "Option::is_none")]
397    pub worker_session: Option<String>,
398    /// Token of the in-flight dispatch.
399    #[serde(skip_serializing_if = "Option::is_none")]
400    pub token: Option<String>,
401    /// Every token issued for this unit, with its attempt number. Results for
402    /// any of them are accepted while the unit is unresolved (reconciliation).
403    #[serde(default, skip_serializing_if = "Vec::is_empty")]
404    pub tokens: Vec<(String, u32)>,
405    /// Commit order of success, used to compensate in reverse.
406    #[serde(skip_serializing_if = "Option::is_none")]
407    pub succeeded_seq: Option<u64>,
408    pub dispatch_count: u32,
409}
410
411impl UnitState {
412    pub fn status(&self) -> UnitStatus {
413        self.status.unwrap_or(UnitStatus::Pending)
414    }
415}
416
417#[derive(Debug, Clone, Serialize, Deserialize)]
418pub struct ExecutionState {
419    pub id: String,
420    /// Run number: 1, plus one for every `continue_as_new`. Each run has its
421    /// own history; this state describes the run named here.
422    #[serde(default = "first_run", skip_serializing_if = "is_first_run")]
423    pub run: u32,
424    #[serde(default, skip_serializing_if = "Option::is_none")]
425    pub parent: Option<ParentRef>,
426    /// The schedule that started this execution, if any.
427    #[serde(default, skip_serializing_if = "Option::is_none")]
428    pub schedule_id: Option<String>,
429    pub workflow: String,
430    pub version: String,
431    pub definition_hash: String,
432    pub input: Value,
433    pub status: WorkflowStatus,
434    pub created_at_ms: u64,
435    pub updated_at_ms: u64,
436    #[serde(skip_serializing_if = "Option::is_none")]
437    pub finished_at_ms: Option<u64>,
438    pub units: BTreeMap<String, UnitState>,
439    #[serde(skip_serializing_if = "Option::is_none")]
440    pub abort: Option<AbortReason>,
441    #[serde(default, skip_serializing_if = "Vec::is_empty")]
442    pub compensation_order: Vec<String>,
443    #[serde(skip_serializing_if = "Option::is_none")]
444    pub compensation: Option<CompensationOutcome>,
445    #[serde(skip_serializing_if = "Option::is_none")]
446    pub output: Option<Value>,
447    #[serde(skip_serializing_if = "Option::is_none")]
448    pub error: Option<TaskError>,
449    /// Signals received so far, in arrival order.
450    #[serde(default, skip_serializing_if = "Vec::is_empty")]
451    pub signals: Vec<ReceivedSignal>,
452    /// Set while `on failure|cancel|timeout` handler steps run.
453    #[serde(default, skip_serializing_if = "Option::is_none")]
454    pub handlers: Option<HandlerPhase>,
455    pub last_seq: u64,
456}
457
458fn first_run() -> u32 {
459    1
460}
461
462fn is_first_run(run: &u32) -> bool {
463    *run == 1
464}
465
466/// The handler block running while an execution finishes.
467#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
468pub struct HandlerPhase {
469    pub kind: String,
470    pub compensation: CompensationOutcome,
471}
472
473/// True for units that keep running while an execution winds down:
474/// compensations and `on` handler steps.
475pub fn is_cleanup_key(key: &str) -> bool {
476    key.starts_with("compensate:") || orbit_ir::parse_handler_key(key).is_some()
477}
478
479/// A signal in an execution's buffer.
480#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
481pub struct ReceivedSignal {
482    /// History sequence number of the `SignalReceived` event.
483    pub seq: u64,
484    pub name: String,
485    pub payload: Value,
486    pub received_at_ms: u64,
487    /// The `await` unit that took it, once delivered.
488    #[serde(default, skip_serializing_if = "Option::is_none")]
489    pub delivered_to: Option<String>,
490}
491
492/// Link from a child execution to the parent unit that started it.
493#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
494pub struct ParentRef {
495    pub execution_id: String,
496    pub step: String,
497}
498
499/// Unit key of element `i` of a `for` step.
500pub fn element_key(step: &str, i: usize) -> String {
501    format!("{step}[{i}]")
502}
503
504/// A unit key decomposed: `[compensate:]step[[index]]`.
505#[derive(Debug, Clone, Copy, PartialEq, Eq)]
506pub struct UnitKey<'a> {
507    pub compensation: bool,
508    pub step: &'a str,
509    pub index: Option<usize>,
510}
511
512pub fn parse_key(key: &str) -> UnitKey<'_> {
513    let (compensation, rest) = match key.strip_prefix("compensate:") {
514        Some(r) => (true, r),
515        None => (false, key),
516    };
517    if let Some(open) = rest.find('[') {
518        if let Some(index) = rest[open + 1..]
519            .strip_suffix(']')
520            .and_then(|n| n.parse().ok())
521        {
522            return UnitKey {
523                compensation,
524                step: &rest[..open],
525                index: Some(index),
526            };
527        }
528    }
529    UnitKey {
530        compensation,
531        step: rest,
532        index: None,
533    }
534}
535
536pub fn compensation_key(step: &str) -> String {
537    format!("compensate:{step}")
538}
539
540impl ExecutionState {
541    pub fn new(id: &str) -> Self {
542        ExecutionState {
543            id: id.into(),
544            run: 1,
545            parent: None,
546            schedule_id: None,
547            workflow: String::new(),
548            version: String::new(),
549            definition_hash: String::new(),
550            input: Value::Null,
551            status: WorkflowStatus::Running,
552            created_at_ms: 0,
553            updated_at_ms: 0,
554            finished_at_ms: None,
555            units: BTreeMap::new(),
556            abort: None,
557            compensation_order: vec![],
558            compensation: None,
559            output: None,
560            error: None,
561            signals: vec![],
562            handlers: None,
563            last_seq: 0,
564        }
565    }
566
567    /// Rebuild from a full history.
568    pub fn replay(id: &str, history: &[HistoryEvent]) -> Self {
569        let mut s = ExecutionState::new(id);
570        for h in history {
571            apply(&mut s, h);
572        }
573        s
574    }
575
576    pub fn unit(&self, key: &str) -> Option<&UnitState> {
577        self.units.get(key)
578    }
579}
580
581/// Apply one event. Pure and total: unknown combinations are ignored so that
582/// replaying an older history never panics.
583pub fn apply(s: &mut ExecutionState, h: &HistoryEvent) {
584    let ts = h.timestamp_ms;
585    s.last_seq = h.seq;
586    s.updated_at_ms = ts;
587    match &h.event {
588        Event::WorkflowCreated {
589            workflow,
590            version,
591            definition_hash,
592            input,
593            parent,
594            schedule_id,
595            run,
596        } => {
597            s.parent = parent.clone();
598            s.schedule_id = schedule_id.clone();
599            s.run = run.unwrap_or(1);
600            s.workflow = workflow.clone();
601            s.version = version.clone();
602            s.definition_hash = definition_hash.clone();
603            s.input = input.clone();
604            s.created_at_ms = ts;
605            s.status = WorkflowStatus::Running;
606        }
607        Event::WorkflowStarted {} => {}
608        Event::StepScheduled {
609            step,
610            attempt,
611            input,
612            idempotency_key,
613            redelivery,
614        } => {
615            let u = s.units.entry(step.clone()).or_default();
616            u.status = Some(UnitStatus::Scheduled);
617            u.attempt = *attempt;
618            u.input = Some(input.clone());
619            u.idempotency_key = Some(idempotency_key.clone());
620            u.redelivery = *redelivery;
621            u.scheduled_at_ms = Some(ts);
622            u.token = None;
623            u.deadline_ms = None;
624            u.wake_at_ms = None;
625        }
626        Event::StepFanOut { step, items } => {
627            let u = s.units.entry(step.clone()).or_default();
628            u.status = Some(UnitStatus::Running);
629            u.input = Some(Value::Array(items.clone()));
630            u.scheduled_at_ms = Some(ts);
631            u.attempt = 1;
632            for i in 0..items.len() {
633                s.units.entry(element_key(step, i)).or_default();
634            }
635        }
636        Event::StepSkipped { step, .. } => {
637            let u = s.units.entry(step.clone()).or_default();
638            u.status = Some(UnitStatus::Skipped);
639            u.output = Some(Value::Null);
640            u.finished_at_ms = Some(ts);
641        }
642        Event::StepDispatched {
643            step,
644            attempt,
645            token,
646            worker_id,
647            worker_session,
648            deadline_ms,
649        } => {
650            let u = s.units.entry(step.clone()).or_default();
651            u.status = Some(UnitStatus::Dispatched);
652            u.attempt = *attempt;
653            u.token = Some(token.clone());
654            u.tokens.push((token.clone(), *attempt));
655            u.worker_id = Some(worker_id.clone());
656            u.worker_session = Some(worker_session.clone());
657            u.dispatched_at_ms = Some(ts);
658            u.deadline_ms = *deadline_ms;
659            u.dispatch_count += 1;
660        }
661        Event::StepSucceeded {
662            step,
663            attempt,
664            output,
665        } => {
666            let u = s.units.entry(step.clone()).or_default();
667            u.status = Some(UnitStatus::Succeeded);
668            u.attempt = *attempt;
669            u.output = Some(output.clone());
670            u.finished_at_ms = Some(ts);
671            u.succeeded_seq = Some(h.seq);
672            u.token = None;
673            u.wake_at_ms = None;
674        }
675        Event::StepAttemptFailed { step, error, .. } => {
676            let u = s.units.entry(step.clone()).or_default();
677            u.status = Some(UnitStatus::AttemptFailed);
678            u.last_error = Some(error.clone());
679            u.token = None;
680        }
681        Event::StepAttemptUncertain { step, .. } => {
682            let u = s.units.entry(step.clone()).or_default();
683            u.status = Some(UnitStatus::Uncertain);
684            u.token = None;
685        }
686        Event::RetryScheduled {
687            step,
688            next_attempt,
689            fire_at_ms,
690            ..
691        } => {
692            let u = s.units.entry(step.clone()).or_default();
693            u.status = Some(UnitStatus::RetryWait);
694            u.attempt = *next_attempt;
695            u.wake_at_ms = Some(*fire_at_ms);
696        }
697        Event::StepFailed { step, error } => {
698            let u = s.units.entry(step.clone()).or_default();
699            u.status = Some(UnitStatus::Failed);
700            u.error = Some(error.clone());
701            u.finished_at_ms = Some(ts);
702            u.token = None;
703            if s.abort.is_none() && !step.starts_with("compensate:") {
704                s.abort = Some(AbortReason::Failed {
705                    error: TaskError {
706                        message: format!("step `{step}` failed: {}", error.message),
707                        ..error.clone()
708                    },
709                });
710            }
711        }
712        Event::StepCancelled { step } => {
713            let u = s.units.entry(step.clone()).or_default();
714            u.status = Some(UnitStatus::Cancelled);
715            u.finished_at_ms = Some(ts);
716            u.token = None;
717        }
718        Event::TimerStarted { step, fire_at_ms } => {
719            let u = s.units.entry(step.clone()).or_default();
720            u.status = Some(UnitStatus::Waiting);
721            u.wake_at_ms = Some(*fire_at_ms);
722            u.scheduled_at_ms = Some(ts);
723            u.attempt = 1;
724        }
725        Event::TimerFired { step } => {
726            let u = s.units.entry(step.clone()).or_default();
727            u.status = Some(UnitStatus::Succeeded);
728            u.output = Some(Value::Object(Default::default()));
729            u.finished_at_ms = Some(ts);
730            u.succeeded_seq = Some(h.seq);
731            u.wake_at_ms = None;
732        }
733        Event::StaleResultIgnored { .. } => {}
734        Event::CancelRequested { reason } => {
735            if s.abort.is_none() {
736                s.abort = Some(AbortReason::Cancelled {
737                    reason: reason.clone(),
738                });
739            }
740        }
741        Event::WorkflowTimeoutReached {} => {
742            if s.abort.is_none() {
743                s.abort = Some(AbortReason::TimedOut);
744            }
745        }
746        Event::CompensationStarted { steps, .. } => {
747            s.status = WorkflowStatus::Compensating;
748            s.compensation_order = steps.clone();
749            for st in steps {
750                s.units.entry(compensation_key(st)).or_default();
751            }
752        }
753        Event::CompensationCompleted {} => s.compensation = Some(CompensationOutcome::Completed),
754        Event::CompensationFailed { .. } => s.compensation = Some(CompensationOutcome::Failed),
755        Event::WorkflowCompleted { output } => {
756            s.status = WorkflowStatus::Completed;
757            s.output = Some(output.clone());
758            s.finished_at_ms = Some(ts);
759        }
760        Event::WorkflowFailed {
761            error,
762            compensation,
763        } => {
764            s.status = WorkflowStatus::Failed;
765            s.error = Some(error.clone());
766            s.compensation = Some(*compensation);
767            s.finished_at_ms = Some(ts);
768        }
769        Event::WorkflowCancelled { compensation } => {
770            s.status = WorkflowStatus::Cancelled;
771            s.compensation = Some(*compensation);
772            s.finished_at_ms = Some(ts);
773        }
774        Event::WorkflowTimedOut { compensation } => {
775            s.status = WorkflowStatus::TimedOut;
776            s.compensation = Some(*compensation);
777            s.finished_at_ms = Some(ts);
778        }
779        Event::LimitExceeded { message, .. } => {
780            if s.abort.is_none() {
781                s.abort = Some(AbortReason::Failed {
782                    error: TaskError::new(codes::LIMIT_EXCEEDED, message.clone(), false),
783                });
784            }
785        }
786        Event::HandlersStarted { kind, compensation } => {
787            s.handlers = Some(HandlerPhase {
788                kind: kind.clone(),
789                compensation: *compensation,
790            });
791        }
792        Event::SignalReceived { name, payload } => s.signals.push(ReceivedSignal {
793            seq: h.seq,
794            name: name.clone(),
795            payload: payload.clone(),
796            received_at_ms: ts,
797            delivered_to: None,
798        }),
799        Event::SignalAwaitStarted {
800            step, deadline_ms, ..
801        } => {
802            let u = s.units.entry(step.clone()).or_default();
803            u.status = Some(UnitStatus::Waiting);
804            u.wake_at_ms = *deadline_ms;
805            u.scheduled_at_ms = Some(ts);
806            u.attempt = 1;
807        }
808        Event::WorkflowContinuedAsNew { .. } => {
809            s.status = WorkflowStatus::ContinuedAsNew;
810            s.finished_at_ms = Some(ts);
811        }
812        Event::SignalDelivered { step, signal_seq } => {
813            let payload = match s.signals.iter_mut().find(|r| r.seq == *signal_seq) {
814                Some(r) => {
815                    r.delivered_to = Some(step.clone());
816                    r.payload.clone()
817                }
818                None => Value::Null,
819            };
820            let u = s.units.entry(step.clone()).or_default();
821            u.status = Some(UnitStatus::Succeeded);
822            u.output = Some(payload);
823            u.finished_at_ms = Some(ts);
824            u.succeeded_seq = Some(h.seq);
825            u.wake_at_ms = None;
826        }
827    }
828}
orbit_runtime/model.rs

orbit_runtime/
model.rs
