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

orbit_runtime/
model.rs
