diff --git a/services/broker_gateway_service/src/error_handler.rs b/services/broker_gateway_service/src/error_handler.rs index 21cca0805..8953d4aaf 100644 --- a/services/broker_gateway_service/src/error_handler.rs +++ b/services/broker_gateway_service/src/error_handler.rs @@ -6,6 +6,9 @@ //! - Dead letter queue for unrecoverable orders //! - Error classification and routing +use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; +#[cfg(test)] +use common::resilience::circuit_breaker::CircuitBreakerState; use std::collections::VecDeque; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -43,17 +46,6 @@ pub enum ErrorRecoveryStrategy { Fallback, } -/// Circuit breaker state for fault isolation -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum CircuitBreakerState { - /// Circuit is closed, requests pass through normally - Closed, - /// Circuit is open, requests fail fast to prevent cascading failures - Open, - /// Circuit is half-open, testing if service has recovered - HalfOpen, -} - /// Dead letter queue entry for unrecoverable orders #[derive(Debug, Clone)] pub struct DeadLetterEntry { @@ -67,124 +59,6 @@ pub struct DeadLetterEntry { pub retry_attempts: u32, } -/// Circuit breaker for fault isolation and cascading failure prevention -pub struct CircuitBreaker { - /// Current circuit breaker state - state: Arc>, - /// Consecutive failure count - failure_count: Arc>, - /// Timestamp when circuit breaker opened - opened_at: Arc>>, -} - -impl CircuitBreaker { - /// Create a new circuit breaker in CLOSED state - pub fn new() -> Self { - Self { - state: Arc::new(RwLock::new(CircuitBreakerState::Closed)), - failure_count: Arc::new(RwLock::new(0)), - opened_at: Arc::new(RwLock::new(None)), - } - } - - /// Get current circuit breaker state - pub async fn state(&self) -> CircuitBreakerState { - *self.state.read().await - } - - /// Record a successful operation (resets failure count) - pub async fn record_success(&self) { - let mut failure_count = self.failure_count.write().await; - *failure_count = 0; - - let mut state = self.state.write().await; - if *state == CircuitBreakerState::HalfOpen { - info!("Circuit breaker transitioning: HALF_OPEN → CLOSED"); - *state = CircuitBreakerState::Closed; - } - } - - /// Record a failed operation (increments failure count) - pub async fn record_failure(&self) { - let mut failure_count = self.failure_count.write().await; - *failure_count += 1; - - let current_state = *self.state.read().await; - - if current_state == CircuitBreakerState::Closed - && *failure_count >= CIRCUIT_BREAKER_THRESHOLD - { - warn!( - "Circuit breaker OPEN after {} failures (threshold: {})", - *failure_count, CIRCUIT_BREAKER_THRESHOLD - ); - - let mut state = self.state.write().await; - *state = CircuitBreakerState::Open; - - let mut opened_at = self.opened_at.write().await; - *opened_at = Some(Instant::now()); - } else if current_state == CircuitBreakerState::HalfOpen { - warn!("Circuit breaker transitioning: HALF_OPEN → OPEN (failure during test)"); - - let mut state = self.state.write().await; - *state = CircuitBreakerState::Open; - - let mut opened_at = self.opened_at.write().await; - *opened_at = Some(Instant::now()); - } - } - - /// Check if circuit breaker allows requests (handles state transitions) - pub async fn allow_request(&self) -> bool { - let current_state = *self.state.read().await; - - match current_state { - CircuitBreakerState::Closed => true, - CircuitBreakerState::Open => { - // Check if timeout has elapsed - let opened_at = self.opened_at.read().await; - if let Some(opened_time) = *opened_at { - if opened_time.elapsed() >= CIRCUIT_BREAKER_TIMEOUT { - info!("Circuit breaker transitioning: OPEN → HALF_OPEN (timeout elapsed)"); - drop(opened_at); - - let mut state = self.state.write().await; - *state = CircuitBreakerState::HalfOpen; - - let mut failure_count = self.failure_count.write().await; - *failure_count = 0; - - return true; - } - } - false - } - CircuitBreakerState::HalfOpen => true, - } - } - - /// Reset circuit breaker to CLOSED state (manual recovery) - pub async fn reset(&self) { - let mut state = self.state.write().await; - *state = CircuitBreakerState::Closed; - - let mut failure_count = self.failure_count.write().await; - *failure_count = 0; - - let mut opened_at = self.opened_at.write().await; - *opened_at = None; - - info!("Circuit breaker manually reset to CLOSED state"); - } -} - -impl Default for CircuitBreaker { - fn default() -> Self { - Self::new() - } -} - /// Dead letter queue for unrecoverable orders pub struct DeadLetterQueue { /// Queue of failed orders @@ -275,8 +149,13 @@ pub struct ErrorHandler { impl ErrorHandler { /// Create a new error handler pub fn new() -> Self { + let cb_config = CircuitBreakerConfig { + failure_threshold: CIRCUIT_BREAKER_THRESHOLD as u32, + timeout: CIRCUIT_BREAKER_TIMEOUT, + success_threshold: 1, + }; Self { - circuit_breaker: CircuitBreaker::new(), + circuit_breaker: CircuitBreaker::new("broker_gateway", cb_config), dead_letter_queue: DeadLetterQueue::new(), } } @@ -422,13 +301,21 @@ impl Default for ErrorHandler { mod tests { use super::*; + fn test_cb_config() -> CircuitBreakerConfig { + CircuitBreakerConfig { + failure_threshold: CIRCUIT_BREAKER_THRESHOLD as u32, + timeout: CIRCUIT_BREAKER_TIMEOUT, + success_threshold: 1, + } + } + #[tokio::test] async fn test_circuit_breaker_closed_to_open() { - let cb = CircuitBreaker::new(); + let cb = CircuitBreaker::new("test_broker", test_cb_config()); // Initially CLOSED assert_eq!(cb.state().await, CircuitBreakerState::Closed); - assert!(cb.allow_request().await); + assert!(cb.can_execute().await); // Record failures up to threshold for _ in 0..CIRCUIT_BREAKER_THRESHOLD { @@ -437,12 +324,12 @@ mod tests { // Should now be OPEN assert_eq!(cb.state().await, CircuitBreakerState::Open); - assert!(!cb.allow_request().await); + assert!(!cb.can_execute().await); } #[tokio::test] async fn test_circuit_breaker_half_open_success() { - let cb = CircuitBreaker::new(); + let cb = CircuitBreaker::new("test_broker", test_cb_config()); // Force HALF_OPEN state for _ in 0..CIRCUIT_BREAKER_THRESHOLD { @@ -451,7 +338,7 @@ mod tests { // Wait for timeout to transition to HALF_OPEN tokio::time::sleep(CIRCUIT_BREAKER_TIMEOUT + Duration::from_millis(10)).await; - assert!(cb.allow_request().await); + assert!(cb.can_execute().await); assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen); // Success should close circuit @@ -461,7 +348,7 @@ mod tests { #[tokio::test] async fn test_circuit_breaker_half_open_failure() { - let cb = CircuitBreaker::new(); + let cb = CircuitBreaker::new("test_broker", test_cb_config()); // Force HALF_OPEN state for _ in 0..CIRCUIT_BREAKER_THRESHOLD { @@ -469,7 +356,7 @@ mod tests { } tokio::time::sleep(CIRCUIT_BREAKER_TIMEOUT + Duration::from_millis(10)).await; - assert!(cb.allow_request().await); // Trigger transition to HALF_OPEN + assert!(cb.can_execute().await); // Trigger transition to HALF_OPEN assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen); // Failure should reopen circuit