feldera_rest_api/
retry.rs

1//! Retry layer for the generated client.
2//!
3//! Every request the generated `Client` sends passes through
4//! [`execute_with_retry`] (wired up by the `ClientHooks` impl in `lib.rs`),
5//! which resends requests that failed for reasons known to be transient:
6//! transport failures and the HTTP statuses in [`RETRYABLE_STATUSES`]. Waits
7//! between attempts grow exponentially; a `Retry-After` header overrides the
8//! computed wait, and a 502 waits based on a cluster-health probe (see
9//! [`cluster_is_healthy`]).
10//!
11//! Repeating a request is only safe when the first attempt provably caused no
12//! server-side effect, or when the operation yields the same state however
13//! often it runs. [`is_idempotent`] classifies operations; non-idempotent
14//! ones retry only failures where the request never reached its target.
15
16use std::time::Duration;
17
18use reqwest::header::HeaderMap;
19use reqwest::{Method, Request, Response, StatusCode};
20
21/// Retry behavior for transient request failures.
22///
23/// Carried by `Client` as its inner state: construct one and pass it to
24/// `Client::new` / `Client::new_with_client`.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct RetryPolicy {
27    /// Retries after the initial attempt. `3` means up to 4 attempts total;
28    /// `0` disables retrying.
29    pub max_retries: u32,
30    /// Wait before the first retry. Each further retry doubles the wait.
31    pub initial_backoff: Duration,
32    /// Upper bound on the wait between attempts.
33    pub max_backoff: Duration,
34    /// Flat wait between 502 retries while the cluster reports unhealthy on
35    /// `/v0/cluster_healthz`: the cluster is likely upgrading or restarting,
36    /// so a flat pause beats an exponential ramp. Not capped by
37    /// `max_backoff`.
38    pub unhealthy_backoff: Duration,
39}
40
41impl Default for RetryPolicy {
42    fn default() -> Self {
43        Self {
44            max_retries: 3,
45            initial_backoff: Duration::from_secs(2),
46            max_backoff: Duration::from_secs(60),
47            unhealthy_backoff: Duration::from_secs(90),
48        }
49    }
50}
51
52impl RetryPolicy {
53    /// A policy that never retries.
54    pub fn none() -> Self {
55        Self {
56            max_retries: 0,
57            ..Self::default()
58        }
59    }
60
61    /// Wait before retry number `retry_index` (zero-based).
62    fn backoff(&self, retry_index: u32) -> Duration {
63        self.initial_backoff
64            .saturating_mul(2u32.saturating_pow(retry_index))
65            .min(self.max_backoff)
66    }
67}
68
69/// Statuses that signal a transient condition worth retrying (for idempotent
70/// operations): request timeout, rate limit, and gateway/service failures.
71const RETRYABLE_STATUSES: [StatusCode; 5] = [
72    StatusCode::REQUEST_TIMEOUT,
73    StatusCode::TOO_MANY_REQUESTS,
74    StatusCode::BAD_GATEWAY,
75    StatusCode::SERVICE_UNAVAILABLE,
76    StatusCode::GATEWAY_TIMEOUT,
77];
78
79/// Mutating operations that are nonetheless safe to repeat: desired-state
80/// setters (repeating yields the same desired state) and pure functions of
81/// their input. Operations absent from this list retry only failures where
82/// the request never reached its target.
83const IDEMPOTENT_OPERATIONS: &[&str] = &[
84    "post_pipeline_start",
85    "post_pipeline_pause",
86    "post_pipeline_resume",
87    "post_pipeline_stop",
88    "post_pipeline_clear",
89    "post_pipeline_activate",
90    "post_pipeline_approve",
91    "post_pipeline_dismiss_error",
92    "post_pipeline_input_connector_action",
93    "post_pipeline_diff",
94    "post_validate_program",
95    // Set-based updates: repeating writes the same field values.
96    "patch_pipeline",
97    "patch_tenant",
98];
99
100/// GET/HEAD/PUT/DELETE are idempotent by HTTP semantics; POST and PATCH only
101/// when the operation is known to be safe to repeat.
102fn is_idempotent(method: &Method, operation_id: &str) -> bool {
103    matches!(
104        *method,
105        Method::GET | Method::HEAD | Method::PUT | Method::DELETE
106    ) || IDEMPOTENT_OPERATIONS.contains(&operation_id)
107}
108
109/// Whether a 503 body proves the request never reached the pipeline, making a
110/// retry safe even for non-idempotent operations.
111///
112/// The api-server proxies pipeline-interaction endpoints and answers 503
113/// `PipelineInteractionUnreachable` for every proxy failure. Only the
114/// connect-phase failure ("Failed to connect to host", the awc connect error
115/// prefix) guarantees the pipeline never saw the request; an exchange that
116/// timed out or disconnected mid-flight carries different wording and may
117/// already have been applied. If the manager ever rewords the message, this
118/// check fails closed: the error surfaces as before, nothing double-applies.
119fn is_never_dispatched_503(body: &[u8]) -> bool {
120    serde_json::from_slice::<serde_json::Value>(body).is_ok_and(|v| {
121        v.get("error_code").and_then(|c| c.as_str()) == Some("PipelineInteractionUnreachable")
122            && v.get("message")
123                .and_then(|m| m.as_str())
124                .is_some_and(|m| m.contains("Failed to connect to host"))
125    })
126}
127
128/// Rebuild a response consumed while inspecting its body, so the caller can
129/// parse the error as if the response arrived untouched. The rebuilt response
130/// loses request metadata such as the URL, which error handling never reads.
131fn rebuild_response(status: StatusCode, headers: HeaderMap, body: bytes::Bytes) -> Response {
132    let mut rebuilt = http::Response::new(body);
133    *rebuilt.status_mut() = status;
134    *rebuilt.headers_mut() = headers;
135    Response::from(rebuilt)
136}
137
138/// Server-requested wait from a `Retry-After` header (seconds form only; the
139/// HTTP-date form is rare and falls back to the computed backoff).
140fn retry_after(headers: &HeaderMap) -> Option<Duration> {
141    let secs = headers
142        .get(reqwest::header::RETRY_AFTER)?
143        .to_str()
144        .ok()?
145        .trim()
146        .parse()
147        .ok()?;
148    Some(Duration::from_secs(secs))
149}
150
151/// How long the `/v0/cluster_healthz` probe may take before the cluster
152/// counts as unhealthy.
153const HEALTH_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
154
155/// Probe `/v0/cluster_healthz` to tell a spurious 502 (retry immediately)
156/// from an unhealthy cluster, e.g. one whose upgrade is in progress (flat
157/// long wait). The endpoint answers 200 only when every service is healthy;
158/// any other status or a probe failure counts as unhealthy.
159async fn cluster_is_healthy(client: &reqwest::Client, baseurl: &str) -> bool {
160    let url = format!("{}/v0/cluster_healthz", baseurl.trim_end_matches('/'));
161    match client.get(url).timeout(HEALTH_PROBE_TIMEOUT).send().await {
162        Ok(response) => response.status().is_success(),
163        Err(_) => false,
164    }
165}
166
167/// Pick the wait before the next retry:
168/// a `Retry-After` value from the server wins (capped at `max_backoff`);
169/// a 502 from a healthy cluster was spurious, so retry immediately;
170/// a 502 from an unhealthy cluster (e.g. an upgrade in progress) waits the
171/// flat `unhealthy_backoff`; everything else backs off exponentially.
172fn next_wait(
173    policy: &RetryPolicy,
174    retry_index: u32,
175    server_wait: Option<Duration>,
176    cluster_healthy_after_502: Option<bool>,
177) -> Duration {
178    match (server_wait, cluster_healthy_after_502) {
179        (Some(server_wait), _) => server_wait.min(policy.max_backoff),
180        (None, Some(true)) => Duration::ZERO,
181        (None, Some(false)) => policy.unhealthy_backoff,
182        (None, None) => policy.backoff(retry_index),
183    }
184}
185
186enum Verdict {
187    Return(Response),
188    Retry {
189        reason: &'static str,
190        /// Wait requested by the server via `Retry-After`; overrides the
191        /// computed backoff (still capped at `max_backoff`).
192        server_wait: Option<Duration>,
193        status: StatusCode,
194    },
195}
196
197/// Decide whether a completed exchange warrants a retry. Consumes the
198/// response only when a body inspection is needed; hands it back otherwise.
199async fn judge_response(response: Response, idempotent: bool) -> reqwest::Result<Verdict> {
200    let status = response.status();
201    if !RETRYABLE_STATUSES.contains(&status) {
202        return Ok(Verdict::Return(response));
203    }
204    let server_wait = retry_after(response.headers());
205    if idempotent {
206        return Ok(Verdict::Retry {
207            reason: "transient HTTP status",
208            server_wait,
209            status,
210        });
211    }
212    if status != StatusCode::SERVICE_UNAVAILABLE {
213        return Ok(Verdict::Return(response));
214    }
215    let headers = response.headers().clone();
216    let body = response.bytes().await?;
217    if is_never_dispatched_503(&body) {
218        Ok(Verdict::Retry {
219            reason: "pipeline unreachable, request never sent",
220            server_wait,
221            status,
222        })
223    } else {
224        Ok(Verdict::Return(rebuild_response(status, headers, body)))
225    }
226}
227
228/// Execute `request`, retrying transient failures per `policy`.
229///
230/// Requests with streaming bodies cannot be cloned and get a single attempt.
231pub(crate) async fn execute_with_retry(
232    client: &reqwest::Client,
233    policy: &RetryPolicy,
234    baseurl: &str,
235    mut request: Request,
236    operation_id: &str,
237) -> reqwest::Result<Response> {
238    let idempotent = is_idempotent(request.method(), operation_id);
239    let mut retry_index = 0u32;
240    loop {
241        // Clone before executing: execute() consumes the request, and a
242        // failed attempt leaves nothing to resend.
243        let retry_request = if retry_index < policy.max_retries {
244            request.try_clone()
245        } else {
246            None
247        };
248
249        let retries_remain = retry_index < policy.max_retries;
250        let outcome = client.execute(request).await;
251
252        let Some(retry_request) = retry_request else {
253            // Final attempt: retries exhausted, or a streaming body that
254            // cannot be resent.
255            let failed_transiently = match &outcome {
256                Ok(response) => RETRYABLE_STATUSES.contains(&response.status()),
257                Err(_) => true,
258            };
259            if retries_remain && failed_transiently {
260                log::info!(
261                    "{operation_id}: not retrying a transient failure because the request body is a stream and cannot be resent"
262                );
263            }
264            return outcome;
265        };
266
267        let (reason, detail, server_wait, retried_status) = match outcome {
268            Ok(response) => match judge_response(response, idempotent).await? {
269                Verdict::Return(response) => return Ok(response),
270                Verdict::Retry {
271                    reason,
272                    server_wait,
273                    status,
274                } => (reason, String::new(), server_wait, Some(status)),
275            },
276            Err(error) => {
277                // Idempotent operations repeat on any transport failure
278                // (timeouts, resets mid-exchange). A connect failure never
279                // reached the server, so it is safe for every operation;
280                // other failures may already have been applied.
281                if !(idempotent || error.is_connect()) {
282                    return Err(error);
283                }
284                ("transport error", format!(": {error}"), None, None)
285            }
286        };
287
288        request = retry_request;
289        // A 502 comes from in front of the api-server; probe cluster health
290        // to pick the right wait (see `next_wait`). Skipped when the server
291        // already prescribed a wait via `Retry-After`.
292        let cluster_healthy_after_502 =
293            if server_wait.is_none() && retried_status == Some(StatusCode::BAD_GATEWAY) {
294                Some(cluster_is_healthy(client, baseurl).await)
295            } else {
296                None
297            };
298        let wait = next_wait(policy, retry_index, server_wait, cluster_healthy_after_502);
299        retry_index += 1;
300        log::debug!(
301            "{operation_id}: {reason}{detail} - retrying in {}s (attempt {} of {})",
302            wait.as_secs(),
303            retry_index + 1,
304            policy.max_retries + 1,
305        );
306        tokio::time::sleep(wait).await;
307    }
308}
309
310#[cfg(test)]
311mod tests {
312    use super::*;
313
314    /// Backoff doubles per retry and is capped at `max_backoff`.
315    #[test]
316    fn backoff_doubles_and_caps() {
317        let policy = RetryPolicy {
318            max_retries: 5,
319            initial_backoff: Duration::from_secs(2),
320            max_backoff: Duration::from_secs(6),
321            ..RetryPolicy::default()
322        };
323        assert_eq!(policy.backoff(0), Duration::from_secs(2));
324        assert_eq!(policy.backoff(1), Duration::from_secs(4));
325        assert_eq!(policy.backoff(2), Duration::from_secs(6));
326        assert_eq!(policy.backoff(30), Duration::from_secs(6));
327    }
328
329    /// Idempotency follows the HTTP method except for allowlisted operations.
330    #[test]
331    fn idempotency_classification() {
332        assert!(is_idempotent(&Method::GET, "get_pipeline"));
333        assert!(is_idempotent(&Method::PUT, "put_pipeline"));
334        assert!(is_idempotent(&Method::DELETE, "delete_pipeline"));
335        assert!(is_idempotent(&Method::POST, "post_pipeline_start"));
336        assert!(!is_idempotent(&Method::POST, "clock_advance"));
337        assert!(!is_idempotent(&Method::POST, "http_input"));
338        assert!(!is_idempotent(&Method::PATCH, "patch_something_new"));
339    }
340
341    /// Wait selection: `Retry-After` wins (capped), a 502 waits per cluster
342    /// health, everything else backs off exponentially.
343    #[test]
344    fn next_wait_selection() {
345        let policy = RetryPolicy {
346            max_retries: 3,
347            initial_backoff: Duration::from_secs(2),
348            max_backoff: Duration::from_secs(60),
349            unhealthy_backoff: Duration::from_secs(90),
350        };
351        // Server-prescribed wait wins, capped at max_backoff.
352        assert_eq!(
353            next_wait(&policy, 0, Some(Duration::from_secs(7)), None),
354            Duration::from_secs(7)
355        );
356        assert_eq!(
357            next_wait(&policy, 0, Some(Duration::from_secs(600)), None),
358            Duration::from_secs(60)
359        );
360        // 502 with healthy cluster: immediate; unhealthy: flat, uncapped.
361        assert_eq!(next_wait(&policy, 0, None, Some(true)), Duration::ZERO);
362        assert_eq!(
363            next_wait(&policy, 0, None, Some(false)),
364            Duration::from_secs(90)
365        );
366        // Otherwise exponential.
367        assert_eq!(next_wait(&policy, 1, None, None), Duration::from_secs(4));
368    }
369
370    /// Only the seconds form of `Retry-After` yields a wait.
371    #[test]
372    fn retry_after_parses_seconds_only() {
373        let with = |v: &str| {
374            let mut h = HeaderMap::new();
375            h.insert(reqwest::header::RETRY_AFTER, v.parse().unwrap());
376            h
377        };
378        assert_eq!(retry_after(&with("7")), Some(Duration::from_secs(7)));
379        assert_eq!(retry_after(&with("Wed, 21 Oct 2026 07:28:00 GMT")), None);
380        assert_eq!(retry_after(&with("-3")), None);
381        assert_eq!(retry_after(&HeaderMap::new()), None);
382    }
383
384    /// Only the connect-phase wording marks a 503 as safe for any operation.
385    #[test]
386    fn never_dispatched_detection() {
387        let connect = br#"{"message": "Failed to connect to host: Timeout while establishing connection", "error_code": "PipelineInteractionUnreachable", "details": {}}"#;
388        let timeout = br#"{"message": "timeout (5s) was reached", "error_code": "PipelineInteractionUnreachable", "details": {}}"#;
389        let other_code = br#"{"message": "Failed to connect to host", "error_code": "PipelineInteractionNotDeployed", "details": {}}"#;
390        assert!(is_never_dispatched_503(connect));
391        assert!(!is_never_dispatched_503(timeout));
392        assert!(!is_never_dispatched_503(other_code));
393        assert!(!is_never_dispatched_503(b"not json"));
394    }
395}
feldera_rest_api/retry.rs

feldera_rest_api/
retry.rs
