ironflow_core/
retry.rs

1//! Retry policy for transient failures with exponential backoff.
2//!
3//! The [`RetryPolicy`] struct configures how operations (HTTP requests, agent
4//! invocations) should be retried when they encounter transient errors. Retries
5//! use exponential backoff with optional jitter to avoid thundering-herd effects.
6//!
7//! # Retryable errors
8//!
9//! Not all errors are retried. Only *transient* failures are eligible:
10//!
11//! | Operation | Retried | Not retried |
12//! |-----------|---------|-------------|
13//! | **HTTP** | Transport errors (DNS, timeout, connection refused), 5xx, 429 | SSRF blocks, 4xx (except 429), response too large |
14//! | **Agent** | Process failures, timeouts, schema validation (CLI non-determinism) | Prompt too large |
15//!
16//! # Examples
17//!
18//! ```no_run
19//! use ironflow_core::operations::http::Http;
20//! use ironflow_core::retry::RetryPolicy;
21//!
22//! # async fn example() -> Result<(), ironflow_core::error::OperationError> {
23//! // Simple: retry up to 3 times with default backoff
24//! let output = Http::get("https://api.example.com/data")
25//!     .retry(3)
26//!     .await?;
27//!
28//! // Advanced: custom backoff settings
29//! let output = Http::get("https://api.example.com/data")
30//!     .retry_policy(
31//!         RetryPolicy::new(3)
32//!             .backoff(std::time::Duration::from_millis(500))
33//!             .max_backoff(std::time::Duration::from_secs(30))
34//!             .multiplier(3.0)
35//!     )
36//!     .await?;
37//! # Ok(())
38//! # }
39//! ```
40
41use std::time::Duration;
42
43use serde::{Deserialize, Serialize};
44
45use crate::error::{AgentError, OperationError};
46
47mod serde_duration_ms {
48    use std::time::Duration;
49
50    use serde::{Deserialize, Deserializer, Serializer};
51
52    pub fn serialize<S: Serializer>(duration: &Duration, s: S) -> Result<S::Ok, S::Error> {
53        let ms = u64::try_from(duration.as_millis())
54            .map_err(|_| serde::ser::Error::custom("duration exceeds u64::MAX milliseconds"))?;
55        s.serialize_u64(ms)
56    }
57
58    pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Duration, D::Error> {
59        let ms = u64::deserialize(d)?;
60        Ok(Duration::from_millis(ms))
61    }
62}
63
64/// Default initial backoff between retries.
65const DEFAULT_INITIAL_BACKOFF: Duration = Duration::from_millis(200);
66
67/// Default maximum backoff between retries.
68const DEFAULT_MAX_BACKOFF: Duration = Duration::from_secs(30);
69
70/// Default backoff multiplier (doubles each retry).
71const DEFAULT_MULTIPLIER: f64 = 2.0;
72
73/// Retry policy with exponential backoff for transient failures.
74///
75/// Created via [`RetryPolicy::new`] with the maximum number of retries.
76/// All other parameters have sensible defaults and can be customised with
77/// builder methods.
78///
79/// # Defaults
80///
81/// | Parameter | Default |
82/// |-----------|---------|
83/// | `max_retries` | (required) |
84/// | `initial_backoff` | 200ms |
85/// | `max_backoff` | 30s |
86/// | `multiplier` | 2.0 |
87///
88/// # Examples
89///
90/// ```
91/// use std::time::Duration;
92/// use ironflow_core::retry::RetryPolicy;
93///
94/// let policy = RetryPolicy::new(3)
95///     .backoff(Duration::from_millis(100))
96///     .max_backoff(Duration::from_secs(10))
97///     .multiplier(3.0);
98/// ```
99#[derive(Debug, Clone, Serialize, Deserialize)]
100#[serde(try_from = "RetryPolicyRaw")]
101pub struct RetryPolicy {
102    pub(crate) max_retries: u32,
103    #[serde(serialize_with = "serde_duration_ms::serialize")]
104    pub(crate) initial_backoff: Duration,
105    #[serde(serialize_with = "serde_duration_ms::serialize")]
106    pub(crate) max_backoff: Duration,
107    pub(crate) multiplier: f64,
108}
109
110/// Wire format for [`RetryPolicy`] deserialization with post-validation.
111#[derive(Deserialize)]
112struct RetryPolicyRaw {
113    max_retries: u32,
114    #[serde(deserialize_with = "serde_duration_ms::deserialize")]
115    initial_backoff: Duration,
116    #[serde(deserialize_with = "serde_duration_ms::deserialize")]
117    max_backoff: Duration,
118    multiplier: f64,
119}
120
121impl TryFrom<RetryPolicyRaw> for RetryPolicy {
122    type Error = String;
123
124    fn try_from(raw: RetryPolicyRaw) -> Result<Self, Self::Error> {
125        if raw.max_retries == 0 {
126            return Err("max_retries must be greater than 0".into());
127        }
128        if raw.initial_backoff.is_zero() {
129            return Err("initial backoff must not be zero".into());
130        }
131        if raw.max_backoff.is_zero() {
132            return Err("max backoff must not be zero".into());
133        }
134        if raw.multiplier < 1.0 || !raw.multiplier.is_finite() {
135            return Err(format!(
136                "multiplier must be >= 1.0 and finite, got {}",
137                raw.multiplier
138            ));
139        }
140        Ok(Self {
141            max_retries: raw.max_retries,
142            initial_backoff: raw.initial_backoff,
143            max_backoff: raw.max_backoff,
144            multiplier: raw.multiplier,
145        })
146    }
147}
148
149impl RetryPolicy {
150    /// Create a new retry policy with the given maximum number of retries.
151    ///
152    /// The initial attempt is not counted as a retry, so `max_retries(3)` means
153    /// up to 4 total attempts (1 initial + 3 retries).
154    ///
155    /// # Panics
156    ///
157    /// Panics if `max_retries` is `0`.
158    ///
159    /// # Examples
160    ///
161    /// ```
162    /// use ironflow_core::retry::RetryPolicy;
163    ///
164    /// let policy = RetryPolicy::new(3);
165    /// assert_eq!(policy.max_retries(), 3);
166    /// ```
167    pub fn new(max_retries: u32) -> Self {
168        assert!(max_retries > 0, "max_retries must be greater than 0");
169        Self {
170            max_retries,
171            initial_backoff: DEFAULT_INITIAL_BACKOFF,
172            max_backoff: DEFAULT_MAX_BACKOFF,
173            multiplier: DEFAULT_MULTIPLIER,
174        }
175    }
176
177    /// Set the initial backoff duration before the first retry.
178    ///
179    /// Subsequent retries multiply this duration by the [`multiplier`](RetryPolicy::multiplier).
180    ///
181    /// # Panics
182    ///
183    /// Panics if `duration` is zero.
184    pub fn backoff(mut self, duration: Duration) -> Self {
185        assert!(!duration.is_zero(), "initial backoff must not be zero");
186        self.initial_backoff = duration;
187        self
188    }
189
190    /// Set the maximum backoff duration between retries.
191    ///
192    /// Even after many retries with exponential growth, the delay will never
193    /// exceed this value.
194    ///
195    /// # Panics
196    ///
197    /// Panics if `duration` is zero.
198    pub fn max_backoff(mut self, duration: Duration) -> Self {
199        assert!(!duration.is_zero(), "max backoff must not be zero");
200        self.max_backoff = duration;
201        self
202    }
203
204    /// Set the backoff multiplier applied after each retry.
205    ///
206    /// For example, with `multiplier(2.0)` and `backoff(200ms)`:
207    /// - Retry 1: 200ms
208    /// - Retry 2: 400ms
209    /// - Retry 3: 800ms
210    ///
211    /// # Panics
212    ///
213    /// Panics if `multiplier` is less than `1.0`, NaN, or infinity.
214    pub fn multiplier(mut self, multiplier: f64) -> Self {
215        assert!(
216            multiplier >= 1.0 && multiplier.is_finite(),
217            "multiplier must be >= 1.0 and finite, got {multiplier}"
218        );
219        self.multiplier = multiplier;
220        self
221    }
222
223    /// Return the maximum number of retries.
224    pub fn max_retries(&self) -> u32 {
225        self.max_retries
226    }
227
228    /// Compute the backoff duration for the given retry attempt (0-indexed).
229    ///
230    /// Returns the delay capped at [`max_backoff`](RetryPolicy::max_backoff).
231    pub fn delay_for_attempt(&self, attempt: u32) -> Duration {
232        let delay = self.initial_backoff.as_secs_f64() * self.multiplier.powi(attempt as i32);
233        let capped = delay.min(self.max_backoff.as_secs_f64());
234        Duration::from_secs_f64(capped)
235    }
236}
237
238/// Returns `true` if the given [`OperationError`] is transient and should be
239/// retried.
240///
241/// # Retryable errors
242///
243/// - [`OperationError::Http`] with no status (transport error) or status 5xx / 429
244/// - [`OperationError::Agent`] wrapping [`AgentError::ProcessFailed`] or [`AgentError::Timeout`]
245/// - [`OperationError::Timeout`] (operation-level timeout)
246///
247/// # Non-retryable errors
248///
249/// - [`OperationError::Http`] with 4xx status (except 429)
250/// - [`OperationError::Agent`] wrapping [`AgentError::PromptTooLarge`] or
251///   [`AgentError::BudgetExceeded`] (the budget is already spent, replaying it
252///   costs money and cannot succeed)
253/// - [`OperationError::Shell`]
254/// - [`OperationError::Deserialize`]
255/// - [`OperationError::External`]
256pub fn is_retryable(error: &OperationError) -> bool {
257    match error {
258        OperationError::Http { status, .. } => match status {
259            None => true,
260            Some(code) => *code >= 500 || *code == 429,
261        },
262        OperationError::Agent(agent_err) => match agent_err {
263            AgentError::ProcessFailed { .. }
264            | AgentError::Timeout { .. }
265            | AgentError::SchemaValidation { .. }
266            | AgentError::RateLimited { .. } => true,
267            AgentError::HttpProvider { status_code, .. } => {
268                *status_code == 0 || *status_code >= 500
269            }
270            AgentError::PromptTooLarge { .. } | AgentError::BudgetExceeded { .. } => false,
271        },
272        OperationError::Timeout { .. } => true,
273        OperationError::Shell { .. }
274        | OperationError::Deserialize { .. }
275        | OperationError::Secret { .. }
276        | OperationError::External { .. } => false,
277    }
278}
279
280/// Returns `true` if the given HTTP status code indicates a transient server
281/// error that should be retried (5xx or 429 Too Many Requests).
282pub(crate) fn is_retryable_status(status: u16) -> bool {
283    status >= 500 || status == 429
284}
285
286#[cfg(test)]
287mod tests {
288    use super::*;
289    use std::time::Duration;
290
291    // --- RetryPolicy builder ---
292
293    #[test]
294    fn new_creates_policy_with_defaults() {
295        let policy = RetryPolicy::new(3);
296        assert_eq!(policy.max_retries, 3);
297        assert_eq!(policy.initial_backoff, DEFAULT_INITIAL_BACKOFF);
298        assert_eq!(policy.max_backoff, DEFAULT_MAX_BACKOFF);
299        assert!((policy.multiplier - DEFAULT_MULTIPLIER).abs() < f64::EPSILON);
300    }
301
302    #[test]
303    #[should_panic(expected = "max_retries must be greater than 0")]
304    fn new_zero_retries_panics() {
305        let _ = RetryPolicy::new(0);
306    }
307
308    #[test]
309    fn backoff_sets_initial_backoff() {
310        let policy = RetryPolicy::new(1).backoff(Duration::from_secs(1));
311        assert_eq!(policy.initial_backoff, Duration::from_secs(1));
312    }
313
314    #[test]
315    #[should_panic(expected = "initial backoff must not be zero")]
316    fn backoff_zero_panics() {
317        let _ = RetryPolicy::new(1).backoff(Duration::ZERO);
318    }
319
320    #[test]
321    fn max_backoff_sets_cap() {
322        let policy = RetryPolicy::new(1).max_backoff(Duration::from_secs(60));
323        assert_eq!(policy.max_backoff, Duration::from_secs(60));
324    }
325
326    #[test]
327    #[should_panic(expected = "max backoff must not be zero")]
328    fn max_backoff_zero_panics() {
329        let _ = RetryPolicy::new(1).max_backoff(Duration::ZERO);
330    }
331
332    #[test]
333    fn multiplier_sets_value() {
334        let policy = RetryPolicy::new(1).multiplier(3.0);
335        assert!((policy.multiplier - 3.0).abs() < f64::EPSILON);
336    }
337
338    #[test]
339    #[should_panic(expected = "multiplier must be >= 1.0")]
340    fn multiplier_below_one_panics() {
341        let _ = RetryPolicy::new(1).multiplier(0.5);
342    }
343
344    #[test]
345    #[should_panic(expected = "multiplier must be >= 1.0 and finite")]
346    fn multiplier_nan_panics() {
347        let _ = RetryPolicy::new(1).multiplier(f64::NAN);
348    }
349
350    #[test]
351    #[should_panic(expected = "multiplier must be >= 1.0 and finite")]
352    fn multiplier_infinity_panics() {
353        let _ = RetryPolicy::new(1).multiplier(f64::INFINITY);
354    }
355
356    #[test]
357    fn max_retries_accessor() {
358        assert_eq!(RetryPolicy::new(5).max_retries(), 5);
359    }
360
361    // --- delay_for_attempt ---
362
363    #[test]
364    fn delay_for_attempt_zero_is_initial_backoff() {
365        let policy = RetryPolicy::new(3).backoff(Duration::from_millis(100));
366        let delay = policy.delay_for_attempt(0);
367        assert_eq!(delay, Duration::from_millis(100));
368    }
369
370    #[test]
371    fn delay_for_attempt_grows_exponentially() {
372        let policy = RetryPolicy::new(5)
373            .backoff(Duration::from_millis(100))
374            .multiplier(2.0);
375
376        assert_eq!(policy.delay_for_attempt(0), Duration::from_millis(100));
377        assert_eq!(policy.delay_for_attempt(1), Duration::from_millis(200));
378        assert_eq!(policy.delay_for_attempt(2), Duration::from_millis(400));
379        assert_eq!(policy.delay_for_attempt(3), Duration::from_millis(800));
380    }
381
382    #[test]
383    fn delay_for_attempt_capped_at_max_backoff() {
384        let policy = RetryPolicy::new(10)
385            .backoff(Duration::from_secs(1))
386            .max_backoff(Duration::from_secs(5))
387            .multiplier(10.0);
388
389        // attempt 0: 1s, attempt 1: 10s (capped to 5s)
390        assert_eq!(policy.delay_for_attempt(0), Duration::from_secs(1));
391        assert_eq!(policy.delay_for_attempt(1), Duration::from_secs(5));
392        assert_eq!(policy.delay_for_attempt(2), Duration::from_secs(5));
393    }
394
395    #[test]
396    fn delay_for_attempt_with_multiplier_one_is_constant() {
397        let policy = RetryPolicy::new(3)
398            .backoff(Duration::from_millis(500))
399            .multiplier(1.0);
400
401        assert_eq!(policy.delay_for_attempt(0), Duration::from_millis(500));
402        assert_eq!(policy.delay_for_attempt(1), Duration::from_millis(500));
403        assert_eq!(policy.delay_for_attempt(2), Duration::from_millis(500));
404    }
405
406    // --- is_retryable ---
407
408    #[test]
409    fn http_transport_error_is_retryable() {
410        let err = OperationError::Http {
411            status: None,
412            message: "connection refused".to_string(),
413        };
414        assert!(is_retryable(&err));
415    }
416
417    #[test]
418    fn http_500_is_retryable() {
419        let err = OperationError::Http {
420            status: Some(500),
421            message: "internal server error".to_string(),
422        };
423        assert!(is_retryable(&err));
424    }
425
426    #[test]
427    fn http_502_is_retryable() {
428        let err = OperationError::Http {
429            status: Some(502),
430            message: "bad gateway".to_string(),
431        };
432        assert!(is_retryable(&err));
433    }
434
435    #[test]
436    fn http_503_is_retryable() {
437        let err = OperationError::Http {
438            status: Some(503),
439            message: "service unavailable".to_string(),
440        };
441        assert!(is_retryable(&err));
442    }
443
444    #[test]
445    fn http_429_is_retryable() {
446        let err = OperationError::Http {
447            status: Some(429),
448            message: "too many requests".to_string(),
449        };
450        assert!(is_retryable(&err));
451    }
452
453    #[test]
454    fn http_400_is_not_retryable() {
455        let err = OperationError::Http {
456            status: Some(400),
457            message: "bad request".to_string(),
458        };
459        assert!(!is_retryable(&err));
460    }
461
462    #[test]
463    fn http_404_is_not_retryable() {
464        let err = OperationError::Http {
465            status: Some(404),
466            message: "not found".to_string(),
467        };
468        assert!(!is_retryable(&err));
469    }
470
471    #[test]
472    fn agent_process_failed_is_retryable() {
473        let err = OperationError::Agent(AgentError::ProcessFailed {
474            exit_code: 1,
475            stderr: "crash".to_string(),
476        });
477        assert!(is_retryable(&err));
478    }
479
480    #[test]
481    fn agent_timeout_is_retryable() {
482        let err = OperationError::Agent(AgentError::Timeout {
483            limit: Duration::from_secs(60),
484        });
485        assert!(is_retryable(&err));
486    }
487
488    #[test]
489    fn agent_prompt_too_large_is_not_retryable() {
490        let err = OperationError::Agent(AgentError::PromptTooLarge {
491            chars: 1_000_000,
492            estimated_tokens: 250_000,
493            model_limit: 200_000,
494        });
495        assert!(!is_retryable(&err));
496    }
497
498    #[test]
499    fn agent_budget_exceeded_is_not_retryable() {
500        let err = OperationError::Agent(AgentError::BudgetExceeded {
501            spent_usd: 0.30,
502            limit_usd: 0.25,
503            debug_messages: Vec::new(),
504            partial_usage: Box::default(),
505        });
506        assert!(!is_retryable(&err));
507    }
508
509    #[test]
510    fn agent_schema_validation_is_retryable() {
511        let err = OperationError::Agent(AgentError::SchemaValidation {
512            expected: "object".to_string(),
513            got: "string".to_string(),
514            debug_messages: Vec::new(),
515            partial_usage: Box::default(),
516            raw_response: None,
517        });
518        assert!(is_retryable(&err));
519    }
520
521    #[test]
522    fn operation_timeout_is_retryable() {
523        let err = OperationError::Timeout {
524            step: "fetch".to_string(),
525            limit: Duration::from_secs(30),
526        };
527        assert!(is_retryable(&err));
528    }
529
530    #[test]
531    fn shell_error_is_not_retryable() {
532        let err = OperationError::Shell {
533            exit_code: 1,
534            stderr: "fail".to_string(),
535        };
536        assert!(!is_retryable(&err));
537    }
538
539    #[test]
540    fn deserialize_error_is_not_retryable() {
541        let err = OperationError::Deserialize {
542            target_type: "MyStruct".to_string(),
543            reason: "missing field".to_string(),
544        };
545        assert!(!is_retryable(&err));
546    }
547
548    // --- is_retryable_status ---
549
550    #[test]
551    fn retryable_status_codes() {
552        assert!(is_retryable_status(500));
553        assert!(is_retryable_status(502));
554        assert!(is_retryable_status(503));
555        assert!(is_retryable_status(504));
556        assert!(is_retryable_status(429));
557    }
558
559    #[test]
560    fn non_retryable_status_codes() {
561        assert!(!is_retryable_status(200));
562        assert!(!is_retryable_status(201));
563        assert!(!is_retryable_status(301));
564        assert!(!is_retryable_status(400));
565        assert!(!is_retryable_status(401));
566        assert!(!is_retryable_status(403));
567        assert!(!is_retryable_status(404));
568        assert!(!is_retryable_status(422));
569        assert!(!is_retryable_status(428));
570    }
571
572    // --- builder chaining ---
573
574    #[test]
575    fn builder_chain_all_methods() {
576        let policy = RetryPolicy::new(5)
577            .backoff(Duration::from_millis(100))
578            .max_backoff(Duration::from_secs(10))
579            .multiplier(3.0);
580
581        assert_eq!(policy.max_retries, 5);
582        assert_eq!(policy.initial_backoff, Duration::from_millis(100));
583        assert_eq!(policy.max_backoff, Duration::from_secs(10));
584        assert!((policy.multiplier - 3.0).abs() < f64::EPSILON);
585    }
586
587    #[test]
588    fn clone_produces_independent_copy() {
589        let policy = RetryPolicy::new(3).backoff(Duration::from_millis(100));
590        let cloned = policy.clone();
591        assert_eq!(policy.max_retries, cloned.max_retries);
592        assert_eq!(policy.initial_backoff, cloned.initial_backoff);
593    }
594
595    #[test]
596    fn debug_does_not_panic() {
597        let policy = RetryPolicy::new(1);
598        let debug = format!("{:?}", policy);
599        assert!(debug.contains("RetryPolicy"));
600    }
601
602    // --- serde ---
603
604    #[test]
605    fn serde_roundtrip() {
606        let policy = RetryPolicy::new(3)
607            .backoff(Duration::from_millis(500))
608            .max_backoff(Duration::from_secs(10))
609            .multiplier(2.5);
610
611        let json = serde_json::to_string(&policy).expect("serialize");
612        let back: RetryPolicy = serde_json::from_str(&json).expect("deserialize");
613
614        assert_eq!(back.max_retries, 3);
615        assert_eq!(back.initial_backoff, Duration::from_millis(500));
616        assert_eq!(back.max_backoff, Duration::from_secs(10));
617        assert!((back.multiplier - 2.5).abs() < f64::EPSILON);
618    }
619
620    #[test]
621    fn serde_duration_is_millis() {
622        let policy = RetryPolicy::new(1).backoff(Duration::from_secs(2));
623        let json = serde_json::to_string(&policy).expect("serialize");
624        assert!(json.contains("2000"), "expected 2000ms, got: {json}");
625    }
626
627    #[test]
628    fn serde_rejects_zero_max_retries() {
629        let json =
630            r#"{"max_retries":0,"initial_backoff":200,"max_backoff":30000,"multiplier":2.0}"#;
631        let err = serde_json::from_str::<RetryPolicy>(json).unwrap_err();
632        assert!(
633            err.to_string()
634                .contains("max_retries must be greater than 0")
635        );
636    }
637
638    #[test]
639    fn serde_rejects_invalid_multiplier() {
640        let json =
641            r#"{"max_retries":3,"initial_backoff":200,"max_backoff":30000,"multiplier":0.5}"#;
642        let err = serde_json::from_str::<RetryPolicy>(json).unwrap_err();
643        assert!(err.to_string().contains("multiplier must be >= 1.0"));
644    }
645
646    #[test]
647    fn serde_rejects_zero_backoff() {
648        let json = r#"{"max_retries":3,"initial_backoff":0,"max_backoff":30000,"multiplier":2.0}"#;
649        let err = serde_json::from_str::<RetryPolicy>(json).unwrap_err();
650        assert!(err.to_string().contains("initial backoff must not be zero"));
651    }
652}
ironflow_core/retry.rs

ironflow_core/
retry.rs
