faucet_core/resilience/
policy.rs

1//! Configuration value types for the resilience policy.
2
3use crate::error::FaucetError;
4use crate::resilience::classify::{RetryClassSet, classify};
5use std::time::Duration;
6
7/// How the inter-attempt delay grows.
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
9pub enum BackoffKind {
10    /// No delay between attempts.
11    None,
12    /// Constant `base` delay, capped at `max`.
13    Fixed,
14    /// `base * 2^attempt`, capped at `max`.
15    #[default]
16    Exponential,
17}
18
19impl BackoffKind {
20    /// Pre-jitter delay for a zero-based `attempt` index.
21    pub fn delay(self, base: Duration, max: Duration, attempt: u32) -> Duration {
22        match self {
23            BackoffKind::None => Duration::ZERO,
24            BackoffKind::Fixed => base.min(max),
25            BackoffKind::Exponential => base.saturating_mul(2u32.saturating_pow(attempt)).min(max),
26        }
27    }
28}
29
30/// Retry configuration shared by the pipeline loop and source connectors.
31#[derive(Debug, Clone)]
32pub struct RetryPolicy {
33    /// Total attempts including the first (1 = no retry).
34    pub max_attempts: u32,
35    /// Backoff growth shape.
36    pub backoff: BackoffKind,
37    /// Base delay.
38    pub base: Duration,
39    /// Per-sleep cap (pre-jitter).
40    pub max: Duration,
41    /// Whether to apply `[0.5, 1.5)` jitter.
42    pub jitter: bool,
43    /// Which error classes are retried.
44    pub retry_on: RetryClassSet,
45}
46
47impl Default for RetryPolicy {
48    fn default() -> Self {
49        Self {
50            max_attempts: 5,
51            backoff: BackoffKind::Exponential,
52            base: Duration::from_millis(200),
53            max: Duration::from_secs(30),
54            jitter: true,
55            retry_on: RetryClassSet::default(),
56        }
57    }
58}
59
60impl RetryPolicy {
61    /// Whether `err` should be retried under this policy: it must classify into
62    /// a class that is in `retry_on`.
63    pub fn is_retriable(&self, err: &FaucetError) -> bool {
64        classify(err).is_some_and(|c| self.retry_on.contains(c))
65    }
66}
67
68/// Circuit-breaker tuning. Counts consecutive page-level write failures.
69#[derive(Debug, Clone, Copy)]
70pub struct CircuitBreakerConfig {
71    /// Consecutive exhausted-retry failures before the circuit opens.
72    pub consecutive_failures: u32,
73    /// Re-entry cooldown (honored by the orchestration layer, e.g. `schedule`).
74    pub cooldown: Duration,
75}
76
77/// What to do with a row that keeps failing after `max_row_attempts`.
78#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
79pub enum PoisonAction {
80    /// Route to the DLQ (requires a DLQ to be configured).
81    #[default]
82    Dlq,
83    /// Discard the row (counts + warns).
84    Drop,
85    /// Propagate the row error and abort the run.
86    Fail,
87}
88
89/// Poison-pill (per-row) policy.
90#[derive(Debug, Clone, Copy)]
91pub struct PoisonPolicy {
92    /// Per-row write attempts before applying `action`.
93    pub max_row_attempts: u32,
94    /// Terminal action for a persistently failing row.
95    pub action: PoisonAction,
96}
97
98/// The umbrella policy attached to a run.
99#[derive(Debug, Clone, Default)]
100pub struct ResiliencePolicy {
101    /// Retry/backoff applied to sink-write, flush, and state-store I/O.
102    pub retry: RetryPolicy,
103    /// Optional circuit breaker.
104    pub circuit_breaker: Option<CircuitBreakerConfig>,
105    /// Optional poison-pill row handling (DLQ path only).
106    pub poison: Option<PoisonPolicy>,
107}
108
109#[cfg(test)]
110mod tests {
111    use super::*;
112    use crate::resilience::classify::RetryClass;
113
114    #[test]
115    fn default_policy_retries_all_classes() {
116        let p = RetryPolicy::default();
117        assert_eq!(p.max_attempts, 5);
118        assert!(p.is_retriable(&FaucetError::HttpStatus {
119            status: 500,
120            url: "u".into(),
121            body: "".into()
122        }));
123        assert!(!p.is_retriable(&FaucetError::Auth("x".into())));
124    }
125
126    #[test]
127    fn restricted_retry_on_excludes_unlisted_classes() {
128        let p = RetryPolicy {
129            retry_on: RetryClassSet::from_iter([RetryClass::Http5xx]),
130            ..RetryPolicy::default()
131        };
132        assert!(p.is_retriable(&FaucetError::HttpStatus {
133            status: 500,
134            url: "u".into(),
135            body: "".into()
136        }));
137        // 429 classifies RateLimited, which is not opted in.
138        assert!(!p.is_retriable(&FaucetError::HttpStatus {
139            status: 429,
140            url: "u".into(),
141            body: "".into()
142        }));
143    }
144
145    #[test]
146    fn backoff_none_is_zero_fixed_is_constant_exp_grows() {
147        let base = Duration::from_millis(100);
148        let max = Duration::from_secs(10);
149        assert_eq!(BackoffKind::None.delay(base, max, 0), Duration::ZERO);
150        assert_eq!(BackoffKind::Fixed.delay(base, max, 3), base);
151        // Fixed is capped at max when base exceeds it.
152        assert_eq!(
153            BackoffKind::Fixed.delay(Duration::from_secs(20), Duration::from_secs(10), 0),
154            Duration::from_secs(10)
155        );
156        assert_eq!(BackoffKind::Exponential.delay(base, max, 0), base);
157        assert_eq!(BackoffKind::Exponential.delay(base, max, 2), base * 4);
158        // capped at max
159        assert_eq!(BackoffKind::Exponential.delay(base, max, 30), max);
160    }
161}
faucet_core/resilience/policy.rs

faucet_core/resilience/
policy.rs
