feat: Implement circuit breaker pattern for external APIs

Implements comprehensive circuit breaker protection for Ozone API and
PDS blob downloads to prevent cascading failures during API degradation.

Changes:
- Created generic CircuitBreaker with states: Closed → Open → Half-Open → Closed
- Implemented three circuit breaker instances:
* Ozone API: 5 consecutive failures → open, 60s timeout, 1 success to close
* PDS Blob: 3 consecutive failures → open, 5m timeout, 1 success to close
* PLC Resolution: 3 consecutive failures → open, 5m timeout (created but not yet integrated)

Implementation Details:
- Circuit breaker checks added before all Ozone moderation API calls
- Circuit breaker checks added before PDS blob fallback downloads
- Automatic state transitions with configurable timeouts
- Thread-safe design using Arc<RwLock<>>
- Metrics tracking for state transitions and rejections
- Comprehensive unit tests covering all state transitions

Architecture:
- src/resilience/circuit_breaker.rs: Core circuit breaker implementation
- src/main.rs: Initialize three circuit breaker instances with metrics
- src/queue/worker.rs: Pass circuit breakers to moderation and download functions
- src/processor/matcher.rs: Circuit breaker protection for PDS blob downloads
- src/moderation/helpers.rs: Circuit breaker protection for Ozone API calls
- src/metrics/mod.rs: Track circuit_breaker_transitions and circuit_breaker_rejections

Tests:
- All 7 circuit breaker unit tests passing
- Verified state transitions (closed → open → half-open → closed)
- Tested failure threshold triggers opening
- Tested success during half-open closes circuit
- Tested failure during half-open reopens circuit
- Tested timeout calculation and half-open transition
- Tested half_open_max_calls limiting

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[e-mail]>

PLAN_REMAINING.md

···
+# Production Resilience Improvements - Implementation Plan
+
+## Overview
+Complete the remaining critical issues needed for full production readiness. Each task includes implementation, testing, and verification.
+
+---
+
+## Task 1: Implement Circuit Breaker Pattern for External APIs
+
+**Objective:** Prevent cascading failures when external APIs (Ozone, PDS, PLC) degrade or fail.
+
+**Requirements:**
+1. Create a circuit breaker module (`src/resilience/circuit_breaker.rs`)
+2. Implement three independent circuit breakers:
+   - **Ozone API** - Opens after 5 consecutive failures, half-opens after 60s
+   - **PDS Blob Fetch** - Opens after 3 consecutive failures per endpoint, 5m timeout
+   - **PLC Resolution** - Opens after 3 consecutive failures per endpoint, 5m timeout
+3. Circuit breaker states: Closed → Open → Half-Open → Closed
+4. Add metrics tracking: circuit_breaker_state, circuit_breaker_transitions
+5. Update error handling to respect circuit breaker state
+6. Add comprehensive tests:
+   - State transitions (closed → open → half-open → closed)
+   - Failure threshold triggers
+   - Success during half-open closes circuit
+   - Failure during half-open reopens circuit
+   - Timeout calculation
+
+**Files to Modify:**
+- Create: `src/resilience/circuit_breaker.rs`
+- Create: `src/resilience/mod.rs`
+- Modify: `src/moderation/helpers.rs` - Wrap Ozone calls with circuit breaker
+- Modify: `src/processor/matcher.rs` - Wrap PDS/CDN calls with circuit breaker
+- Modify: `src/plc/mod.rs` - Wrap PLC calls with circuit breaker
+- Modify: `src/main.rs` - Initialize circuit breakers
+- Modify: `src/metrics/mod.rs` - Add circuit breaker metrics
+
+**Tests Required:**
+- Unit tests for circuit breaker state machine
+- Integration tests for API call wrapping
+- Test timeout and recovery scenarios
+
+---
+
+## Task 2: Add Redis Connection Failure Backoff and Recovery
+
+**Objective:** Handle Redis unavailability gracefully instead of crashing or spinning in error loops.
+
+**Requirements:**
+1. Implement connection retry logic with exponential backoff
+2. Add connection pooling via `ConnectionManager` (redis crate provides this)
+3. Add health check mechanism with configurable interval
+4. Implement graceful degradation:
+   - First failure: Log and retry with 100ms delay
+   - Exponential backoff: Cap at 10s
+   - After 5 consecutive failures: Switch to circuit breaker state
+5. Add configuration:
+   - `REDIS_HEALTH_CHECK_INTERVAL_SECS` (default: 30s)
+   - `REDIS_MAX_BACKOFF_SECS` (default: 10s)
+6. Add metrics: `redis_connection_failures`, `redis_reconnect_attempts`, `redis_health_status`
+7. Add logging for connection state changes
+
+**Files to Modify:**
+- Modify: `src/config/mod.rs` - Add redis retry configuration
+- Create: `src/redis_pool.rs` - Redis connection manager with backoff
+- Modify: `src/queue/redis_queue.rs` - Use connection manager
+- Modify: `src/cache/mod.rs` - Use connection manager
+- Modify: `src/main.rs` - Initialize redis pool
+- Modify: `src/metrics/mod.rs` - Add redis metrics
+
+**Tests Required:**
+- Unit tests for backoff calculation
+- Connection failure retry tests
+- Health check tests
+- Graceful degradation tests
+
+---
+
+## Task 3: Add Integration Tests for Critical Paths
+
+**Objective:** Achieve >50% test coverage for critical paths (up from <5% currently).
+
+**Requirements:**
+1. Use `testcontainers` for Redis integration tests
+2. Test critical paths:
+   - **Worker job processing flow** - Mock blob download, verify cache, verify moderation
+   - **Jetstream message → Job Queue flow** - Verify message parsing, job creation
+   - **Moderation actions** - Test with mock Ozone API, verify retry logic
+   - **Blob download fallback** - Test CDN failure → PDS success
+   - **Cache operations** - Test cache hit/miss, TTL expiration
+3. Tests must be isolated (no shared state between tests)
+4. Use `tokio::test` for async tests
+5. Mock external services (Ozone, PDS, PLC) where appropriate
+6. Test error scenarios:
+   - Blob download timeout
+   - Cache miss
+   - Moderation API retry exhaustion
+   - Redis unavailability
+7. Organize tests in `tests/integration/` directory
+
+**Files to Create:**
+- Create: `tests/integration/mod.rs`
+- Create: `tests/integration/worker_test.rs` - Worker job processing
+- Create: `tests/integration/cache_test.rs` - Cache operations
+- Create: `tests/integration/blob_download_test.rs` - Download with fallback
+- Create: `tests/integration/moderation_test.rs` - Moderation actions with retry
+- Create: `tests/integration/helpers.rs` - Test fixtures and mocks
+
+**Acceptance Criteria:**
+- All 10+ integration tests pass
+- >50% code coverage for critical modules (worker, cache, moderation)
+- Tests are deterministic (no flakiness)
+- Tests complete in <30s total
+
+---
+
+## Success Criteria
+
+- [ ] All tasks implemented and tests passing
+- [ ] No panics or unwraps in production code
+- [ ] Circuit breaker prevents cascading failures
+- [ ] Redis connection failures handled gracefully
+- [ ] Integration tests provide >50% coverage
+- [ ] Code review passes for all changes
+- [ ] All changes committed to `feat/production-resilience-improvements` branch

src/main.rs

···
     processor::matcher,
     queue::{JobQueue, WorkerPool},
     redis_pool::RedisPool,
+    resilience::CircuitBreaker,
 };
 
 #[tokio::main]
···
     let queue = JobQueue::new(redis_pool.clone(), config.processing.retry_attempts);
     info!("Job queue initialized");
 
+    // Create circuit breakers with metrics
+    let ozone_circuit_breaker = CircuitBreaker::with_metrics(
+        "ozone-api",
+        5,  // 5 consecutive failures
+        60, // 60s timeout
+        1,  // 1 success to close
+        metrics.clone(),
+    );
+    info!("Ozone API circuit breaker initialized");
+
+    let pds_circuit_breaker = CircuitBreaker::with_metrics(
+        "pds-blob",
+        3,   // 3 consecutive failures
+        300, // 5m timeout
+        1,   // 1 success to close
+        metrics.clone(),
+    );
+    info!("PDS blob circuit breaker initialized");
+
+    let _plc_circuit_breaker = CircuitBreaker::with_metrics(
+        "plc-resolution",
+        3,   // 3 consecutive failures
+        300, // 5m timeout
+        1,   // 1 success to close
+        metrics.clone(),
+    );
+    info!("PLC resolution circuit breaker initialized (not yet integrated)");
+
     // Create worker pool
     let worker_pool = WorkerPool::new(
         config.clone(),
···
         agent.clone(),
         blob_checks.clone(),
         metrics.clone(),
+        ozone_circuit_breaker,
+        pds_circuit_breaker,
     )?;
     info!(
         "Worker pool created with {} workers",

src/metrics/mod.rs

···
     redis_connection_failures: AtomicU64,
     redis_reconnect_attempts: AtomicU64,
     redis_health_status: AtomicU64,
+
+    // Circuit breaker metrics
+    circuit_breaker_transitions: AtomicU64,
+    circuit_breaker_rejections: AtomicU64,
 }
 
 impl Metrics {
···
                 redis_connection_failures: AtomicU64::new(0),
                 redis_reconnect_attempts: AtomicU64::new(0),
                 redis_health_status: AtomicU64::new(1),
+                circuit_breaker_transitions: AtomicU64::new(0),
+                circuit_breaker_rejections: AtomicU64::new(0),
             }),
         }
     }
···
         self.inner.redis_health_status.store(if healthy { 1 } else { 0 }, Ordering::Relaxed);
     }
 
+    // Circuit breaker metrics
+    pub fn inc_circuit_breaker_transitions(&self) {
+        self.inner.circuit_breaker_transitions.fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn inc_circuit_breaker_rejections(&self) {
+        self.inner.circuit_breaker_rejections.fetch_add(1, Ordering::Relaxed);
+    }
+
     // Getters
     pub fn jobs_received(&self) -> u64 {
         self.inner.jobs_received.load(Ordering::Relaxed)
···
         self.inner.redis_health_status.load(Ordering::Relaxed) == 1
     }
 
+    pub fn circuit_breaker_transitions(&self) -> u64 {
+        self.inner.circuit_breaker_transitions.load(Ordering::Relaxed)
+    }
+
+    pub fn circuit_breaker_rejections(&self) -> u64 {
+        self.inner.circuit_breaker_rejections.load(Ordering::Relaxed)
+    }
+
     /// Log current metrics
     pub fn log_stats(&self) {
         info!("=== Metrics ===");
···
             self.redis_reconnect_attempts(),
             if self.redis_health_status() { "healthy" } else { "degraded" }
         );
+        info!("Circuit breakers: transitions={}, rejections={}",
+            self.circuit_breaker_transitions(),
+            self.circuit_breaker_rejections()
+        );
     }
 
     /// Calculate cache hit rate
···
             redis_connection_failures: self.redis_connection_failures(),
             redis_reconnect_attempts: self.redis_reconnect_attempts(),
             redis_health_status: if self.redis_health_status() { 1 } else { 0 },
+            circuit_breaker_transitions: self.circuit_breaker_transitions(),
+            circuit_breaker_rejections: self.circuit_breaker_rejections(),
         }
     }
 }
···
     pub redis_connection_failures: u64,
     pub redis_reconnect_attempts: u64,
     pub redis_health_status: u64,
+    pub circuit_breaker_transitions: u64,
+    pub circuit_breaker_rejections: u64,
 }
 
 #[cfg(test)]

src/moderation/account.rs

···
 use crate::moderation::{
     build_mod_tool_meta, build_timestamped_comment, send_moderation_event,
 };
+use crate::resilience::CircuitBreaker;
 
 /// Label an account with a specific label via Ozone moderation API
 pub async fn label_account<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
+    circuit_breaker: Option<&CircuitBreaker>,
     did: &Did<'a>,
     label_val: &str,
     check_comment: &str,
···
         })
         .build();
 
-    send_moderation_event(agent, config, rate_limiter, event).await?;
+    send_moderation_event(agent, config, rate_limiter, circuit_breaker, event).await?;
 
     debug!("Successfully labeled account: {}", did);
 
···
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
+    circuit_breaker: Option<&CircuitBreaker>,
     did: &Did<'a>,
     reason: ReasonType<'static>,
     check_comment: &str,
···
         })
         .build();
 
-    send_moderation_event(agent, config, rate_limiter, event).await?;
+    send_moderation_event(agent, config, rate_limiter, circuit_breaker, event).await?;
 
     debug!("Successfully reported account: {}", did);
 
···
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
+    circuit_breaker: Option<&CircuitBreaker>,
     did: &Did<'a>,
     comment: &str,
     created_by: &Did<'a>,
···
         )))
         .build();
 
-    send_moderation_event(agent, config, rate_limiter, event).await?;
+    send_moderation_event(agent, config, rate_limiter, circuit_breaker, event).await?;
 
     debug!("Successfully took down account: {}", did);
 

src/moderation/helpers.rs

···
 
 use crate::config::Config;
 use crate::moderation::rate_limiter::RateLimiter;
+use crate::resilience::CircuitBreaker;
 
 pub fn build_timestamped_comment(check_comment: &str, uri: &str, phash: &str) -> String {
     let timestamp = chrono::Utc::now().to_rfc3339();
···
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
+    circuit_breaker: Option<&CircuitBreaker>,
     event: EmitEvent<'a>,
 ) -> Result<()> {
     const MAX_RETRIES: u32 = 3;
···
     let mut backoff = Duration::from_millis(100);
 
     loop {
+        if let Some(cb) = circuit_breaker {
+            if !cb.is_available().await {
+                warn!("Circuit breaker '{}' is open, rejecting Ozone API call", cb.name());
+                return Err(miette::miette!("Circuit breaker open for Ozone API"));
+            }
+        }
+
         rate_limiter.wait().await;
 
         let opts = build_moderation_call_opts(config);
         match agent.send_with_opts(event.clone(), opts).await.into_diagnostic() {
             Ok(_) => {
                 debug!("Moderation event sent successfully");
+                if let Some(cb) = circuit_breaker {
+                    cb.record_success().await;
+                }
                 return Ok(());
             }
             Err(e) => {
···
                         "Moderation API call failed (attempt {}/{}): {} (transient: {})",
                         retry_count, MAX_RETRIES, error_msg, is_transient
                     );
+                    if let Some(cb) = circuit_breaker {
+                        cb.record_failure().await;
+                    }
                     return Err(e);
                 }
 

src/moderation/post.rs

···
 use crate::moderation::{
     build_mod_tool_meta, build_timestamped_comment, send_moderation_event,
 };
+use crate::resilience::CircuitBreaker;
 
 /// Label a post with a specific label via Ozone moderation API
 pub async fn label_post<'a>(
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
+    circuit_breaker: Option<&CircuitBreaker>,
     post_uri: &AtUri<'a>,
     post_cid: &Cid<'a>,
     label_val: &str,
···
         })
         .build();
 
-    send_moderation_event(agent, config, rate_limiter, event).await?;
+    send_moderation_event(agent, config, rate_limiter, circuit_breaker, event).await?;
 
     debug!("Successfully labeled post: {}", post_uri);
 
···
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
+    circuit_breaker: Option<&CircuitBreaker>,
     post_uri: &AtUri<'a>,
     _post_cid: &Cid<'a>,
     post_did: &Did<'a>,
···
         })
         .build();
 
-    send_moderation_event(agent, config, rate_limiter, event).await?;
+    send_moderation_event(agent, config, rate_limiter, circuit_breaker, event).await?;
 
     debug!("Successfully reported post: {}", post_uri);
 
···
     agent: &Agent<MemoryCredentialSession>,
     config: &Config,
     rate_limiter: &RateLimiter,
+    circuit_breaker: Option<&CircuitBreaker>,
     post_uri: &AtUri<'a>,
     post_cid: &Cid<'a>,
     comment: &str,
···
         )))
         .build();
 
-    send_moderation_event(agent, config, rate_limiter, event).await?;
+    send_moderation_event(agent, config, rate_limiter, circuit_breaker, event).await?;
 
     debug!("Successfully took down post: {}", post_uri);
 

src/moderation/rate_limiter.rs

···
     #[tokio::test]
     async fn test_rate_limiter() {
         // 100ms between requests = 10 requests per second
-        let limiter = RateLimiter::new(100);
+        let limiter = RateLimiter::new(100).unwrap();
 
         let start = Instant::now();
 
···
     #[tokio::test]
     async fn test_rate_limiter_concurrent() {
         // 100ms between requests = 10 requests per second
-        let limiter = RateLimiter::new(100);
+        let limiter = RateLimiter::new(100).unwrap();
 
         let start = Instant::now();
 

src/plc/mod.rs

···
 use tracing::{debug, error, info, warn};
 
 use crate::config::PlcConfig;
+use crate::resilience::CircuitBreaker;
 
 #[derive(Debug, Deserialize)]
 pub struct DidDocument {
···
 pub struct PlcClient {
     client: Client,
     endpoints: Vec<String>,
+    circuit_breaker: Option<CircuitBreaker>,
 }
 
 impl PlcClient {
···
         let mut endpoints = vec![config.endpoint.clone()];
         endpoints.extend(config.fallback_endpoints.clone());
 
-        Self { client, endpoints }
+        Self {
+            client,
+            endpoints,
+            circuit_breaker: None,
+        }
+    }
+
+    /// Create a new PLC client with circuit breaker protection
+    pub fn with_circuit_breaker(
+        client: Client,
+        config: &PlcConfig,
+        circuit_breaker: CircuitBreaker,
+    ) -> Self {
+        let mut endpoints = vec![config.endpoint.clone()];
+        endpoints.extend(config.fallback_endpoints.clone());
+
+        Self {
+            client,
+            endpoints,
+            circuit_breaker: Some(circuit_breaker),
+        }
     }
 
     /// Resolve a DID to its DID document with automatic failover
     pub async fn resolve_did(&self, did: &str) -> Result<DidDocument> {
+        if let Some(cb) = &self.circuit_breaker {
+            if !cb.is_available().await {
+                warn!("Circuit breaker '{}' is open, rejecting PLC resolution", cb.name());
+                return Err(miette::miette!("Circuit breaker open for PLC resolution"));
+            }
+        }
+
         let mut last_error = None;
 
         for (idx, endpoint) in self.endpoints.iter().enumerate() {
···
                                     idx,
                                     did
                                 );
+                            }
+                            if let Some(cb) = &self.circuit_breaker {
+                                cb.record_success().await;
                             }
                             return Ok(doc);
                         }
···
             did,
             self.endpoints.len()
         );
+
+        if let Some(cb) = &self.circuit_breaker {
+            cb.record_failure().await;
+        }
 
         Err(last_error.unwrap_or_else(|| {
             miette::miette!("All PLC endpoints failed for DID: {}", did)

src/processor/matcher.rs

···
 
 use crate::config::Config;
 use crate::processor::phash;
+use crate::resilience::CircuitBreaker;
 use crate::types::{BlobCheck, BlobReference, ImageJob, MatchResult};
 
 /// Load blob checks from a JSON file
···
 pub async fn download_blob(
     client: &Client,
     config: &Config,
+    circuit_breaker: &CircuitBreaker,
     did: &str,
     cid: &str,
 ) -> Result<Vec<u8>> {
···
         start.elapsed().as_secs_f64()
     );
 
+    // Check circuit breaker before attempting PDS
+    if !circuit_breaker.is_available().await {
+        warn!("Circuit breaker '{}' is open, rejecting PDS blob download", circuit_breaker.name());
+        return Err(miette::miette!("Circuit breaker open for PDS blob download"));
+    }
+
     // Check if we've exceeded total timeout before PDS attempt
     if start.elapsed() > total_timeout {
         warn!("Blob download total timeout exceeded before PDS fallback");
···
 
     debug!("Downloading from PDS: {} (timeout: {}s)", pds_url, config.processing.blob_download_timeout_secs);
 
-    let response = client
+    match client
         .get(&pds_url)
         .timeout(per_attempt_timeout)
         .send()
         .await
         .into_diagnostic()
-        .map_err(|e| miette::miette!("PDS blob download failed: {}", e))?
-        .error_for_status()
-        .into_diagnostic()?;
-
-    let bytes = response.bytes().await.into_diagnostic()?;
-    Ok(bytes.to_vec())
+        .and_then(|resp| resp.error_for_status().into_diagnostic())
+    {
+        Ok(response) => {
+            let bytes = response.bytes().await.into_diagnostic()?;
+            circuit_breaker.record_success().await;
+            Ok(bytes.to_vec())
+        }
+        Err(e) => {
+            circuit_breaker.record_failure().await;
+            Err(miette::miette!("PDS blob download failed: {}", e))
+        }
+    }
 }
 
 /// Match a computed phash against blob checks
···
 pub async fn process_blob(
     client: &Client,
     config: &Config,
+    circuit_breaker: &CircuitBreaker,
     blob_checks: &[BlobCheck],
     did: &str,
     blob: &BlobReference,
 ) -> Result<Option<MatchResult>> {
-    let image_bytes = download_blob(client, config, did, &blob.cid).await?;
+    let image_bytes = download_blob(client, config, circuit_breaker, did, &blob.cid).await?;
     let phash = phash::compute_phash(&image_bytes)?;
     debug!("Computed phash for blob {}: {}", blob.cid, phash);
 
···
 pub async fn process_image_job(
     client: &Client,
     config: &Config,
+    circuit_breaker: &CircuitBreaker,
     blob_checks: &[BlobCheck],
     job: &ImageJob,
 ) -> Result<Vec<MatchResult>> {
···
     let mut matches = Vec::new();
 
     for blob in &job.blobs {
-        match process_blob(client, config, blob_checks, &job.post_did, blob).await {
+        match process_blob(client, config, circuit_breaker, blob_checks, &job.post_did, blob).await {
             Ok(Some(result)) => {
                 matches.push(result);
             }

src/queue/worker.rs

···
 use crate::moderation::{account, claims, post, rate_limiter::RateLimiter};
 use crate::processor::matcher;
 use crate::queue::redis_queue::JobQueue;
+use crate::resilience::CircuitBreaker;
 use crate::types::{BlobCheck, ImageJob, MatchResult};
 
 /// Macro to handle moderation actions with claim checking
···
     blob_checks: Vec<BlobCheck>,
     metrics: Metrics,
     rate_limiter: RateLimiter,
+    ozone_circuit_breaker: CircuitBreaker,
+    pds_circuit_breaker: CircuitBreaker,
 }
 
 impl WorkerPool {
···
         agent: AgentSession,
         blob_checks: Vec<BlobCheck>,
         metrics: Metrics,
+        ozone_circuit_breaker: CircuitBreaker,
+        pds_circuit_breaker: CircuitBreaker,
     ) -> Result<Self> {
         let rate_limiter = RateLimiter::new(config.moderation.rate_limit)?;
 
···
             blob_checks,
             metrics,
             rate_limiter,
+            ozone_circuit_breaker,
+            pds_circuit_breaker,
         })
     }
 
···
     /// Concurrency is achieved by running multiple instances of this concurrently
     pub async fn start(
         &self,
-        mut queue: JobQueue,
+        queue: JobQueue,
         mut cache: PhashCache,
         mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
     ) -> Result<()> {
···
                                 &self.blob_checks,
                                 &self.metrics,
                                 &self.rate_limiter,
+                                &self.ozone_circuit_breaker,
+                                &self.pds_circuit_breaker,
                                 &mut cache,
                                 &mut redis_conn,
                                 job,
···
         blob_checks: &[BlobCheck],
         metrics: &Metrics,
         rate_limiter: &RateLimiter,
+        ozone_circuit_breaker: &CircuitBreaker,
+        pds_circuit_breaker: &CircuitBreaker,
         cache: &mut PhashCache,
         redis_conn: &mut redis::aio::MultiplexedConnection,
         job: ImageJob,
···
         debug!("Processing job: {}", job.post_uri);
 
         let matches =
-            Self::process_job_blobs(config, client, blob_checks, metrics, cache, &job).await?;
+            Self::process_job_blobs(config, client, blob_checks, metrics, pds_circuit_breaker, cache, &job).await?;
 
         if matches.is_empty() {
             debug!("No matches found for job: {}", job.post_uri);
···
                 agent,
                 metrics,
                 rate_limiter,
+                ozone_circuit_breaker,
                 redis_conn,
                 &job,
                 &match_result,
···
         client: &Client,
         blob_checks: &[BlobCheck],
         metrics: &Metrics,
+        pds_circuit_breaker: &CircuitBreaker,
         cache: &mut PhashCache,
         job: &ImageJob,
     ) -> Result<Vec<MatchResult>> {
···
 
                 // Download and compute
                 let image_bytes =
-                    matcher::download_blob(client, config, &job.post_did, &blob.cid).await?;
+                    matcher::download_blob(client, config, pds_circuit_breaker, &job.post_did, &blob.cid).await?;
                 let computed_phash = crate::processor::phash::compute_phash(&image_bytes)?;
 
                 // Store in cache
···
         agent: &Arc<Agent<MemoryCredentialSession>>,
         metrics: &Metrics,
         rate_limiter: &RateLimiter,
+        ozone_circuit_breaker: &CircuitBreaker,
         redis_conn: &mut redis::aio::MultiplexedConnection,
         job: &ImageJob,
         match_result: &MatchResult,
···
                 agent.as_ref(),
                 config,
                 rate_limiter,
+                Some(ozone_circuit_breaker),
                 &job.post_uri,
                 &job.post_cid,
                 &job.post_did,
···
                 agent.as_ref(),
                 config,
                 rate_limiter,
+                Some(ozone_circuit_breaker),
                 &job.post_uri,
                 &job.post_cid,
                 &check.label,
···
                 agent.as_ref(),
                 config,
                 rate_limiter,
+                Some(ozone_circuit_breaker),
                 &job.post_did,
                 ReasonType::ComAtprotoModerationDefsReasonSpam,
                 &check.comment,
···
                 agent.as_ref(),
                 config,
                 rate_limiter,
+                Some(ozone_circuit_breaker),
                 &job.post_did,
                 &check.label,
                 &check.comment,
···
             blob_checks: self.blob_checks.clone(),
             metrics: self.metrics.clone(),
             rate_limiter: self.rate_limiter.clone(),
+            ozone_circuit_breaker: self.ozone_circuit_breaker.clone(),
+            pds_circuit_breaker: self.pds_circuit_breaker.clone(),
         }
     }
 }

src/resilience/circuit_breaker.rs

···
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use tokio::sync::RwLock;
+use tracing::{debug, warn};
+
+use crate::metrics::Metrics;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CircuitState {
+    Closed,
+    Open,
+    HalfOpen,
+}
+
+#[derive(Debug, Clone)]
+struct CircuitBreakerState {
+    state: CircuitState,
+    failure_count: u32,
+    success_count: u32,
+    last_failure_time: Option<Instant>,
+    half_open_attempts: u32,
+}
+
+#[derive(Clone)]
+pub struct CircuitBreaker {
+    name: String,
+    failure_threshold: u32,
+    timeout: Duration,
+    half_open_max_calls: u32,
+    state: Arc<RwLock<CircuitBreakerState>>,
+    metrics: Option<Metrics>,
+}
+
+impl CircuitBreaker {
+    pub fn new(
+        name: impl Into<String>,
+        failure_threshold: u32,
+        timeout_secs: u64,
+        half_open_max_calls: u32,
+    ) -> Self {
+        Self {
+            name: name.into(),
+            failure_threshold,
+            timeout: Duration::from_secs(timeout_secs),
+            half_open_max_calls,
+            state: Arc::new(RwLock::new(CircuitBreakerState {
+                state: CircuitState::Closed,
+                failure_count: 0,
+                success_count: 0,
+                last_failure_time: None,
+                half_open_attempts: 0,
+            })),
+            metrics: None,
+        }
+    }
+
+    pub fn with_metrics(
+        name: impl Into<String>,
+        failure_threshold: u32,
+        timeout_secs: u64,
+        half_open_max_calls: u32,
+        metrics: Metrics,
+    ) -> Self {
+        Self {
+            name: name.into(),
+            failure_threshold,
+            timeout: Duration::from_secs(timeout_secs),
+            half_open_max_calls,
+            state: Arc::new(RwLock::new(CircuitBreakerState {
+                state: CircuitState::Closed,
+                failure_count: 0,
+                success_count: 0,
+                last_failure_time: None,
+                half_open_attempts: 0,
+            })),
+            metrics: Some(metrics),
+        }
+    }
+
+    pub async fn is_available(&self) -> bool {
+        let mut state = self.state.write().await;
+
+        match state.state {
+            CircuitState::Closed => true,
+            CircuitState::Open => {
+                if let Some(last_failure) = state.last_failure_time {
+                    if last_failure.elapsed() >= self.timeout {
+                        debug!(
+                            "Circuit breaker '{}' transitioning from Open to HalfOpen (timeout elapsed)",
+                            self.name
+                        );
+                        state.state = CircuitState::HalfOpen;
+                        state.half_open_attempts = 1;
+                        state.success_count = 0;
+                        if let Some(metrics) = &self.metrics {
+                            metrics.inc_circuit_breaker_transitions();
+                        }
+                        true
+                    } else {
+                        if let Some(metrics) = &self.metrics {
+                            metrics.inc_circuit_breaker_rejections();
+                        }
+                        false
+                    }
+                } else {
+                    if let Some(metrics) = &self.metrics {
+                        metrics.inc_circuit_breaker_rejections();
+                    }
+                    false
+                }
+            }
+            CircuitState::HalfOpen => {
+                if state.half_open_attempts < self.half_open_max_calls {
+                    state.half_open_attempts += 1;
+                    true
+                } else {
+                    if let Some(metrics) = &self.metrics {
+                        metrics.inc_circuit_breaker_rejections();
+                    }
+                    false
+                }
+            }
+        }
+    }
+
+    pub async fn record_success(&self) {
+        let mut state = self.state.write().await;
+
+        match state.state {
+            CircuitState::Closed => {
+                state.failure_count = 0;
+            }
+            CircuitState::HalfOpen => {
+                state.success_count += 1;
+                if state.success_count >= 1 {
+                    debug!(
+                        "Circuit breaker '{}' transitioning from HalfOpen to Closed (success threshold met)",
+                        self.name
+                    );
+                    state.state = CircuitState::Closed;
+                    state.failure_count = 0;
+                    state.success_count = 0;
+                    state.half_open_attempts = 0;
+                    state.last_failure_time = None;
+                    if let Some(metrics) = &self.metrics {
+                        metrics.inc_circuit_breaker_transitions();
+                    }
+                }
+            }
+            CircuitState::Open => {}
+        }
+    }
+
+    pub async fn record_failure(&self) {
+        let mut state = self.state.write().await;
+
+        match state.state {
+            CircuitState::Closed => {
+                state.failure_count += 1;
+                state.last_failure_time = Some(Instant::now());
+
+                if state.failure_count >= self.failure_threshold {
+                    warn!(
+                        "Circuit breaker '{}' transitioning from Closed to Open (failure threshold {} reached)",
+                        self.name, self.failure_threshold
+                    );
+                    state.state = CircuitState::Open;
+                    if let Some(metrics) = &self.metrics {
+                        metrics.inc_circuit_breaker_transitions();
+                    }
+                }
+            }
+            CircuitState::HalfOpen => {
+                warn!(
+                    "Circuit breaker '{}' transitioning from HalfOpen to Open (failure during half-open)",
+                    self.name
+                );
+                state.state = CircuitState::Open;
+                state.failure_count = self.failure_threshold;
+                state.success_count = 0;
+                state.half_open_attempts = 0;
+                state.last_failure_time = Some(Instant::now());
+                if let Some(metrics) = &self.metrics {
+                    metrics.inc_circuit_breaker_transitions();
+                }
+            }
+            CircuitState::Open => {
+                state.last_failure_time = Some(Instant::now());
+            }
+        }
+    }
+
+    pub async fn get_state(&self) -> CircuitState {
+        self.state.read().await.state
+    }
+
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_circuit_breaker_starts_closed() {
+        let cb = CircuitBreaker::new("test", 3, 60, 1);
+        assert_eq!(cb.get_state().await, CircuitState::Closed);
+        assert!(cb.is_available().await);
+    }
+
+    #[tokio::test]
+    async fn test_circuit_breaker_opens_after_threshold() {
+        let cb = CircuitBreaker::new("test", 3, 60, 1);
+
+        assert!(cb.is_available().await);
+        cb.record_failure().await;
+        assert_eq!(cb.get_state().await, CircuitState::Closed);
+
+        assert!(cb.is_available().await);
+        cb.record_failure().await;
+        assert_eq!(cb.get_state().await, CircuitState::Closed);
+
+        assert!(cb.is_available().await);
+        cb.record_failure().await;
+        assert_eq!(cb.get_state().await, CircuitState::Open);
+
+        assert!(!cb.is_available().await);
+    }
+
+    #[tokio::test]
+    async fn test_circuit_breaker_half_open_after_timeout() {
+        let cb = CircuitBreaker::new("test", 3, 1, 1);
+
+        for _ in 0..3 {
+            assert!(cb.is_available().await);
+            cb.record_failure().await;
+        }
+
+        assert_eq!(cb.get_state().await, CircuitState::Open);
+        assert!(!cb.is_available().await);
+
+        tokio::time::sleep(Duration::from_secs(2)).await;
+
+        assert!(cb.is_available().await);
+        assert_eq!(cb.get_state().await, CircuitState::HalfOpen);
+    }
+
+    #[tokio::test]
+    async fn test_circuit_breaker_closes_on_success_during_half_open() {
+        let cb = CircuitBreaker::new("test", 3, 1, 1);
+
+        for _ in 0..3 {
+            assert!(cb.is_available().await);
+            cb.record_failure().await;
+        }
+
+        assert_eq!(cb.get_state().await, CircuitState::Open);
+
+        tokio::time::sleep(Duration::from_secs(2)).await;
+
+        assert!(cb.is_available().await);
+        assert_eq!(cb.get_state().await, CircuitState::HalfOpen);
+
+        cb.record_success().await;
+        assert_eq!(cb.get_state().await, CircuitState::Closed);
+        assert!(cb.is_available().await);
+    }
+
+    #[tokio::test]
+    async fn test_circuit_breaker_reopens_on_failure_during_half_open() {
+        let cb = CircuitBreaker::new("test", 3, 1, 1);
+
+        for _ in 0..3 {
+            assert!(cb.is_available().await);
+            cb.record_failure().await;
+        }
+
+        assert_eq!(cb.get_state().await, CircuitState::Open);
+
+        tokio::time::sleep(Duration::from_secs(2)).await;
+
+        assert!(cb.is_available().await);
+        assert_eq!(cb.get_state().await, CircuitState::HalfOpen);
+
+        cb.record_failure().await;
+        assert_eq!(cb.get_state().await, CircuitState::Open);
+        assert!(!cb.is_available().await);
+    }
+
+    #[tokio::test]
+    async fn test_circuit_breaker_half_open_max_calls() {
+        let cb = CircuitBreaker::new("test", 3, 1, 2);
+
+        for _ in 0..3 {
+            assert!(cb.is_available().await);
+            cb.record_failure().await;
+        }
+
+        tokio::time::sleep(Duration::from_secs(2)).await;
+
+        assert!(cb.is_available().await);
+        assert_eq!(cb.get_state().await, CircuitState::HalfOpen);
+
+        assert!(cb.is_available().await);
+        assert_eq!(cb.get_state().await, CircuitState::HalfOpen);
+
+        assert!(!cb.is_available().await);
+    }
+
+    #[tokio::test]
+    async fn test_circuit_breaker_success_resets_failures_when_closed() {
+        let cb = CircuitBreaker::new("test", 3, 60, 1);
+
+        assert!(cb.is_available().await);
+        cb.record_failure().await;
+        assert!(cb.is_available().await);
+        cb.record_failure().await;
+
+        assert_eq!(cb.get_state().await, CircuitState::Closed);
+
+        cb.record_success().await;
+
+        assert!(cb.is_available().await);
+        cb.record_failure().await;
+        assert!(cb.is_available().await);
+        cb.record_failure().await;
+
+        assert_eq!(cb.get_state().await, CircuitState::Closed);
+    }
+}

src/resilience/mod.rs

···
+pub mod circuit_breaker;
+
+pub use circuit_breaker::CircuitBreaker;