brainos_backends/
resilience.rs

1//! Resilience helpers for HTTP backends — pairs the canonical
2//! [`resilience::CircuitBreaker`] with retry + reqwest-aware
3//! transient-error detection. Re-exports `CircuitBreaker` so existing
4//! backend code keeps the `use crate::resilience::CircuitBreaker` import
5//! shape.
6//!
7//! Metrics: each HTTP backend wires its breaker to a [`MetricsObserver`]
8//! so `BreakerStateChange` events drive the per-subsystem
9//! `inc_circuit_open` / `inc_circuit_reset` counters exported on
10//! `/metrics`. The observer also rebroadcasts via a small local channel
11//! so anything that does subscribe — today nothing — still sees the
12//! events.
13
14use std::sync::Arc;
15use std::time::Duration;
16
17use async_trait::async_trait;
18use metrics::SubsystemMetrics;
19use observe::{BrainEvent, ObserveError, Observer};
20use resilience::BreakerConfig;
21use tokio::sync::broadcast;
22
23pub use resilience::{Breaker, CircuitBreaker};
24
25/// Default capacity for the metrics-observer rebroadcast channel. Tiny
26/// because nothing in-tree actually subscribes; the value exists only to
27/// satisfy `Observer::subscribe`.
28const REBROADCAST_CAPACITY: usize = 16;
29
30/// `Observer` that converts `BreakerStateChange` events from a circuit
31/// breaker into backend-side `SubsystemMetrics` counters. Anything else
32/// is rebroadcast unchanged so the observer composes cleanly if another
33/// listener is ever attached.
34pub struct MetricsObserver {
35    metrics: Arc<SubsystemMetrics>,
36    tx: broadcast::Sender<BrainEvent>,
37}
38
39impl MetricsObserver {
40    pub fn new(metrics: Arc<SubsystemMetrics>) -> Arc<Self> {
41        let (tx, _) = broadcast::channel(REBROADCAST_CAPACITY);
42        Arc::new(Self { metrics, tx })
43    }
44}
45
46#[async_trait]
47impl Observer for MetricsObserver {
48    async fn publish(&self, ev: BrainEvent) -> Result<(), ObserveError> {
49        if let BrainEvent::BreakerStateChange { to, .. } = &ev {
50            match to.as_str() {
51                "open" => self.metrics.inc_circuit_open(),
52                "closed" => self.metrics.inc_circuit_reset(),
53                _ => {}
54            }
55        }
56        // send() returns Err when there are no receivers; not a problem.
57        let _ = self.tx.send(ev);
58        Ok(())
59    }
60
61    fn subscribe(&self) -> broadcast::Receiver<BrainEvent> {
62        self.tx.subscribe()
63    }
64}
65
66/// Build a `CircuitBreaker` for an HTTP backend with a familiar
67/// `(name, failure_threshold, cooldown_secs)` shape and an optional
68/// metrics handle. `cooldown_secs` becomes
69/// [`BreakerConfig::open_duration`]; the half-open probe needs one
70/// success to close, matching the prior atomic-breaker semantics.
71pub fn http_breaker(
72    name: &str,
73    failure_threshold: u32,
74    cooldown_secs: u64,
75    metrics: Option<Arc<SubsystemMetrics>>,
76) -> CircuitBreaker {
77    let cfg = BreakerConfig {
78        failure_threshold,
79        open_duration: Duration::from_secs(cooldown_secs),
80        half_open_required_successes: 1,
81    };
82    let mut cb = CircuitBreaker::new(name, cfg);
83    if let Some(m) = metrics {
84        cb = cb.with_observer(MetricsObserver::new(m));
85    }
86    cb
87}
88
89/// Returns true if the HTTP status is transient (worth retrying):
90/// any 5xx, 408 Request Timeout, or 429 Too Many Requests.
91fn is_transient_status(status: reqwest::StatusCode) -> bool {
92    status.is_server_error()
93        || status == reqwest::StatusCode::TOO_MANY_REQUESTS // 429
94        || status == reqwest::StatusCode::REQUEST_TIMEOUT // 408
95}
96
97/// Returns true if the request-error is transient (worth retrying).
98/// Treats timeouts and connect errors as transient; for status-bearing
99/// errors, defers to [`is_transient_status`].
100fn is_transient(err: &reqwest::Error) -> bool {
101    if err.is_timeout() || err.is_connect() {
102        return true;
103    }
104    err.status().is_some_and(is_transient_status)
105}
106
107/// Failure from [`resilient_send`]. Kept local to `backends` so this
108/// low-level HTTP resilience primitive does not depend on the higher-level
109/// `cortex::actions::ActionError` (resilience sits *below* actions in the
110/// layering). Callers map it into their own domain error — the action
111/// backends do so via `ActionError::ExecutionFailed(e.to_string())`, which
112/// preserves the original message verbatim.
113#[derive(Debug, thiserror::Error)]
114pub enum ResilientSendError {
115    /// The circuit breaker for this backend is open; the call was not issued.
116    #[error("{0} circuit breaker is open — backend disabled until cooldown elapses")]
117    BreakerOpen(String),
118    /// All attempts failed (transient retries exhausted, or a non-transient
119    /// error). Carries the last underlying error message.
120    #[error("{0}")]
121    Failed(String),
122}
123
124/// Send an HTTP request with retry + circuit breaker.
125pub async fn resilient_send<F>(
126    build_request: F,
127    circuit_breaker: &impl Breaker,
128    max_retries: u32,
129    retry_base_ms: u64,
130) -> Result<reqwest::Response, ResilientSendError>
131where
132    F: Fn() -> reqwest::RequestBuilder,
133{
134    if circuit_breaker.is_open().await {
135        return Err(ResilientSendError::BreakerOpen(
136            circuit_breaker.tool_id().to_string(),
137        ));
138    }
139
140    let attempts = 1 + max_retries;
141    let mut last_err = None;
142
143    for attempt in 0..attempts {
144        if attempt > 0 {
145            let delay = retry_base_ms * (1u64 << (attempt - 1).min(5));
146            tokio::time::sleep(tokio::time::Duration::from_millis(delay)).await;
147        }
148
149        match build_request().send().await {
150            Ok(response) => {
151                if response.status().is_success() || !is_transient_status(response.status()) {
152                    circuit_breaker.record_success().await;
153                    return Ok(response);
154                }
155                let status = response.status();
156                tracing::debug!(
157                    backend = %circuit_breaker.tool_id(),
158                    attempt = attempt + 1,
159                    status = %status,
160                    "Transient HTTP error, will retry"
161                );
162                last_err = Some(format!("HTTP {}", status));
163            }
164            Err(e) => {
165                if !is_transient(&e) {
166                    circuit_breaker.record_failure().await;
167                    return Err(ResilientSendError::Failed(e.to_string()));
168                }
169                tracing::debug!(
170                    backend = %circuit_breaker.tool_id(),
171                    attempt = attempt + 1,
172                    error = %e,
173                    "Transient error, will retry"
174                );
175                last_err = Some(e.to_string());
176            }
177        }
178    }
179
180    circuit_breaker.record_failure().await;
181    Err(ResilientSendError::Failed(last_err.unwrap_or_else(|| {
182        "all retry attempts exhausted".to_string()
183    })))
184}
185
186#[cfg(test)]
187mod tests {
188    use super::*;
189
190    fn cfg(threshold: u32, cooldown_secs: u64) -> BreakerConfig {
191        BreakerConfig {
192            failure_threshold: threshold,
193            open_duration: Duration::from_secs(cooldown_secs),
194            half_open_required_successes: 1,
195        }
196    }
197
198    #[tokio::test]
199    async fn http_breaker_closed_by_default() {
200        let cb = http_breaker("test", 3, 60, None);
201        assert!(!cb.is_open().await);
202    }
203
204    #[tokio::test]
205    async fn breaker_opens_after_threshold() {
206        let cb = CircuitBreaker::new("test", cfg(3, 60));
207        cb.record_failure().await;
208        cb.record_failure().await;
209        assert!(
210            !cb.is_open().await,
211            "should still be closed below threshold"
212        );
213        cb.record_failure().await;
214        assert!(cb.is_open().await, "should be open at threshold");
215    }
216
217    #[tokio::test]
218    async fn breaker_resets_on_success() {
219        let cb = CircuitBreaker::new("test", cfg(3, 60));
220        cb.record_failure().await;
221        cb.record_failure().await;
222        cb.record_success().await;
223        assert!(!cb.is_open().await);
224        cb.record_failure().await;
225        cb.record_failure().await;
226        assert!(!cb.is_open().await, "should be closed — counter was reset");
227    }
228
229    #[tokio::test]
230    async fn breaker_half_open_after_cooldown() {
231        let cb = CircuitBreaker::new("test", cfg(2, 0));
232        cb.record_failure().await;
233        cb.record_failure().await;
234        // open_duration == 0 means the next is_open() call transitions to
235        // HalfOpen and returns false.
236        assert!(
237            !cb.is_open().await,
238            "should be half-open after zero cooldown"
239        );
240    }
241}
brainos_backends/resilience.rs

brainos_backends/
resilience.rs
