sz_orm_core/
circuit_breaker.rs

1//! 断路器抽象（P1-4：抽象提升到核心层，供连接池执行路径集成）
2//!
3//! 连接池在 `circuit-breaker` feature 下通过 `configure_circuit_breaker()` 配置
4//! [`DefaultCircuitBreaker`]，在获取连接/执行查询前调用 [`CircuitBreaker::can_execute`]
5//! 拦截失败请求，成功/失败时记录反馈，防止故障级联。
6//!
7//! 本模块为自包含实现（不依赖 sz-orm-health），消除核心层对上层 crate 的反向依赖。
8
9use std::time::{Duration, Instant};
10
11/// 断路器状态机状态。
12#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
13pub enum CircuitState {
14    /// 正常运行：请求放行。
15    Closed,
16    /// 已熔断：在 `reset_timeout` 内拦截所有请求。
17    Open,
18    /// 试探：熔断超时后放行单个试探请求。
19    HalfOpen,
20}
21
22/// 断路器抽象 trait。
23///
24/// 实现方需维护失败计数与状态机；连接池通过该 trait 统一驱动
25/// （不依赖具体实现类型，便于测试替换）。
26pub trait CircuitBreaker: Send + Sync {
27    /// 请求是否可执行。`Open` 且熔断超时后自动转为 `HalfOpen` 并放行。
28    fn can_execute(&mut self) -> bool;
29    /// 记录一次成功（复位失败计数，回到 `Closed`）。
30    fn record_success(&mut self);
31    /// 记录一次失败（达到阈值后熔断为 `Open`）。
32    fn record_failure(&mut self);
33    /// 当前状态。
34    fn state(&self) -> CircuitState;
35    /// 手动重置到 `Closed`。返回是否实际发生了状态变更。
36    fn reset(&mut self) -> bool;
37}
38
39/// 默认断路器实现：连续失败 `failure_threshold` 次后熔断，
40/// `reset_timeout` 过后进入 `HalfOpen` 试探，成功恢复 `Closed`，失败回到 `Open`。
41pub struct DefaultCircuitBreaker {
42    failure_threshold: usize,
43    reset_timeout: Duration,
44    state: CircuitState,
45    consecutive_failures: usize,
46    last_failure_at: Option<Instant>,
47    /// v3.8.0: 累计熔断次数
48    total_trips: u64,
49}
50
51impl DefaultCircuitBreaker {
52    /// 创建断路器。
53    ///
54    /// - `failure_threshold`：连续失败多少次后熔断（`Open`）；
55    /// - `reset_timeout`：熔断后等待多久进入 `HalfOpen` 试探。
56    pub fn new(failure_threshold: usize, reset_timeout: Duration) -> Self {
57        Self {
58            failure_threshold,
59            reset_timeout,
60            state: CircuitState::Closed,
61            consecutive_failures: 0,
62            last_failure_at: None,
63            total_trips: 0,
64        }
65    }
66
67    /// v3.8.0: 查询熔断器统计信息
68    #[cfg(feature = "prod-circuit-tuning")]
69    pub fn stats(&self) -> CircuitBreakerStats {
70        CircuitBreakerStats {
71            state: self.state,
72            consecutive_failures: self.consecutive_failures,
73            total_trips: self.total_trips,
74        }
75    }
76}
77
78impl CircuitBreaker for DefaultCircuitBreaker {
79    fn can_execute(&mut self) -> bool {
80        match self.state {
81            CircuitState::Closed => true,
82            CircuitState::HalfOpen => true,
83            CircuitState::Open => {
84                let elapsed = self
85                    .last_failure_at
86                    .map(|t| t.elapsed())
87                    .unwrap_or(Duration::ZERO);
88                if elapsed >= self.reset_timeout {
89                    self.state = CircuitState::HalfOpen;
90                    true
91                } else {
92                    false
93                }
94            }
95        }
96    }
97
98    fn record_success(&mut self) {
99        self.consecutive_failures = 0;
100        self.state = CircuitState::Closed;
101        self.last_failure_at = None;
102    }
103
104    fn record_failure(&mut self) {
105        self.consecutive_failures += 1;
106        self.last_failure_at = Some(Instant::now());
107        if self.consecutive_failures >= self.failure_threshold && self.state != CircuitState::Open {
108            self.state = CircuitState::Open;
109            self.total_trips += 1;
110        }
111    }
112
113    fn state(&self) -> CircuitState {
114        self.state
115    }
116
117    fn reset(&mut self) -> bool {
118        let changed = self.state != CircuitState::Closed || self.consecutive_failures != 0;
119        self.state = CircuitState::Closed;
120        self.consecutive_failures = 0;
121        self.last_failure_at = None;
122        changed
123    }
124}
125
126// ============================================================================
127// v3.8.0: 熔断器生产配置（prod-circuit-tuning feature）
128// ============================================================================
129
130#[cfg(feature = "prod-circuit-tuning")]
131mod prod {
132    use super::CircuitState;
133    use serde::{Deserialize, Serialize};
134    use std::time::Duration;
135
136    /// 熔断器生产配置错误
137    #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
138    pub enum CircuitBreakerProdError {
139        /// 失败阈值非正
140        #[error("circuit breaker failure_threshold must be positive")]
141        FailureThresholdNotPositive,
142        /// 重置超时非正
143        #[error("circuit breaker reset_timeout must be positive")]
144        ResetTimeoutNotPositive,
145    }
146
147    /// 熔断器生产配置
148    #[derive(Debug, Clone, Serialize, Deserialize)]
149    pub struct CircuitBreakerProdConfig {
150        /// 失败阈值
151        pub failure_threshold: u32,
152        /// 重置超时
153        pub reset_timeout: Duration,
154    }
155
156    impl Default for CircuitBreakerProdConfig {
157        fn default() -> Self {
158            Self {
159                failure_threshold: 5,
160                reset_timeout: Duration::from_secs(30),
161            }
162        }
163    }
164
165    impl CircuitBreakerProdConfig {
166        /// 创建配置
167        pub fn new(failure_threshold: u32, reset_timeout: Duration) -> Self {
168            Self {
169                failure_threshold,
170                reset_timeout,
171            }
172        }
173
174        /// 验证配置合法性
175        pub fn validate(&self) -> Result<(), CircuitBreakerProdError> {
176            if self.failure_threshold == 0 {
177                return Err(CircuitBreakerProdError::FailureThresholdNotPositive);
178            }
179            if self.reset_timeout.is_zero() {
180                return Err(CircuitBreakerProdError::ResetTimeoutNotPositive);
181            }
182            Ok(())
183        }
184    }
185
186    /// 熔断器统计信息
187    #[derive(Debug, Clone, Serialize, Deserialize)]
188    pub struct CircuitBreakerStats {
189        /// 当前状态
190        pub state: CircuitState,
191        /// 连续失败次数
192        pub consecutive_failures: usize,
193        /// 总跳闸次数
194        pub total_trips: u64,
195    }
196}
197
198#[cfg(feature = "prod-circuit-tuning")]
199pub use prod::{CircuitBreakerProdConfig, CircuitBreakerProdError, CircuitBreakerStats};
200
201#[cfg(test)]
202mod tests {
203    use super::*;
204
205    #[test]
206    fn test_circuit_breaker_starts_closed() {
207        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
208        assert_eq!(cb.state(), CircuitState::Closed);
209        assert!(cb.can_execute());
210    }
211
212    #[test]
213    fn test_circuit_breaker_trips_after_threshold() {
214        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
215        cb.record_failure();
216        cb.record_failure();
217        assert_eq!(cb.state(), CircuitState::Closed);
218        cb.record_failure();
219        assert_eq!(cb.state(), CircuitState::Open);
220        assert!(!cb.can_execute());
221    }
222
223    #[test]
224    fn test_circuit_breaker_success_resets() {
225        let mut cb = DefaultCircuitBreaker::new(2, Duration::from_secs(60));
226        cb.record_failure();
227        cb.record_success();
228        assert_eq!(cb.state(), CircuitState::Closed);
229        // 成功后失败计数已清零，需重新积累阈值
230        cb.record_failure();
231        assert_eq!(cb.state(), CircuitState::Closed);
232        cb.record_failure();
233        assert_eq!(cb.state(), CircuitState::Open);
234    }
235
236    #[test]
237    fn test_circuit_breaker_half_open_after_timeout() {
238        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_millis(10));
239        cb.record_failure();
240        assert_eq!(cb.state(), CircuitState::Open);
241        // 超时后 can_execute 自动进入 HalfOpen 并放行
242        std::thread::sleep(Duration::from_millis(20));
243        assert!(cb.can_execute());
244        assert_eq!(cb.state(), CircuitState::HalfOpen);
245    }
246
247    #[test]
248    fn test_circuit_breaker_half_open_success_closes() {
249        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_millis(10));
250        cb.record_failure();
251        std::thread::sleep(Duration::from_millis(20));
252        assert!(cb.can_execute()); // HalfOpen
253        cb.record_success();
254        assert_eq!(cb.state(), CircuitState::Closed);
255    }
256
257    #[test]
258    fn test_circuit_breaker_half_open_failure_reopens() {
259        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_millis(10));
260        cb.record_failure();
261        std::thread::sleep(Duration::from_millis(20));
262        assert!(cb.can_execute()); // HalfOpen
263        cb.record_failure();
264        assert_eq!(cb.state(), CircuitState::Open);
265        assert!(!cb.can_execute());
266    }
267
268    #[test]
269    fn test_circuit_breaker_reset() {
270        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_secs(60));
271        cb.record_failure();
272        assert_eq!(cb.state(), CircuitState::Open);
273        assert!(cb.reset());
274        assert_eq!(cb.state(), CircuitState::Closed);
275        assert!(cb.can_execute());
276        // 已关闭时 reset 返回 false
277        assert!(!cb.reset());
278    }
279}
280
281#[cfg(all(test, feature = "prod-circuit-tuning"))]
282mod prod_tests {
283    use super::*;
284
285    #[test]
286    fn test_circuit_breaker_prod_config_validate_ok() {
287        let config = CircuitBreakerProdConfig::new(10, Duration::from_secs(60));
288        assert!(config.validate().is_ok());
289    }
290
291    #[test]
292    fn test_circuit_breaker_prod_config_threshold_zero_rejected() {
293        let config = CircuitBreakerProdConfig::new(0, Duration::from_secs(60));
294        let err = config.validate().unwrap_err();
295        assert!(err
296            .to_string()
297            .contains("failure_threshold must be positive"));
298    }
299
300    #[test]
301    fn test_circuit_breaker_prod_config_timeout_zero_rejected() {
302        let config = CircuitBreakerProdConfig::new(10, Duration::ZERO);
303        assert!(config.validate().is_err());
304    }
305
306    #[test]
307    fn test_circuit_breaker_stats_after_trips() {
308        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
309        cb.record_failure();
310        cb.record_failure();
311        cb.record_failure();
312        let stats = cb.stats();
313        assert_eq!(stats.state, CircuitState::Open);
314        assert_eq!(stats.consecutive_failures, 3);
315        assert_eq!(stats.total_trips, 1);
316    }
317
318    #[test]
319    fn test_circuit_breaker_stats_no_trips() {
320        let cb = DefaultCircuitBreaker::new(5, Duration::from_secs(60));
321        let stats = cb.stats();
322        assert_eq!(stats.state, CircuitState::Closed);
323        assert_eq!(stats.total_trips, 0);
324    }
325
326    #[test]
327    fn test_circuit_breaker_total_trips_increments() {
328        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_secs(60));
329        cb.record_failure();
330        assert_eq!(cb.stats().total_trips, 1);
331        cb.record_success();
332        cb.record_failure();
333        assert_eq!(cb.stats().total_trips, 2);
334    }
335}
sz_orm_core/circuit_breaker.rs

sz_orm_core/
circuit_breaker.rs
