ironflow_api/entities/
create_run.rs

1//! Request type for triggering a workflow.
2
3use std::collections::HashMap;
4
5use chrono::{DateTime, Utc};
6use ironflow_store::models::{
7    ConcurrencyLimit, MAX_CONCURRENCY_KEY_LEN, MAX_IDEMPOTENCY_KEY_LEN,
8    validate_concurrency_limits, validate_worker_tags,
9};
10use rust_decimal::Decimal;
11use serde::Deserialize;
12use serde_json::Value;
13
14/// Request to trigger a workflow.
15///
16/// # Examples
17///
18/// ```
19/// use ironflow_api::entities::CreateRunRequest;
20/// use serde_json::json;
21///
22/// let req = CreateRunRequest {
23///     workflow: "deploy".to_string(),
24///     payload: Some(json!({"env": "prod"})),
25///     labels: None,
26///     scheduled_at: None,
27///     max_retries: Some(2),
28///     max_cost_usd: None,
29///     concurrency_key: Some("issue:12".to_string()),
30///     concurrency_limits: Vec::new(),
31///     worker_tags: vec!["gpu".to_string()],
32/// };
33/// assert_eq!(req.workflow, "deploy");
34/// ```
35#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
36#[derive(Debug, Deserialize)]
37pub struct CreateRunRequest {
38    /// The workflow name to trigger.
39    pub workflow: String,
40    /// Optional input payload for the workflow.
41    #[cfg_attr(feature = "openapi", schema(value_type = Option<std::collections::HashMap<String, serde_json::Value>>))]
42    pub payload: Option<Value>,
43    /// Optional key-value labels for categorization and filtering.
44    #[serde(default)]
45    pub labels: Option<HashMap<String, String>>,
46    /// Optional deferred execution time. `None` means run immediately.
47    #[serde(default)]
48    pub scheduled_at: Option<DateTime<Utc>>,
49    /// How many times the run may be replayed automatically after a transient
50    /// failure. Defaults to `0`, meaning no automatic retry.
51    ///
52    /// Each retry waits an exponential backoff (30 s, 2 min, 8 min, capped at
53    /// 15 min) before the run is replayed from the start. Failures that cannot
54    /// succeed on replay -- an unknown workflow, an invalid payload, an
55    /// exhausted agent budget, a rejected approval, a manual cancellation --
56    /// consume no attempt.
57    #[serde(default)]
58    pub max_retries: Option<u32>,
59    /// Optional cumulative cost cap for this run, in USD.
60    ///
61    /// Overrides the workflow default and the server default. `None` falls back
62    /// to those. Must be zero or positive.
63    #[cfg_attr(feature = "openapi", schema(value_type = Option<f64>))]
64    #[serde(default)]
65    pub max_cost_usd: Option<Decimal>,
66    /// Optional exclusivity key, at most 255 bytes.
67    ///
68    /// While a non-terminal run (pending, running, sleeping, retrying,
69    /// awaiting approval) holds the same key, the request is refused with
70    /// `409 CONCURRENCY_CONFLICT` naming that run. The key is released when
71    /// the run completes, fails, ends with a warning or is cancelled.
72    #[serde(default)]
73    pub concurrency_key: Option<String>,
74    /// Concurrency groups the run belongs to, each with its own limit.
75    ///
76    /// The run is always created, but a worker only starts it once, for every
77    /// group, fewer root runs of that group than its `limit` are running.
78    /// Until then it stays pending, and later runs of other groups go ahead.
79    /// Each group is non-empty, at most 255 bytes and listed once; each limit
80    /// is at least 1. Empty means no limit.
81    #[serde(default)]
82    pub concurrency_limits: Vec<ConcurrencyLimit>,
83    /// Tags a worker must carry to take this run, added to the ones the
84    /// workflow requires.
85    ///
86    /// The run stays pending until a worker carrying all of them asks for
87    /// work. Each tag is 1 to 64 bytes of ASCII letters, digits and
88    /// `- _ . : / =`; at most 32 tags. Empty means only the workflow's own
89    /// requirements apply.
90    #[serde(default)]
91    pub worker_tags: Vec<String>,
92}
93
94impl CreateRunRequest {
95    /// Validate the request body.
96    ///
97    /// # Errors
98    ///
99    /// Returns a human-readable message when `max_cost_usd` is negative, or
100    /// when `concurrency_key` is blank or longer than
101    /// [`MAX_CONCURRENCY_KEY_LEN`] bytes, or when `concurrency_limits` holds an
102    /// empty or too long group, a zero limit or a group listed twice, or when a
103    /// worker tag is invalid.
104    ///
105    /// # Examples
106    ///
107    /// ```
108    /// use ironflow_api::entities::CreateRunRequest;
109    /// use rust_decimal::Decimal;
110    ///
111    /// let req = CreateRunRequest {
112    ///     workflow: "deploy".to_string(),
113    ///     payload: None,
114    ///     labels: None,
115    ///     scheduled_at: None,
116    ///     max_retries: None,
117    ///     max_cost_usd: Some(Decimal::new(-1, 0)),
118    ///     concurrency_key: None,
119    ///     concurrency_limits: Vec::new(),
120    ///     worker_tags: Vec::new(),
121    /// };
122    /// assert!(req.validate().is_err());
123    /// ```
124    pub fn validate(&self) -> Result<(), String> {
125        if let Some(cap) = self.max_cost_usd
126            && cap < Decimal::ZERO
127        {
128            return Err("max_cost_usd must be zero or positive".to_string());
129        }
130        match self.concurrency_key.as_deref() {
131            Some(key) if key.trim().is_empty() => {
132                return Err("concurrency_key must not be empty".to_string());
133            }
134            Some(key) if key.len() > MAX_CONCURRENCY_KEY_LEN => {
135                return Err(format!(
136                    "concurrency_key must be at most {MAX_CONCURRENCY_KEY_LEN} bytes"
137                ));
138            }
139            _ => {}
140        }
141        validate_concurrency_limits(&self.concurrency_limits)
142            .map_err(|e| format!("concurrency_limits: {e}"))?;
143        validate_worker_tags(&self.worker_tags).map_err(|e| format!("worker_tags: {e}"))
144    }
145}
146
147/// Why an `Idempotency-Key` header value was rejected.
148///
149/// # Examples
150///
151/// ```
152/// use ironflow_api::entities::{IdempotencyKeyError, validate_idempotency_key};
153///
154/// assert_eq!(
155///     validate_idempotency_key(""),
156///     Err(IdempotencyKeyError::Empty),
157/// );
158/// ```
159#[derive(Debug, Clone, Copy, PartialEq, Eq)]
160pub enum IdempotencyKeyError {
161    /// The header was present but carried no value.
162    Empty,
163    /// The value exceeds [`MAX_IDEMPOTENCY_KEY_LEN`] bytes.
164    TooLong,
165    /// The value contains a byte outside printable ASCII.
166    NotPrintableAscii,
167}
168
169impl IdempotencyKeyError {
170    /// Client-facing explanation of the rejection.
171    ///
172    /// # Examples
173    ///
174    /// ```
175    /// use ironflow_api::entities::IdempotencyKeyError;
176    ///
177    /// assert!(IdempotencyKeyError::Empty.message().contains("empty"));
178    /// ```
179    pub fn message(&self) -> String {
180        match self {
181            IdempotencyKeyError::Empty => "Idempotency-Key must not be empty".to_string(),
182            IdempotencyKeyError::TooLong => {
183                format!("Idempotency-Key must be at most {MAX_IDEMPOTENCY_KEY_LEN} bytes")
184            }
185            IdempotencyKeyError::NotPrintableAscii => {
186                "Idempotency-Key must contain only printable ASCII characters".to_string()
187            }
188        }
189    }
190}
191
192/// Validate an `Idempotency-Key` header value.
193///
194/// A key must be non-empty, at most [`MAX_IDEMPOTENCY_KEY_LEN`] bytes, and made
195/// only of printable ASCII. Empty keys are rejected because they would otherwise
196/// become a single key shared by every client.
197///
198/// # Errors
199///
200/// Returns [`IdempotencyKeyError`] describing which rule the value broke.
201///
202/// # Examples
203///
204/// ```
205/// use ironflow_api::entities::{IdempotencyKeyError, validate_idempotency_key};
206///
207/// assert!(validate_idempotency_key("github:abc-123").is_ok());
208/// assert_eq!(
209///     validate_idempotency_key("clé"),
210///     Err(IdempotencyKeyError::NotPrintableAscii),
211/// );
212/// ```
213pub fn validate_idempotency_key(key: &str) -> Result<(), IdempotencyKeyError> {
214    if key.is_empty() {
215        return Err(IdempotencyKeyError::Empty);
216    }
217    if key.len() > MAX_IDEMPOTENCY_KEY_LEN {
218        return Err(IdempotencyKeyError::TooLong);
219    }
220    if !key.bytes().all(|b| b.is_ascii_graphic()) {
221        return Err(IdempotencyKeyError::NotPrintableAscii);
222    }
223    Ok(())
224}
225
226#[cfg(test)]
227mod tests {
228    use serde_json::from_str;
229
230    use super::*;
231
232    fn request(max_cost_usd: Option<Decimal>) -> CreateRunRequest {
233        CreateRunRequest {
234            workflow: "deploy".to_string(),
235            payload: None,
236            labels: None,
237            scheduled_at: None,
238            max_retries: None,
239            max_cost_usd,
240            concurrency_key: None,
241            concurrency_limits: Vec::new(),
242            worker_tags: Vec::new(),
243        }
244    }
245
246    fn keyed(concurrency_key: &str) -> CreateRunRequest {
247        CreateRunRequest {
248            concurrency_key: Some(concurrency_key.to_string()),
249            ..request(None)
250        }
251    }
252
253    #[test]
254    fn validate_accepts_absent_zero_and_positive_caps() {
255        assert!(request(None).validate().is_ok());
256        assert!(request(Some(Decimal::ZERO)).validate().is_ok());
257        assert!(request(Some(Decimal::new(150, 2))).validate().is_ok());
258    }
259
260    #[test]
261    fn validate_rejects_negative_cap() {
262        let err = request(Some(Decimal::new(-1, 2)))
263            .validate()
264            .expect_err("negative cap must be rejected");
265        assert!(err.contains("max_cost_usd"));
266    }
267
268    #[test]
269    fn max_cost_usd_defaults_to_none_when_absent() {
270        let req: CreateRunRequest =
271            serde_json::from_str(r#"{"workflow":"deploy"}"#).expect("deserialize");
272        assert!(req.max_cost_usd.is_none());
273    }
274
275    #[test]
276    fn max_cost_usd_parses_from_json_number() {
277        let req: CreateRunRequest =
278            serde_json::from_str(r#"{"workflow":"deploy","max_cost_usd":2.5}"#)
279                .expect("deserialize");
280        assert_eq!(req.max_cost_usd, Some(Decimal::new(25, 1)));
281    }
282
283    #[test]
284    fn concurrency_key_defaults_to_none_when_absent() {
285        let req: CreateRunRequest = from_str(r#"{"workflow":"deploy"}"#).expect("deserialize");
286        assert!(req.concurrency_key.is_none());
287        assert!(req.validate().is_ok());
288    }
289
290    #[test]
291    fn concurrency_key_parses_from_json() {
292        let req: CreateRunRequest =
293            from_str(r#"{"workflow":"deploy","concurrency_key":"issue:12"}"#).expect("deserialize");
294        assert_eq!(req.concurrency_key.as_deref(), Some("issue:12"));
295    }
296
297    #[test]
298    fn validate_accepts_a_concurrency_key_up_to_the_limit() {
299        assert!(keyed("issue:12").validate().is_ok());
300        assert!(keyed("cl\u{e9}:\u{e9}lodie").validate().is_ok());
301        let longest = "k".repeat(MAX_CONCURRENCY_KEY_LEN);
302        assert!(keyed(&longest).validate().is_ok());
303    }
304
305    #[test]
306    fn validate_rejects_an_empty_or_blank_concurrency_key() {
307        for key in ["", "   ", "\t\n"] {
308            let err = keyed(key)
309                .validate()
310                .expect_err("blank key must be rejected");
311            assert_eq!(err, "concurrency_key must not be empty");
312        }
313    }
314
315    #[test]
316    fn validate_rejects_a_concurrency_key_over_the_limit() {
317        let err = keyed(&"k".repeat(MAX_CONCURRENCY_KEY_LEN + 1))
318            .validate()
319            .expect_err("over-long key must be rejected");
320        assert_eq!(err, "concurrency_key must be at most 255 bytes");
321    }
322
323    #[test]
324    fn validate_counts_the_concurrency_key_limit_in_bytes() {
325        // 128 two-byte characters: 128 chars but 256 bytes.
326        let err = keyed(&"\u{e9}".repeat(128))
327            .validate()
328            .expect_err("the limit is in bytes, not characters");
329        assert!(err.contains("255 bytes"));
330    }
331
332    #[test]
333    fn max_retries_defaults_to_none_when_absent() {
334        let req: CreateRunRequest =
335            serde_json::from_str(r#"{"workflow":"deploy"}"#).expect("deserialize");
336        assert!(req.max_retries.is_none());
337    }
338
339    #[test]
340    fn max_retries_parses_from_json_number() {
341        let req: CreateRunRequest =
342            serde_json::from_str(r#"{"workflow":"deploy","max_retries":3}"#).expect("deserialize");
343        assert_eq!(req.max_retries, Some(3));
344    }
345
346    #[test]
347    fn accepts_a_provider_delivery_id() {
348        assert!(validate_idempotency_key("github:8f4e2a10-1234-4bcd-9876-abcdef012345").is_ok());
349    }
350
351    #[test]
352    fn rejects_an_empty_key() {
353        assert_eq!(
354            validate_idempotency_key(""),
355            Err(IdempotencyKeyError::Empty)
356        );
357    }
358
359    #[test]
360    fn accepts_a_key_at_the_length_limit() {
361        let key = "a".repeat(MAX_IDEMPOTENCY_KEY_LEN);
362        assert!(validate_idempotency_key(&key).is_ok());
363    }
364
365    #[test]
366    fn rejects_a_key_one_byte_over_the_limit() {
367        let key = "a".repeat(MAX_IDEMPOTENCY_KEY_LEN + 1);
368        assert_eq!(
369            validate_idempotency_key(&key),
370            Err(IdempotencyKeyError::TooLong)
371        );
372    }
373
374    #[test]
375    fn rejects_non_ascii() {
376        assert_eq!(
377            validate_idempotency_key("clé-🚀"),
378            Err(IdempotencyKeyError::NotPrintableAscii)
379        );
380    }
381
382    #[test]
383    fn rejects_control_characters() {
384        assert_eq!(
385            validate_idempotency_key("abc\ndef"),
386            Err(IdempotencyKeyError::NotPrintableAscii)
387        );
388    }
389
390    #[test]
391    fn rejects_a_space() {
392        // `is_ascii_graphic` excludes the space: a bare space is not a usable key.
393        assert_eq!(
394            validate_idempotency_key("abc def"),
395            Err(IdempotencyKeyError::NotPrintableAscii)
396        );
397    }
398
399    #[test]
400    fn error_messages_name_the_broken_rule() {
401        assert!(IdempotencyKeyError::Empty.message().contains("empty"));
402        assert!(
403            IdempotencyKeyError::TooLong
404                .message()
405                .contains(&MAX_IDEMPOTENCY_KEY_LEN.to_string())
406        );
407        assert!(
408            IdempotencyKeyError::NotPrintableAscii
409                .message()
410                .contains("ASCII")
411        );
412    }
413
414    fn limited(concurrency_limits: Vec<ConcurrencyLimit>) -> CreateRunRequest {
415        CreateRunRequest {
416            concurrency_limits,
417            ..request(None)
418        }
419    }
420
421    #[test]
422    fn concurrency_limits_default_to_empty_when_absent() {
423        let req: CreateRunRequest = from_str(r#"{"workflow":"deploy"}"#).expect("deserialize");
424        assert!(req.concurrency_limits.is_empty());
425        assert!(req.validate().is_ok());
426    }
427
428    #[test]
429    fn concurrency_limits_parse_from_json() {
430        let req: CreateRunRequest = from_str(
431            r#"{"workflow":"deploy","concurrency_limits":[{"group":"repo:acme","limit":2}]}"#,
432        )
433        .expect("deserialize");
434        assert_eq!(
435            req.concurrency_limits,
436            vec![ConcurrencyLimit::new("repo:acme", 2)]
437        );
438        assert!(req.validate().is_ok());
439    }
440
441    #[test]
442    fn concurrency_limits_reject_a_negative_limit_at_deserialization() {
443        let res: Result<CreateRunRequest, _> = from_str(
444            r#"{"workflow":"deploy","concurrency_limits":[{"group":"repo:acme","limit":-1}]}"#,
445        );
446        assert!(res.is_err());
447    }
448
449    #[test]
450    fn validate_rejects_invalid_concurrency_limits() {
451        let cases = [
452            (
453                vec![ConcurrencyLimit::new("repo:acme", 0)],
454                "concurrency_limits: concurrency limit for group 'repo:acme' must be at least 1",
455            ),
456            (
457                vec![ConcurrencyLimit::new("  ", 1)],
458                "concurrency_limits: concurrency group must not be empty",
459            ),
460            (
461                vec![
462                    ConcurrencyLimit::new("repo:acme", 1),
463                    ConcurrencyLimit::new("repo:acme", 2),
464                ],
465                "concurrency_limits: concurrency group 'repo:acme' is listed more than once",
466            ),
467        ];
468        for (limits, expected) in cases {
469            let err = limited(limits)
470                .validate()
471                .expect_err("invalid limits must be rejected");
472            assert_eq!(err, expected);
473        }
474
475        let err = limited(vec![ConcurrencyLimit::new("g".repeat(256), 1)])
476            .validate()
477            .expect_err("over-long group must be rejected");
478        assert!(err.contains("255 bytes"), "{err}");
479    }
480
481    #[test]
482    fn worker_tags_default_to_empty_when_absent() {
483        let req: CreateRunRequest = from_str(r#"{"workflow":"deploy"}"#).expect("deserialize");
484        assert!(req.worker_tags.is_empty());
485        assert!(req.validate().is_ok());
486    }
487
488    #[test]
489    fn worker_tags_parse_from_json_and_validate() {
490        let req: CreateRunRequest =
491            from_str(r#"{"workflow":"transcode","worker_tags":["gpu","region:eu"]}"#)
492                .expect("deserialize");
493        assert_eq!(
494            req.worker_tags,
495            vec!["gpu".to_string(), "region:eu".to_string()]
496        );
497        assert!(req.validate().is_ok());
498    }
499
500    #[test]
501    fn validate_rejects_invalid_worker_tags() {
502        let invalid = [
503            vec![String::new()],
504            vec!["two words".to_string()],
505            vec!["a".repeat(65)],
506            (0..33).map(|i| format!("tag-{i}")).collect(),
507        ];
508        for worker_tags in invalid {
509            let req = CreateRunRequest {
510                worker_tags: worker_tags.clone(),
511                ..request(None)
512            };
513            let err = req.validate().expect_err("invalid tags must be rejected");
514            assert!(err.starts_with("worker_tags: "), "{worker_tags:?}: {err}");
515        }
516    }
517}
ironflow_api/entities/create_run.rs

ironflow_api/entities/
create_run.rs
