forge_core/http/

mod.rs

//! HTTP client with circuit breaker pattern.
//!
//! Wraps `reqwest::Client` with automatic failure tracking per host.
//! After repeated failures, requests fail fast to prevent cascade failures.

use std::collections::HashMap;
use std::sync::RwLock;
use std::time::{Duration, Instant};

use reqwest::{Request, Response};

/// Circuit breaker state for a single host.
#[derive(Debug, Clone)]
pub struct CircuitState {
    /// Current state of the circuit.
    pub state: CircuitStatus,
    /// Number of consecutive failures.
    pub failure_count: u32,
    /// Number of consecutive successes (used in half-open state).
    pub success_count: u32,
    /// When the circuit was opened (for timeout calculation).
    pub opened_at: Option<Instant>,
    /// Current backoff duration.
    pub current_backoff: Duration,
}

/// Circuit breaker status.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CircuitStatus {
    /// Normal operation, requests pass through.
    Closed,
    /// Circuit tripped, requests fail fast.
    Open,
    /// Testing if service recovered, limited requests allowed.
    HalfOpen,
}

impl Default for CircuitState {
    fn default() -> Self {
        Self {
            state: CircuitStatus::Closed,
            failure_count: 0,
            success_count: 0,
            opened_at: None,
            current_backoff: Duration::from_secs(30),
        }
    }
}

/// Configuration for the circuit breaker.
#[derive(Debug, Clone)]
pub struct CircuitBreakerConfig {
    /// Number of failures before opening the circuit.
    pub failure_threshold: u32,
    /// Number of successes in half-open state before closing.
    pub success_threshold: u32,
    /// Initial timeout before trying half-open.
    pub base_timeout: Duration,
    /// Maximum backoff duration.
    pub max_backoff: Duration,
    /// Backoff multiplier for exponential backoff.
    pub backoff_multiplier: f64,
    /// Whether the circuit breaker is enabled.
    pub enabled: bool,
}

impl Default for CircuitBreakerConfig {
    fn default() -> Self {
        Self {
            failure_threshold: 5,
            success_threshold: 2,
            base_timeout: Duration::from_secs(30),
            max_backoff: Duration::from_secs(600), // 10 minutes
            backoff_multiplier: 1.5,
            enabled: true,
        }
    }
}

/// Error returned when circuit breaker is open.
#[derive(Debug, Clone)]
pub struct CircuitBreakerOpen {
    /// The host that is being blocked.
    pub host: String,
    /// Time until the circuit may try again.
    pub retry_after: Duration,
}

impl std::fmt::Display for CircuitBreakerOpen {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "Circuit breaker open for {}: retry after {:?}",
            self.host, self.retry_after
        )
    }
}

impl std::error::Error for CircuitBreakerOpen {}

/// HTTP client with circuit breaker pattern.
///
/// Tracks failure rates per host and fails fast when a host is unhealthy.
#[derive(Clone)]
pub struct CircuitBreakerClient {
    inner: reqwest::Client,
    states: std::sync::Arc<RwLock<HashMap<String, CircuitState>>>,
    config: CircuitBreakerConfig,
}

impl CircuitBreakerClient {
    /// Create a new circuit breaker client wrapping the given reqwest client.
    pub fn new(client: reqwest::Client, config: CircuitBreakerConfig) -> Self {
        Self {
            inner: client,
            states: std::sync::Arc::new(RwLock::new(HashMap::new())),
            config,
        }
    }

    /// Create with default configuration.
    pub fn with_defaults(client: reqwest::Client) -> Self {
        Self::new(client, CircuitBreakerConfig::default())
    }

    /// Get the underlying reqwest client for building requests.
    pub fn inner(&self) -> &reqwest::Client {
        &self.inner
    }

    /// Extract host from URL for tracking.
    fn extract_host(url: &reqwest::Url) -> String {
        format!(
            "{}://{}{}",
            url.scheme(),
            url.host_str().unwrap_or("unknown"),
            url.port().map(|p| format!(":{}", p)).unwrap_or_default()
        )
    }

    /// Check if a request to the given host should be allowed.
    pub fn should_allow(&self, host: &str) -> Result<(), CircuitBreakerOpen> {
        if !self.config.enabled {
            return Ok(());
        }

        let states = self.states.read().unwrap();
        let state = match states.get(host) {
            Some(s) => s,
            None => return Ok(()), // No state = first request, allow
        };

        match state.state {
            CircuitStatus::Closed => Ok(()),
            CircuitStatus::HalfOpen => Ok(()), // Allow test requests
            CircuitStatus::Open => {
                let opened_at = state.opened_at.unwrap_or_else(Instant::now);
                let elapsed = opened_at.elapsed();

                if elapsed >= state.current_backoff {
                    // Timeout expired, will transition to half-open
                    Ok(())
                } else {
                    Err(CircuitBreakerOpen {
                        host: host.to_string(),
                        retry_after: state.current_backoff - elapsed,
                    })
                }
            }
        }
    }

    /// Record a successful request.
    pub fn record_success(&self, host: &str) {
        if !self.config.enabled {
            return;
        }

        let mut states = self.states.write().unwrap();
        let state = states.entry(host.to_string()).or_default();

        match state.state {
            CircuitStatus::Closed => {
                // Reset failure count on success
                state.failure_count = 0;
            }
            CircuitStatus::HalfOpen => {
                state.success_count += 1;
                if state.success_count >= self.config.success_threshold {
                    // Service recovered, close the circuit
                    tracing::info!(host = %host, "Circuit breaker closed, service recovered");
                    state.state = CircuitStatus::Closed;
                    state.failure_count = 0;
                    state.success_count = 0;
                    state.opened_at = None;
                    state.current_backoff = self.config.base_timeout;
                }
            }
            CircuitStatus::Open => {
                // Transition to half-open on first success after timeout
                tracing::info!(host = %host, "Circuit breaker half-open, testing service");
                state.state = CircuitStatus::HalfOpen;
                state.success_count = 1;
            }
        }
    }

    /// Record a failed request.
    pub fn record_failure(&self, host: &str) {
        if !self.config.enabled {
            return;
        }

        let mut states = self.states.write().unwrap();
        let state = states.entry(host.to_string()).or_default();

        match state.state {
            CircuitStatus::Closed => {
                state.failure_count += 1;
                if state.failure_count >= self.config.failure_threshold {
                    // Trip the circuit
                    tracing::warn!(
                        host = %host,
                        failures = state.failure_count,
                        "Circuit breaker opened, service unhealthy"
                    );
                    state.state = CircuitStatus::Open;
                    state.opened_at = Some(Instant::now());
                }
            }
            CircuitStatus::HalfOpen => {
                // Failed during test, reopen with increased backoff
                let new_backoff = Duration::from_secs_f64(
                    (state.current_backoff.as_secs_f64() * self.config.backoff_multiplier)
                        .min(self.config.max_backoff.as_secs_f64()),
                );
                tracing::warn!(
                    host = %host,
                    backoff_secs = new_backoff.as_secs(),
                    "Circuit breaker reopened, service still unhealthy"
                );
                state.state = CircuitStatus::Open;
                state.opened_at = Some(Instant::now());
                state.current_backoff = new_backoff;
                state.success_count = 0;
            }
            CircuitStatus::Open => {
                // Already open, just update timestamp
                state.opened_at = Some(Instant::now());
            }
        }
    }

    /// Execute a request with circuit breaker protection.
    pub async fn execute(&self, request: Request) -> Result<Response, CircuitBreakerError> {
        let host = Self::extract_host(request.url());

        // Check circuit state
        self.should_allow(&host)
            .map_err(CircuitBreakerError::CircuitOpen)?;

        // If circuit is open but timeout expired, transition to half-open
        {
            let mut states = self.states.write().unwrap();
            if let Some(state) = states.get_mut(&host) {
                if state.state == CircuitStatus::Open {
                    if let Some(opened_at) = state.opened_at {
                        if opened_at.elapsed() >= state.current_backoff {
                            tracing::info!(host = %host, "Circuit breaker half-open, testing service");
                            state.state = CircuitStatus::HalfOpen;
                            state.success_count = 0;
                        }
                    }
                }
            }
        }

        // Execute the request
        match self.inner.execute(request).await {
            Ok(response) => {
                // Check if response indicates server error
                if response.status().is_server_error() {
                    self.record_failure(&host);
                } else {
                    self.record_success(&host);
                }
                Ok(response)
            }
            Err(e) => {
                self.record_failure(&host);
                Err(CircuitBreakerError::Request(e))
            }
        }
    }

    /// Get the current state for a host.
    pub fn get_state(&self, host: &str) -> Option<CircuitState> {
        self.states.read().unwrap().get(host).cloned()
    }

    /// Reset the circuit breaker state for a host.
    pub fn reset(&self, host: &str) {
        self.states.write().unwrap().remove(host);
    }

    /// Reset all circuit breaker states.
    pub fn reset_all(&self) {
        self.states.write().unwrap().clear();
    }
}

/// Error type for circuit breaker operations.
#[derive(Debug)]
pub enum CircuitBreakerError {
    /// The circuit is open, request was not attempted.
    CircuitOpen(CircuitBreakerOpen),
    /// The request failed.
    Request(reqwest::Error),
}

impl std::fmt::Display for CircuitBreakerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            CircuitBreakerError::CircuitOpen(e) => write!(f, "{}", e),
            CircuitBreakerError::Request(e) => write!(f, "HTTP request failed: {}", e),
        }
    }
}

impl std::error::Error for CircuitBreakerError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            CircuitBreakerError::CircuitOpen(e) => Some(e),
            CircuitBreakerError::Request(e) => Some(e),
        }
    }
}

impl From<reqwest::Error> for CircuitBreakerError {
    fn from(e: reqwest::Error) -> Self {
        CircuitBreakerError::Request(e)
    }
}

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

    #[test]
    fn test_circuit_breaker_defaults() {
        let config = CircuitBreakerConfig::default();
        assert_eq!(config.failure_threshold, 5);
        assert_eq!(config.success_threshold, 2);
        assert!(config.enabled);
    }

    #[test]
    fn test_circuit_state_transitions() {
        let client = reqwest::Client::new();
        let breaker = CircuitBreakerClient::with_defaults(client);
        let host = "https://api.example.com";

        // Initial state should allow
        assert!(breaker.should_allow(host).is_ok());

        // Record failures to trip the circuit
        for _ in 0..5 {
            breaker.record_failure(host);
        }

        // Circuit should be open
        let state = breaker.get_state(host).unwrap();
        assert_eq!(state.state, CircuitStatus::Open);

        // Should be blocked
        assert!(breaker.should_allow(host).is_err());

        // Reset and verify
        breaker.reset(host);
        assert!(breaker.should_allow(host).is_ok());
    }

    #[test]
    fn test_extract_host() {
        let url = reqwest::Url::parse("https://api.example.com:8080/path").unwrap();
        assert_eq!(
            CircuitBreakerClient::extract_host(&url),
            "https://api.example.com:8080"
        );

        let url2 = reqwest::Url::parse("http://localhost/api").unwrap();
        assert_eq!(
            CircuitBreakerClient::extract_host(&url2),
            "http://localhost"
        );
    }
}