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

ironflow_api/entities/
create_run.rs
