diff --git a/Cargo.toml b/Cargo.toml index 0b3c6ca..b62b8f4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ futures-util = "0.3" async-trait = "0.1" # HTTP & WebSocket -reqwest = { version = "0.13", default-features = false, features = ["json", "rustls", "gzip"] } +reqwest = { version = "0.13", default-features = false, features = ["json", "rustls", "gzip", "stream"] } tokio-tungstenite = { version = "0.28", default-features = false, features = ["rustls-tls-native-roots", "connect"] } # Serialization @@ -72,7 +72,6 @@ parking_lot = "0.12" url = "2.5" urlencoding = "2.1" regex = "1.10" -lazy_static = "1.4" bitflags = "2.9" # Security @@ -140,7 +139,6 @@ once_cell = { workspace = true } parking_lot = { workspace = true } url = { workspace = true } regex = { workspace = true } -lazy_static = { workspace = true } paste = { workspace = true } [[bin]] diff --git a/ccxt-core/Cargo.toml b/ccxt-core/Cargo.toml index 1485fd0..fe0f2f2 100644 --- a/ccxt-core/Cargo.toml +++ b/ccxt-core/Cargo.toml @@ -63,6 +63,7 @@ once_cell = { workspace = true } parking_lot = { workspace = true } url = { workspace = true } urlencoding = { workspace = true } +regex = { workspace = true } # Security zeroize = { workspace = true } diff --git a/ccxt-core/src/base_exchange/mod.rs b/ccxt-core/src/base_exchange/mod.rs index 5eb2721..1814e2c 100644 --- a/ccxt-core/src/base_exchange/mod.rs +++ b/ccxt-core/src/base_exchange/mod.rs @@ -93,6 +93,8 @@ impl BaseExchange { strategy_type: crate::retry_strategy::RetryStrategyType::Fixed, ..crate::retry_strategy::RetryConfig::default() }), + max_response_size: 10 * 1024 * 1024, // 10MB default + circuit_breaker: None, // Disabled by default for backward compatibility }; let mut http_client = HttpClient::new(http_config)?; diff --git a/ccxt-core/src/circuit_breaker.rs b/ccxt-core/src/circuit_breaker.rs new file mode 100644 index 0000000..98e08ae --- /dev/null +++ b/ccxt-core/src/circuit_breaker.rs @@ -0,0 +1,970 @@ +//! Circuit Breaker module for preventing cascading failures. +//! +//! This module implements the Circuit Breaker pattern to protect against +//! continuously failing endpoints. When failures exceed a threshold, the +//! circuit "opens" and rejects requests immediately, allowing the system +//! to recover. +//! +//! # States +//! +//! The circuit breaker has three states: +//! +//! - **Closed**: Normal operation, requests are allowed +//! - **Open**: Blocking requests due to failures, requests are rejected immediately +//! - **HalfOpen**: Testing if the service has recovered, allows limited requests +//! +//! # State Transitions +//! +//! ```text +//! ┌─────────┐ failure_count >= threshold ┌──────┐ +//! │ Closed │ ──────────────────────────▶ │ Open │ +//! └─────────┘ └──────┘ +//! ▲ │ +//! │ test request succeeds │ reset_timeout elapsed +//! │ ▼ +//! │ ┌──────────┐ +//! └──────────────────────────────── │ HalfOpen │ +//! └──────────┘ +//! │ +//! │ test request fails +//! ▼ +//! ┌──────┐ +//! │ Open │ +//! └──────┘ +//! ``` +//! +//! # Example +//! +//! ```rust +//! use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitState}; +//! use std::time::Duration; +//! +//! // Create a circuit breaker with custom configuration +//! let config = CircuitBreakerConfig { +//! failure_threshold: 5, +//! reset_timeout: Duration::from_secs(30), +//! success_threshold: 1, +//! }; +//! let breaker = CircuitBreaker::new(config); +//! +//! // Check if request is allowed +//! if breaker.allow_request().is_ok() { +//! // Make the request +//! // On success: +//! breaker.record_success(); +//! // On failure: +//! // breaker.record_failure(); +//! } +//! ``` + +use crate::error::{ConfigValidationError, Error, Result, ValidationResult}; +use std::sync::atomic::{AtomicI64, AtomicU8, AtomicU32, Ordering}; +use std::time::Duration; +use tokio::sync::mpsc; +use tracing::{debug, info, warn}; + +/// Circuit breaker states. +/// +/// The circuit breaker transitions between these states based on +/// request success/failure patterns. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum CircuitState { + /// Normal operation, requests are allowed. + /// Failures are counted, and if they exceed the threshold, + /// the circuit transitions to Open. + Closed = 0, + + /// Blocking requests due to failures. + /// All requests are rejected immediately with a `ResourceExhausted` error. + /// After `reset_timeout` elapses, transitions to HalfOpen. + Open = 1, + + /// Testing if the service has recovered. + /// Allows a limited number of test requests. + /// If successful, transitions to Closed; if failed, transitions back to Open. + HalfOpen = 2, +} + +impl CircuitState { + /// Converts a u8 value to a CircuitState. + fn from_u8(value: u8) -> Self { + match value { + 1 => CircuitState::Open, + 2 => CircuitState::HalfOpen, + _ => CircuitState::Closed, // Default to Closed for invalid values (including 0) + } + } +} + +/// Circuit breaker configuration. +/// +/// # Example +/// +/// ```rust +/// use ccxt_core::circuit_breaker::CircuitBreakerConfig; +/// use std::time::Duration; +/// +/// let config = CircuitBreakerConfig { +/// failure_threshold: 5, +/// reset_timeout: Duration::from_secs(30), +/// success_threshold: 1, +/// }; +/// ``` +#[derive(Debug, Clone)] +pub struct CircuitBreakerConfig { + /// Number of consecutive failures before opening the circuit. + /// + /// When the failure count reaches this threshold, the circuit + /// transitions from Closed to Open state. + /// + /// Default: 5 + pub failure_threshold: u32, + + /// Time to wait in Open state before transitioning to HalfOpen. + /// + /// After this duration elapses, the circuit breaker will allow + /// a test request to check if the service has recovered. + /// + /// Default: 30 seconds + pub reset_timeout: Duration, + + /// Number of consecutive successes needed to close the circuit from HalfOpen. + /// + /// When in HalfOpen state, this many successful requests are required + /// before transitioning back to Closed state. + /// + /// Default: 1 + pub success_threshold: u32, +} + +impl Default for CircuitBreakerConfig { + fn default() -> Self { + Self { + failure_threshold: 5, + reset_timeout: Duration::from_secs(30), + success_threshold: 1, + } + } +} + +impl CircuitBreakerConfig { + /// Creates a new circuit breaker configuration with the given parameters. + /// + /// # Arguments + /// + /// * `failure_threshold` - Number of failures before opening the circuit + /// * `reset_timeout` - Time to wait before testing recovery + /// * `success_threshold` - Number of successes needed to close the circuit + pub fn new(failure_threshold: u32, reset_timeout: Duration, success_threshold: u32) -> Self { + Self { + failure_threshold, + reset_timeout, + success_threshold, + } + } + + /// Validates the circuit breaker configuration. + /// + /// # Returns + /// + /// Returns `Ok(ValidationResult)` if the configuration is valid. + /// Returns `Err(ConfigValidationError)` if the configuration is invalid. + /// + /// # Validation Rules + /// + /// - `failure_threshold` must be > 0 + /// - `success_threshold` must be > 0 + /// - `reset_timeout` must be >= 1 second + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::CircuitBreakerConfig; + /// + /// let config = CircuitBreakerConfig::default(); + /// assert!(config.validate().is_ok()); + /// ``` + pub fn validate(&self) -> std::result::Result { + let mut warnings = Vec::new(); + + if self.failure_threshold == 0 { + return Err(ConfigValidationError::invalid( + "failure_threshold", + "failure_threshold must be greater than 0", + )); + } + + if self.success_threshold == 0 { + return Err(ConfigValidationError::invalid( + "success_threshold", + "success_threshold must be greater than 0", + )); + } + + if self.reset_timeout < Duration::from_secs(1) { + warnings.push(format!( + "reset_timeout {:?} is very short, may cause rapid state transitions", + self.reset_timeout + )); + } + + Ok(ValidationResult::with_warnings(warnings)) + } +} + +/// Circuit breaker events for observability. +/// +/// These events are emitted when the circuit breaker state changes +/// or when significant actions occur. +#[derive(Debug, Clone)] +pub enum CircuitBreakerEvent { + /// The circuit breaker state has changed. + StateChanged { + /// The previous state + from: CircuitState, + /// The new state + to: CircuitState, + }, + + /// A request was rejected because the circuit is open. + RequestRejected { + /// The reason for rejection + reason: String, + }, + + /// A failure was recorded. + FailureRecorded { + /// The current failure count + count: u32, + }, + + /// A success was recorded. + SuccessRecorded { + /// The current success count (in HalfOpen state) + count: u32, + }, +} + +/// Circuit breaker for preventing cascading failures. +/// +/// The circuit breaker monitors request success/failure patterns and +/// automatically blocks requests to failing endpoints, allowing the +/// system to recover. +/// +/// # Thread Safety +/// +/// This implementation uses atomic operations for all state management, +/// making it safe to use from multiple threads without external locking. +/// +/// # Example +/// +/// ```rust +/// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; +/// +/// let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); +/// +/// // Before making a request +/// if let Err(e) = breaker.allow_request() { +/// // Circuit is open, handle the error +/// println!("Circuit is open: {}", e); +/// } else { +/// // Make the request +/// let success = true; // result of the request +/// if success { +/// breaker.record_success(); +/// } else { +/// breaker.record_failure(); +/// } +/// } +/// ``` +#[derive(Debug)] +pub struct CircuitBreaker { + /// Current state (stored as u8 for atomic operations) + state: AtomicU8, + + /// Number of consecutive failures in Closed state + failure_count: AtomicU32, + + /// Number of consecutive successes in HalfOpen state + success_count: AtomicU32, + + /// Timestamp of the last failure (milliseconds since Unix epoch) + last_failure_time: AtomicI64, + + /// Configuration + config: CircuitBreakerConfig, + + /// Optional event sender for observability + event_tx: Option>, +} + +impl CircuitBreaker { + /// Creates a new circuit breaker with the given configuration. + /// + /// # Arguments + /// + /// * `config` - Circuit breaker configuration + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; + /// + /// let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + /// ``` + pub fn new(config: CircuitBreakerConfig) -> Self { + Self { + state: AtomicU8::new(CircuitState::Closed as u8), + failure_count: AtomicU32::new(0), + success_count: AtomicU32::new(0), + last_failure_time: AtomicI64::new(0), + config, + event_tx: None, + } + } + + /// Creates a new circuit breaker with an event channel for observability. + /// + /// # Arguments + /// + /// * `config` - Circuit breaker configuration + /// * `event_tx` - Channel sender for circuit breaker events + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitBreakerEvent}; + /// use tokio::sync::mpsc; + /// + /// let (tx, mut rx) = mpsc::unbounded_channel::(); + /// let breaker = CircuitBreaker::with_events(CircuitBreakerConfig::default(), tx); + /// + /// // Events will be sent to the channel when state changes occur + /// ``` + pub fn with_events( + config: CircuitBreakerConfig, + event_tx: mpsc::UnboundedSender, + ) -> Self { + Self { + state: AtomicU8::new(CircuitState::Closed as u8), + failure_count: AtomicU32::new(0), + success_count: AtomicU32::new(0), + last_failure_time: AtomicI64::new(0), + config, + event_tx: Some(event_tx), + } + } + + /// Checks if a request should be allowed. + /// + /// # Returns + /// + /// Returns `Ok(())` if the request is allowed. + /// Returns `Err(Error::ResourceExhausted)` if the circuit is open. + /// + /// # State Transitions + /// + /// - **Closed**: Always allows requests + /// - **Open**: Checks if `reset_timeout` has elapsed; if so, transitions to HalfOpen + /// - **HalfOpen**: Allows requests (for testing recovery) + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; + /// + /// let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + /// + /// match breaker.allow_request() { + /// Ok(()) => println!("Request allowed"), + /// Err(e) => println!("Request blocked: {}", e), + /// } + /// ``` + #[allow(clippy::cast_possible_truncation)] + pub fn allow_request(&self) -> Result<()> { + let state = self.state(); + + match state { + CircuitState::Closed => { + debug!("Circuit breaker: Closed state, allowing request"); + Ok(()) + } + CircuitState::Open => { + // Check if reset timeout has elapsed + let last_failure = self.last_failure_time.load(Ordering::Acquire); + let now = chrono::Utc::now().timestamp_millis(); + let elapsed_ms = (now - last_failure).max(0) as u64; + let elapsed = Duration::from_millis(elapsed_ms); + + if elapsed >= self.config.reset_timeout { + // Safe truncation: reset_timeout is typically seconds/minutes, not years + let reset_timeout_ms = + self.config + .reset_timeout + .as_millis() + .min(u128::from(u64::MAX)) as u64; + info!( + elapsed_ms = elapsed_ms, + reset_timeout_ms = reset_timeout_ms, + "Circuit breaker: Reset timeout elapsed, transitioning to HalfOpen" + ); + self.transition_to(CircuitState::HalfOpen); + Ok(()) + } else { + // Safe truncation: reset_timeout is typically seconds/minutes, not years + let reset_timeout_ms = + self.config + .reset_timeout + .as_millis() + .min(u128::from(u64::MAX)) as u64; + let remaining_ms = reset_timeout_ms.saturating_sub(elapsed_ms); + warn!( + remaining_ms = remaining_ms, + "Circuit breaker: Open state, rejecting request" + ); + self.emit_event(CircuitBreakerEvent::RequestRejected { + reason: format!("Circuit breaker is open, retry after {remaining_ms}ms"), + }); + Err(Error::resource_exhausted("Circuit breaker is open")) + } + } + CircuitState::HalfOpen => { + debug!("Circuit breaker: HalfOpen state, allowing test request"); + Ok(()) + } + } + } + + /// Records a successful request. + /// + /// # State Transitions + /// + /// - **Closed**: Resets the failure count to 0 + /// - **HalfOpen**: Increments success count; if it reaches `success_threshold`, + /// transitions to Closed + /// - **Open**: No effect + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; + /// + /// let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + /// + /// // After a successful request + /// breaker.record_success(); + /// ``` + pub fn record_success(&self) { + let state = self.state(); + + match state { + CircuitState::Closed => { + // Reset failure count on success + self.failure_count.store(0, Ordering::Release); + debug!("Circuit breaker: Success recorded in Closed state, failure count reset"); + } + CircuitState::HalfOpen => { + let count = self.success_count.fetch_add(1, Ordering::AcqRel) + 1; + debug!( + success_count = count, + threshold = self.config.success_threshold, + "Circuit breaker: Success recorded in HalfOpen state" + ); + + self.emit_event(CircuitBreakerEvent::SuccessRecorded { count }); + + if count >= self.config.success_threshold { + info!( + success_count = count, + "Circuit breaker: Success threshold reached, transitioning to Closed" + ); + self.transition_to(CircuitState::Closed); + self.failure_count.store(0, Ordering::Release); + self.success_count.store(0, Ordering::Release); + } + } + CircuitState::Open => { + // No effect in Open state + debug!("Circuit breaker: Success recorded in Open state (no effect)"); + } + } + } + + /// Records a failed request. + /// + /// # State Transitions + /// + /// - **Closed**: Increments failure count; if it reaches `failure_threshold`, + /// transitions to Open + /// - **HalfOpen**: Transitions immediately to Open (test request failed) + /// - **Open**: No effect + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; + /// + /// let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + /// + /// // After a failed request + /// breaker.record_failure(); + /// ``` + pub fn record_failure(&self) { + let state = self.state(); + + match state { + CircuitState::Closed => { + let count = self.failure_count.fetch_add(1, Ordering::AcqRel) + 1; + debug!( + failure_count = count, + threshold = self.config.failure_threshold, + "Circuit breaker: Failure recorded in Closed state" + ); + + self.emit_event(CircuitBreakerEvent::FailureRecorded { count }); + + if count >= self.config.failure_threshold { + warn!( + failure_count = count, + "Circuit breaker: Failure threshold reached, transitioning to Open" + ); + self.transition_to(CircuitState::Open); + self.last_failure_time + .store(chrono::Utc::now().timestamp_millis(), Ordering::Release); + } + } + CircuitState::HalfOpen => { + warn!("Circuit breaker: Failure in HalfOpen state, transitioning back to Open"); + self.transition_to(CircuitState::Open); + self.last_failure_time + .store(chrono::Utc::now().timestamp_millis(), Ordering::Release); + self.success_count.store(0, Ordering::Release); + } + CircuitState::Open => { + // No effect in Open state + debug!("Circuit breaker: Failure recorded in Open state (no effect)"); + } + } + } + + /// Returns the current state of the circuit breaker. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitState}; + /// + /// let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + /// assert_eq!(breaker.state(), CircuitState::Closed); + /// ``` + pub fn state(&self) -> CircuitState { + CircuitState::from_u8(self.state.load(Ordering::Acquire)) + } + + /// Returns the current failure count. + /// + /// This is primarily useful for monitoring and debugging. + pub fn failure_count(&self) -> u32 { + self.failure_count.load(Ordering::Acquire) + } + + /// Returns the current success count (in HalfOpen state). + /// + /// This is primarily useful for monitoring and debugging. + pub fn success_count(&self) -> u32 { + self.success_count.load(Ordering::Acquire) + } + + /// Returns a reference to the configuration. + pub fn config(&self) -> &CircuitBreakerConfig { + &self.config + } + + /// Resets the circuit breaker to its initial state (Closed). + /// + /// This can be useful for testing or manual intervention. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; + /// + /// let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + /// // ... some operations that opened the circuit ... + /// breaker.reset(); + /// // Circuit is now Closed again + /// ``` + pub fn reset(&self) { + let old_state = self.state(); + self.state + .store(CircuitState::Closed as u8, Ordering::Release); + self.failure_count.store(0, Ordering::Release); + self.success_count.store(0, Ordering::Release); + self.last_failure_time.store(0, Ordering::Release); + + if old_state != CircuitState::Closed { + info!( + from = ?old_state, + "Circuit breaker: Manual reset to Closed state" + ); + self.emit_event(CircuitBreakerEvent::StateChanged { + from: old_state, + to: CircuitState::Closed, + }); + } + } + + /// Transitions to a new state and emits an event. + fn transition_to(&self, new_state: CircuitState) { + let old_state = self.state(); + self.state.store(new_state as u8, Ordering::Release); + + self.emit_event(CircuitBreakerEvent::StateChanged { + from: old_state, + to: new_state, + }); + } + + /// Emits an event if an event channel is configured. + fn emit_event(&self, event: CircuitBreakerEvent) { + if let Some(ref tx) = self.event_tx { + // Ignore send errors (receiver may have been dropped) + let _ = tx.send(event); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_circuit_state_from_u8() { + assert_eq!(CircuitState::from_u8(0), CircuitState::Closed); + assert_eq!(CircuitState::from_u8(1), CircuitState::Open); + assert_eq!(CircuitState::from_u8(2), CircuitState::HalfOpen); + assert_eq!(CircuitState::from_u8(255), CircuitState::Closed); // Invalid defaults to Closed + } + + #[test] + fn test_circuit_breaker_config_default() { + let config = CircuitBreakerConfig::default(); + assert_eq!(config.failure_threshold, 5); + assert_eq!(config.reset_timeout, Duration::from_secs(30)); + assert_eq!(config.success_threshold, 1); + } + + #[test] + fn test_circuit_breaker_config_new() { + let config = CircuitBreakerConfig::new(10, Duration::from_secs(60), 2); + assert_eq!(config.failure_threshold, 10); + assert_eq!(config.reset_timeout, Duration::from_secs(60)); + assert_eq!(config.success_threshold, 2); + } + + #[test] + fn test_circuit_breaker_config_validate_default() { + let config = CircuitBreakerConfig::default(); + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + } + + #[test] + fn test_circuit_breaker_config_validate_zero_failure_threshold() { + let config = CircuitBreakerConfig { + failure_threshold: 0, + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.field_name(), "failure_threshold"); + } + + #[test] + fn test_circuit_breaker_config_validate_zero_success_threshold() { + let config = CircuitBreakerConfig { + success_threshold: 0, + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.field_name(), "success_threshold"); + } + + #[test] + fn test_circuit_breaker_config_validate_short_reset_timeout_warning() { + let config = CircuitBreakerConfig { + reset_timeout: Duration::from_millis(500), + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_ok()); + let validation_result = result.unwrap(); + assert!(!validation_result.warnings.is_empty()); + assert!(validation_result.warnings[0].contains("reset_timeout")); + } + + #[test] + fn test_circuit_breaker_initial_state() { + let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + assert_eq!(breaker.state(), CircuitState::Closed); + assert_eq!(breaker.failure_count(), 0); + assert_eq!(breaker.success_count(), 0); + } + + #[test] + fn test_circuit_breaker_allow_request_closed() { + let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + assert!(breaker.allow_request().is_ok()); + } + + #[test] + fn test_circuit_breaker_record_success_closed() { + let breaker = CircuitBreaker::new(CircuitBreakerConfig::default()); + + // Record some failures first + breaker.record_failure(); + breaker.record_failure(); + assert_eq!(breaker.failure_count(), 2); + + // Success should reset failure count + breaker.record_success(); + assert_eq!(breaker.failure_count(), 0); + assert_eq!(breaker.state(), CircuitState::Closed); + } + + #[test] + fn test_circuit_breaker_transition_to_open() { + let config = CircuitBreakerConfig { + failure_threshold: 3, + ..Default::default() + }; + let breaker = CircuitBreaker::new(config); + + // Record failures up to threshold + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Closed); + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Closed); + breaker.record_failure(); + + // Should transition to Open after reaching threshold + assert_eq!(breaker.state(), CircuitState::Open); + } + + #[test] + fn test_circuit_breaker_reject_when_open() { + let config = CircuitBreakerConfig { + failure_threshold: 1, + reset_timeout: Duration::from_secs(60), // Long timeout + ..Default::default() + }; + let breaker = CircuitBreaker::new(config); + + // Open the circuit + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Open); + + // Request should be rejected + let result = breaker.allow_request(); + assert!(result.is_err()); + + if let Err(e) = result { + assert!(e.as_resource_exhausted().is_some()); + } + } + + #[test] + fn test_circuit_breaker_transition_to_half_open() { + let config = CircuitBreakerConfig { + failure_threshold: 1, + reset_timeout: Duration::from_millis(10), // Very short timeout for testing + ..Default::default() + }; + let breaker = CircuitBreaker::new(config); + + // Open the circuit + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Open); + + // Wait for reset timeout + std::thread::sleep(Duration::from_millis(20)); + + // Request should be allowed and state should transition to HalfOpen + assert!(breaker.allow_request().is_ok()); + assert_eq!(breaker.state(), CircuitState::HalfOpen); + } + + #[test] + fn test_circuit_breaker_half_open_success_closes() { + let config = CircuitBreakerConfig { + failure_threshold: 1, + reset_timeout: Duration::from_millis(10), + success_threshold: 1, + }; + let breaker = CircuitBreaker::new(config); + + // Open the circuit + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Open); + + // Wait for reset timeout + std::thread::sleep(Duration::from_millis(20)); + + // Transition to HalfOpen + assert!(breaker.allow_request().is_ok()); + assert_eq!(breaker.state(), CircuitState::HalfOpen); + + // Success should close the circuit + breaker.record_success(); + assert_eq!(breaker.state(), CircuitState::Closed); + assert_eq!(breaker.failure_count(), 0); + assert_eq!(breaker.success_count(), 0); + } + + #[test] + fn test_circuit_breaker_half_open_failure_opens() { + let config = CircuitBreakerConfig { + failure_threshold: 1, + reset_timeout: Duration::from_millis(10), + success_threshold: 1, + }; + let breaker = CircuitBreaker::new(config); + + // Open the circuit + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Open); + + // Wait for reset timeout + std::thread::sleep(Duration::from_millis(20)); + + // Transition to HalfOpen + assert!(breaker.allow_request().is_ok()); + assert_eq!(breaker.state(), CircuitState::HalfOpen); + + // Failure should reopen the circuit + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Open); + } + + #[test] + fn test_circuit_breaker_multiple_successes_required() { + let config = CircuitBreakerConfig { + failure_threshold: 1, + reset_timeout: Duration::from_millis(10), + success_threshold: 3, // Require 3 successes + }; + let breaker = CircuitBreaker::new(config); + + // Open the circuit + breaker.record_failure(); + std::thread::sleep(Duration::from_millis(20)); + assert!(breaker.allow_request().is_ok()); + assert_eq!(breaker.state(), CircuitState::HalfOpen); + + // First success - still HalfOpen + breaker.record_success(); + assert_eq!(breaker.state(), CircuitState::HalfOpen); + assert_eq!(breaker.success_count(), 1); + + // Second success - still HalfOpen + breaker.record_success(); + assert_eq!(breaker.state(), CircuitState::HalfOpen); + assert_eq!(breaker.success_count(), 2); + + // Third success - should close + breaker.record_success(); + assert_eq!(breaker.state(), CircuitState::Closed); + } + + #[test] + fn test_circuit_breaker_reset() { + let config = CircuitBreakerConfig { + failure_threshold: 1, + ..Default::default() + }; + let breaker = CircuitBreaker::new(config); + + // Open the circuit + breaker.record_failure(); + assert_eq!(breaker.state(), CircuitState::Open); + + // Reset should close the circuit + breaker.reset(); + assert_eq!(breaker.state(), CircuitState::Closed); + assert_eq!(breaker.failure_count(), 0); + assert_eq!(breaker.success_count(), 0); + } + + #[tokio::test] + async fn test_circuit_breaker_with_events() { + let (tx, mut rx) = mpsc::unbounded_channel::(); + let config = CircuitBreakerConfig { + failure_threshold: 2, + reset_timeout: Duration::from_millis(10), + success_threshold: 1, + }; + let breaker = CircuitBreaker::with_events(config, tx); + + // Record failures + breaker.record_failure(); + breaker.record_failure(); // This should trigger state change + + // Check events + let event1 = rx.recv().await.unwrap(); + assert!(matches!( + event1, + CircuitBreakerEvent::FailureRecorded { count: 1 } + )); + + let event2 = rx.recv().await.unwrap(); + assert!(matches!( + event2, + CircuitBreakerEvent::FailureRecorded { count: 2 } + )); + + let event3 = rx.recv().await.unwrap(); + assert!(matches!( + event3, + CircuitBreakerEvent::StateChanged { + from: CircuitState::Closed, + to: CircuitState::Open + } + )); + } + + #[test] + fn test_circuit_breaker_thread_safety() { + use std::sync::Arc; + use std::thread; + + let breaker = Arc::new(CircuitBreaker::new(CircuitBreakerConfig { + failure_threshold: 100, + ..Default::default() + })); + + let mut handles = vec![]; + + // Spawn multiple threads recording failures + for _ in 0..10 { + let breaker_clone = Arc::clone(&breaker); + handles.push(thread::spawn(move || { + for _ in 0..10 { + breaker_clone.record_failure(); + } + })); + } + + // Wait for all threads to complete + for handle in handles { + handle.join().unwrap(); + } + + // Should have recorded 100 failures and transitioned to Open + assert_eq!(breaker.state(), CircuitState::Open); + } +} diff --git a/ccxt-core/src/error/config.rs b/ccxt-core/src/error/config.rs new file mode 100644 index 0000000..7233910 --- /dev/null +++ b/ccxt-core/src/error/config.rs @@ -0,0 +1,444 @@ +//! Configuration validation error types. +//! +//! This module provides error types for configuration validation, allowing +//! developers to catch invalid configurations early with clear error messages. +//! +//! # Example +//! +//! ```rust +//! use ccxt_core::error::{ConfigValidationError, ValidationResult}; +//! +//! fn validate_max_retries(value: u32) -> Result { +//! if value > 10 { +//! return Err(ConfigValidationError::ValueTooHigh { +//! field: "max_retries", +//! value: value.to_string(), +//! max: "10".to_string(), +//! }); +//! } +//! Ok(ValidationResult::new()) +//! } +//! ``` + +use std::fmt; +use thiserror::Error; + +/// Configuration validation error types. +/// +/// This enum represents different types of validation failures that can occur +/// when validating configuration parameters. Each variant includes the field +/// name and relevant values for debugging. +/// +/// # Example +/// +/// ```rust +/// use ccxt_core::error::ConfigValidationError; +/// +/// let err = ConfigValidationError::ValueTooHigh { +/// field: "max_retries", +/// value: "15".to_string(), +/// max: "10".to_string(), +/// }; +/// assert!(err.to_string().contains("max_retries")); +/// assert!(err.to_string().contains("15")); +/// ``` +#[derive(Error, Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum ConfigValidationError { + /// Field value exceeds the maximum allowed value. + #[error("Field '{field}' value {value} exceeds maximum {max}")] + ValueTooHigh { + /// The name of the configuration field + field: &'static str, + /// The actual value that was provided + value: String, + /// The maximum allowed value + max: String, + }, + + /// Field value is below the minimum allowed value. + #[error("Field '{field}' value {value} is below minimum {min}")] + ValueTooLow { + /// The name of the configuration field + field: &'static str, + /// The actual value that was provided + value: String, + /// The minimum allowed value + min: String, + }, + + /// Field value is invalid for reasons other than range. + #[error("Field '{field}' has invalid value: {reason}")] + ValueInvalid { + /// The name of the configuration field + field: &'static str, + /// The reason why the value is invalid + reason: String, + }, + + /// Required field is missing. + #[error("Required field '{field}' is missing")] + ValueMissing { + /// The name of the missing configuration field + field: &'static str, + }, +} + +impl ConfigValidationError { + /// Returns the field name associated with this error. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ConfigValidationError; + /// + /// let err = ConfigValidationError::ValueTooHigh { + /// field: "max_retries", + /// value: "15".to_string(), + /// max: "10".to_string(), + /// }; + /// assert_eq!(err.field_name(), "max_retries"); + /// ``` + #[must_use] + pub fn field_name(&self) -> &'static str { + match self { + ConfigValidationError::ValueTooHigh { field, .. } + | ConfigValidationError::ValueTooLow { field, .. } + | ConfigValidationError::ValueInvalid { field, .. } + | ConfigValidationError::ValueMissing { field } => field, + } + } + + /// Creates a new `ValueTooHigh` error. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ConfigValidationError; + /// + /// let err = ConfigValidationError::too_high("max_retries", 15, 10); + /// assert!(err.to_string().contains("max_retries")); + /// ``` + pub fn too_high( + field: &'static str, + value: V, + max: M, + ) -> Self { + ConfigValidationError::ValueTooHigh { + field, + value: value.to_string(), + max: max.to_string(), + } + } + + /// Creates a new `ValueTooLow` error. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ConfigValidationError; + /// + /// let err = ConfigValidationError::too_low("base_delay_ms", 5, 10); + /// assert!(err.to_string().contains("base_delay_ms")); + /// ``` + pub fn too_low( + field: &'static str, + value: V, + min: M, + ) -> Self { + ConfigValidationError::ValueTooLow { + field, + value: value.to_string(), + min: min.to_string(), + } + } + + /// Creates a new `ValueInvalid` error. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ConfigValidationError; + /// + /// let err = ConfigValidationError::invalid("capacity", "capacity cannot be zero"); + /// assert!(err.to_string().contains("capacity")); + /// ``` + pub fn invalid(field: &'static str, reason: impl Into) -> Self { + ConfigValidationError::ValueInvalid { + field, + reason: reason.into(), + } + } + + /// Creates a new `ValueMissing` error. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ConfigValidationError; + /// + /// let err = ConfigValidationError::missing("api_key"); + /// assert!(err.to_string().contains("api_key")); + /// ``` + pub fn missing(field: &'static str) -> Self { + ConfigValidationError::ValueMissing { field } + } +} + +/// Result of a successful configuration validation. +/// +/// This struct contains any warnings that were generated during validation. +/// Warnings indicate potential issues that don't prevent the configuration +/// from being used, but may cause suboptimal behavior. +/// +/// # Example +/// +/// ```rust +/// use ccxt_core::error::ValidationResult; +/// +/// let mut result = ValidationResult::new(); +/// result.add_warning("refill_period is very short, may cause high CPU usage"); +/// assert!(!result.warnings.is_empty()); +/// ``` +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ValidationResult { + /// Warnings generated during validation. + /// + /// These are non-fatal issues that the user should be aware of. + pub warnings: Vec, +} + +impl ValidationResult { + /// Creates a new empty validation result. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ValidationResult; + /// + /// let result = ValidationResult::new(); + /// assert!(result.warnings.is_empty()); + /// ``` + #[must_use] + pub fn new() -> Self { + Self { + warnings: Vec::new(), + } + } + + /// Creates a validation result with the given warnings. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ValidationResult; + /// + /// let result = ValidationResult::with_warnings(vec![ + /// "Warning 1".to_string(), + /// "Warning 2".to_string(), + /// ]); + /// assert_eq!(result.warnings.len(), 2); + /// ``` + #[must_use] + pub fn with_warnings(warnings: Vec) -> Self { + Self { warnings } + } + + /// Adds a warning to the validation result. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ValidationResult; + /// + /// let mut result = ValidationResult::new(); + /// result.add_warning("This is a warning"); + /// assert_eq!(result.warnings.len(), 1); + /// ``` + pub fn add_warning(&mut self, warning: impl Into) { + self.warnings.push(warning.into()); + } + + /// Returns `true` if there are no warnings. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ValidationResult; + /// + /// let result = ValidationResult::new(); + /// assert!(result.is_ok()); + /// ``` + #[must_use] + pub fn is_ok(&self) -> bool { + self.warnings.is_empty() + } + + /// Returns `true` if there are any warnings. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ValidationResult; + /// + /// let mut result = ValidationResult::new(); + /// result.add_warning("Warning"); + /// assert!(result.has_warnings()); + /// ``` + #[must_use] + pub fn has_warnings(&self) -> bool { + !self.warnings.is_empty() + } + + /// Merges another validation result into this one. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::ValidationResult; + /// + /// let mut result1 = ValidationResult::new(); + /// result1.add_warning("Warning 1"); + /// + /// let mut result2 = ValidationResult::new(); + /// result2.add_warning("Warning 2"); + /// + /// result1.merge(result2); + /// assert_eq!(result1.warnings.len(), 2); + /// ``` + pub fn merge(&mut self, other: ValidationResult) { + self.warnings.extend(other.warnings); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_value_too_high_display() { + let err = ConfigValidationError::ValueTooHigh { + field: "max_retries", + value: "15".to_string(), + max: "10".to_string(), + }; + let msg = err.to_string(); + assert!(msg.contains("max_retries")); + assert!(msg.contains("15")); + assert!(msg.contains("10")); + } + + #[test] + fn test_value_too_low_display() { + let err = ConfigValidationError::ValueTooLow { + field: "base_delay_ms", + value: "5".to_string(), + min: "10".to_string(), + }; + let msg = err.to_string(); + assert!(msg.contains("base_delay_ms")); + assert!(msg.contains("5")); + assert!(msg.contains("10")); + } + + #[test] + fn test_value_invalid_display() { + let err = ConfigValidationError::ValueInvalid { + field: "capacity", + reason: "capacity cannot be zero".to_string(), + }; + let msg = err.to_string(); + assert!(msg.contains("capacity")); + assert!(msg.contains("cannot be zero")); + } + + #[test] + fn test_value_missing_display() { + let err = ConfigValidationError::ValueMissing { field: "api_key" }; + let msg = err.to_string(); + assert!(msg.contains("api_key")); + assert!(msg.contains("missing")); + } + + #[test] + fn test_field_name() { + let err1 = ConfigValidationError::too_high("max_retries", 15, 10); + assert_eq!(err1.field_name(), "max_retries"); + + let err2 = ConfigValidationError::too_low("base_delay_ms", 5, 10); + assert_eq!(err2.field_name(), "base_delay_ms"); + + let err3 = ConfigValidationError::invalid("capacity", "cannot be zero"); + assert_eq!(err3.field_name(), "capacity"); + + let err4 = ConfigValidationError::missing("api_key"); + assert_eq!(err4.field_name(), "api_key"); + } + + #[test] + fn test_helper_constructors() { + let err1 = ConfigValidationError::too_high("field1", 100, 50); + assert!(matches!(err1, ConfigValidationError::ValueTooHigh { .. })); + + let err2 = ConfigValidationError::too_low("field2", 5, 10); + assert!(matches!(err2, ConfigValidationError::ValueTooLow { .. })); + + let err3 = ConfigValidationError::invalid("field3", "invalid reason"); + assert!(matches!(err3, ConfigValidationError::ValueInvalid { .. })); + + let err4 = ConfigValidationError::missing("field4"); + assert!(matches!(err4, ConfigValidationError::ValueMissing { .. })); + } + + #[test] + fn test_validation_result_new() { + let result = ValidationResult::new(); + assert!(result.warnings.is_empty()); + assert!(result.is_ok()); + assert!(!result.has_warnings()); + } + + #[test] + fn test_validation_result_with_warnings() { + let result = + ValidationResult::with_warnings(vec!["Warning 1".to_string(), "Warning 2".to_string()]); + assert_eq!(result.warnings.len(), 2); + assert!(!result.is_ok()); + assert!(result.has_warnings()); + } + + #[test] + fn test_validation_result_add_warning() { + let mut result = ValidationResult::new(); + result.add_warning("Test warning"); + assert_eq!(result.warnings.len(), 1); + assert_eq!(result.warnings[0], "Test warning"); + } + + #[test] + fn test_validation_result_merge() { + let mut result1 = ValidationResult::new(); + result1.add_warning("Warning 1"); + + let mut result2 = ValidationResult::new(); + result2.add_warning("Warning 2"); + result2.add_warning("Warning 3"); + + result1.merge(result2); + assert_eq!(result1.warnings.len(), 3); + } + + #[test] + fn test_config_validation_error_is_error() { + // Verify that ConfigValidationError implements std::error::Error + fn assert_error() {} + assert_error::(); + } + + #[test] + fn test_config_validation_error_clone() { + let err = ConfigValidationError::too_high("field", 100, 50); + let cloned = err.clone(); + assert_eq!(err, cloned); + } +} diff --git a/ccxt-core/src/error/convert.rs b/ccxt-core/src/error/convert.rs index f87c0fb..e6edec1 100644 --- a/ccxt-core/src/error/convert.rs +++ b/ccxt-core/src/error/convert.rs @@ -1,6 +1,6 @@ //! From implementations for converting between error types. -use crate::error::{Error, NetworkError, OrderError, ParseError}; +use crate::error::{ConfigValidationError, Error, NetworkError, OrderError, ParseError}; /// Maximum length for error messages to prevent memory bloat from large HTTP responses. pub(crate) const MAX_ERROR_MESSAGE_LEN: usize = 1024; @@ -84,3 +84,15 @@ impl From for Error { Error::Network(Box::new(NetworkError::from(e))) } } + +impl From for Error { + fn from(e: ConfigValidationError) -> Self { + Error::ConfigValidation(Box::new(e)) + } +} + +impl From> for Error { + fn from(e: Box) -> Self { + Error::ConfigValidation(e) + } +} diff --git a/ccxt-core/src/error/mod.rs b/ccxt-core/src/error/mod.rs index 5d3cfc3..cef2710 100644 --- a/ccxt-core/src/error/mod.rs +++ b/ccxt-core/src/error/mod.rs @@ -134,6 +134,7 @@ //! } //! ``` +mod config; mod context; mod convert; mod details; @@ -148,6 +149,7 @@ use std::time::Duration; use thiserror::Error; // Re-export all public types for backward compatibility +pub use config::{ConfigValidationError, ValidationResult}; pub use context::ContextExt; #[allow(deprecated)] pub use context::ErrorContext; @@ -275,6 +277,23 @@ pub enum Error { #[error("Resource exhausted: {0}")] ResourceExhausted(Cow<'static, str>), + /// Configuration validation error. + /// + /// This error is returned when configuration parameters fail validation, + /// such as values being out of range or missing required fields. + /// Boxed to reduce enum size. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::{Error, ConfigValidationError}; + /// + /// let err = Error::config_validation(ConfigValidationError::too_high("max_retries", 15, 10)); + /// assert!(err.to_string().contains("Configuration error")); + /// ``` + #[error("Configuration error: {0}")] + ConfigValidation(Box), + /// Error with additional context, preserving the error chain. #[error("{context}")] Context { @@ -411,6 +430,22 @@ impl Error { Self::ResourceExhausted(msg.into()) } + /// Creates a configuration validation error. + /// + /// Use this when configuration parameters fail validation. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::{Error, ConfigValidationError}; + /// + /// let err = Error::config_validation(ConfigValidationError::too_high("max_retries", 15, 10)); + /// assert!(err.to_string().contains("Configuration error")); + /// ``` + pub fn config_validation(err: ConfigValidationError) -> Self { + Self::ConfigValidation(Box::new(err)) + } + /// Creates an invalid request error. pub fn invalid_request(msg: impl Into>) -> Self { Self::InvalidRequest(msg.into()) @@ -626,6 +661,30 @@ impl Error { } } + /// Checks if this is a configuration validation error (penetrates Context layers). + /// Returns a reference to the `ConfigValidationError`. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::error::{Error, ConfigValidationError}; + /// + /// let err = Error::config_validation(ConfigValidationError::too_high("max_retries", 15, 10)); + /// assert!(err.as_config_validation().is_some()); + /// + /// // Works through context layers + /// let wrapped = err.context("Wrapped error"); + /// assert!(wrapped.as_config_validation().is_some()); + /// ``` + #[must_use] + pub fn as_config_validation(&self) -> Option<&ConfigValidationError> { + match self { + Error::ConfigValidation(err) => Some(err.as_ref()), + Error::Context { source, .. } => source.as_config_validation(), + _ => None, + } + } + /// Attempts to downcast the WebSocket error to a specific type. #[must_use] pub fn downcast_websocket(&self) -> Option<&T> { diff --git a/ccxt-core/src/http_client.rs b/ccxt-core/src/http_client.rs index 05a1d00..f377665 100644 --- a/ccxt-core/src/http_client.rs +++ b/ccxt-core/src/http_client.rs @@ -17,12 +17,14 @@ //! - HTTP response status and body preview //! - Error details with structured fields +use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; use crate::config::ProxyConfig; -use crate::error::{Error, Result}; +use crate::error::{ConfigValidationError, Error, Result, ValidationResult}; use crate::rate_limiter::RateLimiter; use crate::retry_strategy::{RetryConfig, RetryStrategy}; use reqwest::{Client, Method, Response, StatusCode, header::HeaderMap}; use serde_json::Value; +use std::sync::Arc; use std::time::Duration; use tracing::{debug, error, info, instrument, warn}; @@ -48,6 +50,20 @@ pub struct HttpConfig { pub enable_rate_limit: bool, /// Optional retry configuration (uses default if `None`) pub retry_config: Option, + /// Maximum response body size in bytes (default: 10MB) + /// + /// Responses exceeding this limit will be rejected with an `InvalidRequest` error. + /// This protects against malicious or abnormal responses that could exhaust memory. + pub max_response_size: usize, + + /// Optional circuit breaker configuration. + /// + /// When enabled, the circuit breaker will track request failures and + /// automatically block requests to failing endpoints, allowing the + /// system to recover. + /// + /// Default: `None` (disabled for backward compatibility) + pub circuit_breaker: Option, } impl Default for HttpConfig { @@ -63,7 +79,66 @@ impl Default for HttpConfig { proxy: None, enable_rate_limit: true, retry_config: None, // Uses default retry configuration + max_response_size: 10 * 1024 * 1024, // 10MB default + circuit_breaker: None, // Disabled by default for backward compatibility + } + } +} + +impl HttpConfig { + /// Validates the HTTP configuration parameters. + /// + /// # Returns + /// + /// Returns `Ok(ValidationResult)` if the configuration is valid. + /// The `ValidationResult` may contain warnings for suboptimal but valid configurations. + /// + /// Returns `Err(ConfigValidationError)` if the configuration is invalid. + /// + /// # Validation Rules + /// + /// - `timeout` > 5 minutes returns an error (excessive timeout) + /// - `timeout` < 1 second generates a warning (may cause frequent timeouts) + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::http_client::HttpConfig; + /// use std::time::Duration; + /// + /// let config = HttpConfig::default(); + /// let result = config.validate(); + /// assert!(result.is_ok()); + /// + /// let invalid_config = HttpConfig { + /// timeout: Duration::from_secs(600), // 10 minutes - too long + /// ..Default::default() + /// }; + /// let result = invalid_config.validate(); + /// assert!(result.is_err()); + /// ``` + pub fn validate(&self) -> std::result::Result { + let mut warnings = Vec::new(); + + // Validate timeout <= 5 minutes + let max_timeout = Duration::from_secs(300); // 5 minutes + if self.timeout > max_timeout { + return Err(ConfigValidationError::too_high( + "timeout", + format!("{:?}", self.timeout), + "5 minutes", + )); + } + + // Warn if timeout < 1 second + if self.timeout < Duration::from_secs(1) { + warnings.push(format!( + "timeout {:?} is very short, may cause frequent timeouts", + self.timeout + )); } + + Ok(ValidationResult::with_warnings(warnings)) } } @@ -74,6 +149,7 @@ pub struct HttpClient { config: HttpConfig, rate_limiter: Option, retry_strategy: RetryStrategy, + circuit_breaker: Option>, } impl HttpClient { @@ -117,11 +193,18 @@ impl HttpClient { let retry_strategy = RetryStrategy::new(config.retry_config.clone().unwrap_or_default()); + // Create circuit breaker if configured + let circuit_breaker = config + .circuit_breaker + .as_ref() + .map(|cb_config| Arc::new(CircuitBreaker::new(cb_config.clone()))); + Ok(Self { client, config, rate_limiter: None, retry_strategy, + circuit_breaker, }) } @@ -163,6 +246,35 @@ impl HttpClient { self.retry_strategy = strategy; } + /// Sets the circuit breaker for this client. + /// + /// # Arguments + /// + /// * `circuit_breaker` - Circuit breaker instance to use + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::http_client::{HttpClient, HttpConfig}; + /// use ccxt_core::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; + /// use std::sync::Arc; + /// + /// let mut client = HttpClient::new(HttpConfig::default()).unwrap(); + /// let cb = Arc::new(CircuitBreaker::new(CircuitBreakerConfig::default())); + /// client.set_circuit_breaker(cb); + /// ``` + pub fn set_circuit_breaker(&mut self, circuit_breaker: Arc) { + self.circuit_breaker = Some(circuit_breaker); + } + + /// Returns a reference to the circuit breaker, if configured. + /// + /// This can be used to check the circuit breaker state or manually + /// record success/failure. + pub fn circuit_breaker(&self) -> Option<&Arc> { + self.circuit_breaker.as_ref() + } + /// Executes an HTTP request with automatic retry mechanism and timeout control. /// /// The entire retry flow is wrapped with `tokio::time::timeout` to ensure @@ -199,6 +311,11 @@ impl HttpClient { headers: Option, body: Option, ) -> Result { + // Check circuit breaker before making request + if let Some(ref cb) = self.circuit_breaker { + cb.allow_request()?; + } + // Lint: collapsible_if // Reason: Keeping separate for clarity - first check if rate limiting is enabled, then check limiter #[allow(clippy::collapsible_if)] @@ -214,7 +331,7 @@ impl HttpClient { let total_timeout = self.config.timeout; let url_for_error = url.to_string(); - match tokio::time::timeout( + let result = match tokio::time::timeout( total_timeout, self.execute_with_retry(|| { let url = url.to_string(); @@ -240,7 +357,17 @@ impl HttpClient { total_timeout.as_millis() ))) } + }; + + // Record success/failure with circuit breaker + if let Some(ref cb) = self.circuit_breaker { + match &result { + Ok(_) => cb.record_success(), + Err(_) => cb.record_failure(), + } } + + result } /// Executes an async operation with automatic retry mechanism. @@ -331,25 +458,56 @@ impl HttpClient { Error::network(format!("Request failed: {e}")) })?; - self.process_response(response).await + self.process_response_with_limit(response, url).await } - /// Processes an HTTP response, handling errors and parsing JSON. - #[instrument(name = "http_process_response", skip(self, response), fields(status))] - async fn process_response(&self, response: Response) -> Result { + /// Processes an HTTP response with size limit enforcement. + /// + /// This method checks the Content-Length header first and rejects responses + /// that exceed the configured `max_response_size`. For responses without + /// Content-Length, it streams the body while tracking accumulated size. + /// + /// # Arguments + /// + /// * `response` - The HTTP response to process + /// * `url` - The request URL (for logging purposes) + /// + /// # Returns + /// + /// Returns the response body as a JSON `Value`. + /// + /// # Errors + /// + /// Returns `InvalidRequest` error if the response exceeds `max_response_size`. + #[instrument(name = "http_process_response_with_limit", skip(self, response), fields(status, url = %url))] + async fn process_response_with_limit(&self, response: Response, url: &str) -> Result { let status = response.status(); let headers = response.headers().clone(); + let max_size = self.config.max_response_size; // Record the status in the span tracing::Span::current().record("status", status.as_u16()); - let body_text = response.text().await.map_err(|e| { - error!( - error = %e, - "Failed to read response body" + // Check Content-Length header first (Requirement 1.2) + if let Some(content_length) = response.content_length() + && content_length > max_size as u64 + { + warn!( + url = %url, + content_length = content_length, + max_size = max_size, + "Response exceeds size limit (Content-Length check)" ); - Error::network(format!("Failed to read response body: {e}")) - })?; + return Err(Error::invalid_request(format!( + "Response size {content_length} bytes exceeds limit {max_size} bytes" + ))); + } + + // Stream response body with size tracking (Requirement 1.3) + let body_bytes = self + .stream_response_with_limit(response, url, max_size) + .await?; + let body_text = String::from_utf8_lossy(&body_bytes).to_string(); // Log response details (truncate body preview for large responses) let body_preview: String = body_text.chars().take(200).collect(); @@ -384,6 +542,66 @@ impl HttpClient { Ok(result) } + /// Streams response body while enforcing size limit. + /// + /// This method reads the response body in chunks and aborts if the + /// accumulated size exceeds the configured limit. + /// + /// # Arguments + /// + /// * `response` - The HTTP response to stream + /// * `url` - The request URL (for logging purposes) + /// * `max_size` - Maximum allowed response size in bytes + /// + /// # Returns + /// + /// Returns the complete response body as bytes. + /// + /// # Errors + /// + /// Returns `InvalidRequest` error if accumulated size exceeds `max_size`. + async fn stream_response_with_limit( + &self, + response: Response, + url: &str, + max_size: usize, + ) -> Result> { + use futures_util::StreamExt; + + let mut stream = response.bytes_stream(); + let mut body = Vec::new(); + let mut accumulated_size: usize = 0; + + while let Some(chunk_result) = stream.next().await { + let chunk = chunk_result.map_err(|e| { + error!( + error = %e, + "Failed to read response chunk" + ); + Error::network(format!("Failed to read response chunk: {e}")) + })?; + + accumulated_size = accumulated_size.saturating_add(chunk.len()); + + // Check if accumulated size exceeds limit (Requirement 1.3) + if accumulated_size > max_size { + warn!( + url = %url, + accumulated_size = accumulated_size, + max_size = max_size, + "Response exceeds size limit during streaming" + ); + return Err(Error::invalid_request(format!( + "Response size {accumulated_size} bytes exceeds limit {max_size} bytes (streaming)" + ))); + } + + body.extend_from_slice(&chunk); + } + + Ok(body) + } + /// Handles HTTP error status codes and converts them to appropriate errors. #[instrument( name = "http_handle_error", @@ -602,6 +820,7 @@ mod tests { assert!(!config.verbose); assert_eq!(config.user_agent, "ccxt-rust/1.0"); assert!(config.enable_rate_limit); + assert_eq!(config.max_response_size, 10 * 1024 * 1024); // 10MB } #[tokio::test] @@ -850,4 +1069,238 @@ mod tests { assert!(result.is_err()); } + + #[test] + fn test_http_config_validate_default() { + let config = HttpConfig::default(); + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + } + + #[test] + fn test_http_config_validate_timeout_too_high() { + let config = HttpConfig { + timeout: Duration::from_secs(600), // 10 minutes + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.field_name(), "timeout"); + assert!(matches!( + err, + crate::error::ConfigValidationError::ValueTooHigh { .. } + )); + } + + #[test] + fn test_http_config_validate_timeout_boundary() { + // timeout = 5 minutes should be valid + let config = HttpConfig { + timeout: Duration::from_secs(300), + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + + // timeout = 301 seconds should be invalid + let config = HttpConfig { + timeout: Duration::from_secs(301), + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_err()); + } + + #[test] + fn test_http_config_validate_short_timeout_warning() { + let config = HttpConfig { + timeout: Duration::from_millis(500), + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_ok()); + let validation_result = result.unwrap(); + assert!(!validation_result.warnings.is_empty()); + assert!(validation_result.warnings[0].contains("timeout")); + assert!(validation_result.warnings[0].contains("very short")); + } + + #[test] + fn test_http_config_validate_timeout_warning_boundary() { + // timeout = 1 second should not generate warning + let config = HttpConfig { + timeout: Duration::from_secs(1), + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + + // timeout = 999ms should generate warning + let config = HttpConfig { + timeout: Duration::from_millis(999), + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_ok()); + assert!(!result.unwrap().warnings.is_empty()); + } + + #[test] + fn test_http_config_max_response_size_default() { + let config = HttpConfig::default(); + assert_eq!(config.max_response_size, 10 * 1024 * 1024); // 10MB + } + + #[test] + fn test_http_config_max_response_size_custom() { + let config = HttpConfig { + max_response_size: 1024 * 1024, // 1MB + ..Default::default() + }; + assert_eq!(config.max_response_size, 1024 * 1024); + } + + #[tokio::test] + async fn test_response_size_limit_small_response() { + // Test that small responses pass through normally + let config = HttpConfig { + max_response_size: 10 * 1024 * 1024, // 10MB + ..Default::default() + }; + let client = HttpClient::new(config).unwrap(); + + // httpbin.org/get returns a small JSON response + match client.get("https://httpbin.org/get", None).await { + Ok(value) => { + assert!(value.is_object()); + } + Err(e) => { + // Network errors are acceptable in CI environments + warn!("Network test skipped due to: {:?}", e); + } + } + } + + #[tokio::test] + async fn test_response_size_limit_with_content_length() { + // Test that responses with Content-Length exceeding limit are rejected + let config = HttpConfig { + max_response_size: 100, // Very small limit (100 bytes) + retry_config: Some(RetryConfig { + max_retries: 0, + ..RetryConfig::default() + }), + ..Default::default() + }; + let client = HttpClient::new(config).unwrap(); + + // httpbin.org/bytes/1000 returns 1000 bytes with Content-Length header + let result = client.get("https://httpbin.org/bytes/1000", None).await; + + match result { + Err(Error::InvalidRequest(msg)) => { + // Expected: response should be rejected due to size limit + assert!( + msg.contains("exceeds limit") || msg.contains("size"), + "Error message should mention size limit: {}", + msg + ); + } + Err(Error::Network(_)) => { + // Network errors are acceptable in CI environments + warn!("Network test skipped due to network error"); + } + Ok(_) => { + // If the response somehow passed, it might be because httpbin + // returned a smaller response or the Content-Length was not set + warn!("Response passed size check - httpbin may have returned smaller response"); + } + Err(e) => { + panic!("Unexpected error type: {:?}", e); + } + } + } + + #[test] + fn test_http_config_circuit_breaker_default() { + let config = HttpConfig::default(); + assert!(config.circuit_breaker.is_none()); + } + + #[test] + fn test_http_config_circuit_breaker_custom() { + use crate::circuit_breaker::CircuitBreakerConfig; + + let config = HttpConfig { + circuit_breaker: Some(CircuitBreakerConfig::default()), + ..Default::default() + }; + assert!(config.circuit_breaker.is_some()); + } + + #[test] + fn test_http_client_circuit_breaker_disabled_by_default() { + let config = HttpConfig::default(); + let client = HttpClient::new(config).unwrap(); + assert!(client.circuit_breaker().is_none()); + } + + #[test] + fn test_http_client_circuit_breaker_enabled() { + use crate::circuit_breaker::CircuitBreakerConfig; + + let config = HttpConfig { + circuit_breaker: Some(CircuitBreakerConfig::default()), + ..Default::default() + }; + let client = HttpClient::new(config).unwrap(); + assert!(client.circuit_breaker().is_some()); + } + + #[test] + fn test_http_client_set_circuit_breaker() { + use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig}; + + let config = HttpConfig::default(); + let mut client = HttpClient::new(config).unwrap(); + assert!(client.circuit_breaker().is_none()); + + let cb = Arc::new(CircuitBreaker::new(CircuitBreakerConfig::default())); + client.set_circuit_breaker(cb); + assert!(client.circuit_breaker().is_some()); + } + + #[tokio::test] + async fn test_http_client_circuit_breaker_blocks_when_open() { + use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitState}; + + let config = HttpConfig::default(); + let mut client = HttpClient::new(config).unwrap(); + + // Create a circuit breaker that's already open + let cb_config = CircuitBreakerConfig { + failure_threshold: 1, + reset_timeout: Duration::from_secs(60), // Long timeout + success_threshold: 1, + }; + let cb = Arc::new(CircuitBreaker::new(cb_config)); + + // Open the circuit + cb.record_failure(); + assert_eq!(cb.state(), CircuitState::Open); + + client.set_circuit_breaker(cb); + + // Request should be blocked + let result = client.get("https://httpbin.org/get", None).await; + assert!(result.is_err()); + + if let Err(e) = result { + assert!(e.as_resource_exhausted().is_some()); + } + } } diff --git a/ccxt-core/src/lib.rs b/ccxt-core/src/lib.rs index 9944607..22aa022 100644 --- a/ccxt-core/src/lib.rs +++ b/ccxt-core/src/lib.rs @@ -85,6 +85,7 @@ pub use serde_json; pub mod auth; pub mod base_exchange; pub mod capability; +pub mod circuit_breaker; pub mod config; pub mod credentials; pub mod error; @@ -113,6 +114,9 @@ pub mod test_config; // Re-exports of core types for convenience pub use base_exchange::{BaseExchange, ExchangeConfig, ExchangeConfigBuilder, MarketCache}; +pub use circuit_breaker::{ + CircuitBreaker, CircuitBreakerConfig, CircuitBreakerEvent, CircuitState, +}; pub use credentials::{SecretBytes, SecretString}; // Re-export unified Exchange trait from exchange module pub use exchange::{ArcExchange, BoxedExchange, Exchange, ExchangeExt}; @@ -160,6 +164,7 @@ pub mod prelude { pub use crate::base_exchange::{ BaseExchange, ExchangeConfig, ExchangeConfigBuilder, MarketCache, }; + pub use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitState}; pub use crate::error::{ContextExt, Error, Result}; // Re-export unified Exchange trait and capabilities pub use crate::exchange::{ diff --git a/ccxt-core/src/rate_limiter.rs b/ccxt-core/src/rate_limiter.rs index d413fed..6d74a23 100644 --- a/ccxt-core/src/rate_limiter.rs +++ b/ccxt-core/src/rate_limiter.rs @@ -27,6 +27,7 @@ //! # } //! ``` +use crate::error::{ConfigValidationError, ValidationResult}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::Mutex; @@ -91,6 +92,58 @@ impl Default for RateLimiterConfig { } } +impl RateLimiterConfig { + /// Validates the rate limiter configuration parameters. + /// + /// # Returns + /// + /// Returns `Ok(ValidationResult)` if the configuration is valid. + /// The `ValidationResult` may contain warnings for suboptimal but valid configurations. + /// + /// Returns `Err(ConfigValidationError)` if the configuration is invalid. + /// + /// # Validation Rules + /// + /// - `capacity` must be > 0 (zero capacity is invalid) + /// - `refill_period` < 100ms generates a warning (may cause high CPU usage) + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::rate_limiter::RateLimiterConfig; + /// use std::time::Duration; + /// + /// let config = RateLimiterConfig::new(10, Duration::from_secs(1)); + /// let result = config.validate(); + /// assert!(result.is_ok()); + /// + /// let invalid_config = RateLimiterConfig::new(0, Duration::from_secs(1)); + /// let result = invalid_config.validate(); + /// assert!(result.is_err()); + /// ``` + pub fn validate(&self) -> Result { + let mut warnings = Vec::new(); + + // Validate capacity > 0 + if self.capacity == 0 { + return Err(ConfigValidationError::invalid( + "capacity", + "capacity cannot be zero", + )); + } + + // Warn if refill_period < 100ms + if self.refill_period < Duration::from_millis(100) { + warnings.push(format!( + "refill_period {:?} is very short, may cause high CPU usage", + self.refill_period + )); + } + + Ok(ValidationResult::with_warnings(warnings)) + } +} + /// Internal state of the rate limiter #[derive(Debug)] struct RateLimiterState { @@ -98,6 +151,10 @@ struct RateLimiterState { tokens: u32, /// Last time tokens were refilled last_refill: Instant, + /// Accumulated nanoseconds for fractional token tracking (precision optimization) + /// This field tracks the remainder when elapsed time doesn't evenly divide into refill periods, + /// preventing floating-point precision drift over long-running applications. + remainder_nanos: u64, /// Configuration config: RateLimiterConfig, } @@ -107,23 +164,60 @@ impl RateLimiterState { Self { tokens: config.capacity, last_refill: Instant::now(), + remainder_nanos: 0, config, } } - /// Refill tokens based on elapsed time + /// Refill tokens based on elapsed time using integer arithmetic + /// + /// This implementation uses nanosecond-based integer arithmetic instead of + /// floating-point seconds to avoid precision drift over long-running applications. + /// The formula used is: `elapsed_nanos / period_nanos * refill_amount` + /// + /// The `remainder_nanos` field tracks fractional periods that haven't yet + /// accumulated to a full refill period, ensuring no time is lost between refills. + #[allow(clippy::cast_possible_truncation)] fn refill(&mut self) { let now = Instant::now(); - let elapsed = now.duration_since(self.last_refill); + // Note: as_nanos() returns u128, but for practical durations (< 584 years), + // u64 is sufficient. We use saturating conversion to handle edge cases. + let elapsed_nanos = now + .duration_since(self.last_refill) + .as_nanos() + .min(u128::from(u64::MAX)) as u64; + let period_nanos = self + .config + .refill_period + .as_nanos() + .min(u128::from(u64::MAX)) as u64; + + // Avoid division by zero (should not happen with valid config) + if period_nanos == 0 { + return; + } + + // Add elapsed time to accumulated remainder + let total_nanos = self.remainder_nanos.saturating_add(elapsed_nanos); - if elapsed >= self.config.refill_period { - // Calculate how many refill periods have passed - let periods = elapsed.as_secs_f64() / self.config.refill_period.as_secs_f64(); - #[allow(clippy::cast_possible_truncation)] - let tokens_to_add = (periods * f64::from(self.config.refill_amount)) as u32; + // Calculate complete periods using integer division + let complete_periods = total_nanos / period_nanos; + + if complete_periods > 0 { + // Calculate tokens to add using integer arithmetic + // Use u64 for intermediate calculation to avoid overflow + let tokens_to_add = complete_periods + .saturating_mul(u64::from(self.config.refill_amount)) + .min(u64::from(u32::MAX)) as u32; // Add tokens up to capacity - self.tokens = (self.tokens + tokens_to_add).min(self.config.capacity); + self.tokens = self + .tokens + .saturating_add(tokens_to_add) + .min(self.config.capacity); + + // Store remainder for next refill (preserves fractional periods) + self.remainder_nanos = total_nanos % period_nanos; self.last_refill = now; } } @@ -278,6 +372,7 @@ impl RateLimiter { let mut state = self.state.lock().await; state.tokens = state.config.capacity; state.last_refill = Instant::now(); + state.remainder_nanos = 0; } } @@ -503,4 +598,160 @@ mod tests { // All tokens should be consumed assert_eq!(limiter.available_tokens().await, 0); } + + #[test] + fn test_rate_limiter_config_validate_default() { + let config = RateLimiterConfig::default(); + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + } + + #[test] + fn test_rate_limiter_config_validate_zero_capacity() { + let config = RateLimiterConfig::new(0, Duration::from_secs(1)); + let result = config.validate(); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.field_name(), "capacity"); + assert!(matches!( + err, + crate::error::ConfigValidationError::ValueInvalid { .. } + )); + } + + #[test] + fn test_rate_limiter_config_validate_short_refill_period_warning() { + let config = RateLimiterConfig::new(10, Duration::from_millis(50)); + let result = config.validate(); + assert!(result.is_ok()); + let validation_result = result.unwrap(); + assert!(!validation_result.warnings.is_empty()); + assert!(validation_result.warnings[0].contains("refill_period")); + assert!(validation_result.warnings[0].contains("very short")); + } + + #[test] + fn test_rate_limiter_config_validate_refill_period_boundary() { + // refill_period = 100ms should not generate warning + let config = RateLimiterConfig::new(10, Duration::from_millis(100)); + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + + // refill_period = 99ms should generate warning + let config = RateLimiterConfig::new(10, Duration::from_millis(99)); + let result = config.validate(); + assert!(result.is_ok()); + assert!(!result.unwrap().warnings.is_empty()); + } + + #[test] + fn test_rate_limiter_config_validate_valid_config() { + let config = RateLimiterConfig::new(100, Duration::from_secs(60)); + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + } + + /// Test that the integer-based refill calculation maintains precision + /// over many iterations without drift. + #[test] + fn test_rate_limiter_integer_precision() { + // Create a state with a specific configuration + let config = RateLimiterConfig::new(100, Duration::from_millis(100)).with_refill_amount(10); + let mut state = RateLimiterState::new(config); + + // Consume all tokens + state.tokens = 0; + + // Simulate many small time increments that don't complete a full period + // This tests the remainder_nanos accumulation + let period_nanos = 100_000_000u64; // 100ms in nanos + let small_increment = 33_333_333u64; // ~33.3ms in nanos + + // After 3 increments of ~33.3ms, we should have ~100ms total + // which equals 1 complete period = 10 tokens + state.remainder_nanos = small_increment; + state.remainder_nanos += small_increment; + state.remainder_nanos += small_increment; + + // Calculate complete periods + let complete_periods = state.remainder_nanos / period_nanos; + assert_eq!(complete_periods, 0); // 99.9ms < 100ms, no complete period yet + + // Add one more nanosecond to push over the threshold + state.remainder_nanos += 1; + let complete_periods = state.remainder_nanos / period_nanos; + assert_eq!(complete_periods, 1); // Now we have 1 complete period + + // Verify remainder is preserved correctly + let expected_remainder = (small_increment * 3 + 1) % period_nanos; + assert_eq!(state.remainder_nanos % period_nanos, expected_remainder); + } + + /// Test that remainder_nanos is properly reset when reset() is called + #[tokio::test] + async fn test_rate_limiter_reset_clears_remainder() { + let config = RateLimiterConfig::new(5, Duration::from_secs(1)); + let limiter = RateLimiter::new(config); + + // Use all tokens + for _ in 0..5 { + limiter.wait().await; + } + + // Wait a bit to accumulate some remainder + sleep(Duration::from_millis(50)).await; + + // Reset should clear everything including remainder + limiter.reset().await; + + // Verify tokens are back to capacity + assert_eq!(limiter.available_tokens().await, 5); + } + + /// Test that the refill calculation handles zero period_nanos gracefully + #[test] + fn test_rate_limiter_refill_zero_period_protection() { + // This tests the edge case protection in refill() + // A zero period should not cause division by zero + let config = RateLimiterConfig { + capacity: 10, + refill_period: Duration::ZERO, // Edge case: zero duration + refill_amount: 5, + cost_per_request: 1, + }; + let mut state = RateLimiterState::new(config); + state.tokens = 0; + + // This should not panic due to division by zero + state.refill(); + + // Tokens should remain unchanged (no refill with zero period) + assert_eq!(state.tokens, 0); + } + + /// Test that integer arithmetic doesn't overflow with large values + #[test] + fn test_rate_limiter_refill_overflow_protection() { + let config = + RateLimiterConfig::new(u32::MAX, Duration::from_nanos(1)).with_refill_amount(u32::MAX); + let mut state = RateLimiterState::new(config); + state.tokens = 0; + + // Simulate a large elapsed time + state.remainder_nanos = u64::MAX / 2; + + // This should not panic due to overflow + // The saturating_mul and min operations should protect against overflow + let period_nanos = 1u64; + let complete_periods = state.remainder_nanos / period_nanos; + let tokens_to_add = complete_periods + .saturating_mul(u64::from(state.config.refill_amount)) + .min(u64::from(u32::MAX)) as u32; + + // Should be capped at u32::MAX + assert_eq!(tokens_to_add, u32::MAX); + } } diff --git a/ccxt-core/src/retry_strategy.rs b/ccxt-core/src/retry_strategy.rs index f7befc9..f1554e9 100644 --- a/ccxt-core/src/retry_strategy.rs +++ b/ccxt-core/src/retry_strategy.rs @@ -7,9 +7,21 @@ //! - Configurable retry conditions //! - Retry budget mechanism -use crate::error::Error; +use crate::error::{ConfigValidationError, Error, ValidationResult}; +use regex::Regex; +use std::sync::LazyLock; use std::time::Duration; +/// Regex pattern for detecting server error messages using word boundary matching. +/// Matches: +/// - HTTP status codes 500, 502, 503, 504 as standalone numbers (not part of larger numbers) +/// - Common server error phrases +static SERVER_ERROR_PATTERN: LazyLock = LazyLock::new(|| { + Regex::new( + r"(?i)\b(500|502|503|504)\b|internal server error|bad gateway|service unavailable|gateway timeout" + ).expect("Invalid server error regex pattern") +}); + /// Retry strategy type. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RetryStrategyType { @@ -105,6 +117,60 @@ impl RetryConfig { jitter_factor: 0.0, } } + + /// Validates the retry configuration parameters. + /// + /// # Returns + /// + /// Returns `Ok(ValidationResult)` if the configuration is valid. + /// The `ValidationResult` may contain warnings for suboptimal but valid configurations. + /// + /// Returns `Err(ConfigValidationError)` if the configuration is invalid. + /// + /// # Validation Rules + /// + /// - `max_retries` must be <= 10 (excessive retries can cause issues) + /// - `base_delay_ms` must be >= 10 (too short delays can cause thundering herd) + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::retry_strategy::RetryConfig; + /// + /// let config = RetryConfig::default(); + /// let result = config.validate(); + /// assert!(result.is_ok()); + /// + /// let invalid_config = RetryConfig { + /// max_retries: 15, // Too high + /// ..Default::default() + /// }; + /// let result = invalid_config.validate(); + /// assert!(result.is_err()); + /// ``` + pub fn validate(&self) -> Result { + let warnings = Vec::new(); + + // Validate max_retries <= 10 + if self.max_retries > 10 { + return Err(ConfigValidationError::too_high( + "max_retries", + self.max_retries, + 10, + )); + } + + // Validate base_delay_ms >= 10 + if self.base_delay_ms < 10 { + return Err(ConfigValidationError::too_low( + "base_delay_ms", + self.base_delay_ms, + 10, + )); + } + + Ok(ValidationResult::with_warnings(warnings)) + } } /// Retry strategy. @@ -197,16 +263,103 @@ impl RetryStrategy { } /// Checks if the message indicates a server error (5xx). - fn is_server_error(msg: &str) -> bool { - let msg_lower = msg.to_lowercase(); - msg_lower.contains("500") - || msg_lower.contains("502") - || msg_lower.contains("503") - || msg_lower.contains("504") - || msg_lower.contains("internal server error") - || msg_lower.contains("bad gateway") - || msg_lower.contains("service unavailable") - || msg_lower.contains("gateway timeout") + /// + /// Uses word boundary matching to avoid false positives like "order_id: 15001234" + /// being misidentified as containing a 500 error. + /// + /// # Arguments + /// + /// * `msg` - The error message to check. + /// + /// # Returns + /// + /// `true` if the message indicates a server error, `false` otherwise. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::retry_strategy::RetryStrategy; + /// + /// // True positives + /// assert!(RetryStrategy::is_server_error("500 Internal Server Error")); + /// assert!(RetryStrategy::is_server_error("HTTP 502 Bad Gateway")); + /// + /// // False positives avoided + /// assert!(!RetryStrategy::is_server_error("order_id: 15001234")); + /// assert!(!RetryStrategy::is_server_error("amount: 5000")); + /// ``` + pub fn is_server_error(msg: &str) -> bool { + Self::is_server_error_message(msg) + } + + /// Checks if an HTTP status code indicates a server error (5xx). + /// + /// Server errors are HTTP status codes in the range 500-599. + /// + /// # Arguments + /// + /// * `status` - The HTTP status code to check. + /// + /// # Returns + /// + /// `true` if the status code is in the 500-599 range, `false` otherwise. + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::retry_strategy::RetryStrategy; + /// + /// assert!(RetryStrategy::is_server_error_code(500)); + /// assert!(RetryStrategy::is_server_error_code(502)); + /// assert!(RetryStrategy::is_server_error_code(503)); + /// assert!(RetryStrategy::is_server_error_code(504)); + /// assert!(RetryStrategy::is_server_error_code(599)); + /// + /// assert!(!RetryStrategy::is_server_error_code(200)); + /// assert!(!RetryStrategy::is_server_error_code(400)); + /// assert!(!RetryStrategy::is_server_error_code(404)); + /// assert!(!RetryStrategy::is_server_error_code(499)); + /// assert!(!RetryStrategy::is_server_error_code(600)); + /// ``` + pub fn is_server_error_code(status: u16) -> bool { + (500..600).contains(&status) + } + + /// Checks if a message indicates a server error using word boundary matching. + /// + /// This method uses regex with word boundaries to precisely detect server error + /// patterns without false positives from numbers that happen to contain 500, 502, etc. + /// + /// # Arguments + /// + /// * `msg` - The error message to check. + /// + /// # Returns + /// + /// `true` if the message matches server error patterns, `false` otherwise. + /// + /// # Patterns Matched + /// + /// - Status codes: `500`, `502`, `503`, `504` (as standalone words) + /// - Phrases: "internal server error", "bad gateway", "service unavailable", "gateway timeout" + /// + /// # Example + /// + /// ```rust + /// use ccxt_core::retry_strategy::RetryStrategy; + /// + /// // Matches server error patterns + /// assert!(RetryStrategy::is_server_error_message("500 Internal Server Error")); + /// assert!(RetryStrategy::is_server_error_message("Error: 502")); + /// assert!(RetryStrategy::is_server_error_message("Service Unavailable")); + /// + /// // Does NOT match numbers containing 500, 502, etc. + /// assert!(!RetryStrategy::is_server_error_message("order_id: 15001234")); + /// assert!(!RetryStrategy::is_server_error_message("balance: 5020.50")); + /// assert!(!RetryStrategy::is_server_error_message("id=25030")); + /// ``` + pub fn is_server_error_message(msg: &str) -> bool { + SERVER_ERROR_PATTERN.is_match(msg) } /// Checks if the message indicates a timeout error. @@ -358,12 +511,100 @@ mod tests { #[test] fn test_is_server_error() { + // True positives - should detect server errors assert!(RetryStrategy::is_server_error("500 Internal Server Error")); assert!(RetryStrategy::is_server_error("502 Bad Gateway")); assert!(RetryStrategy::is_server_error("503 Service Unavailable")); assert!(RetryStrategy::is_server_error("504 Gateway Timeout")); + assert!(RetryStrategy::is_server_error("HTTP 500")); + assert!(RetryStrategy::is_server_error("Error: 502")); + assert!(RetryStrategy::is_server_error("Status 503")); + assert!(RetryStrategy::is_server_error("internal server error")); + assert!(RetryStrategy::is_server_error("bad gateway")); + assert!(RetryStrategy::is_server_error("service unavailable")); + assert!(RetryStrategy::is_server_error("gateway timeout")); + + // True negatives - should NOT detect as server errors assert!(!RetryStrategy::is_server_error("400 Bad Request")); assert!(!RetryStrategy::is_server_error("404 Not Found")); + assert!(!RetryStrategy::is_server_error("200 OK")); + + // False positive prevention - numbers containing 500, 502, etc. + assert!(!RetryStrategy::is_server_error("order_id: 15001234")); + assert!(!RetryStrategy::is_server_error("balance: 5020.50")); + assert!(!RetryStrategy::is_server_error("id=25030")); + assert!(!RetryStrategy::is_server_error("amount: 5000")); + assert!(!RetryStrategy::is_server_error("price: 50200")); + assert!(!RetryStrategy::is_server_error("timestamp: 1500123456789")); + } + + #[test] + fn test_is_server_error_code() { + // Server error codes (500-599) + assert!(RetryStrategy::is_server_error_code(500)); + assert!(RetryStrategy::is_server_error_code(501)); + assert!(RetryStrategy::is_server_error_code(502)); + assert!(RetryStrategy::is_server_error_code(503)); + assert!(RetryStrategy::is_server_error_code(504)); + assert!(RetryStrategy::is_server_error_code(505)); + assert!(RetryStrategy::is_server_error_code(599)); + + // Non-server error codes + assert!(!RetryStrategy::is_server_error_code(200)); + assert!(!RetryStrategy::is_server_error_code(201)); + assert!(!RetryStrategy::is_server_error_code(301)); + assert!(!RetryStrategy::is_server_error_code(400)); + assert!(!RetryStrategy::is_server_error_code(401)); + assert!(!RetryStrategy::is_server_error_code(403)); + assert!(!RetryStrategy::is_server_error_code(404)); + assert!(!RetryStrategy::is_server_error_code(429)); + assert!(!RetryStrategy::is_server_error_code(499)); + assert!(!RetryStrategy::is_server_error_code(600)); + assert!(!RetryStrategy::is_server_error_code(0)); + } + + #[test] + fn test_is_server_error_message() { + // Standard server error messages + assert!(RetryStrategy::is_server_error_message( + "500 Internal Server Error" + )); + assert!(RetryStrategy::is_server_error_message("502 Bad Gateway")); + assert!(RetryStrategy::is_server_error_message( + "503 Service Unavailable" + )); + assert!(RetryStrategy::is_server_error_message( + "504 Gateway Timeout" + )); + + // Case insensitive matching + assert!(RetryStrategy::is_server_error_message( + "INTERNAL SERVER ERROR" + )); + assert!(RetryStrategy::is_server_error_message("Bad Gateway")); + assert!(RetryStrategy::is_server_error_message( + "SERVICE UNAVAILABLE" + )); + + // Status codes at different positions + assert!(RetryStrategy::is_server_error_message("Error 500")); + assert!(RetryStrategy::is_server_error_message("HTTP/1.1 502")); + assert!(RetryStrategy::is_server_error_message("Status: 503")); + assert!(RetryStrategy::is_server_error_message("504")); + + // Word boundary tests - should NOT match + assert!(!RetryStrategy::is_server_error_message( + "order_id: 15001234" + )); + assert!(!RetryStrategy::is_server_error_message("balance: 5020.50")); + assert!(!RetryStrategy::is_server_error_message("id=25030")); + assert!(!RetryStrategy::is_server_error_message("amount: 5000")); + assert!(!RetryStrategy::is_server_error_message("price: 50200")); + assert!(!RetryStrategy::is_server_error_message( + "timestamp: 1500123456789" + )); + assert!(!RetryStrategy::is_server_error_message("order5020")); + assert!(!RetryStrategy::is_server_error_message("5030items")); } #[test] @@ -374,6 +615,80 @@ mod tests { assert!(!RetryStrategy::is_timeout_error("Connection refused")); } + #[test] + fn test_retry_config_validate_default() { + let config = RetryConfig::default(); + let result = config.validate(); + assert!(result.is_ok()); + assert!(result.unwrap().warnings.is_empty()); + } + + #[test] + fn test_retry_config_validate_max_retries_too_high() { + let config = RetryConfig { + max_retries: 15, + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.field_name(), "max_retries"); + assert!(matches!( + err, + crate::error::ConfigValidationError::ValueTooHigh { .. } + )); + } + + #[test] + fn test_retry_config_validate_max_retries_boundary() { + // max_retries = 10 should be valid + let config = RetryConfig { + max_retries: 10, + ..Default::default() + }; + assert!(config.validate().is_ok()); + + // max_retries = 11 should be invalid + let config = RetryConfig { + max_retries: 11, + ..Default::default() + }; + assert!(config.validate().is_err()); + } + + #[test] + fn test_retry_config_validate_base_delay_too_low() { + let config = RetryConfig { + base_delay_ms: 5, + ..Default::default() + }; + let result = config.validate(); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.field_name(), "base_delay_ms"); + assert!(matches!( + err, + crate::error::ConfigValidationError::ValueTooLow { .. } + )); + } + + #[test] + fn test_retry_config_validate_base_delay_boundary() { + // base_delay_ms = 10 should be valid + let config = RetryConfig { + base_delay_ms: 10, + ..Default::default() + }; + assert!(config.validate().is_ok()); + + // base_delay_ms = 9 should be invalid + let config = RetryConfig { + base_delay_ms: 9, + ..Default::default() + }; + assert!(config.validate().is_err()); + } + #[test] fn test_rate_limit_error_minimum_delay() { let config = RetryConfig { diff --git a/ccxt-core/src/ws_client/mod.rs b/ccxt-core/src/ws_client/mod.rs index 3155628..8c5fdc0 100644 --- a/ccxt-core/src/ws_client/mod.rs +++ b/ccxt-core/src/ws_client/mod.rs @@ -496,6 +496,23 @@ impl WsClient { self.subscription_manager.remaining_capacity() } + /// Returns a list of all active subscription channel names. + /// + /// Each subscription is identified by its channel name, optionally combined + /// with a symbol in the format "channel:symbol" or just "channel". + pub fn subscriptions(&self) -> Vec { + self.subscription_manager + .iter() + .map(|entry| { + let sub = entry.value(); + match &sub.symbol { + Some(sym) => format!("{}:{}", sub.channel, sym), + None => sub.channel.clone(), + } + }) + .collect() + } + /// Sends a raw WebSocket message. #[instrument(name = "ws_send", skip(self, message))] pub async fn send(&self, message: Message) -> Result<()> { diff --git a/ccxt-exchanges/src/binance/mod.rs b/ccxt-exchanges/src/binance/mod.rs index fd183b2..32f08dd 100644 --- a/ccxt-exchanges/src/binance/mod.rs +++ b/ccxt-exchanges/src/binance/mod.rs @@ -7,6 +7,7 @@ use ccxt_core::types::default_type::{DefaultSubType, DefaultType}; use ccxt_core::{BaseExchange, ExchangeConfig, Result}; use std::sync::Arc; use std::time::Duration; +use tokio::sync::RwLock; pub mod auth; pub mod builder; @@ -40,6 +41,8 @@ pub struct Binance { options: BinanceOptions, /// Time synchronization manager for caching server time offset. time_sync: Arc, + /// Persistent WebSocket connection reference for state tracking. + ws_connection: Arc>>, } impl Binance { @@ -94,6 +97,7 @@ impl Binance { base, options, time_sync, + ws_connection: Arc::new(RwLock::new(None)), }) } @@ -136,6 +140,7 @@ impl Binance { base, options, time_sync, + ws_connection: Arc::new(RwLock::new(None)), }) } @@ -173,6 +178,7 @@ impl Binance { base, options, time_sync, + ws_connection: Arc::new(RwLock::new(None)), }) } @@ -196,6 +202,32 @@ impl Binance { self.options = options; } + /// Returns a reference to the persistent WebSocket connection. + /// + /// This allows access to the stored WebSocket instance for state tracking. + /// The connection is stored when `ws_connect()` is called and cleared + /// when `ws_disconnect()` is called. + /// + /// # Example + /// + /// ```no_run + /// use ccxt_exchanges::binance::Binance; + /// use ccxt_core::ExchangeConfig; + /// + /// # async fn example() -> ccxt_core::Result<()> { + /// let binance = Binance::new(ExchangeConfig::default())?; + /// let ws_conn = binance.ws_connection(); + /// let guard = ws_conn.read().await; + /// if guard.is_some() { + /// println!("WebSocket is connected"); + /// } + /// # Ok(()) + /// # } + /// ``` + pub fn ws_connection(&self) -> &Arc>> { + &self.ws_connection + } + /// Returns a reference to the time synchronization manager. /// /// The `TimeSyncManager` caches the time offset between local system time diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index df31f98..eb9715e 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -45,6 +45,15 @@ pub struct BinanceWs { pub(crate) positions: Arc>>>, } +impl std::fmt::Debug for BinanceWs { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BinanceWs") + .field("is_connected", &self.client.is_connected()) + .field("state", &self.client.state()) + .finish_non_exhaustive() + } +} + impl BinanceWs { /// Creates a new Binance WebSocket client pub fn new(url: String) -> Self { @@ -283,6 +292,25 @@ impl BinanceWs { self.client.is_connected() } + /// Returns the current connection state. + pub fn state(&self) -> ccxt_core::ws_client::WsConnectionState { + self.client.state() + } + + /// Returns the list of active subscriptions. + /// + /// Returns a vector of subscription channel names that are currently active. + /// This method retrieves the actual subscriptions from the underlying WsClient's + /// subscription manager, providing accurate state tracking. + pub fn subscriptions(&self) -> Vec { + self.client.subscriptions() + } + + /// Returns the number of active subscriptions. + pub fn subscription_count(&self) -> usize { + self.client.subscription_count() + } + /// Watches a single ticker stream (internal helper) async fn watch_ticker_internal(&self, symbol: &str, channel_name: &str) -> Result { let stream = format!("{}@{}", symbol.to_lowercase(), channel_name); diff --git a/ccxt-exchanges/src/binance/ws_exchange_impl.rs b/ccxt-exchanges/src/binance/ws_exchange_impl.rs index 7162d30..994b387 100644 --- a/ccxt-exchanges/src/binance/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/binance/ws_exchange_impl.rs @@ -50,23 +50,40 @@ impl WsExchange for Binance { // Create a public WebSocket connection let ws = self.create_ws(); ws.connect().await?; + + // Store the connected WebSocket instance for state tracking + let mut conn = self.ws_connection.write().await; + *conn = Some(ws); + Ok(()) } async fn ws_disconnect(&self) -> Result<()> { - // Create a temporary WS to disconnect - let ws = self.create_ws(); - ws.disconnect().await + // Get and clear the stored WebSocket connection + let mut conn = self.ws_connection.write().await; + if let Some(ws) = conn.take() { + ws.disconnect().await?; + } + Ok(()) } fn ws_is_connected(&self) -> bool { - // Without persistent state tracking, return false - // In a full implementation, we'd track the active connection + // Use try_read to avoid blocking + if let Ok(conn) = self.ws_connection.try_read() { + if let Some(ref ws) = *conn { + return ws.is_connected(); + } + } false } fn ws_state(&self) -> WsConnectionState { - // Return disconnected since we don't have persistent state + // Use try_read to avoid blocking + if let Ok(conn) = self.ws_connection.try_read() { + if let Some(ref ws) = *conn { + return ws.state(); + } + } WsConnectionState::Disconnected } @@ -442,8 +459,12 @@ impl WsExchange for Binance { } fn subscriptions(&self) -> Vec { - // Without persistent state, return empty - // In a full implementation, we'd track active subscriptions + // Use try_read to avoid blocking + if let Ok(conn) = self.ws_connection.try_read() { + if let Some(ref ws) = *conn { + return ws.subscriptions(); + } + } Vec::new() } }