a3s_flow/model/
snapshot.rs

1use chrono::{DateTime, Utc};
2use serde::de::DeserializeOwned;
3use serde::{Deserialize, Serialize};
4use std::collections::BTreeMap;
5
6use crate::error::{FlowError, Result};
7use crate::runtime_build::RuntimeBuildId;
8
9use super::{
10    CancellationRequestSnapshot, ChildOperationReference, ChildWorkflowSnapshot, JsonValue,
11    RetryPolicy, SignalWaitSnapshot, SignalWaitStatus, WorkflowContinuation, WorkflowProgress,
12    WorkflowSignalSnapshot, WorkflowSpec, WorkflowTerminalOutcome,
13};
14
15/// Materialized lifecycle state of a workflow run.
16#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
17#[non_exhaustive]
18#[serde(rename_all = "snake_case")]
19pub enum WorkflowRunStatus {
20    /// The run exists but has not started replay.
21    Pending,
22    /// The workflow runtime is actively replaying or dispatching work.
23    Running,
24    /// The run is waiting for durable external work or a timer.
25    Suspended,
26    /// A cancellation request is replaying the workflow's cleanup path.
27    Cancelling,
28    /// The run completed successfully.
29    Completed,
30    /// The run terminated with an error.
31    Failed,
32    /// The run completed cancellation.
33    Cancelled,
34    /// The run closed after creating a successor history segment.
35    ContinuedAsNew,
36}
37
38impl WorkflowRunStatus {
39    /// Returns whether no further events may be appended to this run segment.
40    pub fn is_terminal(self) -> bool {
41        matches!(
42            self,
43            Self::Completed | Self::Failed | Self::Cancelled | Self::ContinuedAsNew
44        )
45    }
46}
47
48/// Materialized lifecycle state of a durable step.
49#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
50#[non_exhaustive]
51#[serde(rename_all = "snake_case")]
52pub enum StepStatus {
53    /// The step is ready now or after a retry deadline.
54    Pending,
55    /// A worker has started the current attempt.
56    Running,
57    /// The step produced a durable output.
58    Completed,
59    /// The step exhausted its retry policy.
60    Failed,
61    /// The owning run cancelled the step before completion.
62    Cancelled,
63}
64
65/// Materialized state of one durable step.
66#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
67#[non_exhaustive]
68pub struct StepSnapshot {
69    /// Replay-stable identity of the step.
70    pub step_id: String,
71    /// Registered step implementation name.
72    pub step_name: String,
73    /// Current lifecycle state.
74    pub status: StepStatus,
75    /// JSON input pinned when the step was created.
76    pub input: JsonValue,
77    /// Retry behavior pinned when the step was created.
78    pub retry: RetryPolicy,
79    /// Durable JSON output, when completed successfully.
80    pub output: Option<JsonValue>,
81    /// Final or most recent attempt error.
82    pub error: Option<String>,
83    /// Latest one-based attempt number observed in history.
84    pub attempt: u32,
85    /// Earliest UTC time for a delayed retry.
86    pub retry_after: Option<DateTime<Utc>>,
87}
88
89impl StepSnapshot {
90    /// Decode the persisted step output into a host-defined serde type.
91    pub fn output_as<T>(&self) -> Result<Option<T>>
92    where
93        T: DeserializeOwned,
94    {
95        self.output
96            .clone()
97            .map(serde_json::from_value)
98            .transpose()
99            .map_err(FlowError::from)
100    }
101}
102
103/// Materialized lifecycle state of a durable timer wait.
104#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
105#[non_exhaustive]
106#[serde(rename_all = "snake_case")]
107pub enum WaitStatus {
108    /// The timer deadline has not been completed.
109    Waiting,
110    /// The timer deadline was durably completed.
111    Completed,
112    /// The owning run cancelled the timer.
113    Cancelled,
114}
115
116/// Materialized state of one durable timer wait.
117#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
118#[non_exhaustive]
119pub struct WaitSnapshot {
120    /// Replay-stable identity of the wait.
121    pub wait_id: String,
122    /// Current lifecycle state.
123    pub status: WaitStatus,
124    /// UTC time at which the wait becomes ready.
125    pub resume_at: DateTime<Utc>,
126}
127
128/// Materialized lifecycle state of an external callback hook.
129#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
130#[non_exhaustive]
131#[serde(rename_all = "snake_case")]
132pub enum HookStatus {
133    /// The hook can accept one external resolution.
134    Active,
135    /// A callback payload was received.
136    Received,
137    /// The hook was explicitly closed without a payload.
138    Disposed,
139    /// The owning run cancelled the hook.
140    Cancelled,
141}
142
143/// Materialized state of one external callback hook.
144#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
145#[non_exhaustive]
146pub struct HookSnapshot {
147    /// Replay-stable identity of the hook.
148    pub hook_id: String,
149    /// Secret bearer token used to resolve the hook.
150    pub token: String,
151    /// Current lifecycle state.
152    pub status: HookStatus,
153    /// Application metadata pinned when the hook was created.
154    pub metadata: JsonValue,
155    /// JSON payload received from the external caller.
156    pub payload: Option<JsonValue>,
157}
158
159impl HookSnapshot {
160    /// Create a materialized hook value for a custom event-store projection.
161    pub fn new(
162        hook_id: impl Into<String>,
163        token: impl Into<String>,
164        status: HookStatus,
165        metadata: JsonValue,
166        payload: Option<JsonValue>,
167    ) -> Self {
168        Self {
169            hook_id: hook_id.into(),
170            token: token.into(),
171            status,
172            metadata,
173            payload,
174        }
175    }
176
177    /// Decode the persisted hook metadata into a host-defined serde type.
178    pub fn metadata_as<T>(&self) -> Result<T>
179    where
180        T: DeserializeOwned,
181    {
182        serde_json::from_value(self.metadata.clone()).map_err(FlowError::from)
183    }
184
185    /// Decode the received hook payload into a host-defined serde type.
186    pub fn payload_as<T>(&self) -> Result<Option<T>>
187    where
188        T: DeserializeOwned,
189    {
190        self.payload
191            .clone()
192            .map(serde_json::from_value)
193            .transpose()
194            .map_err(FlowError::from)
195    }
196}
197
198/// Active external callback hook with the run that owns it.
199#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
200#[non_exhaustive]
201pub struct ActiveHookSnapshot {
202    /// Run that owns the active hook.
203    pub run_id: String,
204    /// Materialized active hook state.
205    pub hook: HookSnapshot,
206}
207
208impl ActiveHookSnapshot {
209    /// Associate a materialized active hook with its owning run.
210    pub fn new(run_id: impl Into<String>, hook: HookSnapshot) -> Self {
211        Self {
212            run_id: run_id.into(),
213            hook,
214        }
215    }
216
217    /// Decode the active hook metadata into a host-defined serde type.
218    pub fn metadata_as<T>(&self) -> Result<T>
219    where
220        T: DeserializeOwned,
221    {
222        self.hook.metadata_as()
223    }
224}
225
226/// Kind of durable timer that can wake a suspended workflow run.
227#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
228#[non_exhaustive]
229#[serde(rename_all = "snake_case")]
230pub enum ScheduledWakeupKind {
231    /// A durable timer wait.
232    Wait,
233    /// A delayed step retry.
234    Retry,
235}
236
237impl ScheduledWakeupKind {
238    #[cfg(any(feature = "postgres", feature = "sqlite"))]
239    pub(crate) fn from_database_code(code: i64) -> Result<Self> {
240        match code {
241            0 => Ok(Self::Wait),
242            2 => Ok(Self::Retry),
243            _ => Err(FlowError::Store(format!(
244                "invalid scheduled wakeup kind code {code}"
245            ))),
246        }
247    }
248}
249
250/// Minimal indexed record for a wait timer or delayed step retry.
251#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
252#[non_exhaustive]
253pub struct ScheduledWakeup {
254    /// Run that owns the scheduled work.
255    pub run_id: String,
256    /// Kind of durable work that becomes ready.
257    pub kind: ScheduledWakeupKind,
258    /// Step or wait identifier within the run.
259    pub subject_id: String,
260    /// UTC time at which the work becomes ready.
261    pub scheduled_at: DateTime<Utc>,
262    /// Runtime build persisted by the owning run, used for indexed dispatch.
263    #[serde(default, skip_serializing_if = "Option::is_none")]
264    pub runtime_build_id: Option<RuntimeBuildId>,
265}
266
267impl ScheduledWakeup {
268    /// Create an indexed scheduled-work record for a custom event store.
269    pub fn new(
270        run_id: impl Into<String>,
271        kind: ScheduledWakeupKind,
272        subject_id: impl Into<String>,
273        scheduled_at: DateTime<Utc>,
274        runtime_build_id: Option<RuntimeBuildId>,
275    ) -> Self {
276        Self {
277            run_id: run_id.into(),
278            kind,
279            subject_id: subject_id.into(),
280            scheduled_at,
281            runtime_build_id,
282        }
283    }
284}
285
286/// Materialized state of a workflow run.
287#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
288#[non_exhaustive]
289pub struct WorkflowRunSnapshot {
290    /// Stable identifier of the run.
291    pub run_id: String,
292    /// Immutable workflow definition pinned at creation.
293    pub spec: WorkflowSpec,
294    /// Initial JSON input supplied to the workflow.
295    pub input: JsonValue,
296    /// Current materialized run state.
297    pub status: WorkflowRunStatus,
298    /// Durable steps indexed by their stable identifiers.
299    pub steps: BTreeMap<String, StepSnapshot>,
300    /// Durable timer waits indexed by their stable identifiers.
301    pub waits: BTreeMap<String, WaitSnapshot>,
302    /// External callback hooks indexed by their stable identifiers.
303    pub hooks: BTreeMap<String, HookSnapshot>,
304    /// Active or completed cleanup-aware cancellation request.
305    #[serde(default)]
306    pub cancellation: Option<CancellationRequestSnapshot>,
307    /// Durable progress updates in event order.
308    #[serde(default)]
309    pub progress: Vec<WorkflowProgress>,
310    /// Linked child operations indexed by parent-local identifiers.
311    #[serde(default)]
312    pub child_operations: BTreeMap<String, ChildOperationReference>,
313    /// First-class child workflows indexed by parent-local identifiers.
314    #[serde(default)]
315    pub child_workflows: BTreeMap<String, ChildWorkflowSnapshot>,
316    /// Received signals in durable delivery order.
317    #[serde(default)]
318    pub signals: Vec<WorkflowSignalSnapshot>,
319    /// Deterministic signal waits indexed by stable wait identifiers.
320    #[serde(default)]
321    pub signal_waits: BTreeMap<String, SignalWaitSnapshot>,
322    /// Final JSON output for a successfully completed run.
323    pub output: Option<JsonValue>,
324    /// Terminal error for a failed run.
325    pub error: Option<String>,
326    /// Typed terminal result projected from the closing event.
327    #[serde(default)]
328    pub terminal_outcome: Option<WorkflowTerminalOutcome>,
329    /// Link to a successor history segment created by continue-as-new.
330    #[serde(default, skip_serializing_if = "Option::is_none")]
331    pub continuation: Option<WorkflowContinuation>,
332    /// Last event sequence included in this materialized state.
333    pub last_sequence: u64,
334}
335
336impl WorkflowRunSnapshot {
337    /// Create the empty pending projection for a newly persisted workflow run.
338    ///
339    /// Custom event stores and downstream tests should start from this
340    /// constructor instead of a struct literal so new projection fields can be
341    /// added without breaking callers.
342    pub fn new(run_id: impl Into<String>, spec: WorkflowSpec, input: JsonValue) -> Self {
343        Self {
344            run_id: run_id.into(),
345            spec,
346            input,
347            status: WorkflowRunStatus::Pending,
348            steps: BTreeMap::new(),
349            waits: BTreeMap::new(),
350            hooks: BTreeMap::new(),
351            cancellation: None,
352            progress: Vec::new(),
353            child_operations: BTreeMap::new(),
354            child_workflows: BTreeMap::new(),
355            signals: Vec::new(),
356            signal_waits: BTreeMap::new(),
357            output: None,
358            error: None,
359            terminal_outcome: None,
360            continuation: None,
361            last_sequence: 0,
362        }
363    }
364
365    /// Decode the workflow input into a host-defined serde type.
366    pub fn input_as<T>(&self) -> Result<T>
367    where
368        T: DeserializeOwned,
369    {
370        serde_json::from_value(self.input.clone()).map_err(FlowError::from)
371    }
372
373    /// Decode the terminal workflow output into a host-defined serde type.
374    pub fn output_as<T>(&self) -> Result<Option<T>>
375    where
376        T: DeserializeOwned,
377    {
378        self.output
379            .clone()
380            .map(serde_json::from_value)
381            .transpose()
382            .map_err(FlowError::from)
383    }
384
385    /// Returns the durable JSON output of a completed step.
386    pub fn step_output(&self, step_id: &str) -> Option<&JsonValue> {
387        self.steps
388            .get(step_id)
389            .and_then(|step| step.output.as_ref())
390    }
391
392    /// Decode a persisted step output into a host-defined serde type.
393    pub fn step_output_as<T>(&self, step_id: &str) -> Result<Option<T>>
394    where
395        T: DeserializeOwned,
396    {
397        match self.steps.get(step_id) {
398            Some(step) => step.output_as(),
399            None => Ok(None),
400        }
401    }
402
403    /// Returns the JSON payload received by a hook.
404    pub fn hook_payload(&self, hook_id: &str) -> Option<&JsonValue> {
405        self.hooks
406            .get(hook_id)
407            .and_then(|hook| hook.payload.as_ref())
408    }
409
410    /// Return a durable progress update by its idempotency identity.
411    pub fn progress(&self, progress_id: &str) -> Option<&WorkflowProgress> {
412        self.progress
413            .iter()
414            .find(|progress| progress.progress_id == progress_id)
415    }
416
417    /// Return the most recently persisted progress update.
418    pub fn latest_progress(&self) -> Option<&WorkflowProgress> {
419        self.progress.last()
420    }
421
422    /// Return a durable child-operation reference by its parent-local id.
423    pub fn child_operation(&self, reference_id: &str) -> Option<&ChildOperationReference> {
424        self.child_operations.get(reference_id)
425    }
426
427    /// Return a first-class child workflow by its stable parent-local id.
428    pub fn child_workflow(&self, child_id: &str) -> Option<&ChildWorkflowSnapshot> {
429        self.child_workflows.get(child_id)
430    }
431
432    /// Return a received signal by its caller-owned idempotency identity.
433    pub fn signal(&self, signal_id: &str) -> Option<&WorkflowSignalSnapshot> {
434        self.signals
435            .iter()
436            .find(|signal| signal.signal_id == signal_id)
437    }
438
439    /// Return the signal payload paired with a deterministic signal wait.
440    pub fn signal_wait_payload(&self, wait_id: &str) -> Option<&JsonValue> {
441        let signal_id = self.signal_waits.get(wait_id)?.signal_id.as_deref()?;
442        self.signal(signal_id).map(|signal| &signal.payload)
443    }
444
445    /// Decode the signal payload paired with a deterministic signal wait.
446    pub fn signal_wait_payload_as<T>(&self, wait_id: &str) -> Result<Option<T>>
447    where
448        T: DeserializeOwned,
449    {
450        self.signal_wait_payload(wait_id)
451            .cloned()
452            .map(serde_json::from_value)
453            .transpose()
454            .map_err(FlowError::from)
455    }
456
457    /// Decode persisted hook metadata into a host-defined serde type.
458    pub fn hook_metadata_as<T>(&self, hook_id: &str) -> Result<Option<T>>
459    where
460        T: DeserializeOwned,
461    {
462        match self.hooks.get(hook_id) {
463            Some(hook) => hook.metadata_as().map(Some),
464            None => Ok(None),
465        }
466    }
467
468    /// Decode a received hook payload into a host-defined serde type.
469    pub fn hook_payload_as<T>(&self, hook_id: &str) -> Result<Option<T>>
470    where
471        T: DeserializeOwned,
472    {
473        match self.hooks.get(hook_id) {
474            Some(hook) => hook.payload_as(),
475            None => Ok(None),
476        }
477    }
478
479    /// Returns whether durable work is still preventing terminal completion.
480    pub fn has_open_suspension(&self) -> bool {
481        self.waits
482            .values()
483            .any(|wait| wait.status == WaitStatus::Waiting)
484            || self
485                .hooks
486                .values()
487                .any(|hook| hook.status == HookStatus::Active)
488            || self.steps.values().any(|step| step.retry_after.is_some())
489            || self
490                .child_workflows
491                .values()
492                .any(ChildWorkflowSnapshot::is_open)
493            || self
494                .signal_waits
495                .values()
496                .any(|wait| wait.status == SignalWaitStatus::Waiting)
497    }
498
499    /// Returns delayed step retries ready at or before `now`.
500    pub fn due_retries(&self, now: DateTime<Utc>) -> Vec<(String, DateTime<Utc>)> {
501        self.steps
502            .values()
503            .filter_map(|step| match step.retry_after {
504                Some(retry_after) if step.status == StepStatus::Pending && retry_after <= now => {
505                    Some((step.step_id.clone(), retry_after))
506                }
507                _ => None,
508            })
509            .collect()
510    }
511
512    /// Returns whether any pending step has a retry deadline after `now`.
513    pub fn has_future_retry(&self, now: DateTime<Utc>) -> bool {
514        self.steps.values().any(|step| {
515            step.status == StepStatus::Pending
516                && step
517                    .retry_after
518                    .map(|retry_after| retry_after > now)
519                    .unwrap_or(false)
520        })
521    }
522}
a3s_flow/model/snapshot.rs

a3s_flow/model/
snapshot.rs
