ironflow_engine/config/
workflow.rs

1//! Configuration for workflow (sub-workflow) steps.
2
3use std::collections::HashMap;
4use std::ops::Not;
5
6use ironflow_core::retry::RetryPolicy;
7use ironflow_store::entities::{MAX_CONCURRENCY_KEY_LEN, validate_priority};
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10
11/// Prefix of the run labels the engine sets itself on a child run.
12const ENGINE_LABEL_PREFIX: &str = "ironflow.io/";
13
14/// Configuration for invoking a registered workflow as a sub-step.
15///
16/// The engine will look up the handler by [`workflow_name`](WorkflowStepConfig::workflow_name)
17/// and execute it as a child run with its own steps and lifecycle.
18///
19/// # Examples
20///
21/// ```
22/// use ironflow_engine::config::WorkflowStepConfig;
23/// use serde_json::json;
24///
25/// let config = WorkflowStepConfig::new("build", json!({"branch": "main"}));
26/// assert_eq!(config.workflow_name, "build");
27/// ```
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct WorkflowStepConfig {
30    /// Name of the registered workflow handler to invoke.
31    pub workflow_name: String,
32    /// Payload to pass to the child workflow run.
33    pub payload: Value,
34    /// Optional step-level retry policy.
35    #[serde(default, skip_serializing_if = "Option::is_none")]
36    pub retry: Option<RetryPolicy>,
37    /// Concurrency key held by the child run while it is not terminal.
38    ///
39    /// Set through [`WorkflowOptions::concurrency_key`] and recorded in the
40    /// step input.
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub concurrency_key: Option<String>,
43    /// Queue priority of the child run.
44    ///
45    /// Set through [`WorkflowOptions::priority`] and recorded in the step
46    /// input. `None` gives the child the priority of its own handler.
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub priority: Option<i16>,
49    /// Tolerate a failed child: the step completes with the child's failure
50    /// in its output instead of failing the parent.
51    #[serde(default, skip_serializing_if = "Not::not")]
52    pub allow_failure: bool,
53    /// Labels set on the child run, over the ones it inherits from the
54    /// parent run.
55    ///
56    /// Set through [`WorkflowOptions::label`] and recorded in the step input.
57    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
58    pub labels: HashMap<String, String>,
59}
60
61impl WorkflowStepConfig {
62    /// Create a new workflow step config.
63    ///
64    /// # Examples
65    ///
66    /// ```
67    /// use ironflow_engine::config::WorkflowStepConfig;
68    /// use serde_json::json;
69    ///
70    /// let config = WorkflowStepConfig::new("deploy", json!({}));
71    /// assert_eq!(config.workflow_name, "deploy");
72    /// ```
73    pub fn new(workflow_name: &str, payload: Value) -> Self {
74        Self {
75            workflow_name: workflow_name.to_string(),
76            payload,
77            retry: None,
78            concurrency_key: None,
79            priority: None,
80            allow_failure: false,
81            labels: HashMap::new(),
82        }
83    }
84
85    /// Set a step-level retry policy.
86    ///
87    /// # Examples
88    ///
89    /// ```
90    /// use ironflow_core::retry::RetryPolicy;
91    /// use ironflow_engine::config::WorkflowStepConfig;
92    /// use serde_json::json;
93    ///
94    /// let config = WorkflowStepConfig::new("build", json!({}))
95    ///     .retry_policy(RetryPolicy::new(3));
96    /// assert!(config.retry.is_some());
97    /// ```
98    pub fn retry_policy(mut self, policy: RetryPolicy) -> Self {
99        self.retry = Some(policy);
100        self
101    }
102
103    /// Tolerate a failed child run: the step completes with a
104    /// [`SubWorkflowOutput`](crate::executor::SubWorkflowOutput) reporting the
105    /// failure and the parent run ends as `Warning`.
106    ///
107    /// # Examples
108    ///
109    /// ```
110    /// use ironflow_engine::config::WorkflowStepConfig;
111    /// use serde_json::json;
112    ///
113    /// let config = WorkflowStepConfig::new("build", json!({})).allow_failure();
114    /// assert!(config.allow_failure);
115    /// ```
116    pub fn allow_failure(mut self) -> Self {
117        self.allow_failure = true;
118        self
119    }
120}
121
122/// Options of a sub-workflow step started with
123/// [`workflow_with`](crate::context::WorkflowContext::workflow_with) or
124/// [`workflows_with`](crate::context::WorkflowContext::workflows_with).
125///
126/// # Examples
127///
128/// ```
129/// use ironflow_engine::config::WorkflowOptions;
130///
131/// let options = WorkflowOptions::new().allow_failure();
132/// assert!(options.allow_failure);
133/// assert!(!WorkflowOptions::default().allow_failure);
134/// ```
135#[derive(Debug, Clone, Default, PartialEq, Eq)]
136pub struct WorkflowOptions {
137    /// Tolerate a failed child run instead of failing the parent.
138    pub allow_failure: bool,
139    concurrency_key: Option<String>,
140    priority: Option<i16>,
141    step_name: Option<String>,
142    labels: HashMap<String, String>,
143}
144
145impl WorkflowOptions {
146    /// Create options with every flag off.
147    ///
148    /// # Examples
149    ///
150    /// ```
151    /// use ironflow_engine::config::WorkflowOptions;
152    ///
153    /// assert_eq!(WorkflowOptions::new(), WorkflowOptions::default());
154    /// ```
155    pub fn new() -> Self {
156        Self::default()
157    }
158
159    /// Tolerate a failed child run: the step completes with the failure in its
160    /// output and the parent run ends as `Warning`. A suspension is never
161    /// tolerated.
162    ///
163    /// # Examples
164    ///
165    /// ```
166    /// use ironflow_engine::config::WorkflowOptions;
167    ///
168    /// assert!(WorkflowOptions::new().allow_failure().allow_failure);
169    /// ```
170    pub fn allow_failure(mut self) -> Self {
171        self.allow_failure = true;
172        self
173    }
174
175    /// Make the child run exclusive on `key`.
176    ///
177    /// While another non-terminal run (Pending, Running, Retrying, Sleeping,
178    /// AwaitingApproval) holds the same key, no child run is created and the
179    /// step completes with a
180    /// [`SubWorkflowOutcome::Conflict`](crate::executor::SubWorkflowOutcome::Conflict).
181    ///
182    /// # Panics
183    ///
184    /// Panics if `key` is empty or only whitespace, or longer than
185    /// [`MAX_CONCURRENCY_KEY_LEN`] bytes.
186    ///
187    /// # Examples
188    ///
189    /// ```
190    /// use ironflow_engine::config::WorkflowOptions;
191    ///
192    /// let options = WorkflowOptions::new().concurrency_key("issue:12");
193    /// assert_eq!(options.concurrency_key_ref(), Some("issue:12"));
194    /// ```
195    ///
196    /// ```should_panic
197    /// use ironflow_engine::config::WorkflowOptions;
198    ///
199    /// WorkflowOptions::new().concurrency_key("  ");
200    /// ```
201    pub fn concurrency_key(mut self, key: impl Into<String>) -> Self {
202        let key = key.into();
203        assert!(!key.trim().is_empty(), "concurrency key must not be empty");
204        assert!(
205            key.len() <= MAX_CONCURRENCY_KEY_LEN,
206            "concurrency key must be at most {MAX_CONCURRENCY_KEY_LEN} bytes"
207        );
208        self.concurrency_key = Some(key);
209        self
210    }
211
212    /// The concurrency key, if one was set.
213    ///
214    /// # Examples
215    ///
216    /// ```
217    /// use ironflow_engine::config::WorkflowOptions;
218    ///
219    /// let options = WorkflowOptions::new().concurrency_key("deploy:prod");
220    /// assert_eq!(options.concurrency_key_ref(), Some("deploy:prod"));
221    /// ```
222    pub fn concurrency_key_ref(&self) -> Option<&str> {
223        self.concurrency_key.as_deref()
224    }
225
226    /// Set the queue priority of the child run, overriding the priority of
227    /// the child handler.
228    ///
229    /// The child executes inside its parent's worker; the priority is
230    /// recorded on the child run and orders it whenever it waits in the queue.
231    ///
232    /// # Panics
233    ///
234    /// Panics if `priority` is outside
235    /// [`MIN_PRIORITY`](ironflow_store::entities::MIN_PRIORITY)`..=`[`MAX_PRIORITY`](ironflow_store::entities::MAX_PRIORITY).
236    ///
237    /// # Examples
238    ///
239    /// ```
240    /// use ironflow_engine::config::WorkflowOptions;
241    ///
242    /// let options = WorkflowOptions::new().priority(20);
243    /// assert_eq!(options.priority_ref(), Some(20));
244    /// ```
245    ///
246    /// ```should_panic
247    /// use ironflow_engine::config::WorkflowOptions;
248    ///
249    /// WorkflowOptions::new().priority(101);
250    /// ```
251    pub fn priority(mut self, priority: i16) -> Self {
252        if let Err(message) = validate_priority(priority) {
253            panic!("{message}");
254        }
255        self.priority = Some(priority);
256        self
257    }
258
259    /// The priority of the child run, if one was set.
260    ///
261    /// # Examples
262    ///
263    /// ```
264    /// use ironflow_engine::config::WorkflowOptions;
265    ///
266    /// assert_eq!(WorkflowOptions::new().priority_ref(), None);
267    /// assert_eq!(WorkflowOptions::new().priority(-5).priority_ref(), Some(-5));
268    /// ```
269    pub fn priority_ref(&self) -> Option<i16> {
270        self.priority
271    }
272
273    /// Name the `Workflow` step of the parent run.
274    ///
275    /// The step is named after the child handler by default. The name is what
276    /// the run timeline shows and what identifies the step on resume: a
277    /// handler that delegates to the same child several times gives each call
278    /// its own name.
279    ///
280    /// # Panics
281    ///
282    /// Panics if `name` is empty or only whitespace.
283    ///
284    /// # Examples
285    ///
286    /// ```
287    /// use ironflow_engine::config::WorkflowOptions;
288    ///
289    /// let options = WorkflowOptions::new().step_name("delegate-1-alice");
290    /// assert_eq!(options.step_name_ref(), Some("delegate-1-alice"));
291    /// ```
292    ///
293    /// ```should_panic
294    /// use ironflow_engine::config::WorkflowOptions;
295    ///
296    /// WorkflowOptions::new().step_name(" ");
297    /// ```
298    pub fn step_name(mut self, name: impl Into<String>) -> Self {
299        let name = name.into();
300        assert!(!name.trim().is_empty(), "step name must not be empty");
301        self.step_name = Some(name);
302        self
303    }
304
305    /// The step name, if one was set.
306    ///
307    /// # Examples
308    ///
309    /// ```
310    /// use ironflow_engine::config::WorkflowOptions;
311    ///
312    /// assert_eq!(WorkflowOptions::new().step_name_ref(), None);
313    /// ```
314    pub fn step_name_ref(&self) -> Option<&str> {
315        self.step_name.as_deref()
316    }
317
318    /// Set a label on the child run.
319    ///
320    /// The child inherits the labels of its parent run; a label set here
321    /// replaces an inherited one with the same key. The last value set for a
322    /// key wins.
323    ///
324    /// # Panics
325    ///
326    /// Panics if `key` starts with `ironflow.io/`: those labels link the child
327    /// to its parent and root runs and are set by the engine.
328    ///
329    /// # Examples
330    ///
331    /// ```
332    /// use ironflow_engine::config::WorkflowOptions;
333    ///
334    /// let options = WorkflowOptions::new().label("paperflow.io/agent", "alice");
335    /// assert_eq!(options.labels_ref()["paperflow.io/agent"], "alice");
336    /// ```
337    ///
338    /// ```should_panic
339    /// use ironflow_engine::config::WorkflowOptions;
340    ///
341    /// WorkflowOptions::new().label("ironflow.io/parent-run-id", "x");
342    /// ```
343    pub fn label(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
344        let key = key.into();
345        assert!(
346            !key.starts_with(ENGINE_LABEL_PREFIX),
347            "label {key:?} is reserved for the engine"
348        );
349        self.labels.insert(key, value.into());
350        self
351    }
352
353    /// Set several labels on the child run, as [`label`](Self::label) does
354    /// for each pair in order.
355    ///
356    /// # Panics
357    ///
358    /// Panics if a key starts with `ironflow.io/`.
359    ///
360    /// # Examples
361    ///
362    /// ```
363    /// use ironflow_engine::config::WorkflowOptions;
364    ///
365    /// let options = WorkflowOptions::new().labels([("tour", "1"), ("agent", "alice")]);
366    /// assert_eq!(options.labels_ref().len(), 2);
367    /// ```
368    pub fn labels<K, V>(self, labels: impl IntoIterator<Item = (K, V)>) -> Self
369    where
370        K: Into<String>,
371        V: Into<String>,
372    {
373        labels
374            .into_iter()
375            .fold(self, |options, (key, value)| options.label(key, value))
376    }
377
378    /// The labels set on the child run.
379    ///
380    /// # Examples
381    ///
382    /// ```
383    /// use ironflow_engine::config::WorkflowOptions;
384    ///
385    /// assert!(WorkflowOptions::new().labels_ref().is_empty());
386    /// ```
387    pub fn labels_ref(&self) -> &HashMap<String, String> {
388        &self.labels
389    }
390
391    /// Consume the options and build the step config for `workflow_name`,
392    /// with the name of the step: the one set, otherwise the handler name.
393    pub(crate) fn into_step_config(
394        self,
395        workflow_name: &str,
396        payload: Value,
397    ) -> (String, WorkflowStepConfig) {
398        let step_name = self.step_name.unwrap_or_else(|| workflow_name.to_string());
399        let config = WorkflowStepConfig {
400            concurrency_key: self.concurrency_key,
401            priority: self.priority,
402            allow_failure: self.allow_failure,
403            labels: self.labels,
404            ..WorkflowStepConfig::new(workflow_name, payload)
405        };
406        (step_name, config)
407    }
408}
409
410#[cfg(test)]
411mod tests {
412    use super::*;
413    use ironflow_store::entities::{MAX_PRIORITY, MIN_PRIORITY};
414    use serde_json::{from_str, json, to_string, to_value};
415
416    #[test]
417    fn new_sets_fields() {
418        let config = WorkflowStepConfig::new("build", json!({"key": "val"}));
419        assert_eq!(config.workflow_name, "build");
420        assert_eq!(config.payload["key"], "val");
421    }
422
423    #[test]
424    fn serde_roundtrip() {
425        let config = WorkflowStepConfig::new("deploy", json!({"env": "prod"}));
426        let json = serde_json::to_string(&config).unwrap();
427        let back: WorkflowStepConfig = serde_json::from_str(&json).unwrap();
428        assert_eq!(back.workflow_name, "deploy");
429        assert_eq!(back.payload["env"], "prod");
430    }
431
432    #[test]
433    fn a_config_predating_retry_still_deserializes() {
434        let config: WorkflowStepConfig =
435            serde_json::from_str(r#"{"workflow_name":"build","payload":{"key":"val"}}"#)
436                .expect("deserialize");
437        assert!(config.retry.is_none());
438    }
439
440    #[test]
441    fn a_config_without_concurrency_key_omits_it() {
442        let config = WorkflowStepConfig::new("build", json!({}));
443        let value = to_value(&config).expect("serialize");
444        assert!(value.get("concurrency_key").is_none());
445    }
446
447    #[test]
448    fn concurrency_key_roundtrip() {
449        let mut config = WorkflowStepConfig::new("build", json!({}));
450        config.concurrency_key = Some("issue:12".to_string());
451        let json = to_string(&config).expect("serialize");
452        let back: WorkflowStepConfig = from_str(&json).expect("deserialize");
453        assert_eq!(back.concurrency_key.as_deref(), Some("issue:12"));
454    }
455
456    #[test]
457    fn options_carry_the_concurrency_key() {
458        let options = WorkflowOptions::new().concurrency_key("issue:12");
459        assert_eq!(options.concurrency_key_ref(), Some("issue:12"));
460        let (_, config) = options.into_step_config("build", json!({}));
461        assert_eq!(config.concurrency_key.as_deref(), Some("issue:12"));
462        assert!(WorkflowOptions::new().concurrency_key_ref().is_none());
463    }
464
465    #[test]
466    fn options_accept_a_key_at_the_length_limit() {
467        let key = "a".repeat(MAX_CONCURRENCY_KEY_LEN);
468        let options = WorkflowOptions::new().concurrency_key(key.clone());
469        assert_eq!(options.concurrency_key_ref(), Some(key.as_str()));
470    }
471
472    #[test]
473    #[should_panic(expected = "concurrency key must not be empty")]
474    fn options_reject_an_empty_key() {
475        let _ = WorkflowOptions::new().concurrency_key("");
476    }
477
478    #[test]
479    #[should_panic(expected = "concurrency key must not be empty")]
480    fn options_reject_a_blank_key() {
481        let _ = WorkflowOptions::new().concurrency_key(" \t ");
482    }
483
484    #[test]
485    #[should_panic(expected = "concurrency key must be at most")]
486    fn options_reject_a_key_over_the_limit() {
487        let _ = WorkflowOptions::new().concurrency_key("a".repeat(MAX_CONCURRENCY_KEY_LEN + 1));
488    }
489
490    #[test]
491    fn options_priority_is_carried_and_split() {
492        let options = WorkflowOptions::new()
493            .concurrency_key("issue:12")
494            .priority(MAX_PRIORITY);
495        assert_eq!(options.priority_ref(), Some(MAX_PRIORITY));
496        let (_, config) = options.into_step_config("build", json!({}));
497        assert_eq!(config.concurrency_key.as_deref(), Some("issue:12"));
498        assert_eq!(config.priority, Some(MAX_PRIORITY));
499        assert_eq!(
500            WorkflowOptions::new().priority(MIN_PRIORITY).priority_ref(),
501            Some(MIN_PRIORITY)
502        );
503        assert!(WorkflowOptions::new().priority_ref().is_none());
504    }
505
506    #[test]
507    #[should_panic(expected = "priority must be between -100 and 100")]
508    fn options_priority_rejects_above_the_maximum() {
509        let _ = WorkflowOptions::new().priority(MAX_PRIORITY + 1);
510    }
511
512    #[test]
513    #[should_panic(expected = "priority must be between -100 and 100")]
514    fn options_priority_rejects_below_the_minimum() {
515        let _ = WorkflowOptions::new().priority(MIN_PRIORITY - 1);
516    }
517
518    #[test]
519    fn step_config_priority_is_omitted_when_unset_and_round_trips() {
520        let config = WorkflowStepConfig::new("build", json!({}));
521        assert!(
522            to_value(&config)
523                .expect("serialize")
524                .get("priority")
525                .is_none()
526        );
527
528        let mut config = WorkflowStepConfig::new("build", json!({}));
529        config.priority = Some(-30);
530        let back: WorkflowStepConfig =
531            from_str(&to_string(&config).expect("serialize")).expect("deserialize");
532        assert_eq!(back.priority, Some(-30));
533    }
534
535    #[test]
536    fn allow_failure_defaults_to_false() {
537        let config = WorkflowStepConfig::new("build", json!({}));
538        assert!(!config.allow_failure);
539    }
540
541    #[test]
542    fn allow_failure_is_omitted_from_json_when_false() {
543        let config = WorkflowStepConfig::new("build", json!({}));
544        let value = serde_json::to_value(&config).expect("serialize");
545        assert!(value.get("allow_failure").is_none());
546    }
547
548    #[test]
549    fn allow_failure_roundtrip() {
550        let config = WorkflowStepConfig::new("build", json!({})).allow_failure();
551        let json = serde_json::to_string(&config).expect("serialize");
552        let back: WorkflowStepConfig = serde_json::from_str(&json).expect("deserialize");
553        assert!(back.allow_failure);
554    }
555
556    #[test]
557    fn options_builder_sets_allow_failure() {
558        assert!(!WorkflowOptions::new().allow_failure);
559        assert!(WorkflowOptions::new().allow_failure().allow_failure);
560    }
561
562    #[test]
563    fn options_carry_the_step_name() {
564        let options = WorkflowOptions::new().step_name("delegate-1-alice");
565        assert_eq!(options.step_name_ref(), Some("delegate-1-alice"));
566        assert!(WorkflowOptions::new().step_name_ref().is_none());
567    }
568
569    #[test]
570    #[should_panic(expected = "step name must not be empty")]
571    fn options_reject_an_empty_step_name() {
572        let _ = WorkflowOptions::new().step_name("");
573    }
574
575    #[test]
576    #[should_panic(expected = "step name must not be empty")]
577    fn options_reject_a_blank_step_name() {
578        let _ = WorkflowOptions::new().step_name(" \n ");
579    }
580
581    #[test]
582    fn options_merge_labels_and_the_last_value_wins() {
583        let options = WorkflowOptions::new()
584            .label("paperflow.io/tour", "1")
585            .labels([("paperflow.io/agent", "alice"), ("paperflow.io/tour", "2")]);
586        let labels = options.labels_ref();
587        assert_eq!(labels.len(), 2);
588        assert_eq!(labels["paperflow.io/tour"], "2");
589        assert_eq!(labels["paperflow.io/agent"], "alice");
590        assert!(WorkflowOptions::new().labels_ref().is_empty());
591    }
592
593    #[test]
594    #[should_panic(expected = "reserved for the engine")]
595    fn options_reject_an_engine_label() {
596        let _ = WorkflowOptions::new().label("ironflow.io/parent-run-id", "x");
597    }
598
599    #[test]
600    #[should_panic(expected = "reserved for the engine")]
601    fn options_reject_an_engine_label_among_several() {
602        let _ = WorkflowOptions::new().labels([("team", "a"), ("ironflow.io/root-run-id", "x")]);
603    }
604
605    #[test]
606    fn step_config_defaults_the_step_name_to_the_handler() {
607        let (name, config) = WorkflowOptions::new().into_step_config("review", json!({}));
608        assert_eq!(name, "review");
609        assert_eq!(config.workflow_name, "review");
610        assert!(config.labels.is_empty());
611    }
612
613    #[test]
614    fn step_config_takes_every_option() {
615        let (name, config) = WorkflowOptions::new()
616            .step_name("delegate-1-alice")
617            .label("agent", "alice")
618            .concurrency_key("issue:12")
619            .priority(7)
620            .allow_failure()
621            .into_step_config("review", json!({"n": 1}));
622        assert_eq!(name, "delegate-1-alice");
623        assert_eq!(config.workflow_name, "review");
624        assert_eq!(config.payload, json!({"n": 1}));
625        assert_eq!(config.labels["agent"], "alice");
626        assert_eq!(config.concurrency_key.as_deref(), Some("issue:12"));
627        assert_eq!(config.priority, Some(7));
628        assert!(config.allow_failure);
629    }
630
631    #[test]
632    fn step_config_labels_are_omitted_when_empty_and_round_trip() {
633        let config = WorkflowStepConfig::new("build", json!({}));
634        assert!(
635            to_value(&config)
636                .expect("serialize")
637                .get("labels")
638                .is_none()
639        );
640
641        let (_, config) = WorkflowOptions::new()
642            .label("agent", "alice")
643            .into_step_config("build", json!({}));
644        let back: WorkflowStepConfig =
645            from_str(&to_string(&config).expect("serialize")).expect("deserialize");
646        assert_eq!(back.labels["agent"], "alice");
647    }
648
649    #[test]
650    fn a_config_predating_labels_still_deserializes() {
651        let config: WorkflowStepConfig =
652            from_str(r#"{"workflow_name":"build","payload":{}}"#).expect("deserialize");
653        assert!(config.labels.is_empty());
654    }
655
656    #[test]
657    fn retry_policy_roundtrip() {
658        let config = WorkflowStepConfig::new("deploy", json!({})).retry_policy(RetryPolicy::new(3));
659        let json = serde_json::to_string(&config).expect("serialize");
660        let back: WorkflowStepConfig = serde_json::from_str(&json).expect("deserialize");
661        assert_eq!(back.retry.as_ref().unwrap().max_retries(), 3);
662    }
663}
ironflow_engine/config/workflow.rs

ironflow_engine/config/
workflow.rs
