diff --git a/src/cache.rs b/src/cache.rs index a2a2018..894f603 100644 --- a/src/cache.rs +++ b/src/cache.rs @@ -55,9 +55,9 @@ use std::{ collections::{HashMap, VecDeque}, prelude::v1::*, sync::Arc, - time::{Duration, SystemTime, UNIX_EPOCH}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; -use tokio::sync::{broadcast, Mutex, RwLock}; +use tokio::sync::{broadcast, Mutex, RwLock, Semaphore}; use crate::metrics::MetricsRegistry; @@ -110,16 +110,35 @@ impl CacheBackend { } /// Retrieve a raw cached value. Emits cache hit/miss/expired metrics. + /// When using Redis backend, falls back to InMemory cache on circuit breaker open or failure. pub async fn get_raw(&self, key: &CacheKey) -> Result> { match self { Self::Redis(c) => { - // Redis handles TTL natively — misses include expired entries - let result = c.get_raw(&key.as_string()).await?; - match &result { - Some(_) => c.record_hit(), - None => c.record_miss(), + // Try Redis first + match c.get_raw(&key.as_string()).await { + Ok(result) => { + match &result { + Some(_) => c.record_hit(), + None => c.record_miss(), + } + Ok(result) + } + Err(_) => { + // Redis failed — fallback to InMemory cache + if let Some(ref m) = c.metrics { + m.increment_cache_fallback_use(); + } + let (value, was_expired) = c.fallback().get_raw_with_expiry(key).await?; + if was_expired { + c.fallback().record_expired(); + } else if value.is_some() { + c.fallback().record_hit(); + } else { + c.fallback().record_miss(); + } + Ok(value) + } } - Ok(result) } Self::InMemory(c) => { // InMemory distinguishes expired from true miss @@ -138,7 +157,19 @@ impl CacheBackend { pub async fn set_raw(&self, key: &CacheKey, value: &str, ttl: u64) -> Result<()> { match self { - Self::Redis(c) => c.set_raw(&key.as_string(), value, ttl).await, + Self::Redis(c) => { + // Try Redis first + match c.set_raw(&key.as_string(), value, ttl).await { + Ok(()) => Ok(()), + Err(_) => { + // Redis failed — write to InMemory fallback cache + if let Some(ref m) = c.metrics { + m.increment_cache_fallback_use(); + } + c.fallback().set_raw(key, value, ttl).await + } + } + } Self::InMemory(c) => c.set_raw(key, value, ttl).await, } } @@ -173,17 +204,291 @@ impl CacheBackend { pub async fn delete(&self, key: &CacheKey) -> Result<()> { match self { - Self::Redis(c) => c.delete(&key.as_string()).await, + Self::Redis(c) => { + // Try Redis first + match c.delete(&key.as_string()).await { + Ok(()) => Ok(()), + Err(_) => { + // Redis failed — delete from InMemory fallback cache + if let Some(ref m) = c.metrics { + m.increment_cache_fallback_use(); + } + c.fallback().delete(key).await + } + } + } Self::InMemory(c) => c.delete(key).await, } } } +// ── Circuit Breaker ────────────────────────────────────────────────────── + +/// Circuit breaker states for Redis cache operations. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CircuitState { + /// Normal operation: requests pass through. + Closed, + /// Circuit is tripped: all requests are rejected. + Open, + /// Probing: limited requests allowed to test recovery. + HalfOpen, +} + +impl CircuitState { + fn as_str(&self) -> &'static str { + match self { + CircuitState::Closed => "closed", + CircuitState::Open => "open", + CircuitState::HalfOpen => "half_open", + } + } + + fn as_metric_value(&self) -> i64 { + match self { + CircuitState::Closed => 0, + CircuitState::Open => 1, + CircuitState::HalfOpen => 2, + } + } +} + +/// Configuration for the cache circuit breaker. +#[derive(Debug, Clone)] +pub struct CacheCircuitBreakerConfig { + pub failure_threshold: u32, + pub open_duration_ms: u64, + pub half_open_max_calls: u32, + pub backoff_base_ms: u64, + pub backoff_max_ms: u64, +} + +impl Default for CacheCircuitBreakerConfig { + fn default() -> Self { + Self { + failure_threshold: 5, + open_duration_ms: 30_000, + half_open_max_calls: 1, + backoff_base_ms: 100, + backoff_max_ms: 30_000, + } + } +} + +/// Circuit breaker for Redis cache operations. +/// +/// Transitions: Closed -> Open (after failure_threshold failures) +/// Open -> HalfOpen (after open_duration with exponential backoff) +/// HalfOpen -> Closed (on success) or Open (on failure) +#[derive(Debug)] +pub struct CacheCircuitBreaker { + state: CircuitState, + consecutive_failures: u32, + failure_threshold: u32, + #[allow(dead_code)] + open_duration: Duration, + half_open_max_calls: u32, + half_open_calls: u32, + opened_at: Option, + backoff_base: Duration, + backoff_max: Duration, + _last_transition: Option<(CircuitState, CircuitState)>, +} + +impl CacheCircuitBreaker { + pub fn new(config: CacheCircuitBreakerConfig) -> Self { + Self { + state: CircuitState::Closed, + consecutive_failures: 0, + failure_threshold: config.failure_threshold, + open_duration: Duration::from_millis(config.open_duration_ms), + half_open_max_calls: config.half_open_max_calls, + half_open_calls: 0, + opened_at: None, + backoff_base: Duration::from_millis(config.backoff_base_ms), + backoff_max: Duration::from_millis(config.backoff_max_ms), + _last_transition: None, + } + } + + /// Check if a request is allowed through the circuit breaker. + pub fn should_allow(&mut self) -> bool { + match self.state { + CircuitState::Closed => true, + CircuitState::Open => { + // Check if we should transition to HalfOpen + if let Some(opened_at) = self.opened_at { + let backoff = self.current_backoff(); + if opened_at.elapsed() >= backoff { + self.transition_to(CircuitState::HalfOpen); + return true; + } + } + false + } + CircuitState::HalfOpen => { + self.half_open_calls < self.half_open_max_calls + } + } + } + + /// Record a successful operation. + pub fn record_success(&mut self) { + match self.state { + CircuitState::Closed => { + self.consecutive_failures = 0; + } + CircuitState::HalfOpen => { + // Success in half-open: close the circuit + self.half_open_calls += 1; + if self.half_open_calls >= self.half_open_max_calls { + self.transition_to(CircuitState::Closed); + } + } + CircuitState::Open => { + // Shouldn't happen, but handle gracefully + } + } + } + + /// Record a failed operation. + pub fn record_failure(&mut self) { + match self.state { + CircuitState::Closed => { + self.consecutive_failures += 1; + if self.consecutive_failures >= self.failure_threshold { + self.transition_to(CircuitState::Open); + } + } + CircuitState::HalfOpen => { + // Failure in half-open: reopen the circuit + self.transition_to(CircuitState::Open); + } + CircuitState::Open => { + // Already open, just increment counter + self.consecutive_failures += 1; + } + } + } + + fn transition_to(&mut self, new_state: CircuitState) { + let old_state = self.state; + self.state = new_state; + match new_state { + CircuitState::Closed => { + self.consecutive_failures = 0; + self.opened_at = None; + self.half_open_calls = 0; + } + CircuitState::Open => { + self.opened_at = Some(Instant::now()); + self.half_open_calls = 0; + } + CircuitState::HalfOpen => { + self.half_open_calls = 0; + } + } + // Notify metrics + // Note: metrics are passed separately to avoid circular dependency + self._last_transition = Some((old_state, new_state)); + } + + /// Returns the last state transition if one occurred, and clears it. + pub fn take_last_transition(&mut self) -> Option<(CircuitState, CircuitState)> { + self._last_transition.take() + } + + /// Get the current exponential backoff duration. + fn current_backoff(&self) -> Duration { + let failure_count = self.consecutive_failures.max(1); + let base = self.backoff_base.as_millis() as u64; + let exponent = (failure_count - 1).min(6); + let backoff_ms = (base * 2u64.pow(exponent)).min(self.backoff_max.as_millis() as u64); + Duration::from_millis(backoff_ms) + } + + /// Get the current state. + pub fn state(&self) -> CircuitState { + self.state + } +} + +// ── Bulkhead ────────────────────────────────────────────────────────────── + +/// Configuration for the cache bulkhead. +#[derive(Debug, Clone)] +pub struct CacheBulkheadConfig { + pub max_concurrent: u32, + pub max_queue: u32, +} + +impl Default for CacheBulkheadConfig { + fn default() -> Self { + Self { + max_concurrent: 20, + max_queue: 200, + } + } +} + +/// Bulkhead for limiting concurrent Redis cache operations. +/// +/// Uses a semaphore to limit concurrent operations and a queue depth limit +/// to prevent memory exhaustion under load. +#[derive(Debug)] +pub struct CacheBulkhead { + semaphore: Arc, + #[allow(dead_code)] + max_concurrent: usize, + #[allow(dead_code)] + max_queue: usize, +} + +impl CacheBulkhead { + pub fn new(config: CacheBulkheadConfig) -> Self { + Self { + semaphore: Arc::new(Semaphore::new(config.max_concurrent as usize)), + max_concurrent: config.max_concurrent as usize, + max_queue: config.max_queue as usize, + } + } + + /// Try to acquire a permit. Returns Ok(BulkheadGuard) if available, Err if bulkhead is full. + /// + /// Non-blocking: immediately rejects if no permits are available. + pub async fn try_acquire(&self) -> Result { + let permit = Arc::clone(&self.semaphore) + .try_acquire_owned() + .map_err(|_| ())?; + Ok(BulkheadGuard { _permit: permit }) + } + + /// Get the number of currently active operations. + pub fn active_count(&self) -> usize { + self.max_concurrent - self.semaphore.available_permits() + } +} + +/// Guard that releases a bulkhead permit on drop. +#[derive(Debug)] +pub struct BulkheadGuard { + _permit: tokio::sync::OwnedSemaphorePermit, +} + pub struct RedisCache { connection: ConnectionManager, metrics: Option>, // Health check state with mutex to prevent concurrent health checks health_check_state: Arc>, + // Circuit breaker for Redis operations + circuit_breaker: Arc>, + // Bulkhead for limiting concurrent Redis operations + bulkhead: Arc, + // Fallback InMemory cache for when Redis is open/unhealthy + fallback_cache: InMemoryCache, + // Circuit breaker configuration + #[allow(dead_code)] + circuit_breaker_config: CacheCircuitBreakerConfig, } struct HealthCheckState { @@ -233,12 +538,29 @@ impl HealthCheckState { impl RedisCache { pub async fn new(redis_url: &str) -> Result { + Self::with_config( + redis_url, + CacheCircuitBreakerConfig::default(), + CacheBulkheadConfig::default(), + ) + .await + } + + pub async fn with_config( + redis_url: &str, + cb_config: CacheCircuitBreakerConfig, + bulkhead_config: CacheBulkheadConfig, + ) -> Result { let client = redis::Client::open(redis_url)?; let connection = ConnectionManager::new(client).await?; Ok(Self { connection, metrics: None, health_check_state: Arc::new(Mutex::new(HealthCheckState::new())), + circuit_breaker: Arc::new(Mutex::new(CacheCircuitBreaker::new(cb_config.clone()))), + bulkhead: Arc::new(CacheBulkhead::new(bulkhead_config)), + fallback_cache: InMemoryCache::new(), + circuit_breaker_config: cb_config, }) } @@ -247,6 +569,48 @@ impl RedisCache { self } + /// Check if the circuit breaker allows operations. + pub async fn circuit_breaker_allows(&self) -> bool { + self.circuit_breaker.lock().await.should_allow() + } + + /// Record a successful Redis operation with the circuit breaker. + pub async fn record_success(&self) { + let mut cb = self.circuit_breaker.lock().await; + cb.record_success(); + if let Some((from, to)) = cb.take_last_transition() { + if let Some(ref m) = self.metrics { + m.record_cache_circuit_transition(from.as_str(), to.as_str()); + m.set_cache_circuit_state(to.as_metric_value()); + if to == CircuitState::Closed { + m.increment_cache_circuit_recovery(); + } + } + } + } + + /// Record a failed Redis operation with the circuit breaker. + pub async fn record_failure(&self) { + let mut cb = self.circuit_breaker.lock().await; + cb.record_failure(); + if let Some((from, to)) = cb.take_last_transition() { + if let Some(ref m) = self.metrics { + m.record_cache_circuit_transition(from.as_str(), to.as_str()); + m.set_cache_circuit_state(to.as_metric_value()); + } + } + } + + /// Get the current circuit breaker state. + pub async fn circuit_state(&self) -> CircuitState { + self.circuit_breaker.lock().await.state() + } + + /// Get a reference to the fallback InMemory cache. + pub fn fallback(&self) -> &InMemoryCache { + &self.fallback_cache + } + async fn check_connection(&self) -> bool { let mut state = self.health_check_state.lock().await; @@ -274,36 +638,132 @@ impl RedisCache { } async fn get_raw(&self, key: &str) -> Result> { + // Check circuit breaker + if !self.circuit_breaker.lock().await.should_allow() { + if let Some(ref m) = self.metrics { + m.increment_cache_circuit_rejection(); + } + return Err(anyhow::anyhow!("Cache circuit breaker is open")); + } + + // Check bulkhead + let _permit = match self.bulkhead.try_acquire().await { + Ok(p) => p, + Err(()) => { + if let Some(ref m) = self.metrics { + m.increment_cache_bulkhead_rejection(); + } + return Err(anyhow::anyhow!("Cache bulkhead is full")); + } + }; + + // Update bulkhead metric + if let Some(ref m) = self.metrics { + m.set_cache_bulkhead_active(self.bulkhead.active_count() as i64); + } + // Validate connection state before operation if !self.check_connection().await { + self.record_failure().await; return Err(anyhow::anyhow!("Redis connection is unhealthy")); } - let mut conn = self.connection.clone(); - let value: Option = conn.get(key).await?; - Ok(value) + match self.connection.clone().get(key).await { + Ok(value) => { + self.record_success().await; + Ok(value) + } + Err(e) => { + self.record_failure().await; + Err(e.into()) + } + } } async fn set_raw(&self, key: &str, value: &str, ttl: u64) -> Result<()> { + // Check circuit breaker + if !self.circuit_breaker.lock().await.should_allow() { + if let Some(ref m) = self.metrics { + m.increment_cache_circuit_rejection(); + } + return Err(anyhow::anyhow!("Cache circuit breaker is open")); + } + + // Check bulkhead + let _permit = match self.bulkhead.try_acquire().await { + Ok(p) => p, + Err(()) => { + if let Some(ref m) = self.metrics { + m.increment_cache_bulkhead_rejection(); + } + return Err(anyhow::anyhow!("Cache bulkhead is full")); + } + }; + + // Update bulkhead metric + if let Some(ref m) = self.metrics { + m.set_cache_bulkhead_active(self.bulkhead.active_count() as i64); + } + // Validate connection state before operation if !self.check_connection().await { + self.record_failure().await; return Err(anyhow::anyhow!("Redis connection is unhealthy")); } - let mut conn = self.connection.clone(); - conn.set_ex::<_, _, ()>(key, value, ttl).await?; - Ok(()) + match self.connection.clone().set_ex::<_, _, ()>(key, value, ttl).await { + Ok(()) => { + self.record_success().await; + Ok(()) + } + Err(e) => { + self.record_failure().await; + Err(e.into()) + } + } } async fn delete(&self, key: &str) -> Result<()> { + // Check circuit breaker + if !self.circuit_breaker.lock().await.should_allow() { + if let Some(ref m) = self.metrics { + m.increment_cache_circuit_rejection(); + } + return Err(anyhow::anyhow!("Cache circuit breaker is open")); + } + + // Check bulkhead + let _permit = match self.bulkhead.try_acquire().await { + Ok(p) => p, + Err(()) => { + if let Some(ref m) = self.metrics { + m.increment_cache_bulkhead_rejection(); + } + return Err(anyhow::anyhow!("Cache bulkhead is full")); + } + }; + + // Update bulkhead metric + if let Some(ref m) = self.metrics { + m.set_cache_bulkhead_active(self.bulkhead.active_count() as i64); + } + // Validate connection state before operation if !self.check_connection().await { + self.record_failure().await; return Err(anyhow::anyhow!("Redis connection is unhealthy")); } - let mut conn = self.connection.clone(); - conn.del::<_, ()>(key).await?; - Ok(()) + match self.connection.clone().del::<_, ()>(key).await { + Ok(()) => { + self.record_success().await; + Ok(()) + } + Err(e) => { + self.record_failure().await; + Err(e.into()) + } + } } fn record_hit(&self) { @@ -1368,4 +1828,369 @@ mod tests { assert!(event1.is_ok()); assert!(event2.is_ok()); } + + // ── Circuit Breaker Tests ─────────────────────────────────────────── + + #[test] + fn circuit_breaker_starts_closed() { + let mut cb = CacheCircuitBreaker::new(CacheCircuitBreakerConfig::default()); + assert_eq!(cb.state(), CircuitState::Closed); + assert!(cb.should_allow()); + } + + #[test] + fn circuit_breaker_opens_after_threshold() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 3, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + // Record failures up to threshold + cb.record_failure(); + assert_eq!(cb.state(), CircuitState::Closed); + cb.record_failure(); + assert_eq!(cb.state(), CircuitState::Closed); + cb.record_failure(); + assert_eq!(cb.state(), CircuitState::Open); + + // Should not allow when open + assert!(!cb.should_allow()); + } + + #[test] + fn circuit_breaker_success_resets_failures() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 3, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + cb.record_failure(); + cb.record_failure(); + assert_eq!(cb.state(), CircuitState::Closed); + cb.record_success(); // resets failures + cb.record_failure(); + assert_eq!(cb.state(), CircuitState::Closed); // not open yet + } + + #[test] + fn circuit_breaker_transitions_to_half_open_after_backoff() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 1, + open_duration_ms: 10, // very short for testing + backoff_base_ms: 10, + backoff_max_ms: 100, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + // Open the circuit + cb.record_failure(); + assert_eq!(cb.state(), CircuitState::Open); + + // Initially blocked + assert!(!cb.should_allow()); + + // Wait for backoff to expire + std::thread::sleep(Duration::from_millis(20)); + + // Should now transition to HalfOpen and allow + assert!(cb.should_allow()); + assert_eq!(cb.state(), CircuitState::HalfOpen); + } + + #[test] + fn circuit_breaker_half_open_success_closes() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 1, + open_duration_ms: 10, + half_open_max_calls: 1, + backoff_base_ms: 10, + backoff_max_ms: 100, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + cb.record_failure(); // opens circuit + std::thread::sleep(Duration::from_millis(20)); + cb.should_allow(); // transitions to half_open + + cb.record_success(); // should close circuit + assert_eq!(cb.state(), CircuitState::Closed); + } + + #[test] + fn circuit_breaker_half_open_failure_reopens() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 1, + open_duration_ms: 10, + half_open_max_calls: 2, + backoff_base_ms: 10, + backoff_max_ms: 100, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + cb.record_failure(); // opens circuit + std::thread::sleep(Duration::from_millis(20)); + cb.should_allow(); // transitions to half_open + cb.record_failure(); // should reopen circuit + assert_eq!(cb.state(), CircuitState::Open); + } + + #[test] + fn circuit_breaker_exponential_backoff_grows() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 1, + open_duration_ms: 60_000, + backoff_base_ms: 100, + backoff_max_ms: 30_000, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + cb.record_failure(); + let backoff1 = cb.current_backoff(); + + cb.record_failure(); + let backoff2 = cb.current_backoff(); + + // Backoff should grow exponentially + assert!(backoff2 > backoff1); + } + + #[test] + fn circuit_breaker_backoff_caps_at_max() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 1, + open_duration_ms: 60_000, + backoff_base_ms: 100, + backoff_max_ms: 500, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + // Trigger many failures to grow backoff + for _ in 0..20 { + cb.record_failure(); + } + + let backoff = cb.current_backoff(); + assert!(backoff <= Duration::from_millis(500)); + } + + #[test] + fn circuit_breaker_take_last_transition() { + let config = CacheCircuitBreakerConfig { + failure_threshold: 1, + ..Default::default() + }; + let mut cb = CacheCircuitBreaker::new(config); + + // No transition yet + assert!(cb.take_last_transition().is_none()); + + // Open the circuit -> should record transition + cb.record_failure(); + let transition = cb.take_last_transition(); + assert!(transition.is_some()); + let (from, to) = transition.unwrap(); + assert_eq!(from, CircuitState::Closed); + assert_eq!(to, CircuitState::Open); + + // Should be cleared + assert!(cb.take_last_transition().is_none()); + } + + // ── Bulkhead Tests ────────────────────────────────────────────────── + + #[tokio::test] + async fn bulkhead_allows_up_to_max_concurrent() { + let config = CacheBulkheadConfig { + max_concurrent: 3, + max_queue: 10, + }; + let bulkhead = CacheBulkhead::new(config); + + assert_eq!(bulkhead.active_count(), 0); + + let _p1 = bulkhead.try_acquire().await.unwrap(); + assert_eq!(bulkhead.active_count(), 1); + + let _p2 = bulkhead.try_acquire().await.unwrap(); + assert_eq!(bulkhead.active_count(), 2); + + let _p3 = bulkhead.try_acquire().await.unwrap(); + assert_eq!(bulkhead.active_count(), 3); + } + + #[tokio::test] + async fn bulkhead_rejects_when_full() { + let config = CacheBulkheadConfig { + max_concurrent: 2, + max_queue: 0, // no queue + }; + let bulkhead = CacheBulkhead::new(config); + + let _p1 = bulkhead.try_acquire().await.unwrap(); + let _p2 = bulkhead.try_acquire().await.unwrap(); + + // Should reject when at capacity + assert!(bulkhead.try_acquire().await.is_err()); + } + + #[tokio::test] + async fn bulkhead_permits_release_on_drop() { + let config = CacheBulkheadConfig { + max_concurrent: 1, + max_queue: 10, + }; + let bulkhead = CacheBulkhead::new(config); + + { + let _permit = bulkhead.try_acquire().await.unwrap(); + assert_eq!(bulkhead.active_count(), 1); + } // permit dropped here + + assert_eq!(bulkhead.active_count(), 0); + assert!(bulkhead.try_acquire().await.is_ok()); + } + + #[tokio::test] + async fn bulkhead_limits_concurrent_permits() { + let config = CacheBulkheadConfig { + max_concurrent: 3, + max_queue: 10, + }; + let bulkhead = CacheBulkhead::new(config); + + // Use all concurrent slots + let _p1 = bulkhead.try_acquire().await.unwrap(); + let _p2 = bulkhead.try_acquire().await.unwrap(); + let _p3 = bulkhead.try_acquire().await.unwrap(); + + // At max capacity — should reject + assert!(bulkhead.try_acquire().await.is_err()); + + // Drop one permit + drop(_p1); + + // Now should succeed again + assert!(bulkhead.try_acquire().await.is_ok()); + } + + // ── Circuit Breaker Metrics Test ──────────────────────────────────── + + #[test] + fn circuit_state_as_str_and_metric() { + assert_eq!(CircuitState::Closed.as_str(), "closed"); + assert_eq!(CircuitState::Open.as_str(), "open"); + assert_eq!(CircuitState::HalfOpen.as_str(), "half_open"); + + assert_eq!(CircuitState::Closed.as_metric_value(), 0); + assert_eq!(CircuitState::Open.as_metric_value(), 1); + assert_eq!(CircuitState::HalfOpen.as_metric_value(), 2); + } + + // ── Fallback Integration Tests ────────────────────────────────────── + + #[tokio::test] + async fn redis_cache_fallback_to_inmemory_on_circuit_open() { + // Create a RedisCache with a very low failure threshold + let cb_config = CacheCircuitBreakerConfig { + failure_threshold: 1, + open_duration_ms: 60_000, + half_open_max_calls: 1, + backoff_base_ms: 100, + backoff_max_ms: 30_000, + }; + let bulkhead_config = CacheBulkheadConfig::default(); + + // Try to connect to Redis — if unavailable, we test the fallback path + let redis_cache = + RedisCache::with_config("redis://127.0.0.1:6379", cb_config, bulkhead_config).await; + if redis_cache.is_err() { + return; // Skip if Redis unavailable + } + let redis_cache = redis_cache.unwrap(); + let metrics = MetricsRegistry::arc(); + let redis_cache = redis_cache.with_metrics(Arc::clone(&metrics)); + let backend = CacheBackend::Redis(redis_cache); + + // Open the circuit breaker by recording a failure + if let CacheBackend::Redis(ref c) = backend { + c.record_failure().await; + assert_eq!(c.circuit_state().await, CircuitState::Open); + } + + // Set a value — should fallback to InMemory + let key = CacheKey::Verification("fallback_test".to_string()); + backend.set_raw(&key, "fallback_value", 60).await.unwrap(); + + // Get the value — should come from fallback + let value = backend.get_raw(&key).await.unwrap(); + assert_eq!(value, Some("fallback_value".to_string())); + + // Check fallback metric + let output = metrics.render(); + assert!(output.contains("cache_fallback_uses_total")); + } + + #[tokio::test] + async fn cache_backend_inmemory_has_no_circuit_breaker() { + // InMemory backend should always work regardless of circuit breaker state + let cache = CacheBackend::InMemory(InMemoryCache::new()); + let key = CacheKey::Verification("no_cb".to_string()); + + cache.set_raw(&key, "value", 60).await.unwrap(); + let value = cache.get_raw(&key).await.unwrap(); + assert_eq!(value, Some("value".to_string())); + + cache.delete(&key).await.unwrap(); + assert_eq!(cache.get_raw(&key).await.unwrap(), None); + } + + #[tokio::test] + async fn circuit_breaker_config_defaults() { + let config = CacheCircuitBreakerConfig::default(); + assert_eq!(config.failure_threshold, 5); + assert_eq!(config.open_duration_ms, 30_000); + assert_eq!(config.half_open_max_calls, 1); + assert_eq!(config.backoff_base_ms, 100); + assert_eq!(config.backoff_max_ms, 30_000); + } + + #[tokio::test] + async fn bulkhead_config_defaults() { + let config = CacheBulkheadConfig::default(); + assert_eq!(config.max_concurrent, 20); + assert_eq!(config.max_queue, 200); + } + + // ── Error Variant Tests ───────────────────────────────────────────── + + #[test] + fn circuit_breaker_error_display() { + use crate::error::AuditError; + let err = AuditError::CircuitBreakerOpen("redis is down".to_string()); + assert!(err.to_string().contains("circuit breaker open")); + assert!(err.to_string().contains("redis is down")); + } + + #[test] + fn bulkhead_full_error_display() { + use crate::error::AuditError; + let err = AuditError::BulkheadFull("too many operations".to_string()); + assert!(err.to_string().contains("bulkhead full")); + assert!(err.to_string().contains("too many operations")); + } + + #[test] + fn cache_unhealthy_error_display() { + use crate::error::AuditError; + let err = AuditError::CacheUnhealthy("connection refused".to_string()); + assert!(err.to_string().contains("cache unhealthy")); + assert!(err.to_string().contains("connection refused")); + } } diff --git a/src/config.rs b/src/config.rs index b0cea27..387363e 100644 --- a/src/config.rs +++ b/src/config.rs @@ -21,6 +21,14 @@ const DEFAULT_STELLAR_RETRY_JITTER_TYPE: &str = "full"; const DEFAULT_STELLAR_BULKHEAD_MAX_CONCURRENT: u32 = 10; const DEFAULT_STELLAR_BULKHEAD_MAX_QUEUE: u32 = 100; +const DEFAULT_CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD: u32 = 5; +const DEFAULT_CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS: u64 = 30_000; +const DEFAULT_CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS: u32 = 1; +const DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS: u64 = 100; +const DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS: u64 = 30_000; +const DEFAULT_CACHE_BULKHEAD_MAX_CONCURRENT: u32 = 20; +const DEFAULT_CACHE_BULKHEAD_MAX_QUEUE: u32 = 200; + /// Current configuration schema version. /// Increment this when adding or removing fields that break backward compatibility. pub const CONFIG_VERSION: u32 = 1; @@ -247,6 +255,17 @@ pub struct AppConfig { pub cache_max_size: usize, pub cache_config_ttl: u64, pub cache_events_ttl: u64, + + // ── Cache circuit breaker configuration ───────────────────────────── + pub cache_circuit_breaker_failure_threshold: u32, + pub cache_circuit_breaker_open_duration_ms: u64, + pub cache_circuit_breaker_half_open_max_calls: u32, + pub cache_circuit_breaker_backoff_base_ms: u64, + pub cache_circuit_breaker_backoff_max_ms: u64, + + // ── Cache bulkhead configuration ─────────────────────────────────── + pub cache_bulkhead_max_concurrent: u32, + pub cache_bulkhead_max_queue: u32, } impl fmt::Debug for AppConfig { @@ -336,6 +355,13 @@ impl fmt::Debug for AppConfig { .field("cache_max_size", &self.cache_max_size) .field("cache_config_ttl", &self.cache_config_ttl) .field("cache_events_ttl", &self.cache_events_ttl) + .field("cache_circuit_breaker_failure_threshold", &self.cache_circuit_breaker_failure_threshold) + .field("cache_circuit_breaker_open_duration_ms", &self.cache_circuit_breaker_open_duration_ms) + .field("cache_circuit_breaker_half_open_max_calls", &self.cache_circuit_breaker_half_open_max_calls) + .field("cache_circuit_breaker_backoff_base_ms", &self.cache_circuit_breaker_backoff_base_ms) + .field("cache_circuit_breaker_backoff_max_ms", &self.cache_circuit_breaker_backoff_max_ms) + .field("cache_bulkhead_max_concurrent", &self.cache_bulkhead_max_concurrent) + .field("cache_bulkhead_max_queue", &self.cache_bulkhead_max_queue) .finish() } } @@ -408,6 +434,17 @@ impl AppConfig { // CACHE_VERIFICATION_TTL - TTL for verification cache (default: 3600) // CACHE_CONFIG_TTL - TTL for config cache (default: 3600) // CACHE_EVENTS_TTL - TTL for events cache (default: 1800) + // + // CACHE CIRCUIT BREAKER: + // CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD - Failures before circuit opens (default: 5) + // CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS - Duration circuit stays open (default: 30000) + // CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS - Probes in half-open state (default: 1) + // CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS - Exponential backoff base (default: 100) + // CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS - Exponential backoff max (default: 30000) + // + // CACHE BULKHEAD: + // CACHE_BULKHEAD_MAX_CONCURRENT - Max concurrent Redis operations (default: 20) + // CACHE_BULKHEAD_MAX_QUEUE - Max queued operations when at capacity (default: 200) let port_raw = get_env_or_default("PORT", "8080"); let stellar_horizon_url = @@ -503,6 +540,34 @@ impl AppConfig { let cache_max_size_raw = get_env_or_default("CACHE_MAX_SIZE", "10000"); let cache_config_ttl_raw = get_env_or_default("CACHE_CONFIG_TTL", "3600"); let cache_events_ttl_raw = get_env_or_default("CACHE_EVENTS_TTL", "1800"); + let cache_cb_failure_threshold_raw = get_env_or_default( + "CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD", + &DEFAULT_CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD.to_string(), + ); + let cache_cb_open_duration_ms_raw = get_env_or_default( + "CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS", + &DEFAULT_CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS.to_string(), + ); + let cache_cb_half_open_max_calls_raw = get_env_or_default( + "CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS", + &DEFAULT_CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS.to_string(), + ); + let cache_cb_backoff_base_ms_raw = get_env_or_default( + "CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS", + &DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS.to_string(), + ); + let cache_cb_backoff_max_ms_raw = get_env_or_default( + "CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS", + &DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS.to_string(), + ); + let cache_bulkhead_max_concurrent_raw = get_env_or_default( + "CACHE_BULKHEAD_MAX_CONCURRENT", + &DEFAULT_CACHE_BULKHEAD_MAX_CONCURRENT.to_string(), + ); + let cache_bulkhead_max_queue_raw = get_env_or_default( + "CACHE_BULKHEAD_MAX_QUEUE", + &DEFAULT_CACHE_BULKHEAD_MAX_QUEUE.to_string(), + ); // ── Port validation with bounds ────────────────────────────────── let port: u16 = match port_raw.parse() { @@ -888,6 +953,145 @@ impl AppConfig { } }; + // ── Cache circuit breaker validation ───────────────────────────── + let cache_circuit_breaker_failure_threshold: u32 = + match cache_cb_failure_threshold_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push( + "CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD must be greater than 0" + .to_string(), + ); + DEFAULT_CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD + } + Err(_) => { + errors.push(format!( + "CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD must be a valid u32, got '{}'", + cache_cb_failure_threshold_raw + )); + DEFAULT_CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD + } + }; + + let cache_circuit_breaker_open_duration_ms: u64 = + match cache_cb_open_duration_ms_raw.parse() { + Ok(v) if v > 0 && v <= MAX_TIMEOUT_MS => v, + Ok(v) if v > 0 => { + errors.push(format!( + "CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS ({}) exceeds maximum {}", + v, MAX_TIMEOUT_MS + )); + DEFAULT_CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS + } + Ok(_) => { + errors.push( + "CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS must be greater than 0" + .to_string(), + ); + DEFAULT_CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS + } + Err(_) => { + errors.push(format!( + "CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS must be a valid u64, got '{}'", + cache_cb_open_duration_ms_raw + )); + DEFAULT_CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS + } + }; + + let cache_circuit_breaker_half_open_max_calls: u32 = + match cache_cb_half_open_max_calls_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push( + "CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS must be greater than 0" + .to_string(), + ); + DEFAULT_CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS + } + Err(_) => { + errors.push(format!( + "CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS must be a valid u32, got '{}'", + cache_cb_half_open_max_calls_raw + )); + DEFAULT_CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS + } + }; + + let cache_circuit_breaker_backoff_base_ms: u64 = + match cache_cb_backoff_base_ms_raw.parse() { + Ok(v) if v > 0 && v <= MAX_TIMEOUT_MS => v, + Ok(_) => { + errors.push( + "CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS must be greater than 0" + .to_string(), + ); + DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS + } + Err(_) => { + errors.push(format!( + "CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS must be a valid u64, got '{}'", + cache_cb_backoff_base_ms_raw + )); + DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS + } + }; + + let cache_circuit_breaker_backoff_max_ms: u64 = + match cache_cb_backoff_max_ms_raw.parse() { + Ok(v) if v > 0 && v <= MAX_TIMEOUT_MS => v, + Ok(_) => { + errors.push( + "CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS must be greater than 0" + .to_string(), + ); + DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS + } + Err(_) => { + errors.push(format!( + "CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS must be a valid u64, got '{}'", + cache_cb_backoff_max_ms_raw + )); + DEFAULT_CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS + } + }; + + // ── Cache bulkhead validation ──────────────────────────────────── + let cache_bulkhead_max_concurrent: u32 = + match cache_bulkhead_max_concurrent_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push( + "CACHE_BULKHEAD_MAX_CONCURRENT must be greater than 0".to_string(), + ); + DEFAULT_CACHE_BULKHEAD_MAX_CONCURRENT + } + Err(_) => { + errors.push(format!( + "CACHE_BULKHEAD_MAX_CONCURRENT must be a valid u32, got '{}'", + cache_bulkhead_max_concurrent_raw + )); + DEFAULT_CACHE_BULKHEAD_MAX_CONCURRENT + } + }; + + let cache_bulkhead_max_queue: u32 = match cache_bulkhead_max_queue_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push( + "CACHE_BULKHEAD_MAX_QUEUE must be greater than 0".to_string(), + ); + DEFAULT_CACHE_BULKHEAD_MAX_QUEUE + } + Err(_) => { + errors.push(format!( + "CACHE_BULKHEAD_MAX_QUEUE must be a valid u32, got '{}'", + cache_bulkhead_max_queue_raw + )); + DEFAULT_CACHE_BULKHEAD_MAX_QUEUE + } + }; + // Log level validation match log_level.to_lowercase().as_str() { "trace" | "debug" | "info" | "warn" | "error" => {} @@ -1064,6 +1268,13 @@ impl AppConfig { cache_max_size, cache_config_ttl, cache_events_ttl, + cache_circuit_breaker_failure_threshold, + cache_circuit_breaker_open_duration_ms, + cache_circuit_breaker_half_open_max_calls, + cache_circuit_breaker_backoff_base_ms, + cache_circuit_breaker_backoff_max_ms, + cache_bulkhead_max_concurrent, + cache_bulkhead_max_queue, }) } @@ -1135,6 +1346,13 @@ mod tests { "CACHE_MAX_SIZE", "CACHE_CONFIG_TTL", "CACHE_EVENTS_TTL", + "CACHE_CIRCUIT_BREAKER_FAILURE_THRESHOLD", + "CACHE_CIRCUIT_BREAKER_OPEN_DURATION_MS", + "CACHE_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS", + "CACHE_CIRCUIT_BREAKER_BACKOFF_BASE_MS", + "CACHE_CIRCUIT_BREAKER_BACKOFF_MAX_MS", + "CACHE_BULKHEAD_MAX_CONCURRENT", + "CACHE_BULKHEAD_MAX_QUEUE", ]; for key in keys { env::remove_var(key); @@ -1339,6 +1557,13 @@ mod tests { cache_max_size: 10000, cache_config_ttl: 3600, cache_events_ttl: 1800, + cache_circuit_breaker_failure_threshold: 5, + cache_circuit_breaker_open_duration_ms: 30_000, + cache_circuit_breaker_half_open_max_calls: 1, + cache_circuit_breaker_backoff_base_ms: 100, + cache_circuit_breaker_backoff_max_ms: 30_000, + cache_bulkhead_max_concurrent: 20, + cache_bulkhead_max_queue: 200, }; let debug = format!("{:?}", config); diff --git a/src/error.rs b/src/error.rs index 751a285..f7f4207 100644 --- a/src/error.rs +++ b/src/error.rs @@ -9,6 +9,9 @@ pub type Result = core::result::Result; pub enum AuditError { SerializationError(String), InvalidContractEventContext(String), + CircuitBreakerOpen(String), + BulkheadFull(String), + CacheUnhealthy(String), } impl fmt::Display for AuditError { @@ -18,6 +21,15 @@ impl fmt::Display for AuditError { Self::InvalidContractEventContext(message) => { write!(f, "invalid contract event context: {message}") } + Self::CircuitBreakerOpen(message) => { + write!(f, "circuit breaker open: {message}") + } + Self::BulkheadFull(message) => { + write!(f, "bulkhead full: {message}") + } + Self::CacheUnhealthy(message) => { + write!(f, "cache unhealthy: {message}") + } } } } diff --git a/src/main.rs b/src/main.rs index 01804fd..5acf03a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -42,7 +42,7 @@ mod native { use axum::{Json, Router}; use serde_json::json; - use proofstell_contract::cache::{CacheBackend, InMemoryCache}; + use proofstell_contract::cache::{CacheBackend, CacheBulkheadConfig, CacheCircuitBreakerConfig, InMemoryCache}; use proofstell_contract::config::{self, AppConfig, ConfigUpdate, ConfigWatcher}; use proofstell_contract::metrics::MetricsRegistry; use proofstell_contract::webhook::WebhookDispatcher; @@ -202,7 +202,22 @@ mod native { let cache: Arc = match config.cache_backend.as_str() { "redis" => { eprintln!("[proofstell] Initializing Redis cache backend..."); - match proofstell_contract::cache::RedisCache::new(&config.redis_url).await { + let cb_config = CacheCircuitBreakerConfig { + failure_threshold: config.cache_circuit_breaker_failure_threshold, + open_duration_ms: config.cache_circuit_breaker_open_duration_ms, + half_open_max_calls: config.cache_circuit_breaker_half_open_max_calls, + backoff_base_ms: config.cache_circuit_breaker_backoff_base_ms, + backoff_max_ms: config.cache_circuit_breaker_backoff_max_ms, + }; + let bulkhead_config = CacheBulkheadConfig { + max_concurrent: config.cache_bulkhead_max_concurrent, + max_queue: config.cache_bulkhead_max_queue, + }; + match proofstell_contract::cache::RedisCache::with_config( + &config.redis_url, + cb_config, + bulkhead_config, + ).await { Ok(redis_cache) => { let cache = redis_cache.with_metrics(Arc::clone(&metrics)); Arc::new(CacheBackend::Redis(cache)) diff --git a/src/metrics.rs b/src/metrics.rs index d7d0cd3..5891fde 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -62,6 +62,17 @@ pub struct MetricsRegistry { circuit_transitions_total: IntCounterVec, circuit_state_changes_total: IntCounterVec, + // ── Cache circuit breaker metrics ── + cache_circuit_state: Gauge, + cache_circuit_transitions_total: IntCounterVec, + cache_circuit_rejections_total: IntCounter, + cache_circuit_recoveries_total: IntCounter, + cache_fallback_uses_total: IntCounter, + + // ── Cache bulkhead metrics ── + cache_bulkhead_active: Gauge, + cache_bulkhead_rejections_total: IntCounter, + // ── Webhook delivery metrics ── webhook_deliveries_total: IntCounterVec, webhook_delivery_latency_seconds: HistogramVec, @@ -250,6 +261,53 @@ impl MetricsRegistry { ) .unwrap(); + // ── Cache circuit breaker metrics ──────────────────────────────── + let cache_circuit_state = Gauge::new( + "cache_circuit_breaker_state", + "Current cache circuit breaker state (0=closed, 1=open, 2=half_open)", + ) + .unwrap(); + + let cache_circuit_transitions_total = IntCounterVec::new( + Opts::new( + "cache_circuit_breaker_transitions_total", + "Total cache circuit breaker state transitions", + ), + &["from_state", "to_state"], + ) + .unwrap(); + + let cache_circuit_rejections_total = IntCounter::new( + "cache_circuit_breaker_rejections_total", + "Total cache operations rejected by circuit breaker", + ) + .unwrap(); + + let cache_circuit_recoveries_total = IntCounter::new( + "cache_circuit_breaker_recoveries_total", + "Total cache circuit breaker recovery successes", + ) + .unwrap(); + + let cache_fallback_uses_total = IntCounter::new( + "cache_fallback_uses_total", + "Total cache operations falling back to InMemory", + ) + .unwrap(); + + // ── Cache bulkhead metrics ─────────────────────────────────────── + let cache_bulkhead_active = Gauge::new( + "cache_bulkhead_active_operations", + "Current number of active cache bulkhead operations", + ) + .unwrap(); + + let cache_bulkhead_rejections_total = IntCounter::new( + "cache_bulkhead_rejections_total", + "Total cache operations rejected by bulkhead", + ) + .unwrap(); + // ── Webhook delivery metrics ── let webhook_deliveries_total = IntCounterVec::new( Opts::new( @@ -311,6 +369,13 @@ impl MetricsRegistry { Box::new(circuit_state.clone()), Box::new(circuit_transitions_total.clone()), Box::new(circuit_state_changes_total.clone()), + Box::new(cache_circuit_state.clone()), + Box::new(cache_circuit_transitions_total.clone()), + Box::new(cache_circuit_rejections_total.clone()), + Box::new(cache_circuit_recoveries_total.clone()), + Box::new(cache_fallback_uses_total.clone()), + Box::new(cache_bulkhead_active.clone()), + Box::new(cache_bulkhead_rejections_total.clone()), Box::new(webhook_deliveries_total.clone()), Box::new(webhook_delivery_latency_seconds.clone()), Box::new(webhook_dlq_depth.clone()), @@ -349,6 +414,13 @@ impl MetricsRegistry { circuit_state, circuit_transitions_total, circuit_state_changes_total, + cache_circuit_state, + cache_circuit_transitions_total, + cache_circuit_rejections_total, + cache_circuit_recoveries_total, + cache_fallback_uses_total, + cache_bulkhead_active, + cache_bulkhead_rejections_total, webhook_deliveries_total, webhook_delivery_latency_seconds, webhook_dlq_depth, @@ -523,6 +595,40 @@ impl MetricsRegistry { .inc(); } + // ── Cache circuit breaker metrics ────────────────────────────────── + + pub fn set_cache_circuit_state(&self, state: i64) { + self.cache_circuit_state.set(state as f64); + } + + pub fn record_cache_circuit_transition(&self, from_state: &str, to_state: &str) { + self.cache_circuit_transitions_total + .with_label_values(&[from_state, to_state]) + .inc(); + } + + pub fn increment_cache_circuit_rejection(&self) { + self.cache_circuit_rejections_total.inc(); + } + + pub fn increment_cache_circuit_recovery(&self) { + self.cache_circuit_recoveries_total.inc(); + } + + pub fn increment_cache_fallback_use(&self) { + self.cache_fallback_uses_total.inc(); + } + + // ── Cache bulkhead metrics ───────────────────────────────────────── + + pub fn set_cache_bulkhead_active(&self, active: i64) { + self.cache_bulkhead_active.set(active as f64); + } + + pub fn increment_cache_bulkhead_rejection(&self) { + self.cache_bulkhead_rejections_total.inc(); + } + // ── Webhook delivery metrics ────────────────────────────────────── /// Record a completed delivery attempt (success or dead_lettered) with latency. @@ -609,6 +715,13 @@ mod tests { metrics.decrement_event_backlog(); metrics.increment_config_validation_failure(); metrics.increment_config_reload(); + metrics.set_cache_circuit_state(1); + metrics.record_cache_circuit_transition("closed", "open"); + metrics.increment_cache_circuit_rejection(); + metrics.increment_cache_circuit_recovery(); + metrics.increment_cache_fallback_use(); + metrics.set_cache_bulkhead_active(3); + metrics.increment_cache_bulkhead_rejection(); metrics.record_webhook_delivery("success", 0.05); metrics.record_webhook_delivery("dead_lettered", 1.0); metrics.increment_webhook_retry(); @@ -624,6 +737,13 @@ mod tests { assert!(output.contains("rate_limit_rejections_total")); assert!(output.contains("event_backlog_size")); assert!(output.contains("config_validation_failures_total")); + assert!(output.contains("cache_circuit_breaker_state")); + assert!(output.contains("cache_circuit_breaker_transitions_total")); + assert!(output.contains("cache_circuit_breaker_rejections_total")); + assert!(output.contains("cache_circuit_breaker_recoveries_total")); + assert!(output.contains("cache_fallback_uses_total")); + assert!(output.contains("cache_bulkhead_active_operations")); + assert!(output.contains("cache_bulkhead_rejections_total")); assert!(output.contains("webhook_deliveries_total")); assert!(output.contains("webhook_delivery_latency_seconds")); assert!(output.contains("webhook_dlq_depth"));