rws_server/circuit_breaker/
mod.rs

1//! Circuit breaker state machine and retry middleware.
2//!
3//! # Circuit breaker
4//!
5//! [`CircuitBreaker`] tracks per-backend failure counts and transitions through
6//! three states:
7//!
8//! * **Closed** — the backend is healthy; failures are counted.  When the count
9//!   reaches `failure_threshold` the breaker moves to **Open**.
10//! * **Open** — the backend is considered unhealthy; all requests are rejected
11//!   immediately (no TCP connection is attempted).  After `recovery` seconds the
12//!   breaker moves to **HalfOpen**.
13//! * **HalfOpen** — one probe request is let through.  On success the breaker
14//!   closes; on failure it re-opens and the recovery timer resets.
15//!
16//! # Retry middleware
17//!
18//! [`RetryLayer`] wraps any [`Application`] and re-dispatches the request when
19//! the inner app returns one of the configured status codes (default: 502, 503,
20//! 504) up to `max_retries` additional times.
21//!
22//! # Persistence
23//!
24//! [`CircuitBreaker`] keeps state in a plain in-process `HashMap` — a restart
25//! (or a deploy) resets every backend back to `Closed`, so a backend that
26//! tripped the breaker moments before a restart looks healthy again
27//! immediately, and may cascade failures again before anything notices.
28//! [`RedisCircuitBreaker`] has the same state machine and the same
29//! `is_available`/`record_success`/`record_failure`/`reset`/`state` shape, but
30//! persists each backend's state in Redis (via the same hand-rolled RESP
31//! client [`crate::rate_limit::RedisRateLimiter`] and
32//! [`crate::session::RedisSessionStore`] use) — surviving a restart, and
33//! shared across every `rws` instance pointed at that Redis server.
34//!
35//! # Example
36//!
37//! ```rust,no_run
38//! # extern crate rws_server as rust_web_server;
39//! use rust_web_server::app::App;
40//! use rust_web_server::core::New;
41//! use rust_web_server::circuit_breaker::RetryLayer;
42//! use rust_web_server::middleware::WithMiddleware;
43//!
44//! let app = WithMiddleware::new(App::new())
45//!     .wrap(RetryLayer::new().max_retries(2));
46//! ```
47
48#[cfg(test)]
49mod tests;
50
51use std::collections::HashMap;
52use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
53use std::sync::{Arc, Mutex, OnceLock};
54use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
55
56use crate::application::Application;
57use crate::middleware::Middleware;
58use crate::redis_protocol::{RespConn, RespReply};
59use crate::request::Request;
60use crate::response::Response;
61use crate::server::ConnectionInfo;
62
63// ── BreakerState ─────────────────────────────────────────────────────────────
64
65/// Current state of a single backend's circuit breaker.
66#[derive(Debug, Clone, PartialEq)]
67pub enum BreakerState {
68    /// Healthy — requests are forwarded and failures are counted.
69    Closed,
70    /// Unhealthy — requests are rejected until the recovery window expires.
71    Open,
72    /// Probing — one request is let through to test backend health.
73    HalfOpen,
74}
75
76// ── BackendEntry ──────────────────────────────────────────────────────────────
77
78struct BackendEntry {
79    state: BreakerState,
80    failures: u32,
81    opened_at: Option<Instant>,
82    /// Probes currently let through while `state == HalfOpen`. Capped by
83    /// [`CircuitBreaker::max_half_open_probes`] so a burst of concurrent
84    /// requests arriving the instant a backend transitions to `HalfOpen`
85    /// doesn't all land as "the" trial request at once.
86    half_open_in_flight: u32,
87}
88
89impl BackendEntry {
90    fn new() -> Self {
91        Self { state: BreakerState::Closed, failures: 0, opened_at: None, half_open_in_flight: 0 }
92    }
93}
94
95// ── CircuitBreaker ────────────────────────────────────────────────────────────
96
97/// Per-backend circuit breaker.
98///
99/// # Concurrency
100///
101/// `CircuitBreaker` is not `Sync` on its own — wrap it in a [`Mutex`] for
102/// shared use across threads (see [`global()`]).
103pub struct CircuitBreaker {
104    backends: HashMap<String, BackendEntry>,
105    failure_threshold: u32,
106    recovery: Duration,
107    max_half_open_probes: u32,
108}
109
110impl CircuitBreaker {
111    /// Create a new circuit breaker.
112    ///
113    /// * `failure_threshold` — how many consecutive failures open the circuit.
114    /// * `recovery_secs` — how long the circuit stays Open before testing again.
115    ///
116    /// Defaults to letting exactly one probe through while `HalfOpen` — see
117    /// [`CircuitBreaker::max_half_open_probes`] to change that.
118    pub fn new(failure_threshold: u32, recovery_secs: u64) -> Self {
119        Self {
120            backends: HashMap::new(),
121            failure_threshold,
122            recovery: Duration::from_secs(recovery_secs),
123            max_half_open_probes: 1,
124        }
125    }
126
127    /// Override how many concurrent probe requests are let through while a
128    /// backend is `HalfOpen` (default: 1). Chainable — call before the
129    /// breaker is put behind a shared `Mutex`/`Arc`.
130    pub fn max_half_open_probes(mut self, n: u32) -> Self {
131        self.max_half_open_probes = n.max(1);
132        self
133    }
134
135    /// Returns `true` if a request should be forwarded to `backend`.
136    ///
137    /// Transitions `Open → HalfOpen` when the recovery window has elapsed.
138    /// While `HalfOpen`, at most `max_half_open_probes` concurrent calls
139    /// return `true` — further calls are rejected like `Open` until the
140    /// in-flight probe(s) resolve via `record_success`/`record_failure`.
141    pub fn is_available(&mut self, backend: &str) -> bool {
142        let max_probes = self.max_half_open_probes;
143        let entry = self.backends.entry(backend.to_string()).or_insert_with(BackendEntry::new);
144        match entry.state {
145            BreakerState::Closed => true,
146            BreakerState::HalfOpen => {
147                if entry.half_open_in_flight < max_probes {
148                    entry.half_open_in_flight += 1;
149                    true
150                } else {
151                    false
152                }
153            }
154            BreakerState::Open => {
155                if let Some(opened_at) = entry.opened_at {
156                    if opened_at.elapsed() >= self.recovery {
157                        entry.state = BreakerState::HalfOpen;
158                        entry.opened_at = None;
159                        entry.half_open_in_flight = 1;
160                        return true;
161                    }
162                }
163                false
164            }
165        }
166    }
167
168    /// Record a successful response for `backend`.
169    ///
170    /// Transitions `HalfOpen → Closed` and resets the failure counter and
171    /// the in-flight half-open probe count.
172    pub fn record_success(&mut self, backend: &str) {
173        let entry = self.backends.entry(backend.to_string()).or_insert_with(BackendEntry::new);
174        entry.state = BreakerState::Closed;
175        entry.failures = 0;
176        entry.opened_at = None;
177        entry.half_open_in_flight = 0;
178    }
179
180    /// Record a failed response for `backend`.
181    ///
182    /// In `Closed` state, increments the counter and opens the circuit when
183    /// `failure_threshold` is reached.  In `HalfOpen` state, immediately
184    /// re-opens the circuit and resets the recovery timer.
185    pub fn record_failure(&mut self, backend: &str) {
186        let threshold = self.failure_threshold;
187        let entry = self.backends.entry(backend.to_string()).or_insert_with(BackendEntry::new);
188        match entry.state {
189            BreakerState::Closed => {
190                entry.failures += 1;
191                if entry.failures >= threshold {
192                    entry.state = BreakerState::Open;
193                    entry.opened_at = Some(Instant::now());
194                }
195            }
196            BreakerState::HalfOpen => {
197                entry.state = BreakerState::Open;
198                entry.opened_at = Some(Instant::now());
199                entry.half_open_in_flight = 0;
200            }
201            BreakerState::Open => {
202                // Already open; refresh the timer.
203                entry.opened_at = Some(Instant::now());
204            }
205        }
206    }
207
208    /// Reset `backend` to `Closed` with zero failures.
209    pub fn reset(&mut self, backend: &str) {
210        let entry = self.backends.entry(backend.to_string()).or_insert_with(BackendEntry::new);
211        entry.state = BreakerState::Closed;
212        entry.failures = 0;
213        entry.opened_at = None;
214        entry.half_open_in_flight = 0;
215    }
216
217    /// Return the current state for `backend` (defaults to `Closed` if unseen).
218    pub fn state(&self, backend: &str) -> BreakerState {
219        self.backends
220            .get(backend)
221            .map(|e| e.state.clone())
222            .unwrap_or(BreakerState::Closed)
223    }
224
225    /// Snapshot of every backend this breaker has ever seen, with its
226    /// current state. Powers the `rws_circuit_breaker_state{backend}`
227    /// metric (`crate::metrics::prometheus_text`) — there is no
228    /// `RedisCircuitBreaker` equivalent, since enumerating keys isn't
229    /// something the minimal hand-rolled RESP client supports (no
230    /// `SCAN`/`KEYS` array-reply decoding).
231    pub fn all_states(&self) -> Vec<(String, BreakerState)> {
232        self.backends.iter().map(|(k, v)| (k.clone(), v.state.clone())).collect()
233    }
234}
235
236// ── global() ─────────────────────────────────────────────────────────────────
237
238static GLOBAL_BREAKER: OnceLock<Mutex<CircuitBreaker>> = OnceLock::new();
239
240/// Return the process-wide default circuit breaker (threshold=5, recovery=30 s).
241///
242/// Acquire the mutex before calling any `CircuitBreaker` method:
243///
244/// ```rust
245/// # extern crate rws_server as rust_web_server;
246/// use rust_web_server::circuit_breaker;
247///
248/// let available = circuit_breaker::global().lock().unwrap().is_available("backend-a:8080");
249/// ```
250pub fn global() -> &'static Mutex<CircuitBreaker> {
251    GLOBAL_BREAKER.get_or_init(|| Mutex::new(CircuitBreaker::new(5, 30)))
252}
253
254// ── RedisCircuitBreaker ─────────────────────────────────────────────────────────
255
256/// Per-backend circuit breaker, persisted in Redis.
257///
258/// Same Closed → Open → HalfOpen state machine as [`CircuitBreaker`], and the
259/// same method names, but every operation is a synchronous Redis round trip
260/// (via the shared [`crate::redis_protocol`] RESP client) instead of an
261/// in-memory `HashMap` update — so state survives a process restart, and is
262/// shared across every `rws` instance pointed at the same Redis server.
263///
264/// # Why Redis, not the model layer
265///
266/// The model layer (`DbPool`, `sqlx`) is `async fn`-only, while
267/// `CircuitBreaker`'s methods and `Middleware::handle` (what [`RetryLayer`]
268/// implements) are both synchronous — the identical async/sync mismatch that
269/// left `SqliteRateLimiter` unbuilt after [`crate::rate_limit::RedisRateLimiter`]
270/// shipped. Redis, reached over a plain blocking `TcpStream`, stays fully
271/// synchronous and drops into the same call sites `CircuitBreaker` already
272/// has, with no new Cargo dependency.
273///
274/// # Consistency
275///
276/// Each operation is a read-then-write (`GET` then `SET`) against one Redis
277/// key per backend — not a single atomic command. Two `rws` instances racing
278/// to record a failure for the same backend at the same instant can lose one
279/// of the two increments. This is a deliberate simplification: unlike a rate
280/// limit (a hard resource/security boundary, where `RedisRateLimiter` uses
281/// genuinely atomic `INCR` for exactly this reason), a circuit breaker is a
282/// self-healing heuristic where opening one failure late — or one request
283/// later than a perfectly-synchronized count would — has no real consequence.
284///
285/// # Example
286///
287/// ```rust,no_run
288/// # extern crate rws_server as rust_web_server;
289/// use rust_web_server::circuit_breaker::RedisCircuitBreaker;
290///
291/// let breaker = RedisCircuitBreaker::new("127.0.0.1:6379", None, 5, 30);
292///
293/// match breaker.is_available("backend-a:8080") {
294///     Ok(true) => { /* forward the request */ }
295///     Ok(false) => { /* short-circuit — return 503 without contacting the backend */ }
296///     Err(e) => { /* Redis unreachable — decide fail-open vs fail-closed yourself */ }
297/// }
298/// ```
299pub struct RedisCircuitBreaker {
300    conn: Arc<RespConn>,
301    failure_threshold: AtomicU32,
302    recovery_secs: AtomicU64,
303    max_half_open_probes: AtomicU32,
304}
305
306impl Clone for RedisCircuitBreaker {
307    fn clone(&self) -> Self {
308        RedisCircuitBreaker {
309            conn: Arc::clone(&self.conn),
310            failure_threshold: AtomicU32::new(self.failure_threshold.load(Ordering::Relaxed)),
311            recovery_secs: AtomicU64::new(self.recovery_secs.load(Ordering::Relaxed)),
312            max_half_open_probes: AtomicU32::new(self.max_half_open_probes.load(Ordering::Relaxed)),
313        }
314    }
315}
316
317impl RedisCircuitBreaker {
318    /// Create a breaker that connects to `addr` (e.g. `"127.0.0.1:6379"`).
319    /// `password` is passed to Redis `AUTH` if `Some`. Defaults to letting
320    /// exactly one probe through while `HalfOpen` — see
321    /// [`RedisCircuitBreaker::set_max_half_open_probes`] to change that.
322    pub fn new(addr: impl Into<String>, password: Option<String>, failure_threshold: u32, recovery_secs: u64) -> Self {
323        RedisCircuitBreaker {
324            conn: Arc::new(RespConn::new(addr, password)),
325            failure_threshold: AtomicU32::new(failure_threshold),
326            recovery_secs: AtomicU64::new(recovery_secs),
327            max_half_open_probes: AtomicU32::new(1),
328        }
329    }
330
331    /// Build a breaker from environment variables:
332    /// - `RWS_REDIS_HOST` (default `127.0.0.1`)
333    /// - `RWS_REDIS_PORT` (default `6379`)
334    /// - `RWS_REDIS_PASSWORD` (optional)
335    /// - `RWS_CONFIG_CIRCUIT_BREAKER_FAILURE_THRESHOLD` (default `5`)
336    /// - `RWS_CONFIG_CIRCUIT_BREAKER_RECOVERY_SECS` (default `30`)
337    /// - `RWS_CONFIG_CIRCUIT_BREAKER_MAX_HALF_OPEN_PROBES` (default `1`)
338    pub fn from_env() -> Self {
339        let host = std::env::var("RWS_REDIS_HOST").unwrap_or_else(|_| "127.0.0.1".into());
340        let port = std::env::var("RWS_REDIS_PORT").unwrap_or_else(|_| "6379".into());
341        let addr = format!("{}:{}", host, port);
342        let password = std::env::var("RWS_REDIS_PASSWORD").ok();
343        let failure_threshold = std::env::var("RWS_CONFIG_CIRCUIT_BREAKER_FAILURE_THRESHOLD")
344            .ok()
345            .and_then(|v| v.parse().ok())
346            .unwrap_or(5);
347        let recovery_secs = std::env::var("RWS_CONFIG_CIRCUIT_BREAKER_RECOVERY_SECS")
348            .ok()
349            .and_then(|v| v.parse().ok())
350            .unwrap_or(30);
351        let breaker = Self::new(addr, password, failure_threshold, recovery_secs);
352        if let Some(n) = std::env::var("RWS_CONFIG_CIRCUIT_BREAKER_MAX_HALF_OPEN_PROBES").ok().and_then(|v| v.parse().ok()) {
353            breaker.set_max_half_open_probes(n);
354        }
355        breaker
356    }
357
358    /// Update the thresholds on a live breaker without restarting.
359    pub fn set_limits(&self, failure_threshold: u32, recovery_secs: u64) {
360        self.failure_threshold.store(failure_threshold, Ordering::Relaxed);
361        self.recovery_secs.store(recovery_secs, Ordering::Relaxed);
362    }
363
364    /// Override how many concurrent probe requests are let through while a
365    /// backend is `HalfOpen` (default: 1) — on a live breaker, no restart
366    /// required.
367    pub fn set_max_half_open_probes(&self, n: u32) {
368        self.max_half_open_probes.store(n.max(1), Ordering::Relaxed);
369    }
370
371    fn redis_key(backend: &str) -> Vec<u8> {
372        format!("rws:cb:{}", backend).into_bytes()
373    }
374
375    fn load(&self, backend: &str) -> std::io::Result<(BreakerState, u32, u64, u32)> {
376        match self.conn.cmd(&[b"GET", &Self::redis_key(backend)])? {
377            RespReply::Bulk(Some(bytes)) => Ok(decode_entry(&bytes)),
378            _ => Ok((BreakerState::Closed, 0, 0, 0)),
379        }
380    }
381
382    fn store(&self, backend: &str, state: &BreakerState, failures: u32, opened_at: u64, half_open_in_flight: u32) -> std::io::Result<()> {
383        let encoded = encode_entry(state, failures, opened_at, half_open_in_flight);
384        self.conn.cmd(&[b"SET", &Self::redis_key(backend), encoded.as_bytes()])?;
385        Ok(())
386    }
387
388    /// Returns `Ok(true)` if a request should be forwarded to `backend`.
389    ///
390    /// Transitions `Open → HalfOpen` when the recovery window has elapsed.
391    /// While `HalfOpen`, at most `max_half_open_probes` concurrent calls
392    /// return `Ok(true)` — this is a read-then-write against one Redis key
393    /// (not atomic), the same deliberate simplification already documented
394    /// above for this type, so a race between two instances can very rarely
395    /// let one extra probe through; it cannot let an unbounded burst through.
396    /// Returns `Err` if Redis is unreachable — callers decide whether that
397    /// means fail open (treat as available) or fail closed (treat as not).
398    pub fn is_available(&self, backend: &str) -> std::io::Result<bool> {
399        let (state, failures, opened_at, half_open_in_flight) = self.load(backend)?;
400        match state {
401            BreakerState::Closed => Ok(true),
402            BreakerState::HalfOpen => {
403                let max_probes = self.max_half_open_probes.load(Ordering::Relaxed);
404                if half_open_in_flight < max_probes {
405                    self.store(backend, &BreakerState::HalfOpen, failures, opened_at, half_open_in_flight + 1)?;
406                    Ok(true)
407                } else {
408                    Ok(false)
409                }
410            }
411            BreakerState::Open => {
412                let recovery = self.recovery_secs.load(Ordering::Relaxed);
413                if now_unix().saturating_sub(opened_at) >= recovery {
414                    self.store(backend, &BreakerState::HalfOpen, failures, 0, 1)?;
415                    Ok(true)
416                } else {
417                    Ok(false)
418                }
419            }
420        }
421    }
422
423    /// Record a successful response for `backend`.
424    ///
425    /// Transitions `HalfOpen → Closed` and resets the failure counter and
426    /// the in-flight half-open probe count.
427    pub fn record_success(&self, backend: &str) -> std::io::Result<()> {
428        self.store(backend, &BreakerState::Closed, 0, 0, 0)
429    }
430
431    /// Record a failed response for `backend`.
432    ///
433    /// In `Closed` state, increments the counter and opens the circuit when
434    /// `failure_threshold` is reached. In `HalfOpen` state, immediately
435    /// re-opens the circuit and resets the recovery timer.
436    pub fn record_failure(&self, backend: &str) -> std::io::Result<()> {
437        let (state, failures, _, _) = self.load(backend)?;
438        match state {
439            BreakerState::Closed => {
440                let failures = failures + 1;
441                if failures >= self.failure_threshold.load(Ordering::Relaxed) {
442                    self.store(backend, &BreakerState::Open, failures, now_unix(), 0)
443                } else {
444                    self.store(backend, &BreakerState::Closed, failures, 0, 0)
445                }
446            }
447            BreakerState::HalfOpen | BreakerState::Open => {
448                self.store(backend, &BreakerState::Open, failures, now_unix(), 0)
449            }
450        }
451    }
452
453    /// Reset `backend` to `Closed` with zero failures.
454    pub fn reset(&self, backend: &str) -> std::io::Result<()> {
455        self.conn.cmd(&[b"DEL", &Self::redis_key(backend)])?;
456        Ok(())
457    }
458
459    /// Return the current state for `backend` (defaults to `Closed` if unseen).
460    pub fn state(&self, backend: &str) -> std::io::Result<BreakerState> {
461        Ok(self.load(backend)?.0)
462    }
463}
464
465fn now_unix() -> u64 {
466    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
467}
468
469/// `"state|failures|opened_at|half_open_in_flight"` — a plain-string encoding
470/// chosen so a backend's whole entry fits in one Redis key read via
471/// `GET`/written via `SET`, rather than a hash needing `HGETALL` (which
472/// `redis_protocol`'s minimal RESP client doesn't decode — it only handles
473/// simple/bulk/integer replies, not arrays).
474fn encode_entry(state: &BreakerState, failures: u32, opened_at: u64, half_open_in_flight: u32) -> String {
475    let state_str = match state {
476        BreakerState::Closed => "closed",
477        BreakerState::Open => "open",
478        BreakerState::HalfOpen => "half_open",
479    };
480    format!("{}|{}|{}|{}", state_str, failures, opened_at, half_open_in_flight)
481}
482
483fn decode_entry(raw: &[u8]) -> (BreakerState, u32, u64, u32) {
484    let text = String::from_utf8_lossy(raw);
485    let mut parts = text.splitn(4, '|');
486    let state = match parts.next() {
487        Some("open") => BreakerState::Open,
488        Some("half_open") => BreakerState::HalfOpen,
489        _ => BreakerState::Closed,
490    };
491    let failures = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0);
492    let opened_at = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0);
493    let half_open_in_flight = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0);
494    (state, failures, opened_at, half_open_in_flight)
495}
496
497// ── Breaker trait (ReverseProxy auto-wiring) ─────────────────────────────────
498
499/// Object-safe interface implemented by both `Mutex<CircuitBreaker>` and
500/// [`RedisCircuitBreaker`], so a consumer like
501/// `proxy::ReverseProxy::with_circuit_breaker` can accept either
502/// breaker kind through one `Arc<dyn Breaker>` without a generic parameter
503/// leaking into its own type.
504///
505/// `RedisCircuitBreaker`'s `Err` (Redis unreachable) is treated as *available*
506/// — failing open, not closed. A breaker that can't be reached is a worse
507/// single point of failure than occasionally skipping the check it exists to
508/// perform; callers who need fail-closed semantics should call
509/// `RedisCircuitBreaker::is_available` directly and decide for themselves, as
510/// its own docs already describe.
511pub trait Breaker: Send + Sync {
512    /// Returns `true` if a request should be forwarded to `backend`.
513    fn is_available(&self, backend: &str) -> bool;
514    /// Record a successful response for `backend`.
515    fn record_success(&self, backend: &str);
516    /// Record a failed response for `backend`.
517    fn record_failure(&self, backend: &str);
518}
519
520impl Breaker for Mutex<CircuitBreaker> {
521    fn is_available(&self, backend: &str) -> bool {
522        self.lock().unwrap().is_available(backend)
523    }
524    fn record_success(&self, backend: &str) {
525        self.lock().unwrap().record_success(backend);
526    }
527    fn record_failure(&self, backend: &str) {
528        self.lock().unwrap().record_failure(backend);
529    }
530}
531
532impl Breaker for RedisCircuitBreaker {
533    fn is_available(&self, backend: &str) -> bool {
534        RedisCircuitBreaker::is_available(self, backend).unwrap_or(true)
535    }
536    fn record_success(&self, backend: &str) {
537        let _ = RedisCircuitBreaker::record_success(self, backend);
538    }
539    fn record_failure(&self, backend: &str) {
540        let _ = RedisCircuitBreaker::record_failure(self, backend);
541    }
542}
543
544/// Lets `circuit_breaker::global()` (which returns `&'static Mutex<CircuitBreaker>`,
545/// not an owned `Mutex<CircuitBreaker>`) be used directly as a `Breaker`:
546/// `Arc::new(circuit_breaker::global())` then coerces to `Arc<dyn Breaker>`.
547impl<T: Breaker + ?Sized> Breaker for &T {
548    fn is_available(&self, backend: &str) -> bool {
549        (**self).is_available(backend)
550    }
551    fn record_success(&self, backend: &str) {
552        (**self).record_success(backend);
553    }
554    fn record_failure(&self, backend: &str) {
555        (**self).record_failure(backend);
556    }
557}
558
559// ── RetryLayer ────────────────────────────────────────────────────────────────
560
561/// Retry middleware.
562///
563/// When the inner application returns a response whose status code is in the
564/// configured list, the request is re-dispatched up to `max_retries` additional
565/// times.  If all attempts return a retryable status the last response is
566/// returned as-is.
567pub struct RetryLayer {
568    max_retries: u32,
569    retry_on: Vec<i16>,
570}
571
572impl RetryLayer {
573    /// Create a `RetryLayer` with defaults: retry on 502, 503, 504 up to 3 times.
574    pub fn new() -> Self {
575        Self { max_retries: 3, retry_on: vec![502, 503, 504] }
576    }
577
578    /// Override the maximum number of retry attempts.
579    pub fn max_retries(mut self, n: u32) -> Self {
580        self.max_retries = n;
581        self
582    }
583
584    /// Override the set of status codes that trigger a retry.
585    pub fn retry_on(mut self, codes: Vec<i16>) -> Self {
586        self.retry_on = codes;
587        self
588    }
589}
590
591impl Default for RetryLayer {
592    fn default() -> Self {
593        Self::new()
594    }
595}
596
597impl Middleware for RetryLayer {
598    fn handle(
599        &self,
600        request: &Request,
601        connection: &ConnectionInfo,
602        next: &dyn Application,
603    ) -> Result<Response, String> {
604        let mut response = next.execute(request, connection)?;
605        let mut attempts = 0u32;
606        while attempts < self.max_retries && self.retry_on.contains(&response.status_code) {
607            response = next.execute(request, connection)?;
608            attempts += 1;
609        }
610        Ok(response)
611    }
612}
rws_server/circuit_breaker/mod.rs

rws_server/circuit_breaker/
mod.rs
