amalgam/
circuit.rs

1//! A circuit breaker for L2 / backplane operations.
2//!
3//! After a failure, the breaker "opens" for a fixed duration; while open, guarded
4//! operations are skipped without being attempted (avoiding hammering a known-bad
5//! dependency). A zero-duration breaker is permanently closed — this is
6//! FusionCache's default (`DistributedCacheCircuitBreakerDuration = 0`).
7
8use std::sync::atomic::{AtomicI64, Ordering};
9use std::time::Duration;
10
11use crate::time::{Timestamp, duration_to_ticks};
12
13const CLOSED: i64 = i64::MIN;
14
15/// A time-based circuit breaker. Cheap, lock-free, shareable.
16#[derive(Debug)]
17pub struct CircuitBreaker {
18    /// How long the breaker stays open after a trip; 0 ⇒ disabled (always closed).
19    open_duration_ticks: i64,
20    /// The tick at which the breaker re-closes, or [`CLOSED`] when closed.
21    reopen_at: AtomicI64,
22}
23
24impl CircuitBreaker {
25    /// Creates a breaker that opens for `open_duration` after a failure.
26    /// `Duration::ZERO` disables it (it stays permanently closed).
27    #[must_use]
28    pub fn new(open_duration: Duration) -> Self {
29        Self {
30            open_duration_ticks: duration_to_ticks(open_duration),
31            reopen_at: AtomicI64::new(CLOSED),
32        }
33    }
34
35    /// `true` if this breaker is disabled (zero open-duration).
36    #[must_use]
37    pub fn is_disabled(&self) -> bool {
38        self.open_duration_ticks == 0
39    }
40
41    /// `true` if the breaker is closed (operations may proceed) at `now`.
42    ///
43    /// If the open window has elapsed, the breaker auto-closes as a side effect.
44    #[must_use]
45    pub fn is_closed(&self, now: Timestamp) -> bool {
46        if self.is_disabled() {
47            return true;
48        }
49        let reopen = self.reopen_at.load(Ordering::Acquire);
50        if reopen == CLOSED {
51            return true;
52        }
53        if now.ticks() >= reopen {
54            // Window elapsed: auto-close.
55            let _ = self.reopen_at.compare_exchange(
56                reopen,
57                CLOSED,
58                Ordering::AcqRel,
59                Ordering::Relaxed,
60            );
61            true
62        } else {
63            false
64        }
65    }
66
67    /// Trips the breaker open until `now + open_duration`.
68    ///
69    /// Returns `true` if this call transitioned the breaker from closed to open
70    /// (so the caller can fire a `CircuitBreakerChange` event exactly once).
71    pub fn trip(&self, now: Timestamp) -> bool {
72        if self.is_disabled() {
73            return false;
74        }
75        let reopen = now.ticks().saturating_add(self.open_duration_ticks);
76        let previous = self.reopen_at.swap(reopen, Ordering::AcqRel);
77        previous == CLOSED
78    }
79
80    /// Forces the breaker closed (e.g. after a successful operation or a received
81    /// backplane message). Returns `true` if it transitioned from open to closed.
82    pub fn close(&self) -> bool {
83        let previous = self.reopen_at.swap(CLOSED, Ordering::AcqRel);
84        previous != CLOSED
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91
92    #[test]
93    fn disabled_breaker_is_always_closed() {
94        let cb = CircuitBreaker::new(Duration::ZERO);
95        assert!(cb.is_disabled());
96        assert!(cb.is_closed(Timestamp::from_ticks(0)));
97        assert!(!cb.trip(Timestamp::from_ticks(0)));
98        assert!(cb.is_closed(Timestamp::from_ticks(1_000_000)));
99    }
100
101    #[test]
102    fn trip_opens_then_auto_closes_after_window() {
103        let cb = CircuitBreaker::new(Duration::from_secs(10));
104        let t0 = Timestamp::from_ticks(0);
105        assert!(cb.is_closed(t0));
106        assert!(cb.trip(t0)); // closed -> open transition
107        assert!(!cb.trip(t0)); // already open, no transition
108        assert!(!cb.is_closed(t0.saturating_add(Duration::from_secs(5))));
109        // After the window, it auto-closes.
110        assert!(cb.is_closed(t0.saturating_add(Duration::from_secs(11))));
111    }
112
113    #[test]
114    fn close_reports_transition() {
115        let cb = CircuitBreaker::new(Duration::from_secs(10));
116        let t0 = Timestamp::from_ticks(0);
117        cb.trip(t0);
118        assert!(cb.close()); // open -> closed
119        assert!(!cb.close()); // already closed
120    }
121}
amalgam/circuit.rs

amalgam/
circuit.rs
