sz_orm_pool/

circuit_breaker.rs

//! 断路器抽象 trait
//!
//! P1-4 修复：核心层定义断路器接口，消除对 sz-orm-health 的反向依赖。
//!
//! ## 设计动机
//!
//! 之前 `sz-orm-core` 通过 optional dependency 反向依赖 `sz-orm-health`，
//! 违反了"核心 ← 扩展 ← 集成"的单向依赖原则。本模块将断路器的抽象 trait
//! 提升至核心层，扩展包（如 sz-orm-health）实现该 trait，从而反转依赖方向。
//!
//! ## 使用方式
//!
//! - 池内部默认使用 [`DefaultCircuitBreaker`]（核心层自带实现）
//! - 调用方可通过实现 [`CircuitBreaker`] trait 自定义断路器逻辑
//! - sz-orm-health 包的 `CircuitBreaker` 结构体也实现了本 trait

use std::time::{Duration, Instant};

/// 断路器状态
///
/// 三态有限状态机：
/// ```text
///   Closed ──(失败次数达阈值)──> Open
///   Open ──(reset_timeout 到达)──> HalfOpen
///   HalfOpen ──(一次成功)──> Closed
///   HalfOpen ──(一次失败)──> Open
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CircuitState {
    /// 正常运行：请求放行
    Closed,
    /// 跳闸：请求被阻塞，直到 `reset_timeout` 到达
    Open,
    /// 探测：放行一次试探请求（reset_timeout 到达后进入）
    HalfOpen,
}

/// 断路器抽象 trait
///
/// P1-4：核心层定义的断路器接口，消除对 sz-orm-health 的反向依赖。
///
/// 实现方需保证：
/// - `can_execute()` 在 `Open` 状态下根据 `reset_timeout` 自动迁移到 `HalfOpen`
/// - `record_success()` 将状态重置为 `Closed` 并清空失败计数
/// - `record_failure()` 累计失败次数，达阈值时迁移到 `Open`
pub trait CircuitBreaker: Send + Sync {
    /// 获取当前断路器状态
    fn state(&self) -> CircuitState;

    /// 记录一次成功请求
    ///
    /// 通常将状态重置为 `Closed` 并清空连续失败计数。
    fn record_success(&mut self);

    /// 记录一次失败请求
    ///
    /// 累计连续失败次数，达到阈值时将状态迁移到 `Open`。
    fn record_failure(&mut self);

    /// 判断当前是否允许请求通过
    ///
    /// 在 `Open` 状态下，若 `reset_timeout` 已到达，应迁移到 `HalfOpen` 并返回 `true`。
    fn can_execute(&mut self) -> bool;

    /// 手动重置断路器到 `Closed` 状态
    ///
    /// 与 `record_success` 的区别：`reset` 是运维主动强制重置，
    /// 无视当前状态（含 `Open`），常用于故障排除后手动恢复。
    ///
    /// 返回是否实际发生了状态变更。
    fn reset(&mut self) -> bool;
}

/// 默认断路器实现
///
/// 提供 `failure_threshold`（连续失败阈值）+ `reset_timeout`（重置超时）的经典三态机。
///
/// # 示例
///
/// ```ignore
/// use sz_orm_pool::circuit_breaker::{CircuitBreaker, CircuitState, DefaultCircuitBreaker};
/// use std::time::Duration;
///
/// let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
/// assert_eq!(cb.state(), CircuitState::Closed);
/// assert!(cb.can_execute());
///
/// // 3 次失败后跳闸
/// cb.record_failure();
/// cb.record_failure();
/// cb.record_failure();
/// assert_eq!(cb.state(), CircuitState::Open);
/// assert!(!cb.can_execute());
///
/// // 一次成功后恢复
/// cb.record_success();
/// assert_eq!(cb.state(), CircuitState::Closed);
/// ```
#[derive(Debug)]
pub struct DefaultCircuitBreaker {
    /// 连续失败达到此阈值时跳闸（>= 比较）
    failure_threshold: usize,
    /// Open 状态持续时间，到达后进入 HalfOpen
    reset_timeout: Duration,
    /// 当前状态
    state: CircuitState,
    /// 连续失败次数
    consecutive_failures: usize,
    /// 最后一次失败的时间（用于判断 reset_timeout 是否到达）
    last_failure_at: Option<Instant>,
}

impl DefaultCircuitBreaker {
    /// 创建默认断路器
    ///
    /// - `failure_threshold`：连续失败次数阈值（>= 时跳闸）
    /// - `reset_timeout`：Open 状态持续时间，到达后进入 HalfOpen
    pub fn new(failure_threshold: usize, reset_timeout: Duration) -> Self {
        Self {
            failure_threshold,
            reset_timeout,
            state: CircuitState::Closed,
            consecutive_failures: 0,
            last_failure_at: None,
        }
    }

    /// 获取连续失败次数（主要用于测试与监控）
    pub fn consecutive_failures(&self) -> usize {
        self.consecutive_failures
    }

    /// 获取失败阈值
    pub fn failure_threshold(&self) -> usize {
        self.failure_threshold
    }

    /// 获取重置超时
    pub fn reset_timeout(&self) -> Duration {
        self.reset_timeout
    }
}

impl Default for DefaultCircuitBreaker {
    /// 默认配置：5 次连续失败跳闸，30 秒后进入 HalfOpen
    fn default() -> Self {
        Self::new(5, Duration::from_secs(30))
    }
}

impl CircuitBreaker for DefaultCircuitBreaker {
    fn state(&self) -> CircuitState {
        self.state
    }

    fn record_success(&mut self) {
        self.consecutive_failures = 0;
        self.state = CircuitState::Closed;
        self.last_failure_at = None;
    }

    fn record_failure(&mut self) {
        self.consecutive_failures += 1;
        self.last_failure_at = Some(Instant::now());
        if self.consecutive_failures >= self.failure_threshold {
            self.state = CircuitState::Open;
        }
    }

    fn can_execute(&mut self) -> bool {
        match self.state {
            CircuitState::Closed => true,
            CircuitState::HalfOpen => true,
            CircuitState::Open => {
                let elapsed = self
                    .last_failure_at
                    .map(|t| t.elapsed())
                    .unwrap_or_else(|| Duration::ZERO);
                if elapsed >= self.reset_timeout {
                    self.state = CircuitState::HalfOpen;
                    true
                } else {
                    false
                }
            }
        }
    }

    fn reset(&mut self) -> bool {
        let changed = self.state != CircuitState::Closed || self.consecutive_failures != 0;
        self.state = CircuitState::Closed;
        self.consecutive_failures = 0;
        self.last_failure_at = None;
        changed
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_circuit_breaker_starts_closed() {
        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_millis(100));
        assert_eq!(cb.state(), CircuitState::Closed);
        assert!(cb.can_execute());
    }

    #[test]
    fn test_circuit_breaker_trips_after_threshold() {
        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
        assert!(cb.can_execute());
        cb.record_failure();
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Closed);
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Open);
        assert!(!cb.can_execute());
    }

    #[test]
    fn test_circuit_breaker_success_resets() {
        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
        cb.record_failure();
        cb.record_failure();
        cb.record_success();
        assert_eq!(cb.state(), CircuitState::Closed);
        // After success, should need 3 failures again to trip.
        cb.record_failure();
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn test_circuit_breaker_half_open_after_timeout() {
        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_millis(10));
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Open);
        // Immediately: still open.
        assert!(!cb.can_execute());
        // Wait for reset timeout.
        std::thread::sleep(Duration::from_millis(30));
        assert!(cb.can_execute());
        assert_eq!(cb.state(), CircuitState::HalfOpen);
    }

    #[test]
    fn test_circuit_breaker_half_open_success_closes() {
        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_millis(10));
        cb.record_failure();
        std::thread::sleep(Duration::from_millis(20));
        assert!(cb.can_execute());
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        cb.record_success();
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn test_circuit_breaker_half_open_failure_reopens() {
        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_millis(10));
        cb.record_failure();
        std::thread::sleep(Duration::from_millis(20));
        assert!(cb.can_execute());
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Open);
    }

    #[test]
    fn test_circuit_breaker_boundary_exactly_threshold() {
        // threshold = 3 means 3 failures should trip (>= comparison).
        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
        cb.record_failure();
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Closed);
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Open);
    }

    #[test]
    fn test_circuit_breaker_reset_from_open() {
        let mut cb = DefaultCircuitBreaker::new(2, Duration::from_secs(60));
        cb.record_failure();
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Open);
        assert!(cb.reset());
        assert_eq!(cb.state(), CircuitState::Closed);
        assert_eq!(cb.consecutive_failures(), 0);
        // 重置后立即可执行
        assert!(cb.can_execute());
        // 仍需累计到阈值才会再次跳闸
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn test_circuit_breaker_reset_from_half_open() {
        let mut cb = DefaultCircuitBreaker::new(1, Duration::from_millis(10));
        cb.record_failure();
        assert_eq!(cb.state(), CircuitState::Open);
        std::thread::sleep(Duration::from_millis(20));
        assert!(cb.can_execute());
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        // HalfOpen 状态下手动重置
        assert!(cb.reset());
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn test_circuit_breaker_reset_idempotent_when_closed() {
        let mut cb = DefaultCircuitBreaker::new(3, Duration::from_secs(60));
        assert!(!cb.reset());
        assert_eq!(cb.state(), CircuitState::Closed);
        // 存在失败计数但未跳闸时，reset 视为变更
        cb.record_failure();
        cb.record_failure();
        assert!(cb.reset());
        assert_eq!(cb.consecutive_failures(), 0);
        // 再次 reset 无变更
        assert!(!cb.reset());
    }

    #[test]
    fn test_circuit_state_variants_distinct() {
        assert_ne!(CircuitState::Closed, CircuitState::Open);
        assert_ne!(CircuitState::Open, CircuitState::HalfOpen);
        assert_ne!(CircuitState::Closed, CircuitState::HalfOpen);
    }

    #[test]
    fn test_default_circuit_breaker_default_config() {
        let cb = DefaultCircuitBreaker::default();
        assert_eq!(cb.failure_threshold(), 5);
        assert_eq!(cb.reset_timeout(), Duration::from_secs(30));
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn test_circuit_breaker_send_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<DefaultCircuitBreaker>();
        assert_send_sync::<CircuitState>();
    }

    /// 验证 trait object 可用（动态分发）
    #[test]
    fn test_circuit_breaker_via_trait_object() {
        let cb: Box<dyn CircuitBreaker> =
            Box::new(DefaultCircuitBreaker::new(2, Duration::from_secs(60)));
        // 只能调用 trait 方法
        assert_eq!(cb.state(), CircuitState::Closed);
    }
}