backbone_core/workflow/
spec.rs

1//! What the schema generator writes for a workflow: its steps as data, and the one trait that
2//! performs what only typed code can.
3//!
4//! A [`WorkflowSpec`] is the compiled form of a `schema/workflows/<name>.workflow.yaml` with
5//! `config.executor: engine`. The generated module carries it as JSON (`WorkflowSpec::from_json`)
6//! beside a [`WorkflowDefinition`] implementation, which does the three things a spec cannot say:
7//! call an entity's generated service (create, update, delete, query, fetch), move a record
8//! through its state machine, and call the hand-written verbs and functions of the workflow's
9//! `Actions` trait.
10
11use async_trait::async_trait;
12use serde::{Deserialize, Serialize};
13use serde_json::Value;
14
15use super::expr::{Dur, Expr, Tpl};
16use crate::service::ServiceResult;
17use crate::write_hooks::WriteCtx;
18
19/// How long to wait between attempts of a step that failed for a transient reason.
20#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
21#[serde(rename_all = "snake_case")]
22pub enum Backoff {
23    /// Doubling from the engine's base up to its cap.
24    Exponential,
25    /// The same wait every time, in seconds.
26    Fixed(i64),
27}
28
29/// The retry policy of a workflow or of one step.
30#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
31pub struct RetrySpec {
32    /// Attempts in all, the first included.
33    pub max_attempts: u32,
34    /// The wait between them.
35    pub backoff: Backoff,
36}
37
38/// What a step does when it succeeds (or when a wait is answered or times out).
39#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
40pub struct Outcome {
41    /// Context keys to set, evaluated in order.
42    #[serde(default)]
43    pub set: Vec<(String, Tpl)>,
44    /// The step to go to. `None` ends the run of this step without a successor, which is an
45    /// error for anything but a terminal.
46    #[serde(default)]
47    pub next: Option<String>,
48}
49
50/// What a step does when it fails.
51#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
52pub struct FailureSpec {
53    /// The step to go to.
54    #[serde(default)]
55    pub next: Option<String>,
56    /// Context keys to set when the failure is routed to `next`.
57    #[serde(default)]
58    pub set: Vec<(String, Tpl)>,
59    /// Attempts for a transient failure; the workflow's, else the engine's, when absent.
60    #[serde(default)]
61    pub retry: Option<RetrySpec>,
62    /// What to do once the attempts are used up, when it differs from `next`.
63    #[serde(default)]
64    pub on_exhausted: Option<Outcome>,
65}
66
67/// One branch of a condition step; the first whose `when` holds is taken, a branch with no `when`
68/// is the `else`.
69#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
70pub struct Branch {
71    /// The condition.
72    #[serde(default)]
73    pub when: Option<Expr>,
74    /// Context keys to set when it is taken.
75    #[serde(default)]
76    pub set: Vec<(String, Tpl)>,
77    /// Where it goes.
78    pub next: String,
79}
80
81/// One equality of a wait's condition: the event's field must equal the context expression.
82#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
83pub struct WaitKey {
84    /// A top-level field of the event's payload.
85    pub event_field: String,
86    /// What it must equal, read from the context when the wait is entered.
87    pub context: Expr,
88}
89
90/// How a terminal step ends the instance.
91#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
92#[serde(rename_all = "snake_case")]
93pub enum Terminal {
94    /// `completed`.
95    Completed,
96    /// `failed`.
97    Failed,
98    /// `expired`.
99    Expired,
100    /// `cancelled`.
101    Cancelled,
102}
103
104/// An event a terminal step stages.
105#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
106pub struct EmitSpec {
107    /// The event type.
108    pub event: String,
109    /// The payload.
110    pub data: Tpl,
111}
112
113/// What a step is.
114#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
115#[serde(tag = "kind", rename_all = "snake_case")]
116pub enum StepKind {
117    /// An action: an engine action (`audit_log`, `emit_event`, `send_email`, `notify`,
118    /// `send_notification`, `send_sms`), or one the [`WorkflowDefinition`] performs.
119    Action {
120        /// The verb.
121        action: String,
122        /// The entity of an entity action.
123        #[serde(default)]
124        entity: Option<String>,
125        /// The record an update, delete or fetch is about.
126        #[serde(default)]
127        id: Option<Tpl>,
128        /// The parameters, rendered when the step runs.
129        #[serde(default)]
130        params: Vec<(String, Tpl)>,
131        /// What a success does.
132        on_success: Outcome,
133        /// What a hand call that outlasts the step's `call_timeout` does, instead of failing.
134        #[serde(default)]
135        on_timeout: Option<Outcome>,
136    },
137    /// A state-machine verb on a record.
138    Transition {
139        /// The entity.
140        entity: String,
141        /// The record.
142        id: Tpl,
143        /// The verb.
144        verb: String,
145        /// Extra fields written with the move.
146        #[serde(default)]
147        params: Vec<(String, Tpl)>,
148        /// What a success does.
149        on_success: Outcome,
150    },
151    /// A branch on conditions, in order.
152    Condition {
153        /// The branches.
154        branches: Vec<Branch>,
155    },
156    /// A durable wait.
157    Wait {
158        /// The event waited for; `None` is a pure timer.
159        #[serde(default)]
160        event: Option<String>,
161        /// The equalities that match an event to this instance.
162        #[serde(default)]
163        key: Vec<WaitKey>,
164        /// How long to wait.
165        #[serde(default)]
166        timeout: Option<Dur>,
167        /// What an event does.
168        #[serde(default)]
169        on_event: Option<Outcome>,
170        /// What the timeout does.
171        #[serde(default)]
172        on_timeout: Option<Outcome>,
173    },
174    /// An end.
175    Terminal {
176        /// How it ends.
177        status: Terminal,
178        /// Why, for a person reading the log.
179        #[serde(default)]
180        reason: Option<Tpl>,
181        /// An event to stage.
182        #[serde(default)]
183        emit: Option<EmitSpec>,
184        /// Run compensation after ending as failed.
185        #[serde(default)]
186        compensate: bool,
187        /// A value stored as `context.result` when the instance ends.
188        #[serde(default)]
189        result: Option<Tpl>,
190    },
191}
192
193/// One step.
194#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
195pub struct StepSpec {
196    /// Its name, unique in the workflow.
197    pub name: String,
198    /// A guard: when it does not hold the step is skipped and the run goes to `on_success.next`.
199    #[serde(default)]
200    pub guard: Option<Expr>,
201    /// What it is.
202    #[serde(flatten)]
203    pub kind: StepKind,
204    /// What a failure does.
205    #[serde(default)]
206    pub on_failure: Option<FailureSpec>,
207    /// Seconds a hand call may take (not the wait step's own `timeout`, which is part of its kind).
208    #[serde(default)]
209    pub call_timeout: Option<i64>,
210}
211
212/// One compensation step.
213#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
214pub struct CompStep {
215    /// Its name; the step log records it as `compensate:<name>`.
216    pub name: String,
217    /// It runs only when this holds (`steps.<name>.completed == true`).
218    #[serde(default)]
219    pub when: Option<Expr>,
220    /// The verb.
221    pub action: String,
222    /// The entity of an entity action.
223    #[serde(default)]
224    pub entity: Option<String>,
225    /// The record.
226    #[serde(default)]
227    pub id: Option<Tpl>,
228    /// The parameters.
229    #[serde(default)]
230    pub params: Vec<(String, Tpl)>,
231    /// The state-machine verb, when the action is `transition`.
232    #[serde(default)]
233    pub verb: Option<String>,
234}
235
236/// The compiled workflow.
237#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
238pub struct WorkflowSpec {
239    /// Its name (`PasswordReset`).
240    pub name: String,
241    /// Its version, from the file.
242    pub version: i32,
243    /// The event that starts an instance.
244    pub trigger_event: String,
245    /// Context keys filled from the start event: `company_id: event.company_id`.
246    #[serde(default)]
247    pub extract: Vec<(String, Expr)>,
248    /// The context an instance starts with, before `extract`.
249    #[serde(default)]
250    pub context: Vec<(String, Value)>,
251    /// `false` runs the whole flow in one transaction and allows no wait.
252    pub persistence: bool,
253    /// How long an instance may live; passing it while waiting ends it `expired`.
254    #[serde(default)]
255    pub timeout: Option<Dur>,
256    /// The workflow-wide retry policy.
257    #[serde(default)]
258    pub retry: Option<RetrySpec>,
259    /// The first step is the first of these.
260    pub steps: Vec<StepSpec>,
261    /// Compensation, in declaration order; it runs in reverse.
262    #[serde(default)]
263    pub compensation: Vec<CompStep>,
264}
265
266impl WorkflowSpec {
267    /// Read the JSON a generated module carries.
268    ///
269    /// # Panics
270    /// When the text is not a spec: the generator wrote it, so that is a bug of the generator or a
271    /// version mismatch between it and this crate, and it is loudest at startup.
272    pub fn from_json(text: &str) -> Self {
273        match serde_json::from_str(text) {
274            Ok(spec) => spec,
275            Err(e) => panic!("a generated workflow spec does not parse (generator and backbone-core disagree): {e}"),
276        }
277    }
278
279    /// The step called `name`.
280    pub fn step(&self, name: &str) -> Option<&StepSpec> {
281        self.steps.iter().find(|s| s.name == name)
282    }
283
284    /// The events this workflow reacts to: its trigger and every event a wait names.
285    pub fn events(&self) -> Vec<&str> {
286        let mut out = vec![self.trigger_event.as_str()];
287        for step in &self.steps {
288            if let StepKind::Wait { event: Some(e), .. } = &step.kind {
289                if !out.contains(&e.as_str()) {
290                    out.push(e.as_str());
291                }
292            }
293        }
294        out
295    }
296}
297
298/// What a [`WorkflowDefinition`] is asked to perform.
299#[derive(Debug)]
300pub struct Call<'a> {
301    /// The step (or `compensate:<name>`) asking. A generated definition dispatches on it.
302    pub step: &'a str,
303    /// The verb.
304    pub action: &'a str,
305    /// The entity of an entity action or a transition.
306    pub entity: Option<&'a str>,
307    /// The record, rendered from the step's `id`, or from `params.id`.
308    pub id: Option<String>,
309    /// The state-machine verb of a transition step.
310    pub verb: Option<&'a str>,
311    /// The parameters, rendered.
312    pub params: &'a Value,
313    /// The instance's context (the `wf` of a hand verb).
314    pub context: &'a Value,
315}
316
317/// A workflow the engine can run: its spec, and what only typed code performs.
318#[async_trait]
319pub trait WorkflowDefinition: Send + Sync {
320    /// The compiled workflow.
321    fn spec(&self) -> &WorkflowSpec;
322
323    /// Perform an action the engine does not: an entity action through the entity's generated
324    /// service, a transition, or a hand-written verb. The answer is the step's `result`. An error
325    /// of a kind a caller can fix (`NotFound`, `Validation`, `Violations`, `AlreadyExists`) is a
326    /// business failure and is not retried; any other is transient and is.
327    async fn perform(&self, ctx: &WriteCtx, call: &Call<'_>) -> ServiceResult<Value>;
328
329    /// Call a function the workflow declares under `functions:`.
330    async fn function(&self, ctx: &WriteCtx, name: &str, args: &[Value]) -> ServiceResult<Value> {
331        let _ = (ctx, args);
332        Err(crate::service::ServiceError::Internal(format!(
333            "function `{name}` is not implemented"
334        )))
335    }
336}
337
338#[cfg(test)]
339mod tests {
340    use super::*;
341    use serde_json::json;
342
343    /// A wait's own `timeout` and a step's `call_timeout` live side by side in one JSON object: the
344    /// kind is flattened into the step, so the two must not share a name.
345    #[test]
346    fn a_wait_keeps_its_timeout_apart_from_the_steps_call_timeout() {
347        let spec = WorkflowSpec::from_json(
348            &json!({
349                "name": "W", "version": 1, "trigger_event": "E", "persistence": true,
350                "steps": [
351                    { "name": "w", "kind": "wait", "event": "X", "key": [], "timeout": { "secs": 900 },
352                      "on_event": { "next": "d" }, "on_timeout": { "next": "d" }, "call_timeout": 30 },
353                    { "name": "d", "kind": "terminal", "status": "completed" }
354                ]
355            })
356            .to_string(),
357        );
358        let wait = spec.step("w").unwrap();
359        assert_eq!(wait.call_timeout, Some(30));
360        assert!(matches!(
361            &wait.kind,
362            StepKind::Wait {
363                timeout: Some(Dur::Secs(900)),
364                ..
365            }
366        ));
367        assert_eq!(spec.events(), vec!["E", "X"]);
368    }
369}
backbone_core/workflow/spec.rs

backbone_core/workflow/
spec.rs
