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

orbit_runtime/
model.rs
