rust_zero_core/
breaker.rs

1use std::{
2    collections::VecDeque,
3    convert::Infallible,
4    fmt,
5    future::Future,
6    sync::{
7        atomic::{AtomicU64, Ordering},
8        Arc, Mutex,
9    },
10    time::{Duration, Instant},
11};
12
13/// Circuit-breaker behavior for an unreliable downstream dependency.
14#[derive(Debug, Clone, Copy)]
15pub struct CircuitBreakerConfig {
16    pub max_failures: u32,
17    pub reset_timeout: Duration,
18    pub half_open_max_calls: u32,
19    pub policy: CircuitBreakerPolicy,
20}
21
22impl CircuitBreakerConfig {
23    /// Builds the original consecutive-failure policy.
24    pub fn new(max_failures: u32, reset_timeout: Duration) -> Self {
25        assert!(
26            max_failures > 0,
27            "maximum failures must be greater than zero"
28        );
29        assert!(
30            !reset_timeout.is_zero(),
31            "reset timeout must be greater than zero"
32        );
33
34        Self {
35            max_failures,
36            reset_timeout,
37            half_open_max_calls: 1,
38            policy: CircuitBreakerPolicy::Consecutive,
39        }
40    }
41
42    /// Builds a rolling adaptive breaker. Consecutive-policy fields remain available so callers
43    /// can switch policies without rebuilding the rest of their client configuration.
44    pub fn rolling(config: RollingCircuitBreakerConfig) -> Self {
45        Self {
46            max_failures: 5,
47            reset_timeout: Duration::from_secs(30),
48            half_open_max_calls: 1,
49            policy: CircuitBreakerPolicy::Rolling(config),
50        }
51    }
52
53    pub fn with_half_open_max_calls(mut self, calls: u32) -> Self {
54        assert!(calls > 0, "half-open calls must be greater than zero");
55        self.half_open_max_calls = calls;
56        self
57    }
58
59    pub fn with_policy(mut self, policy: CircuitBreakerPolicy) -> Self {
60        self.policy = policy;
61        self
62    }
63}
64
65/// Selects between the compatibility breaker and go-zero-style adaptive breaking.
66#[derive(Debug, Clone, Copy)]
67pub enum CircuitBreakerPolicy {
68    /// Open after a fixed number of consecutive failures, then use bounded half-open probes.
69    Consecutive,
70    /// Probabilistically reject calls from recent accepted/total request history.
71    Rolling(RollingCircuitBreakerConfig),
72}
73
74/// Settings for the rolling adaptive policy.
75#[derive(Debug, Clone, Copy)]
76pub struct RollingCircuitBreakerConfig {
77    pub window: Duration,
78    pub buckets: usize,
79    pub sensitivity: f64,
80    pub minimum_requests: u64,
81    pub probe_interval: u64,
82    pub random_seed: u64,
83}
84
85impl RollingCircuitBreakerConfig {
86    /// Uses go-zero-equivalent defaults: a five-second, 40-bucket history, a 1.5 acceptance
87    /// multiplier, and five protected observations before adaptive rejection starts.
88    pub fn new() -> Self {
89        Self {
90            window: Duration::from_secs(5),
91            buckets: 40,
92            sensitivity: 1.5,
93            minimum_requests: 5,
94            probe_interval: 100,
95            random_seed: 0,
96        }
97    }
98
99    pub fn with_window(mut self, window: Duration, buckets: usize) -> Self {
100        assert!(
101            !window.is_zero(),
102            "rolling window must be greater than zero"
103        );
104        assert!(
105            buckets > 0,
106            "rolling bucket count must be greater than zero"
107        );
108        assert!(
109            window.as_nanos() >= buckets as u128,
110            "rolling buckets must have non-zero width"
111        );
112        self.window = window;
113        self.buckets = buckets;
114        self
115    }
116
117    pub fn with_sensitivity(mut self, sensitivity: f64) -> Self {
118        assert!(
119            sensitivity.is_finite() && sensitivity > 0.0,
120            "rolling sensitivity must be finite and greater than zero"
121        );
122        self.sensitivity = sensitivity;
123        self
124    }
125
126    pub fn with_minimum_requests(mut self, requests: u64) -> Self {
127        self.minimum_requests = requests;
128        self
129    }
130
131    /// Bounds consecutive adaptive rejections so recovery always receives probe traffic.
132    pub fn with_probe_interval(mut self, requests: u64) -> Self {
133        assert!(requests > 0, "probe interval must be greater than zero");
134        self.probe_interval = requests;
135        self
136    }
137
138    /// Sets the deterministic PRNG seed. This is mainly useful for repeatable fault tests.
139    pub fn with_random_seed(mut self, seed: u64) -> Self {
140        self.random_seed = seed.max(1);
141        self
142    }
143
144    fn validate(self) {
145        assert!(
146            !self.window.is_zero(),
147            "rolling window must be greater than zero"
148        );
149        assert!(
150            self.buckets > 0,
151            "rolling bucket count must be greater than zero"
152        );
153        assert!(
154            self.window.as_nanos() >= self.buckets as u128,
155            "rolling buckets must have non-zero width"
156        );
157        assert!(
158            self.sensitivity.is_finite() && self.sensitivity > 0.0,
159            "rolling sensitivity must be finite and greater than zero"
160        );
161        assert!(
162            self.probe_interval > 0,
163            "probe interval must be greater than zero"
164        );
165    }
166}
167
168impl Default for RollingCircuitBreakerConfig {
169    fn default() -> Self {
170        Self::new()
171    }
172}
173
174/// Externally visible circuit state.
175#[derive(Debug, Clone, Copy, PartialEq, Eq)]
176pub enum BreakerState {
177    Closed,
178    Open,
179    HalfOpen,
180}
181
182/// The observed completion of an admitted dependency call.
183#[derive(Debug, Clone, Copy, PartialEq, Eq)]
184pub enum CircuitOutcome {
185    Success,
186    Failure,
187    Cancellation,
188}
189
190/// Current rolling history and lifetime outcome counters.
191#[derive(Debug, Clone, Copy, Default, PartialEq)]
192pub struct CircuitBreakerSnapshot {
193    pub accepted: u64,
194    pub total: u64,
195    pub drop_ratio: f64,
196    pub successes: u64,
197    pub failures: u64,
198    pub rejections: u64,
199    pub cancellations: u64,
200}
201
202enum ConsecutiveState {
203    Closed { consecutive_failures: u32 },
204    Open { opened_at: Instant },
205    HalfOpen { attempts: u32, successes: u32 },
206}
207
208impl ConsecutiveState {
209    fn status(&self) -> BreakerState {
210        match self {
211            Self::Closed { .. } => BreakerState::Closed,
212            Self::Open { .. } => BreakerState::Open,
213            Self::HalfOpen { .. } => BreakerState::HalfOpen,
214        }
215    }
216}
217
218enum BreakerMode {
219    Consecutive(Mutex<ConsecutiveState>),
220    Rolling(RollingBreaker),
221}
222
223#[derive(Default)]
224struct OutcomeCounters {
225    successes: AtomicU64,
226    failures: AtomicU64,
227    rejections: AtomicU64,
228    cancellations: AtomicU64,
229}
230
231/// Prevents calls to a dependency according to the configured policy.
232pub struct CircuitBreaker {
233    config: CircuitBreakerConfig,
234    mode: BreakerMode,
235    counters: OutcomeCounters,
236}
237
238impl CircuitBreaker {
239    pub fn new(config: CircuitBreakerConfig) -> Self {
240        let mode = match config.policy {
241            CircuitBreakerPolicy::Consecutive => {
242                BreakerMode::Consecutive(Mutex::new(ConsecutiveState::Closed {
243                    consecutive_failures: 0,
244                }))
245            }
246            CircuitBreakerPolicy::Rolling(rolling) => {
247                rolling.validate();
248                BreakerMode::Rolling(RollingBreaker::new(rolling))
249            }
250        };
251        Self {
252            config,
253            mode,
254            counters: OutcomeCounters::default(),
255        }
256    }
257
258    pub fn state(&self) -> BreakerState {
259        match &self.mode {
260            BreakerMode::Consecutive(state) => state
261                .lock()
262                .expect("circuit breaker state lock poisoned")
263                .status(),
264            BreakerMode::Rolling(rolling) => {
265                if rolling.history().2 > 0.0 {
266                    BreakerState::Open
267                } else {
268                    BreakerState::Closed
269                }
270            }
271        }
272    }
273
274    pub fn snapshot(&self) -> CircuitBreakerSnapshot {
275        let (accepted, total, drop_ratio) = match &self.mode {
276            BreakerMode::Consecutive(_) => (0, 0, 0.0),
277            BreakerMode::Rolling(rolling) => rolling.history(),
278        };
279        CircuitBreakerSnapshot {
280            accepted,
281            total,
282            drop_ratio,
283            successes: self.counters.successes.load(Ordering::Relaxed),
284            failures: self.counters.failures.load(Ordering::Relaxed),
285            rejections: self.counters.rejections.load(Ordering::Relaxed),
286            cancellations: self.counters.cancellations.load(Ordering::Relaxed),
287        }
288    }
289
290    /// Reserves a call for wrappers whose final outcome arrives in response trailers.
291    /// Dropping an unfinished permit records caller cancellation without changing breaker health.
292    pub fn acquire(self: &Arc<Self>) -> Option<CircuitBreakerPermit> {
293        if self.before_call::<Infallible>().is_err() {
294            return None;
295        }
296        Some(CircuitBreakerPermit {
297            breaker: Arc::clone(self),
298            finished: false,
299        })
300    }
301
302    pub fn execute<T, E, F>(&self, operation: F) -> Result<T, CircuitBreakerError<E>>
303    where
304        F: FnOnce() -> Result<T, E>,
305    {
306        self.execute_with_accept(operation, Result::is_ok)
307    }
308
309    pub fn execute_with_accept<T, E, F, A>(
310        &self,
311        operation: F,
312        acceptable: A,
313    ) -> Result<T, CircuitBreakerError<E>>
314    where
315        F: FnOnce() -> Result<T, E>,
316        A: FnOnce(&Result<T, E>) -> bool,
317    {
318        self.execute_with_outcome(operation, |result| {
319            if acceptable(result) {
320                CircuitOutcome::Success
321            } else {
322                CircuitOutcome::Failure
323            }
324        })
325    }
326
327    pub fn execute_with_outcome<T, E, F, A>(
328        &self,
329        operation: F,
330        outcome: A,
331    ) -> Result<T, CircuitBreakerError<E>>
332    where
333        F: FnOnce() -> Result<T, E>,
334        A: FnOnce(&Result<T, E>) -> CircuitOutcome,
335    {
336        self.before_call()?;
337        let mut completion = CompletionGuard::new(self);
338        let result = operation();
339        completion.finish(outcome(&result));
340        result.map_err(CircuitBreakerError::Operation)
341    }
342
343    pub async fn execute_async<T, E, F, Fut>(
344        &self,
345        operation: F,
346    ) -> Result<T, CircuitBreakerError<E>>
347    where
348        F: FnOnce() -> Fut,
349        Fut: Future<Output = Result<T, E>>,
350    {
351        self.execute_async_with_accept(operation, Result::is_ok)
352            .await
353    }
354
355    pub async fn execute_async_with_accept<T, E, F, Fut, A>(
356        &self,
357        operation: F,
358        acceptable: A,
359    ) -> Result<T, CircuitBreakerError<E>>
360    where
361        F: FnOnce() -> Fut,
362        Fut: Future<Output = Result<T, E>>,
363        A: FnOnce(&Result<T, E>) -> bool,
364    {
365        self.execute_async_with_outcome(operation, |result| {
366            if acceptable(result) {
367                CircuitOutcome::Success
368            } else {
369                CircuitOutcome::Failure
370            }
371        })
372        .await
373    }
374
375    pub async fn execute_async_with_outcome<T, E, F, Fut, A>(
376        &self,
377        operation: F,
378        outcome: A,
379    ) -> Result<T, CircuitBreakerError<E>>
380    where
381        F: FnOnce() -> Fut,
382        Fut: Future<Output = Result<T, E>>,
383        A: FnOnce(&Result<T, E>) -> CircuitOutcome,
384    {
385        self.before_call()?;
386        let mut completion = CompletionGuard::new(self);
387        let result = operation().await;
388        completion.finish(outcome(&result));
389        result.map_err(CircuitBreakerError::Operation)
390    }
391
392    fn before_call<E>(&self) -> Result<(), CircuitBreakerError<E>> {
393        let admitted = match &self.mode {
394            BreakerMode::Consecutive(state) => self.before_consecutive(state),
395            BreakerMode::Rolling(rolling) => !rolling.should_drop(),
396        };
397        if admitted {
398            Ok(())
399        } else {
400            self.counters.rejections.fetch_add(1, Ordering::Relaxed);
401            Err(CircuitBreakerError::Open)
402        }
403    }
404
405    fn before_consecutive(&self, state: &Mutex<ConsecutiveState>) -> bool {
406        let mut state = state.lock().expect("circuit breaker state lock poisoned");
407        if let ConsecutiveState::Open { opened_at } = *state {
408            if opened_at.elapsed() < self.config.reset_timeout {
409                return false;
410            }
411            *state = ConsecutiveState::HalfOpen {
412                attempts: 0,
413                successes: 0,
414            };
415        }
416        if let ConsecutiveState::HalfOpen { attempts, .. } = &mut *state {
417            if *attempts >= self.config.half_open_max_calls {
418                return false;
419            }
420            *attempts += 1;
421        }
422        true
423    }
424
425    fn record_outcome(&self, outcome: CircuitOutcome) {
426        match outcome {
427            CircuitOutcome::Success => {
428                self.counters.successes.fetch_add(1, Ordering::Relaxed);
429            }
430            CircuitOutcome::Failure => {
431                self.counters.failures.fetch_add(1, Ordering::Relaxed);
432            }
433            CircuitOutcome::Cancellation => {
434                self.counters.cancellations.fetch_add(1, Ordering::Relaxed);
435            }
436        }
437        match &self.mode {
438            BreakerMode::Consecutive(state) => self.record_consecutive(state, outcome),
439            BreakerMode::Rolling(rolling) => rolling.record(outcome),
440        }
441    }
442
443    fn record_consecutive(&self, state: &Mutex<ConsecutiveState>, outcome: CircuitOutcome) {
444        let mut state = state.lock().expect("circuit breaker state lock poisoned");
445        match outcome {
446            CircuitOutcome::Success => match &mut *state {
447                ConsecutiveState::Closed {
448                    consecutive_failures,
449                } => *consecutive_failures = 0,
450                ConsecutiveState::Open { .. } => {}
451                ConsecutiveState::HalfOpen { successes, .. } => {
452                    *successes += 1;
453                    if *successes == self.config.half_open_max_calls {
454                        *state = ConsecutiveState::Closed {
455                            consecutive_failures: 0,
456                        };
457                    }
458                }
459            },
460            CircuitOutcome::Failure => match &mut *state {
461                ConsecutiveState::Closed {
462                    consecutive_failures,
463                } => {
464                    *consecutive_failures += 1;
465                    if *consecutive_failures >= self.config.max_failures {
466                        *state = ConsecutiveState::Open {
467                            opened_at: Instant::now(),
468                        };
469                    }
470                }
471                ConsecutiveState::Open { .. } => {}
472                ConsecutiveState::HalfOpen { .. } => {
473                    *state = ConsecutiveState::Open {
474                        opened_at: Instant::now(),
475                    };
476                }
477            },
478            CircuitOutcome::Cancellation => {
479                if let ConsecutiveState::HalfOpen { attempts, .. } = &mut *state {
480                    *attempts = attempts.saturating_sub(1);
481                }
482            }
483        }
484    }
485}
486
487struct CompletionGuard<'a> {
488    breaker: &'a CircuitBreaker,
489    finished: bool,
490}
491
492impl<'a> CompletionGuard<'a> {
493    fn new(breaker: &'a CircuitBreaker) -> Self {
494        Self {
495            breaker,
496            finished: false,
497        }
498    }
499
500    fn finish(&mut self, outcome: CircuitOutcome) {
501        self.breaker.record_outcome(outcome);
502        self.finished = true;
503    }
504}
505
506impl Drop for CompletionGuard<'_> {
507    fn drop(&mut self) {
508        if !self.finished {
509            self.breaker.record_outcome(CircuitOutcome::Cancellation);
510        }
511    }
512}
513
514struct RollingBreaker {
515    config: RollingCircuitBreakerConfig,
516    bucket_width: Duration,
517    state: Mutex<RollingState>,
518    random: AtomicU64,
519    drop_sequence: AtomicU64,
520}
521
522impl RollingBreaker {
523    fn new(config: RollingCircuitBreakerConfig) -> Self {
524        static NEXT_SEED: AtomicU64 = AtomicU64::new(0x9e37_79b9_7f4a_7c15);
525        let bucket_width = Duration::from_nanos(
526            u64::try_from(config.window.as_nanos() / config.buckets as u128)
527                .unwrap_or(u64::MAX)
528                .max(1),
529        );
530        Self {
531            config,
532            bucket_width,
533            state: Mutex::new(RollingState {
534                current_started: Instant::now(),
535                buckets: VecDeque::from([OutcomeBucket::default()]),
536            }),
537            random: AtomicU64::new(if config.random_seed == 0 {
538                NEXT_SEED.fetch_add(0x9e37_79b9_7f4a_7c15, Ordering::Relaxed)
539            } else {
540                config.random_seed
541            }),
542            drop_sequence: AtomicU64::new(0),
543        }
544    }
545
546    fn should_drop(&self) -> bool {
547        let (_, _, ratio) = self.history();
548        if ratio <= 0.0 {
549            return false;
550        }
551        let sequence = self.drop_sequence.fetch_add(1, Ordering::Relaxed) + 1;
552        if sequence.is_multiple_of(self.config.probe_interval) {
553            return false;
554        }
555        self.random_unit() < ratio
556    }
557
558    fn record(&self, outcome: CircuitOutcome) {
559        if outcome == CircuitOutcome::Cancellation {
560            return;
561        }
562        let mut state = self.state.lock().expect("rolling breaker mutex poisoned");
563        self.rotate(&mut state, Instant::now());
564        let bucket = state
565            .buckets
566            .back_mut()
567            .expect("rolling breaker always has a current bucket");
568        bucket.total = bucket.total.saturating_add(1);
569        if outcome == CircuitOutcome::Success {
570            bucket.accepted = bucket.accepted.saturating_add(1);
571        }
572    }
573
574    fn history(&self) -> (u64, u64, f64) {
575        let mut state = self.state.lock().expect("rolling breaker mutex poisoned");
576        self.rotate(&mut state, Instant::now());
577        let (accepted, total) = state.buckets.iter().fold((0_u64, 0_u64), |sum, bucket| {
578            (
579                sum.0.saturating_add(bucket.accepted),
580                sum.1.saturating_add(bucket.total),
581            )
582        });
583        let unprotected = total.saturating_sub(self.config.minimum_requests) as f64;
584        let weighted_accepted = self.config.sensitivity * accepted as f64;
585        let drop_ratio = ((unprotected - weighted_accepted) / (total as f64 + 1.0)).max(0.0);
586        (accepted, total, drop_ratio)
587    }
588
589    fn rotate(&self, state: &mut RollingState, now: Instant) {
590        let elapsed = now.saturating_duration_since(state.current_started);
591        let elapsed_buckets = usize::try_from(elapsed.as_nanos() / self.bucket_width.as_nanos())
592            .unwrap_or(self.config.buckets);
593        if elapsed_buckets == 0 {
594            return;
595        }
596        if elapsed_buckets >= self.config.buckets {
597            state.buckets.clear();
598            state.buckets.push_back(OutcomeBucket::default());
599            state.current_started = now;
600            return;
601        }
602        for _ in 0..elapsed_buckets {
603            state.buckets.push_back(OutcomeBucket::default());
604            if state.buckets.len() > self.config.buckets {
605                state.buckets.pop_front();
606            }
607        }
608        state.current_started += self.bucket_width * elapsed_buckets as u32;
609    }
610
611    fn random_unit(&self) -> f64 {
612        let mut current = self.random.load(Ordering::Relaxed);
613        loop {
614            let mut next = current;
615            next ^= next << 13;
616            next ^= next >> 7;
617            next ^= next << 17;
618            next = next.max(1);
619            match self.random.compare_exchange_weak(
620                current,
621                next,
622                Ordering::Relaxed,
623                Ordering::Relaxed,
624            ) {
625                Ok(_) => return ((next >> 11) as f64) * (1.0 / 9_007_199_254_740_992.0),
626                Err(observed) => current = observed,
627            }
628        }
629    }
630}
631
632struct RollingState {
633    current_started: Instant,
634    buckets: VecDeque<OutcomeBucket>,
635}
636
637#[derive(Default)]
638struct OutcomeBucket {
639    accepted: u64,
640    total: u64,
641}
642
643/// An admitted circuit-breaker call whose outcome may arrive later.
644pub struct CircuitBreakerPermit {
645    breaker: Arc<CircuitBreaker>,
646    finished: bool,
647}
648
649impl CircuitBreakerPermit {
650    pub fn finish(self, acceptable: bool) {
651        self.finish_with_outcome(if acceptable {
652            CircuitOutcome::Success
653        } else {
654            CircuitOutcome::Failure
655        });
656    }
657
658    pub fn finish_with_outcome(mut self, outcome: CircuitOutcome) {
659        self.breaker.record_outcome(outcome);
660        self.finished = true;
661    }
662}
663
664impl Drop for CircuitBreakerPermit {
665    fn drop(&mut self) {
666        if !self.finished {
667            self.breaker.record_outcome(CircuitOutcome::Cancellation);
668        }
669    }
670}
671
672/// An operation error or the rejection produced by an open circuit.
673#[derive(Debug, PartialEq, Eq)]
674pub enum CircuitBreakerError<E> {
675    Open,
676    Operation(E),
677}
678
679impl<E: fmt::Display> fmt::Display for CircuitBreakerError<E> {
680    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
681        match self {
682            Self::Open => formatter.write_str("circuit breaker is open"),
683            Self::Operation(error) => write!(formatter, "protected operation failed: {error}"),
684        }
685    }
686}
687
688impl<E: std::error::Error + 'static> std::error::Error for CircuitBreakerError<E> {
689    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
690        match self {
691            Self::Open => None,
692            Self::Operation(error) => Some(error),
693        }
694    }
695}
696
697#[cfg(test)]
698mod tests {
699    use super::*;
700    use std::thread;
701
702    #[test]
703    fn opens_after_the_configured_failure_threshold() {
704        let breaker = CircuitBreaker::new(
705            CircuitBreakerConfig::new(2, Duration::from_secs(1))
706                .with_policy(CircuitBreakerPolicy::Consecutive),
707        );
708        assert_eq!(
709            breaker.execute(|| Err::<(), _>("first")),
710            Err(CircuitBreakerError::Operation("first"))
711        );
712        assert_eq!(
713            breaker.execute(|| Err::<(), _>("second")),
714            Err(CircuitBreakerError::Operation("second"))
715        );
716        assert_eq!(breaker.state(), BreakerState::Open);
717        assert_eq!(
718            breaker.execute(|| Ok::<_, ()>(())),
719            Err(CircuitBreakerError::Open)
720        );
721    }
722
723    #[test]
724    fn closes_after_a_successful_half_open_probe() {
725        let breaker = CircuitBreaker::new(CircuitBreakerConfig::new(1, Duration::from_millis(5)));
726        let _ = breaker.execute(|| Err::<(), _>("unavailable"));
727        thread::sleep(Duration::from_millis(10));
728        assert_eq!(breaker.execute(|| Ok::<_, ()>(42)), Ok(42));
729        assert_eq!(breaker.state(), BreakerState::Closed);
730    }
731
732    #[tokio::test]
733    async fn async_acceptance_can_reject_a_successful_result() {
734        let breaker = CircuitBreaker::new(CircuitBreakerConfig::new(1, Duration::from_secs(1)));
735        let response = breaker
736            .execute_async_with_accept(
737                || async { Ok::<_, ()>(503) },
738                |result| result.as_ref().is_ok_and(|status| *status < 500),
739            )
740            .await;
741        assert_eq!(response, Ok(503));
742        assert_eq!(breaker.state(), BreakerState::Open);
743    }
744
745    #[tokio::test]
746    async fn cancelled_async_call_is_neutral_to_rolling_health() {
747        let breaker = Arc::new(CircuitBreaker::new(CircuitBreakerConfig::rolling(
748            RollingCircuitBreakerConfig::new(),
749        )));
750        let task_breaker = Arc::clone(&breaker);
751        let task = tokio::spawn(async move {
752            task_breaker
753                .execute_async(std::future::pending::<Result<(), ()>>)
754                .await
755        });
756        tokio::task::yield_now().await;
757        task.abort();
758        assert!(task.await.unwrap_err().is_cancelled());
759
760        let snapshot = breaker.snapshot();
761        assert_eq!(snapshot.cancellations, 1);
762        assert_eq!((snapshot.accepted, snapshot.total), (0, 0));
763        assert_eq!(breaker.state(), BreakerState::Closed);
764    }
765
766    #[test]
767    fn dropped_transport_permit_records_cancellation_without_healing() {
768        let breaker = Arc::new(CircuitBreaker::new(
769            CircuitBreakerConfig::new(1, Duration::from_millis(5)).with_half_open_max_calls(2),
770        ));
771        breaker.acquire().unwrap().finish(false);
772        thread::sleep(Duration::from_millis(10));
773        drop(breaker.acquire().unwrap());
774        assert_eq!(breaker.state(), BreakerState::HalfOpen);
775        assert_eq!(breaker.snapshot().cancellations, 1);
776        breaker.acquire().unwrap().finish(true);
777        assert_eq!(breaker.state(), BreakerState::HalfOpen);
778    }
779
780    #[test]
781    fn rolling_fault_pattern_uses_recent_accepted_and_total_counts() {
782        let breaker = Arc::new(CircuitBreaker::new(CircuitBreakerConfig::rolling(
783            RollingCircuitBreakerConfig::new()
784                .with_window(Duration::from_millis(40), 4)
785                .with_minimum_requests(2)
786                .with_random_seed(7),
787        )));
788        for _ in 0..3 {
789            breaker.acquire().unwrap().finish(false);
790        }
791        let snapshot = breaker.snapshot();
792        assert_eq!((snapshot.accepted, snapshot.total), (0, 3));
793        assert!(snapshot.drop_ratio > 0.0);
794        assert_eq!(breaker.state(), BreakerState::Open);
795
796        thread::sleep(Duration::from_millis(50));
797        assert_eq!(breaker.state(), BreakerState::Closed);
798        assert_eq!(breaker.snapshot().total, 0);
799    }
800
801    #[test]
802    fn rolling_breaker_guarantees_bounded_probe_traffic() {
803        let breaker = Arc::new(CircuitBreaker::new(CircuitBreakerConfig::rolling(
804            RollingCircuitBreakerConfig::new()
805                .with_sensitivity(0.000_001)
806                .with_minimum_requests(0)
807                .with_probe_interval(4)
808                .with_random_seed(1),
809        )));
810        for _ in 0..20 {
811            if let Some(permit) = breaker.acquire() {
812                permit.finish(false);
813            }
814        }
815        assert!(breaker.snapshot().rejections > 0);
816        let admitted = (0..4).filter(|_| breaker.acquire().is_some()).count();
817        assert!(admitted >= 1);
818    }
819
820    #[test]
821    fn rolling_accounting_is_exact_under_concurrent_completion() {
822        let breaker = Arc::new(CircuitBreaker::new(CircuitBreakerConfig::rolling(
823            RollingCircuitBreakerConfig::new().with_minimum_requests(1_000),
824        )));
825        let workers: Vec<_> = (0..32)
826            .map(|_| {
827                let breaker = Arc::clone(&breaker);
828                thread::spawn(move || breaker.acquire().unwrap().finish(true))
829            })
830            .collect();
831        for worker in workers {
832            worker.join().unwrap();
833        }
834        let snapshot = breaker.snapshot();
835        assert_eq!((snapshot.accepted, snapshot.total), (32, 32));
836        assert_eq!(snapshot.successes, 32);
837    }
838}
rust_zero_core/breaker.rs

rust_zero_core/
breaker.rs
