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

a3s_flow/model/
snapshot.rs
