gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(broker_gateway): replace inline CircuitBreaker with common::resilience::CircuitBreaker

Browse Source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-22 22:50:12 +01:00
parent bb4d4159d8
commit 4a14e53eeb
1 changed files with 24 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
services/broker_gateway_service/src/error_handler.rs
View File

@@ -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<RwLock<CircuitBreakerState>>,
/// Consecutive failure count
failure_count: Arc<RwLock<usize>>,
/// Timestamp when circuit breaker opened
opened_at: Arc<RwLock<Option<Instant>>>,
}
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