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

ironflow_api/entities/
create_run.rs
