dataflow_rs/engine/
retry.rs

1//! Retrying an operation that failed for a reason worth retrying.
2//!
3//! The crate has carried a complete retryability *classification* with no
4//! *mechanism*: [`DataflowError::retryable`] sorts every variant, and
5//! [`ErrorInfo`](crate::ErrorInfo) has `retry_attempted` / `retry_count` fields,
6//! but no engine code path acts on any of it. Every host has written the same
7//! loop.
8//!
9//! This module supplies the loop. It is deliberately **not** engine-level
10//! automatic retry: the engine cannot know which handlers are idempotent — an
11//! SMTP send that times out after `DATA` is indistinguishable from one that
12//! succeeded, and retrying duplicates the mail. Whether to retry stays a
13//! per-handler, per-call-site decision; the crate just supplies the correct
14//! loop for those that opt in.
15//!
16//! # Not available on wasm32
17//!
18//! Backoff needs a timer, and tokio's time driver does not run on
19//! `wasm32-unknown-unknown`. The whole module is `cfg`-gated off that target.
20
21use crate::engine::error::{DataflowError, Result};
22use std::future::Future;
23use std::time::Duration;
24use tokio::time::Instant;
25
26/// The longest a single backoff sleep may grow to, however many attempts have
27/// failed. Without a ceiling, doubling reaches minutes and a caller waiting on
28/// the result has no idea why.
29const MAX_BACKOFF: Duration = Duration::from_secs(60);
30
31/// How hard to retry, and for how long overall.
32///
33/// ```
34/// use dataflow_rs::RetryPolicy;
35/// use std::time::Duration;
36///
37/// // Three retries, 100ms doubling, but never more than 5s in total.
38/// let policy = RetryPolicy {
39///     max_retries: 3,
40///     retry_delay_ms: 100,
41///     deadline: Some(Duration::from_secs(5)),
42/// };
43/// assert_eq!(policy.max_retries, 3);
44/// ```
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub struct RetryPolicy {
47    /// Retries *after* the first attempt. `0` means try once and give up.
48    pub max_retries: u32,
49    /// Base delay. Doubles per attempt, capped at 60s.
50    pub retry_delay_ms: u64,
51    /// Wall-clock ceiling for the **whole** loop, sleeps included.
52    ///
53    /// Without this, a call with a 30s per-attempt timeout and capped backoff
54    /// can run to roughly 127s under a 30s caller budget — the per-attempt
55    /// bound says nothing about the total.
56    pub deadline: Option<Duration>,
57}
58
59impl Default for RetryPolicy {
60    /// Three retries, 100ms base delay, no deadline.
61    fn default() -> Self {
62        Self {
63            max_retries: 3,
64            retry_delay_ms: 100,
65            deadline: None,
66        }
67    }
68}
69
70impl RetryPolicy {
71    /// A policy that never retries. Useful as an explicit opt-out at a call
72    /// site that takes a policy.
73    pub fn none() -> Self {
74        Self {
75            max_retries: 0,
76            retry_delay_ms: 0,
77            deadline: None,
78        }
79    }
80
81    /// The backoff before retry number `retry` (1-based), capped.
82    fn backoff(&self, retry: u32) -> Duration {
83        let doubled = self
84            .retry_delay_ms
85            .checked_mul(1u64 << retry.min(32).saturating_sub(1))
86            .map_or(MAX_BACKOFF, Duration::from_millis);
87        doubled.min(MAX_BACKOFF)
88    }
89}
90
91/// Run `operation`, retrying while it fails retryably and budget remains.
92///
93/// Retries only when [`DataflowError::retryable`] says so — a validation error
94/// fails once and returns immediately, because trying it again cannot help.
95///
96/// ```no_run
97/// use dataflow_rs::{RetryPolicy, retry_with_policy, DataflowError, Result};
98/// # async fn demo() -> Result<String> {
99/// # async fn call_the_service() -> Result<String> { Ok(String::new()) }
100/// let body = retry_with_policy(RetryPolicy::default(), "user_service", || async {
101///     call_the_service().await
102/// })
103/// .await?;
104/// # Ok(body)
105/// # }
106/// ```
107///
108/// # Deadline
109///
110/// The deadline covers the whole loop, sleeps included. A backoff that would
111/// cross it is **skipped** rather than slept: sleeping only to fail afterwards
112/// spends latency the caller is already waiting on. The loop then ends with the
113/// last error.
114///
115/// Timing uses [`tokio::time::Instant`], so the deadline stays coherent with
116/// the sleeps under `tokio::time::pause()`.
117pub async fn retry_with_policy<T, F, Fut>(
118    policy: RetryPolicy,
119    label: &str,
120    operation: F,
121) -> Result<T>
122where
123    F: FnMut() -> Fut,
124    Fut: Future<Output = Result<T>>,
125{
126    retry_with_attempts(policy, label, operation).await.0
127}
128
129/// As [`retry_with_policy`], also reporting how many attempts were made.
130///
131/// The count is what fills [`ErrorInfo::retry_attempted`](crate::ErrorInfo) and
132/// `retry_count` — the two fields the crate has always carried with nothing to
133/// populate them:
134///
135/// ```no_run
136/// use dataflow_rs::{ErrorInfo, RetryPolicy, retry_with_attempts};
137/// # async fn demo() {
138/// # async fn call() -> dataflow_rs::Result<()> { Ok(()) }
139/// let (result, attempts) =
140///     retry_with_attempts(RetryPolicy::default(), "svc", || async { call().await }).await;
141///
142/// if let Err(err) = result {
143///     let mut info = ErrorInfo::simple_ref("SVC_FAILED", &err.to_string(), None);
144///     info.retry_attempted = Some(attempts > 1);
145///     info.retry_count = Some(attempts.saturating_sub(1));
146/// }
147/// # }
148/// ```
149pub async fn retry_with_attempts<T, F, Fut>(
150    policy: RetryPolicy,
151    label: &str,
152    mut operation: F,
153) -> (Result<T>, u32)
154where
155    F: FnMut() -> Fut,
156    Fut: Future<Output = Result<T>>,
157{
158    let started = Instant::now();
159    let mut attempts = 0u32;
160    let mut last: DataflowError;
161
162    loop {
163        attempts += 1;
164        match operation().await {
165            Ok(value) => return (Ok(value), attempts),
166            Err(err) => last = err,
167        }
168
169        if !last.retryable() {
170            log::debug!("{label}: not retryable after {attempts} attempt(s): {last}");
171            return (Err(last), attempts);
172        }
173
174        let retry = attempts; // the retry we are about to consider, 1-based
175        if retry > policy.max_retries {
176            log::debug!("{label}: giving up after {attempts} attempt(s): {last}");
177            return (Err(last), attempts);
178        }
179
180        let backoff = policy.backoff(retry);
181
182        // The deadline covers the sleep too. Sleeping and *then* failing spends
183        // latency the caller is already waiting on, so a backoff that would
184        // cross the line ends the loop instead.
185        if let Some(deadline) = policy.deadline {
186            let elapsed = started.elapsed();
187            if elapsed + backoff >= deadline {
188                log::debug!(
189                    "{label}: deadline {deadline:?} leaves no room for a {backoff:?} backoff \
190                     after {elapsed:?}; stopping at {attempts} attempt(s)"
191                );
192                return (Err(last), attempts);
193            }
194        }
195
196        log::debug!("{label}: retry {retry} in {backoff:?} after: {last}");
197        tokio::time::sleep(backoff).await;
198    }
199}
dataflow_rs/engine/retry.rs

dataflow_rs/engine/
retry.rs
