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/// Aggregated run counts for host dashboards and health probes.
141#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
142#[serde(default)]
143pub struct WorkflowRunSummary {
144    pub total_runs: usize,
145    pub pending_runs: usize,
146    pub running_runs: usize,
147    pub suspended_runs: usize,
148    pub cancelling_runs: usize,
149    pub completed_runs: usize,
150    pub failed_runs: usize,
151    pub cancelled_runs: usize,
152    pub terminal_runs: usize,
153    pub non_terminal_runs: usize,
154    pub open_waits: usize,
155    pub active_hooks: usize,
156    pub pending_retries: usize,
157}
158
159impl WorkflowRunSummary {
160    pub fn from_snapshots(snapshots: &[WorkflowRunSnapshot]) -> Self {
161        let mut summary = Self::default();
162        for snapshot in snapshots {
163            summary.record(snapshot);
164        }
165        summary
166    }
167
168    pub fn record(&mut self, snapshot: &WorkflowRunSnapshot) {
169        self.total_runs += 1;
170        match snapshot.status {
171            WorkflowRunStatus::Pending => self.pending_runs += 1,
172            WorkflowRunStatus::Running => self.running_runs += 1,
173            WorkflowRunStatus::Suspended => self.suspended_runs += 1,
174            WorkflowRunStatus::Cancelling => self.cancelling_runs += 1,
175            WorkflowRunStatus::Completed => self.completed_runs += 1,
176            WorkflowRunStatus::Failed => self.failed_runs += 1,
177            WorkflowRunStatus::Cancelled => self.cancelled_runs += 1,
178        }
179
180        if snapshot.status.is_terminal() {
181            self.terminal_runs += 1;
182            return;
183        }
184
185        self.non_terminal_runs += 1;
186        self.open_waits += snapshot
187            .waits
188            .values()
189            .filter(|wait| wait.status == WaitStatus::Waiting)
190            .count();
191        self.active_hooks += snapshot
192            .hooks
193            .values()
194            .filter(|hook| hook.status == HookStatus::Active)
195            .count();
196        self.pending_retries += snapshot
197            .steps
198            .values()
199            .filter(|step| step.status == StepStatus::Pending && step.retry_after.is_some())
200            .count();
201    }
202}
203
204/// Open suspension projected for host dashboards and operator consoles.
205#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
206#[serde(tag = "kind", rename_all = "snake_case")]
207pub enum WorkflowRunSuspension {
208    Wait {
209        run_id: String,
210        wait: WaitSnapshot,
211        due: bool,
212    },
213    Hook {
214        run_id: String,
215        hook: HookSnapshot,
216    },
217    Retry {
218        run_id: String,
219        step: StepSnapshot,
220        due: bool,
221    },
222}
223
224impl WorkflowRunSuspension {
225    pub fn run_id(&self) -> &str {
226        match self {
227            Self::Wait { run_id, .. } | Self::Hook { run_id, .. } | Self::Retry { run_id, .. } => {
228                run_id
229            }
230        }
231    }
232
233    pub fn subject_id(&self) -> &str {
234        match self {
235            Self::Wait { wait, .. } => &wait.wait_id,
236            Self::Hook { hook, .. } => &hook.hook_id,
237            Self::Retry { step, .. } => &step.step_id,
238        }
239    }
240
241    pub(crate) fn kind_order(&self) -> u8 {
242        match self {
243            Self::Wait { .. } => 0,
244            Self::Hook { .. } => 1,
245            Self::Retry { .. } => 2,
246        }
247    }
248
249    pub fn is_due(&self) -> bool {
250        match self {
251            Self::Wait { due, .. } | Self::Retry { due, .. } => *due,
252            Self::Hook { .. } => false,
253        }
254    }
255
256    /// Scheduled resume time for wait and delayed-retry suspensions.
257    pub fn scheduled_at(&self) -> Option<DateTime<Utc>> {
258        match self {
259            Self::Wait { wait, .. } => Some(wait.resume_at),
260            Self::Retry { step, .. } => step.retry_after,
261            Self::Hook { .. } => None,
262        }
263    }
264}
265
266/// Materialized state of a workflow run.
267#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
268pub struct WorkflowRunSnapshot {
269    pub run_id: String,
270    pub spec: WorkflowSpec,
271    pub input: JsonValue,
272    pub status: WorkflowRunStatus,
273    pub steps: BTreeMap<String, StepSnapshot>,
274    pub waits: BTreeMap<String, WaitSnapshot>,
275    pub hooks: BTreeMap<String, HookSnapshot>,
276    #[serde(default)]
277    pub cancellation: Option<CancellationRequestSnapshot>,
278    #[serde(default)]
279    pub progress: Vec<WorkflowProgress>,
280    #[serde(default)]
281    pub child_operations: BTreeMap<String, ChildOperationReference>,
282    pub output: Option<JsonValue>,
283    pub error: Option<String>,
284    #[serde(default)]
285    pub terminal_outcome: Option<WorkflowTerminalOutcome>,
286    pub last_sequence: u64,
287}
288
289impl WorkflowRunSnapshot {
290    /// Decode the workflow input into a host-defined serde type.
291    pub fn input_as<T>(&self) -> Result<T>
292    where
293        T: DeserializeOwned,
294    {
295        serde_json::from_value(self.input.clone()).map_err(FlowError::from)
296    }
297
298    /// Decode the terminal workflow output into a host-defined serde type.
299    pub fn output_as<T>(&self) -> Result<Option<T>>
300    where
301        T: DeserializeOwned,
302    {
303        self.output
304            .clone()
305            .map(serde_json::from_value)
306            .transpose()
307            .map_err(FlowError::from)
308    }
309
310    pub fn step_output(&self, step_id: &str) -> Option<&JsonValue> {
311        self.steps
312            .get(step_id)
313            .and_then(|step| step.output.as_ref())
314    }
315
316    /// Decode a persisted step output into a host-defined serde type.
317    pub fn step_output_as<T>(&self, step_id: &str) -> Result<Option<T>>
318    where
319        T: DeserializeOwned,
320    {
321        match self.steps.get(step_id) {
322            Some(step) => step.output_as(),
323            None => Ok(None),
324        }
325    }
326
327    pub fn hook_payload(&self, hook_id: &str) -> Option<&JsonValue> {
328        self.hooks
329            .get(hook_id)
330            .and_then(|hook| hook.payload.as_ref())
331    }
332
333    /// Return a durable progress update by its idempotency identity.
334    pub fn progress(&self, progress_id: &str) -> Option<&WorkflowProgress> {
335        self.progress
336            .iter()
337            .find(|progress| progress.progress_id == progress_id)
338    }
339
340    /// Return the most recently persisted progress update.
341    pub fn latest_progress(&self) -> Option<&WorkflowProgress> {
342        self.progress.last()
343    }
344
345    /// Return a durable child-operation reference by its parent-local id.
346    pub fn child_operation(&self, reference_id: &str) -> Option<&ChildOperationReference> {
347        self.child_operations.get(reference_id)
348    }
349
350    /// Decode persisted hook metadata into a host-defined serde type.
351    pub fn hook_metadata_as<T>(&self, hook_id: &str) -> Result<Option<T>>
352    where
353        T: DeserializeOwned,
354    {
355        match self.hooks.get(hook_id) {
356            Some(hook) => hook.metadata_as().map(Some),
357            None => Ok(None),
358        }
359    }
360
361    /// Decode a received hook payload into a host-defined serde type.
362    pub fn hook_payload_as<T>(&self, hook_id: &str) -> Result<Option<T>>
363    where
364        T: DeserializeOwned,
365    {
366        match self.hooks.get(hook_id) {
367            Some(hook) => hook.payload_as(),
368            None => Ok(None),
369        }
370    }
371
372    pub fn has_open_suspension(&self) -> bool {
373        self.waits
374            .values()
375            .any(|wait| wait.status == WaitStatus::Waiting)
376            || self
377                .hooks
378                .values()
379                .any(|hook| hook.status == HookStatus::Active)
380            || self.steps.values().any(|step| step.retry_after.is_some())
381    }
382
383    pub fn due_retries(&self, now: DateTime<Utc>) -> Vec<(String, DateTime<Utc>)> {
384        self.steps
385            .values()
386            .filter_map(|step| match step.retry_after {
387                Some(retry_after) if step.status == StepStatus::Pending && retry_after <= now => {
388                    Some((step.step_id.clone(), retry_after))
389                }
390                _ => None,
391            })
392            .collect()
393    }
394
395    pub fn has_future_retry(&self, now: DateTime<Utc>) -> bool {
396        self.steps.values().any(|step| {
397            step.status == StepStatus::Pending
398                && step
399                    .retry_after
400                    .map(|retry_after| retry_after > now)
401                    .unwrap_or(false)
402        })
403    }
404}
a3s_flow/model/snapshot.rs

a3s_flow/model/
snapshot.rs
