dapr_durabletask/api/
retry_policy.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use crate::api::FailureDetails;
5
6/// A predicate that decides whether a failed task should be retried.
7///
8/// Receives the [`FailureDetails`] of the failure. Return `true` to retry,
9/// `false` to propagate the error immediately.
10pub type RetryHandle = Arc<dyn Fn(&FailureDetails) -> bool + Send + Sync>;
11
12/// Policy that governs automatic retry behaviour for activities and sub-orchestrations.
13#[derive(Clone, serde::Serialize, serde::Deserialize)]
14pub struct RetryPolicy {
15    /// Maximum total number of attempts (first execution + retries).
16    pub max_number_of_attempts: u32,
17    /// Delay before the first retry.
18    pub first_retry_interval: Duration,
19    /// Multiplier applied to the delay on each successive retry.
20    /// `1.0` means constant interval; `2.0` means exponential doubling.
21    pub backoff_coefficient: f64,
22    /// Upper bound on the computed delay between retries.
23    pub max_retry_interval: Option<Duration>,
24    /// Maximum total wall-clock time spent retrying (measured from the first
25    /// failure). Once elapsed, no further retries are scheduled.
26    pub retry_timeout: Option<Duration>,
27    /// Optional predicate called on each failure to decide whether to retry.
28    /// If `None`, all failures are retried (up to `max_number_of_attempts`).
29    #[serde(skip)]
30    pub handle: Option<RetryHandle>,
31}
32
33impl std::fmt::Debug for RetryPolicy {
34    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35        f.debug_struct("RetryPolicy")
36            .field("max_number_of_attempts", &self.max_number_of_attempts)
37            .field("first_retry_interval", &self.first_retry_interval)
38            .field("backoff_coefficient", &self.backoff_coefficient)
39            .field("max_retry_interval", &self.max_retry_interval)
40            .field("retry_timeout", &self.retry_timeout)
41            .field("handle", &self.handle.as_ref().map(|_| "<fn>"))
42            .finish()
43    }
44}
45
46impl RetryPolicy {
47    /// Creates a simple retry policy with the given attempt count and interval.
48    pub fn new(max_number_of_attempts: u32, first_retry_interval: Duration) -> Self {
49        Self {
50            max_number_of_attempts,
51            first_retry_interval,
52            backoff_coefficient: 1.0,
53            max_retry_interval: None,
54            retry_timeout: None,
55            handle: None,
56        }
57    }
58
59    /// Set the backoff coefficient (multiplier applied to the delay each retry).
60    pub fn with_backoff_coefficient(mut self, coefficient: f64) -> Self {
61        self.backoff_coefficient = coefficient;
62        self
63    }
64
65    /// Set the maximum delay between retries.
66    pub fn with_max_retry_interval(mut self, interval: Duration) -> Self {
67        self.max_retry_interval = Some(interval);
68        self
69    }
70
71    /// Set the maximum total wall-clock time spent retrying.
72    pub fn with_retry_timeout(mut self, timeout: Duration) -> Self {
73        self.retry_timeout = Some(timeout);
74        self
75    }
76
77    /// Set a custom predicate to decide whether a failure should be retried.
78    ///
79    /// # Example
80    /// ```ignore
81    /// let policy = RetryPolicy::new(3, Duration::from_secs(1))
82    ///     .with_handle(|details| details.error_type != "NonRetryableError");
83    /// ```
84    pub fn with_handle<F>(mut self, f: F) -> Self
85    where
86        F: Fn(&FailureDetails) -> bool + Send + Sync + 'static,
87    {
88        self.handle = Some(Arc::new(f));
89        self
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[test]
98    fn new_sets_defaults() {
99        let p = RetryPolicy::new(5, Duration::from_secs(2));
100        assert_eq!(p.max_number_of_attempts, 5);
101        assert_eq!(p.first_retry_interval, Duration::from_secs(2));
102        assert!((p.backoff_coefficient - 1.0).abs() < f64::EPSILON);
103        assert!(p.max_retry_interval.is_none());
104        assert!(p.retry_timeout.is_none());
105        assert!(p.handle.is_none());
106    }
107
108    #[test]
109    fn with_backoff_coefficient() {
110        let p = RetryPolicy::new(3, Duration::from_secs(1)).with_backoff_coefficient(2.5);
111        assert!((p.backoff_coefficient - 2.5).abs() < f64::EPSILON);
112    }
113
114    #[test]
115    fn with_max_retry_interval() {
116        let p = RetryPolicy::new(3, Duration::from_secs(1))
117            .with_max_retry_interval(Duration::from_secs(60));
118        assert_eq!(p.max_retry_interval, Some(Duration::from_secs(60)));
119    }
120
121    #[test]
122    fn with_retry_timeout() {
123        let p = RetryPolicy::new(3, Duration::from_secs(1))
124            .with_retry_timeout(Duration::from_secs(300));
125        assert_eq!(p.retry_timeout, Some(Duration::from_secs(300)));
126    }
127
128    #[test]
129    fn with_handle_is_callable() {
130        let p = RetryPolicy::new(3, Duration::from_secs(1)).with_handle(|_details| true);
131        let fd = FailureDetails {
132            message: "err".into(),
133            error_type: "E".into(),
134            stack_trace: None,
135        };
136        assert!((p.handle.unwrap())(&fd));
137    }
138
139    #[test]
140    fn debug_with_handle() {
141        let p = RetryPolicy::new(1, Duration::from_secs(1)).with_handle(|_| false);
142        let dbg = format!("{p:?}");
143        assert!(dbg.contains(r#"Some("<fn>")"#));
144    }
145
146    #[test]
147    fn debug_without_handle() {
148        let p = RetryPolicy::new(1, Duration::from_secs(1));
149        let dbg = format!("{p:?}");
150        assert!(dbg.contains("handle: None"));
151    }
152}
dapr_durabletask/api/retry_policy.rs

dapr_durabletask/api/
retry_policy.rs
