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};
7
8use super::{
9    CancellationRequestSnapshot, ChildOperationReference, JsonValue, RetryPolicy, WorkflowProgress,
10    WorkflowSpec, WorkflowTerminalOutcome,
11};
12
13#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
14#[serde(rename_all = "snake_case")]
15pub enum WorkflowRunStatus {
16    Pending,
17    Running,
18    Suspended,
19    Cancelling,
20    Completed,
21    Failed,
22    Cancelled,
23}
24
25impl WorkflowRunStatus {
26    pub fn is_terminal(self) -> bool {
27        matches!(self, Self::Completed | Self::Failed | Self::Cancelled)
28    }
29}
30
31#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
32#[serde(rename_all = "snake_case")]
33pub enum StepStatus {
34    Pending,
35    Running,
36    Completed,
37    Failed,
38    Cancelled,
39}
40
41#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
42pub struct StepSnapshot {
43    pub step_id: String,
44    pub step_name: String,
45    pub status: StepStatus,
46    pub input: JsonValue,
47    pub retry: RetryPolicy,
48    pub output: Option<JsonValue>,
49    pub error: Option<String>,
50    pub attempt: u32,
51    pub retry_after: Option<DateTime<Utc>>,
52}
53
54impl StepSnapshot {
55    /// Decode the persisted step output into a host-defined serde type.
56    pub fn output_as<T>(&self) -> Result<Option<T>>
57    where
58        T: DeserializeOwned,
59    {
60        self.output
61            .clone()
62            .map(serde_json::from_value)
63            .transpose()
64            .map_err(FlowError::from)
65    }
66}
67
68#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
69#[serde(rename_all = "snake_case")]
70pub enum WaitStatus {
71    Waiting,
72    Completed,
73    Cancelled,
74}
75
76#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
77pub struct WaitSnapshot {
78    pub wait_id: String,
79    pub status: WaitStatus,
80    pub resume_at: DateTime<Utc>,
81}
82
83#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
84#[serde(rename_all = "snake_case")]
85pub enum HookStatus {
86    Active,
87    Received,
88    Disposed,
89    Cancelled,
90}
91
92#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
93pub struct HookSnapshot {
94    pub hook_id: String,
95    pub token: String,
96    pub status: HookStatus,
97    pub metadata: JsonValue,
98    pub payload: Option<JsonValue>,
99}
100
101impl HookSnapshot {
102    /// Decode the persisted hook metadata into a host-defined serde type.
103    pub fn metadata_as<T>(&self) -> Result<T>
104    where
105        T: DeserializeOwned,
106    {
107        serde_json::from_value(self.metadata.clone()).map_err(FlowError::from)
108    }
109
110    /// Decode the received hook payload into a host-defined serde type.
111    pub fn payload_as<T>(&self) -> Result<Option<T>>
112    where
113        T: DeserializeOwned,
114    {
115        self.payload
116            .clone()
117            .map(serde_json::from_value)
118            .transpose()
119            .map_err(FlowError::from)
120    }
121}
122
123/// Active external callback hook with the run that owns it.
124#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
125pub struct ActiveHookSnapshot {
126    pub run_id: String,
127    pub hook: HookSnapshot,
128}
129
130impl ActiveHookSnapshot {
131    /// Decode the active hook metadata into a host-defined serde type.
132    pub fn metadata_as<T>(&self) -> Result<T>
133    where
134        T: DeserializeOwned,
135    {
136        self.hook.metadata_as()
137    }
138}
139
140/// Kind of durable timer that can wake a suspended workflow run.
141#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
142#[serde(rename_all = "snake_case")]
143pub enum ScheduledWakeupKind {
144    Wait,
145    Retry,
146}
147
148impl ScheduledWakeupKind {
149    #[cfg(any(feature = "postgres", feature = "sqlite"))]
150    pub(crate) fn from_database_code(code: i64) -> Result<Self> {
151        match code {
152            0 => Ok(Self::Wait),
153            2 => Ok(Self::Retry),
154            _ => Err(FlowError::Store(format!(
155                "invalid scheduled wakeup kind code {code}"
156            ))),
157        }
158    }
159}
160
161/// Minimal indexed record for a wait timer or delayed step retry.
162#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
163pub struct ScheduledWakeup {
164    pub run_id: String,
165    pub kind: ScheduledWakeupKind,
166    pub subject_id: String,
167    pub scheduled_at: DateTime<Utc>,
168}
169
170/// Aggregated run counts for host dashboards and health probes.
171#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
172#[serde(default)]
173pub struct WorkflowRunSummary {
174    pub total_runs: usize,
175    pub pending_runs: usize,
176    pub running_runs: usize,
177    pub suspended_runs: usize,
178    pub cancelling_runs: usize,
179    pub completed_runs: usize,
180    pub failed_runs: usize,
181    pub cancelled_runs: usize,
182    pub terminal_runs: usize,
183    pub non_terminal_runs: usize,
184    pub open_waits: usize,
185    pub active_hooks: usize,
186    pub pending_retries: usize,
187}
188
189impl WorkflowRunSummary {
190    pub fn from_snapshots(snapshots: &[WorkflowRunSnapshot]) -> Self {
191        let mut summary = Self::default();
192        for snapshot in snapshots {
193            summary.record(snapshot);
194        }
195        summary
196    }
197
198    pub fn record(&mut self, snapshot: &WorkflowRunSnapshot) {
199        self.total_runs += 1;
200        match snapshot.status {
201            WorkflowRunStatus::Pending => self.pending_runs += 1,
202            WorkflowRunStatus::Running => self.running_runs += 1,
203            WorkflowRunStatus::Suspended => self.suspended_runs += 1,
204            WorkflowRunStatus::Cancelling => self.cancelling_runs += 1,
205            WorkflowRunStatus::Completed => self.completed_runs += 1,
206            WorkflowRunStatus::Failed => self.failed_runs += 1,
207            WorkflowRunStatus::Cancelled => self.cancelled_runs += 1,
208        }
209
210        if snapshot.status.is_terminal() {
211            self.terminal_runs += 1;
212            return;
213        }
214
215        self.non_terminal_runs += 1;
216        self.open_waits += snapshot
217            .waits
218            .values()
219            .filter(|wait| wait.status == WaitStatus::Waiting)
220            .count();
221        self.active_hooks += snapshot
222            .hooks
223            .values()
224            .filter(|hook| hook.status == HookStatus::Active)
225            .count();
226        self.pending_retries += snapshot
227            .steps
228            .values()
229            .filter(|step| step.status == StepStatus::Pending && step.retry_after.is_some())
230            .count();
231    }
232}
233
234/// Open suspension projected for host dashboards and operator consoles.
235#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
236#[serde(tag = "kind", rename_all = "snake_case")]
237pub enum WorkflowRunSuspension {
238    Wait {
239        run_id: String,
240        wait: WaitSnapshot,
241        due: bool,
242    },
243    Hook {
244        run_id: String,
245        hook: HookSnapshot,
246    },
247    Retry {
248        run_id: String,
249        step: StepSnapshot,
250        due: bool,
251    },
252}
253
254impl WorkflowRunSuspension {
255    pub fn run_id(&self) -> &str {
256        match self {
257            Self::Wait { run_id, .. } | Self::Hook { run_id, .. } | Self::Retry { run_id, .. } => {
258                run_id
259            }
260        }
261    }
262
263    pub fn subject_id(&self) -> &str {
264        match self {
265            Self::Wait { wait, .. } => &wait.wait_id,
266            Self::Hook { hook, .. } => &hook.hook_id,
267            Self::Retry { step, .. } => &step.step_id,
268        }
269    }
270
271    pub(crate) fn kind_order(&self) -> u8 {
272        match self {
273            Self::Wait { .. } => 0,
274            Self::Hook { .. } => 1,
275            Self::Retry { .. } => 2,
276        }
277    }
278
279    pub fn is_due(&self) -> bool {
280        match self {
281            Self::Wait { due, .. } | Self::Retry { due, .. } => *due,
282            Self::Hook { .. } => false,
283        }
284    }
285
286    /// Scheduled resume time for wait and delayed-retry suspensions.
287    pub fn scheduled_at(&self) -> Option<DateTime<Utc>> {
288        match self {
289            Self::Wait { wait, .. } => Some(wait.resume_at),
290            Self::Retry { step, .. } => step.retry_after,
291            Self::Hook { .. } => None,
292        }
293    }
294}
295
296/// Materialized state of a workflow run.
297#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
298pub struct WorkflowRunSnapshot {
299    pub run_id: String,
300    pub spec: WorkflowSpec,
301    pub input: JsonValue,
302    pub status: WorkflowRunStatus,
303    pub steps: BTreeMap<String, StepSnapshot>,
304    pub waits: BTreeMap<String, WaitSnapshot>,
305    pub hooks: BTreeMap<String, HookSnapshot>,
306    #[serde(default)]
307    pub cancellation: Option<CancellationRequestSnapshot>,
308    #[serde(default)]
309    pub progress: Vec<WorkflowProgress>,
310    #[serde(default)]
311    pub child_operations: BTreeMap<String, ChildOperationReference>,
312    pub output: Option<JsonValue>,
313    pub error: Option<String>,
314    #[serde(default)]
315    pub terminal_outcome: Option<WorkflowTerminalOutcome>,
316    pub last_sequence: u64,
317}
318
319impl WorkflowRunSnapshot {
320    /// Decode the workflow input into a host-defined serde type.
321    pub fn input_as<T>(&self) -> Result<T>
322    where
323        T: DeserializeOwned,
324    {
325        serde_json::from_value(self.input.clone()).map_err(FlowError::from)
326    }
327
328    /// Decode the terminal workflow output into a host-defined serde type.
329    pub fn output_as<T>(&self) -> Result<Option<T>>
330    where
331        T: DeserializeOwned,
332    {
333        self.output
334            .clone()
335            .map(serde_json::from_value)
336            .transpose()
337            .map_err(FlowError::from)
338    }
339
340    pub fn step_output(&self, step_id: &str) -> Option<&JsonValue> {
341        self.steps
342            .get(step_id)
343            .and_then(|step| step.output.as_ref())
344    }
345
346    /// Decode a persisted step output into a host-defined serde type.
347    pub fn step_output_as<T>(&self, step_id: &str) -> Result<Option<T>>
348    where
349        T: DeserializeOwned,
350    {
351        match self.steps.get(step_id) {
352            Some(step) => step.output_as(),
353            None => Ok(None),
354        }
355    }
356
357    pub fn hook_payload(&self, hook_id: &str) -> Option<&JsonValue> {
358        self.hooks
359            .get(hook_id)
360            .and_then(|hook| hook.payload.as_ref())
361    }
362
363    /// Return a durable progress update by its idempotency identity.
364    pub fn progress(&self, progress_id: &str) -> Option<&WorkflowProgress> {
365        self.progress
366            .iter()
367            .find(|progress| progress.progress_id == progress_id)
368    }
369
370    /// Return the most recently persisted progress update.
371    pub fn latest_progress(&self) -> Option<&WorkflowProgress> {
372        self.progress.last()
373    }
374
375    /// Return a durable child-operation reference by its parent-local id.
376    pub fn child_operation(&self, reference_id: &str) -> Option<&ChildOperationReference> {
377        self.child_operations.get(reference_id)
378    }
379
380    /// Decode persisted hook metadata into a host-defined serde type.
381    pub fn hook_metadata_as<T>(&self, hook_id: &str) -> Result<Option<T>>
382    where
383        T: DeserializeOwned,
384    {
385        match self.hooks.get(hook_id) {
386            Some(hook) => hook.metadata_as().map(Some),
387            None => Ok(None),
388        }
389    }
390
391    /// Decode a received hook payload into a host-defined serde type.
392    pub fn hook_payload_as<T>(&self, hook_id: &str) -> Result<Option<T>>
393    where
394        T: DeserializeOwned,
395    {
396        match self.hooks.get(hook_id) {
397            Some(hook) => hook.payload_as(),
398            None => Ok(None),
399        }
400    }
401
402    pub fn has_open_suspension(&self) -> bool {
403        self.waits
404            .values()
405            .any(|wait| wait.status == WaitStatus::Waiting)
406            || self
407                .hooks
408                .values()
409                .any(|hook| hook.status == HookStatus::Active)
410            || self.steps.values().any(|step| step.retry_after.is_some())
411    }
412
413    pub fn due_retries(&self, now: DateTime<Utc>) -> Vec<(String, DateTime<Utc>)> {
414        self.steps
415            .values()
416            .filter_map(|step| match step.retry_after {
417                Some(retry_after) if step.status == StepStatus::Pending && retry_after <= now => {
418                    Some((step.step_id.clone(), retry_after))
419                }
420                _ => None,
421            })
422            .collect()
423    }
424
425    pub fn has_future_retry(&self, now: DateTime<Utc>) -> bool {
426        self.steps.values().any(|step| {
427            step.status == StepStatus::Pending
428                && step
429                    .retry_after
430                    .map(|retry_after| retry_after > now)
431                    .unwrap_or(false)
432        })
433    }
434}
a3s_flow/model/snapshot.rs

a3s_flow/model/
snapshot.rs
