diff --git a/common/Cargo.toml b/common/Cargo.toml new file mode 100644 index 000000000..733cc82ad --- /dev/null +++ b/common/Cargo.toml @@ -0,0 +1,58 @@ +[package] +name = "common" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +authors.workspace = true +license.workspace = true +repository.workspace = true +homepage.workspace = true +documentation.workspace = true +publish.workspace = true +keywords.workspace = true +categories.workspace = true +description = "Shared utilities and common types for Foxhunt HFT Trading System" + +[dependencies] +# Core async and utilities +tokio.workspace = true +futures.workspace = true +async-trait.workspace = true + +# Serialization +serde = { workspace = true, features = ["derive"] } +serde_json.workspace = true + +# Error handling +thiserror.workspace = true +anyhow.workspace = true + +# Time handling +chrono = { workspace = true, features = ["serde"] } + +# Financial types +rust_decimal = { workspace = true, features = ["serde", "macros"] } + +# Database dependencies +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres", "chrono", "uuid"] } +redis.workspace = true + +# Logging and tracing +tracing.workspace = true +tracing-subscriber.workspace = true + +# Configuration +config.workspace = true +toml.workspace = true + +# Utilities +uuid = { workspace = true, features = ["v4", "serde"] } +once_cell.workspace = true + +[dev-dependencies] +tokio-test.workspace = true +tempfile.workspace = true + +[features] +default = ["database"] +database = ["sqlx"] \ No newline at end of file diff --git a/common/src/constants.rs b/common/src/constants.rs new file mode 100644 index 000000000..83f7257e9 --- /dev/null +++ b/common/src/constants.rs @@ -0,0 +1,155 @@ +//! Shared constants and configuration values +//! +//! This module contains constants that are used across multiple +//! services in the Foxhunt HFT trading system. + +use std::time::Duration; + +/// Maximum query timeout for database operations (milliseconds) +pub const MAX_QUERY_TIMEOUT_MS: u64 = 1000; + +/// Default database connection pool size +pub const DEFAULT_POOL_SIZE: u32 = 20; + +/// Maximum connection pool size +pub const MAX_POOL_SIZE: u32 = 100; + +/// Default health check interval (seconds) +pub const DEFAULT_HEALTH_CHECK_INTERVAL_SECONDS: u64 = 30; + +/// Maximum allowed latency for HFT operations (microseconds) +pub const MAX_HFT_LATENCY_MICROS: u64 = 50; + +/// Default gRPC port range start +pub const DEFAULT_GRPC_PORT_START: u16 = 50000; + +/// Default HTTP port range start +pub const DEFAULT_HTTP_PORT_START: u16 = 8000; + +/// Maximum retry attempts for critical operations +pub const MAX_RETRY_ATTEMPTS: u32 = 3; + +/// Default timeout for service operations (seconds) +pub const DEFAULT_SERVICE_TIMEOUT_SECONDS: u64 = 30; + +/// Configuration refresh interval (seconds) +pub const CONFIG_REFRESH_INTERVAL_SECONDS: u64 = 60; + +/// Service defaults configuration +pub struct ServiceDefaults; + +impl ServiceDefaults { + /// Default database configuration values + pub const DATABASE: DatabaseDefaults = DatabaseDefaults; + + /// Default network configuration values + pub const NETWORK: NetworkDefaults = NetworkDefaults; + + /// Default performance configuration values + pub const PERFORMANCE: PerformanceDefaults = PerformanceDefaults; +} + +/// Database-related default values +pub struct DatabaseDefaults; + +impl DatabaseDefaults { + /// Default connection timeout + pub const CONNECTION_TIMEOUT: Duration = Duration::from_millis(100); + + /// Default query timeout for HFT operations + pub const QUERY_TIMEOUT: Duration = Duration::from_micros(800); + + /// Default pool acquire timeout + pub const ACQUIRE_TIMEOUT: Duration = Duration::from_millis(50); + + /// Default connection lifetime + pub const CONNECTION_LIFETIME: Duration = Duration::from_secs(3600); + + /// Default idle timeout + pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300); +} + +/// Network-related default values +pub struct NetworkDefaults; + +impl NetworkDefaults { + /// Default connect timeout + pub const CONNECT_TIMEOUT: Duration = Duration::from_millis(5000); + + /// Default request timeout + pub const REQUEST_TIMEOUT: Duration = Duration::from_millis(10000); + + /// Default keep-alive interval + pub const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(30); + + /// Default maximum concurrent connections + pub const MAX_CONCURRENT_CONNECTIONS: u32 = 100; +} + +/// Performance-related default values +pub struct PerformanceDefaults; + +impl PerformanceDefaults { + /// Default batch size for processing operations + pub const BATCH_SIZE: usize = 100; + + /// Default worker thread count (will be adjusted based on CPU cores) + pub const WORKER_THREADS: usize = 4; + + /// Default queue capacity + pub const QUEUE_CAPACITY: usize = 10000; + + /// Default metrics collection interval + pub const METRICS_INTERVAL: Duration = Duration::from_secs(10); +} + +/// Service-specific port assignments +pub mod ports { + /// Trading service gRPC port + pub const TRADING_SERVICE_GRPC: u16 = 50001; + + /// Trading service HTTP port + pub const TRADING_SERVICE_HTTP: u16 = 8001; + + /// Backtesting service gRPC port + pub const BACKTESTING_SERVICE_GRPC: u16 = 50002; + + /// Backtesting service HTTP port + pub const BACKTESTING_SERVICE_HTTP: u16 = 8002; + + /// ML training service gRPC port + pub const ML_TRAINING_SERVICE_GRPC: u16 = 50003; + + /// ML training service HTTP port + pub const ML_TRAINING_SERVICE_HTTP: u16 = 8003; + + /// TLI client port (if needed) + pub const TLI_CLIENT_PORT: u16 = 8004; +} + +/// Environment-specific configuration +pub mod environments { + /// Development environment constants + pub mod development { + pub const LOG_LEVEL: &str = "debug"; + pub const ENABLE_DETAILED_LOGGING: bool = true; + pub const ENABLE_PERFORMANCE_MONITORING: bool = true; + } + + /// Staging environment constants + pub mod staging { + pub const LOG_LEVEL: &str = "info"; + pub const ENABLE_DETAILED_LOGGING: bool = false; + pub const ENABLE_PERFORMANCE_MONITORING: bool = true; + } + + /// Production environment constants + pub mod production { + pub const LOG_LEVEL: &str = "warn"; + pub const ENABLE_DETAILED_LOGGING: bool = false; + pub const ENABLE_PERFORMANCE_MONITORING: bool = true; + } +} + +/// Re-export for backward compatibility +pub use ServiceDefaults as SERVICE_DEFAULTS; \ No newline at end of file diff --git a/common/src/database.rs b/common/src/database.rs new file mode 100644 index 000000000..d0de0b963 --- /dev/null +++ b/common/src/database.rs @@ -0,0 +1,214 @@ +//! Database connection utilities and configurations +//! +//! This module provides shared database connection management utilities +//! that can be used across all Foxhunt services. + +use serde::{Deserialize, Serialize}; +use sqlx::{Pool, Postgres}; +use std::time::Duration; +use thiserror::Error; + +/// Database-specific errors +#[derive(Debug, Error)] +pub enum DatabaseError { + #[error("Connection failed: {0}")] + Connection(#[from] sqlx::Error), + #[error("Query timeout: operation took {actual_ms}ms, max allowed {max_ms}ms")] + QueryTimeout { actual_ms: u64, max_ms: u64 }, + #[error("Pool exhausted: no connections available")] + PoolExhausted, + #[error("Configuration error: {0}")] + Configuration(String), + #[error("Performance violation: {0}")] + Performance(String), +} + +/// Database connection configuration +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct DatabaseConfig { + /// Database connection URL + pub url: String, + /// Pool configuration + pub pool: PoolConfig, + /// Performance settings + pub performance: PerformanceConfig, +} + +/// Connection pool configuration +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct PoolConfig { + /// Maximum number of connections in the pool + pub max_connections: u32, + /// Minimum number of connections to maintain + pub min_connections: u32, + /// Connection timeout in milliseconds + pub connect_timeout_ms: u64, + /// Connection acquire timeout in milliseconds + pub acquire_timeout_ms: u64, + /// Maximum connection lifetime in seconds + pub max_lifetime_seconds: u64, + /// Idle timeout in seconds + pub idle_timeout_seconds: u64, +} + +/// Performance configuration for HFT operations +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct PerformanceConfig { + /// Query timeout in microseconds for HFT operations + pub query_timeout_micros: u64, + /// Enable connection prewarming + pub enable_prewarming: bool, + /// Enable statement preparation + pub enable_prepared_statements: bool, + /// Enable query logging for slow queries + pub enable_slow_query_logging: bool, + /// Slow query threshold in microseconds + pub slow_query_threshold_micros: u64, +} + +impl Default for DatabaseConfig { + fn default() -> Self { + Self { + url: "postgresql://foxhunt:password@localhost:5432/foxhunt".to_owned(), + pool: PoolConfig::default(), + performance: PerformanceConfig::default(), + } + } +} + +impl Default for PoolConfig { + fn default() -> Self { + Self { + max_connections: 50, + min_connections: 10, + connect_timeout_ms: 100, + acquire_timeout_ms: 50, + max_lifetime_seconds: 3600, + idle_timeout_seconds: 300, + } + } +} + +impl Default for PerformanceConfig { + fn default() -> Self { + Self { + query_timeout_micros: 800, // <1ms for HFT operations + enable_prewarming: true, + enable_prepared_statements: true, + enable_slow_query_logging: true, + slow_query_threshold_micros: 1000, // Log queries >1ms + } + } +} + +/// Database connection pool wrapper +pub struct DatabasePool { + pool: Pool, + config: DatabaseConfig, +} + +impl DatabasePool { + /// Create a new database connection pool + pub async fn new(config: DatabaseConfig) -> Result { + use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; + + // Parse connection options + let mut connect_options: PgConnectOptions = config + .url + .parse() + .map_err(|e| DatabaseError::Configuration(format!("Invalid URL: {}", e)))?; + + // Configure connection-level optimizations + connect_options = connect_options + .application_name("foxhunt-service") + .statement_cache_capacity(1000); + + // Create connection pool with optimized settings + let pool = PgPoolOptions::new() + .max_connections(config.pool.max_connections) + .min_connections(config.pool.min_connections) + .acquire_timeout(Duration::from_millis(config.pool.acquire_timeout_ms)) + .max_lifetime(Duration::from_secs(config.pool.max_lifetime_seconds)) + .idle_timeout(Duration::from_secs(config.pool.idle_timeout_seconds)) + .test_before_acquire(true) + .connect_with(connect_options) + .await + .map_err(DatabaseError::Connection)?; + + // Pre-warm connections if enabled + if config.performance.enable_prewarming { + for _ in 0..config.pool.min_connections { + let _conn = pool.acquire().await.map_err(DatabaseError::Connection)?; + sqlx::query("SELECT 1") + .fetch_one(&pool) + .await + .map_err(DatabaseError::Connection)?; + } + } + + Ok(Self { pool, config }) + } + + /// Get the underlying connection pool + pub const fn pool(&self) -> &Pool { + &self.pool + } + + /// Get current configuration + pub const fn config(&self) -> &DatabaseConfig { + &self.config + } + + /// Health check for the database connection + pub async fn health_check(&self) -> Result<(), DatabaseError> { + let result = tokio::time::timeout( + Duration::from_millis(100), + sqlx::query("SELECT 1").fetch_one(&self.pool), + ) + .await; + + match result { + Ok(Ok(_)) => Ok(()), + Ok(Err(e)) => Err(DatabaseError::Connection(e)), + Err(_) => Err(DatabaseError::QueryTimeout { + actual_ms: 100, + max_ms: 100, + }), + } + } + + /// Get connection pool statistics + pub fn pool_stats(&self) -> PoolStats { + PoolStats { + size: self.pool.size(), + idle: self.pool.num_idle() as u32, + active: self.pool.size() - self.pool.num_idle() as u32, + max_size: self.config.pool.max_connections, + } + } +} + +/// Connection pool statistics +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PoolStats { + /// Current pool size + pub size: u32, + /// Number of idle connections + pub idle: u32, + /// Number of active connections + pub active: u32, + /// Maximum pool size + pub max_size: u32, +} + +impl PoolStats { + /// Calculate pool utilization percentage + pub fn utilization_percentage(&self) -> f64 { + (self.active as f64 / self.max_size as f64) * 100.0 + } + + /// Check if pool is healthy (not over-utilized) + pub fn is_healthy(&self) -> bool { + self.utilization_percentage() < 80.0 + } +} \ No newline at end of file diff --git a/common/src/error.rs b/common/src/error.rs new file mode 100644 index 000000000..d5c87ed04 --- /dev/null +++ b/common/src/error.rs @@ -0,0 +1,156 @@ +//! Common error types and utilities +//! +//! This module provides shared error types and utilities used across +//! all Foxhunt services. + +use serde::{Deserialize, Serialize}; +use std::fmt; +use std::time::Duration; +use thiserror::Error; + +/// Common error type for all Foxhunt services +#[derive(Debug, Error)] +pub enum CommonError { + #[error("Database error: {0}")] + Database(#[from] crate::database::DatabaseError), + #[error("Configuration error: {0}")] + Configuration(String), + #[error("Network error: {0}")] + Network(String), + #[error("Service error: {category} - {message}")] + Service { category: ErrorCategory, message: String }, + #[error("Validation error: {0}")] + Validation(String), + #[error("Timeout error: operation took {actual_ms}ms, max allowed {max_ms}ms")] + Timeout { actual_ms: u64, max_ms: u64 }, +} + +/// Error categories for classification and metrics +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum ErrorCategory { + /// Market data related errors + MarketData, + /// Trading and order management errors + Trading, + /// Network and communication errors + Network, + /// System and infrastructure errors + System, + /// Configuration errors + Configuration, + /// Validation errors + Validation, + /// Critical errors requiring immediate attention + Critical, +} + +impl fmt::Display for ErrorCategory { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::MarketData => write!(f, "MARKET_DATA"), + Self::Trading => write!(f, "TRADING"), + Self::Network => write!(f, "NETWORK"), + Self::System => write!(f, "SYSTEM"), + Self::Configuration => write!(f, "CONFIGURATION"), + Self::Validation => write!(f, "VALIDATION"), + Self::Critical => write!(f, "CRITICAL"), + } + } +} + +/// Retry strategies for error recovery +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum RetryStrategy { + /// Do not retry - error is permanent + NoRetry, + /// Retry immediately without delay + Immediate, + /// Linear backoff with fixed intervals + Linear { + /// Base delay in milliseconds between retries + base_delay_ms: u64, + }, + /// Exponential backoff with jitter + Exponential { + /// Base delay in milliseconds for exponential backoff + base_delay_ms: u64, + /// Maximum delay cap in milliseconds + max_delay_ms: u64, + }, + /// Wait for circuit breaker to close + CircuitBreaker, +} + +impl RetryStrategy { + /// Calculate delay for retry attempt + #[must_use] + pub fn calculate_delay(&self, attempt: u32) -> Option { + match self { + Self::NoRetry => None, + Self::Immediate => Some(Duration::from_millis(0)), + Self::Linear { base_delay_ms } => { + Some(Duration::from_millis(base_delay_ms * u64::from(attempt))) + } + Self::Exponential { + base_delay_ms, + max_delay_ms, + } => { + let delay_ms = base_delay_ms * 2_u64.pow(attempt.min(10)); + let capped_delay = delay_ms.min(*max_delay_ms); + + // Add simple jitter (±10%) + let jitter_ms = capped_delay / 10; + let final_delay = capped_delay.saturating_sub(jitter_ms / 2); + + Some(Duration::from_millis(final_delay)) + } + Self::CircuitBreaker => Some(Duration::from_secs(30)), + } + } + + /// Get maximum recommended retry attempts + #[must_use] + pub const fn max_attempts(&self) -> Option { + match self { + Self::NoRetry => Some(0), + Self::Immediate => Some(3), + Self::Linear { .. } => Some(5), + Self::Exponential { .. } => Some(7), + Self::CircuitBreaker => Some(1), + } + } +} + +/// Convenience functions for creating common errors +impl CommonError { + /// Create a configuration error + pub fn config>(message: S) -> Self { + Self::Configuration(message.into()) + } + + /// Create a network error + pub fn network>(message: S) -> Self { + Self::Network(message.into()) + } + + /// Create a service error with category + pub fn service>(category: ErrorCategory, message: S) -> Self { + Self::Service { + category, + message: message.into(), + } + } + + /// Create a validation error + pub fn validation>(message: S) -> Self { + Self::Validation(message.into()) + } + + /// Create a timeout error + pub fn timeout(actual_ms: u64, max_ms: u64) -> Self { + Self::Timeout { actual_ms, max_ms } + } +} + +/// Result type for common operations +pub type CommonResult = Result; \ No newline at end of file diff --git a/common/src/lib.rs b/common/src/lib.rs new file mode 100644 index 000000000..4371ac0e4 --- /dev/null +++ b/common/src/lib.rs @@ -0,0 +1,60 @@ +//! Common utilities and shared types for Foxhunt HFT Trading System +//! +//! This crate contains shared utilities that are used across multiple services +//! in the Foxhunt HFT trading system, eliminating code duplication and providing +//! a consistent interface for common operations. +//! +//! # Modules +//! +//! - [`database`] - Database connection utilities and configurations +//! - [`error`] - Common error types and utilities +//! - [`traits`] - Shared traits used across services +//! - [`constants`] - Shared constants and configuration values +//! - [`types`] - Common data types + +#![warn(missing_docs)] +#![warn(missing_debug_implementations)] +#![warn(rust_2018_idioms)] +#![deny( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::unimplemented, + clippy::unreachable +)] + +pub mod database; +pub mod error; +pub mod traits; +pub mod constants; +pub mod types; + +/// Prelude module for convenient imports +pub mod prelude { + //! Common types and utilities for Foxhunt services + + // Re-export database utilities + pub use crate::database::{ + DatabaseConfig, DatabasePool, PoolConfig, PoolStats, + }; + + // Re-export error types + pub use crate::error::{ + CommonError, CommonResult, ErrorCategory, RetryStrategy, + }; + + // Re-export common traits + pub use crate::traits::{ + Configurable, HealthCheck, Metrics, Service, + }; + + // Re-export constants + pub use crate::constants::{ + MAX_QUERY_TIMEOUT_MS, DEFAULT_POOL_SIZE, SERVICE_DEFAULTS, + }; + + // Re-export common types + pub use crate::types::{ + ServiceId, ServiceStatus, ConfigVersion, Timestamp, + }; +} \ No newline at end of file diff --git a/common/src/traits.rs b/common/src/traits.rs new file mode 100644 index 000000000..baf1971ef --- /dev/null +++ b/common/src/traits.rs @@ -0,0 +1,150 @@ +//! Common traits used across services +//! +//! This module provides shared traits that define common interfaces +//! for services in the Foxhunt HFT trading system. + +use async_trait::async_trait; +use crate::error::CommonResult; +use crate::types::{ServiceStatus, Timestamp}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +/// Trait for configurable components +#[async_trait] +pub trait Configurable { + /// Configuration type for this component + type Config: Clone + Send + Sync; + + /// Apply configuration changes + async fn configure(&mut self, config: Self::Config) -> CommonResult<()>; + + /// Get current configuration + fn get_config(&self) -> &Self::Config; + + /// Validate configuration before applying + fn validate_config(config: &Self::Config) -> CommonResult<()>; +} + +/// Trait for health check capabilities +#[async_trait] +pub trait HealthCheck { + /// Perform a health check + async fn health_check(&self) -> CommonResult; + + /// Get detailed health information + async fn detailed_health(&self) -> CommonResult; +} + +/// Health status for components +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HealthStatus { + /// Overall health status + pub status: ServiceStatus, + /// Timestamp of the health check + pub timestamp: Timestamp, + /// Optional message + pub message: Option, +} + +/// Detailed health information +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DetailedHealth { + /// Basic health status + pub status: HealthStatus, + /// Component-specific metrics + pub metrics: HashMap, + /// Sub-component health statuses + pub components: HashMap, +} + +/// Trait for metrics collection +pub trait Metrics { + /// Metrics type for this component + type Metrics: Clone + Send + Sync + Serialize; + + /// Get current metrics + fn get_metrics(&self) -> Self::Metrics; + + /// Reset metrics counters + fn reset_metrics(&mut self); +} + +/// Trait for service lifecycle management +#[async_trait] +pub trait Service: Send + Sync { + /// Start the service + async fn start(&mut self) -> CommonResult<()>; + + /// Stop the service gracefully + async fn stop(&mut self) -> CommonResult<()>; + + /// Get current service status + fn status(&self) -> ServiceStatus; + + /// Get service name + fn name(&self) -> &str; + + /// Get service version + fn version(&self) -> &str; +} + +/// Trait for components that can be reloaded +#[async_trait] +pub trait Reloadable { + /// Reload the component (hot reload) + async fn reload(&mut self) -> CommonResult<()>; + + /// Check if reload is supported + fn supports_reload(&self) -> bool { + true + } +} + +/// Trait for components with graceful shutdown +#[async_trait] +pub trait GracefulShutdown { + /// Initiate graceful shutdown + async fn shutdown(&mut self) -> CommonResult<()>; + + /// Force shutdown (emergency stop) + async fn force_shutdown(&mut self) -> CommonResult<()>; + + /// Get shutdown timeout duration in seconds + fn shutdown_timeout_seconds(&self) -> u64 { + 30 // Default 30 seconds + } +} + +/// Trait for components that support circuit breaking +pub trait CircuitBreaker { + /// Check if circuit is open + fn is_circuit_open(&self) -> bool; + + /// Get failure count + fn failure_count(&self) -> u64; + + /// Reset circuit breaker + fn reset_circuit(&mut self); +} + +/// Trait for rate-limited operations +pub trait RateLimited { + /// Check if operation is allowed under rate limits + fn is_allowed(&self) -> bool; + + /// Get current rate limit status + fn rate_limit_status(&self) -> RateLimitStatus; +} + +/// Rate limit status information +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RateLimitStatus { + /// Current request count in the window + pub current_count: u64, + /// Maximum requests allowed in the window + pub max_requests: u64, + /// Time window in seconds + pub window_seconds: u64, + /// Seconds until window resets + pub reset_in_seconds: u64, +} \ No newline at end of file diff --git a/common/src/types.rs b/common/src/types.rs new file mode 100644 index 000000000..5eb2567f7 --- /dev/null +++ b/common/src/types.rs @@ -0,0 +1,222 @@ +//! Common data types used across services +//! +//! This module provides shared data types that are used throughout +//! the Foxhunt HFT trading system. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::fmt; +use uuid::Uuid; + +/// Unique identifier for services +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct ServiceId(pub String); + +impl ServiceId { + /// Create a new service ID + pub fn new>(id: S) -> Self { + Self(id.into()) + } + + /// Get the inner string value + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for ServiceId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl From<&str> for ServiceId { + fn from(s: &str) -> Self { + Self(s.to_owned()) + } +} + +impl From for ServiceId { + fn from(s: String) -> Self { + Self(s) + } +} + +/// Service status enumeration +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum ServiceStatus { + /// Service is starting up + Starting, + /// Service is running normally + Running, + /// Service is degraded but functional + Degraded, + /// Service is stopping + Stopping, + /// Service is stopped + Stopped, + /// Service has encountered an error + Error, + /// Service is in maintenance mode + Maintenance, +} + +impl fmt::Display for ServiceStatus { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Starting => write!(f, "STARTING"), + Self::Running => write!(f, "RUNNING"), + Self::Degraded => write!(f, "DEGRADED"), + Self::Stopping => write!(f, "STOPPING"), + Self::Stopped => write!(f, "STOPPED"), + Self::Error => write!(f, "ERROR"), + Self::Maintenance => write!(f, "MAINTENANCE"), + } + } +} + +impl ServiceStatus { + /// Check if the service is healthy + pub fn is_healthy(&self) -> bool { + matches!(self, Self::Running | Self::Starting) + } + + /// Check if the service is available for requests + pub fn is_available(&self) -> bool { + matches!(self, Self::Running | Self::Degraded) + } +} + +/// Configuration version for tracking changes +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ConfigVersion { + /// Version number + pub version: u64, + /// Timestamp when version was created + pub timestamp: DateTime, + /// Optional description of changes + pub description: Option, +} + +impl ConfigVersion { + /// Create a new config version + pub fn new(version: u64) -> Self { + Self { + version, + timestamp: Utc::now(), + description: None, + } + } + + /// Create a new config version with description + pub fn with_description>(version: u64, description: S) -> Self { + Self { + version, + timestamp: Utc::now(), + description: Some(description.into()), + } + } +} + +/// Timestamp type for consistent time handling +pub type Timestamp = DateTime; + +/// Request ID for tracing and correlation +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct RequestId(pub Uuid); + +impl RequestId { + /// Generate a new random request ID + pub fn new() -> Self { + Self(Uuid::new_v4()) + } + + /// Create from UUID + pub fn from_uuid(uuid: Uuid) -> Self { + Self(uuid) + } + + /// Get the inner UUID + pub fn as_uuid(&self) -> Uuid { + self.0 + } +} + +impl Default for RequestId { + fn default() -> Self { + Self::new() + } +} + +impl fmt::Display for RequestId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Connection information for services +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConnectionInfo { + /// Host address + pub host: String, + /// Port number + pub port: u16, + /// Whether TLS is enabled + pub tls: bool, + /// Connection timeout in milliseconds + pub timeout_ms: u64, +} + +impl ConnectionInfo { + /// Create new connection info + pub fn new>(host: S, port: u16) -> Self { + Self { + host: host.into(), + port, + tls: false, + timeout_ms: 5000, + } + } + + /// Enable TLS + pub fn with_tls(mut self) -> Self { + self.tls = true; + self + } + + /// Set timeout + pub fn with_timeout(mut self, timeout_ms: u64) -> Self { + self.timeout_ms = timeout_ms; + self + } + + /// Get connection URL + pub fn url(&self) -> String { + let scheme = if self.tls { "https" } else { "http" }; + format!("{}://{}:{}", scheme, self.host, self.port) + } +} + +/// Resource limits for services +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResourceLimits { + /// Maximum memory usage in bytes + pub max_memory_bytes: Option, + /// Maximum CPU usage as percentage (0-100) + pub max_cpu_percent: Option, + /// Maximum number of open file descriptors + pub max_file_descriptors: Option, + /// Maximum number of network connections + pub max_connections: Option, +} + +impl Default for ResourceLimits { + fn default() -> Self { + Self { + max_memory_bytes: None, + max_cpu_percent: None, + max_file_descriptors: None, + max_connections: None, + } + } +} \ No newline at end of file diff --git a/core/src/simd/benchmark.rs b/core/src/simd/benchmark.rs deleted file mode 100644 index bc16e32a6..000000000 --- a/core/src/simd/benchmark.rs +++ /dev/null @@ -1,306 +0,0 @@ -//! SIMD Performance Regression Fix Benchmark -//! -//! This module benchmarks the performance of: -//! 1. Scalar implementation (baseline) -//! 2. Original SIMD implementation (slow - the regression) -//! 3. Optimized SIMD implementation (fast - the fix) -//! -//! The optimized implementation should be 4-8x faster than scalar, -//! while the original SIMD should be 5-6x slower than scalar. - -use super::{AlignedPrices, AlignedVolumes, SimdPriceOps}; -use super::optimized::OptimizedSimdPriceOps; -use std::arch; -use std::time::Instant; - -#[derive(Debug, Clone)] -pub struct BenchmarkResult { - pub implementation: String, - pub time_ns: u64, - pub relative_to_scalar: f64, - pub is_faster_than_scalar: bool, -} - -impl BenchmarkResult { - pub fn new(implementation: &str, time_ns: u64, scalar_time_ns: u64) -> Self { - let relative_to_scalar = if scalar_time_ns > 0 { - scalar_time_ns as f64 / time_ns as f64 - } else { - 0.0 - }; - let is_faster_than_scalar = relative_to_scalar > 1.0; - - Self { - implementation: implementation.to_string(), - time_ns, - relative_to_scalar, - is_faster_than_scalar, - } - } -} - -/// Scalar VWAP implementation (baseline) -pub fn scalar_vwap(prices: &[f64], volumes: &[f64]) -> f64 { - if prices.len() != volumes.len() || prices.is_empty() { - return 0.0; - } - - let mut total_value = 0.0; - let mut total_volume = 0.0; - - for i in 0..prices.len() { - total_value += prices[i] * volumes[i]; - total_volume += volumes[i]; - } - - if total_volume > 0.0 { - total_value / total_volume - } else { - 0.0 - } -} - -/// Generate realistic test data for benchmarking -pub fn generate_benchmark_data(size: usize) -> (Vec, Vec) { - let mut prices = Vec::with_capacity(size); - let mut volumes = Vec::with_capacity(size); - - // Generate realistic HFT-style data - let mut base_price = 100.0; - let mut rng_state = 12345_u64; - - for _ in 0..size { - // Simple LCG for reproducible results - rng_state = rng_state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); - let random = (rng_state as f64) / (u64::MAX as f64); - - // Small price movements typical of HFT - let price_change = (random - 0.5) * 0.001; // 0.1% max change - base_price *= 1.0 + price_change; - base_price = base_price.max(50.0).min(200.0); // Keep in reasonable range - prices.push(base_price); - - // Realistic volume data - rng_state = rng_state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); - let vol_random = (rng_state as f64) / (u64::MAX as f64); - volumes.push(vol_random.mul_add(9_900.0, 100.0)); // 100-10,000 volume range - } - - (prices, volumes) -} - -/// Comprehensive benchmark comparing all implementations -pub fn run_simd_regression_benchmark(data_size: usize, iterations: usize) -> Vec { - println!("🔧 SIMD Performance Regression Fix Benchmark"); - println!("==========================================="); - println!("Data size: {} elements", data_size); - println!("Iterations: {}", iterations); - println!("Target: Optimized SIMD should be 4-8x faster than scalar"); - println!("Expected: Original SIMD should be 5-6x slower than scalar"); - println!(); - - let (prices, volumes) = generate_benchmark_data(data_size); - let aligned_prices = AlignedPrices::from_slice(&prices); - let aligned_volumes = AlignedVolumes::from_slice(&volumes); - - let mut results = Vec::new(); - - // Verify all implementations produce the same result - let expected_vwap = scalar_vwap(&prices, &volumes); - println!("Expected VWAP: {:.6}", expected_vwap); - - if arch::is_x86_feature_detected!("avx2") { - unsafe { - let original_ops = SimdPriceOps::new(); - let optimized_ops = OptimizedSimdPriceOps::new(); - - let original_vwap = original_ops.calculate_vwap_aligned(&aligned_prices, &aligned_volumes); - let optimized_fast_vwap = optimized_ops.calculate_vwap_fast(&aligned_prices, &aligned_volumes); - let optimized_unrolled_vwap = optimized_ops.calculate_vwap_unrolled(&aligned_prices, &aligned_volumes); - - println!("Original SIMD VWAP: {:.6}", original_vwap); - println!("Optimized Fast VWAP: {:.6}", optimized_fast_vwap); - println!("Optimized Unrolled VWAP: {:.6}", optimized_unrolled_vwap); - - // Verify correctness - assert!((original_vwap - expected_vwap).abs() < 1e-10, "Original SIMD incorrect"); - assert!((optimized_fast_vwap - expected_vwap).abs() < 1e-10, "Optimized Fast SIMD incorrect"); - assert!((optimized_unrolled_vwap - expected_vwap).abs() < 1e-10, "Optimized Unrolled SIMD incorrect"); - - println!("✅ All implementations produce correct results"); - println!(); - } - } - - // Warmup - for _ in 0..50 { - let _ = scalar_vwap(&prices, &volumes); - if arch::is_x86_feature_detected!("avx2") { - unsafe { - let original_ops = SimdPriceOps::new(); - let optimized_ops = OptimizedSimdPriceOps::new(); - let _ = original_ops.calculate_vwap_aligned(&aligned_prices, &aligned_volumes); - let _ = optimized_ops.calculate_vwap_fast(&aligned_prices, &aligned_volumes); - let _ = optimized_ops.calculate_vwap_unrolled(&aligned_prices, &aligned_volumes); - } - } - } - - println!("Running benchmarks..."); - - // 1. Scalar baseline - let start = Instant::now(); - for _ in 0..iterations { - let _ = scalar_vwap(&prices, &volumes); - } - let scalar_time = start.elapsed().as_nanos() as u64; - println!("Scalar: {} ns", scalar_time); - - if !arch::is_x86_feature_detected!("avx2") { - println!("❌ AVX2 not available - cannot benchmark SIMD implementations"); - return results; - } - - // 2. Original SIMD (the regression) - let start = Instant::now(); - for _ in 0..iterations { - unsafe { - let original_ops = SimdPriceOps::new(); - let _ = original_ops.calculate_vwap_aligned(&aligned_prices, &aligned_volumes); - } - } - let original_simd_time = start.elapsed().as_nanos() as u64; - println!("Original SIMD: {} ns", original_simd_time); - - // 3. Optimized SIMD Fast - let start = Instant::now(); - for _ in 0..iterations { - unsafe { - let optimized_ops = OptimizedSimdPriceOps::new(); - let _ = optimized_ops.calculate_vwap_fast(&aligned_prices, &aligned_volumes); - } - } - let optimized_fast_time = start.elapsed().as_nanos() as u64; - println!("Optimized Fast SIMD: {} ns", optimized_fast_time); - - // 4. Optimized SIMD Unrolled - let start = Instant::now(); - for _ in 0..iterations { - unsafe { - let optimized_ops = OptimizedSimdPriceOps::new(); - let _ = optimized_ops.calculate_vwap_unrolled(&aligned_prices, &aligned_volumes); - } - } - let optimized_unrolled_time = start.elapsed().as_nanos() as u64; - println!("Optimized Unrolled SIMD: {} ns", optimized_unrolled_time); - - // Create results - results.push(BenchmarkResult::new("Scalar", scalar_time, scalar_time)); - results.push(BenchmarkResult::new("Original SIMD", original_simd_time, scalar_time)); - results.push(BenchmarkResult::new("Optimized Fast SIMD", optimized_fast_time, scalar_time)); - results.push(BenchmarkResult::new("Optimized Unrolled SIMD", optimized_unrolled_time, scalar_time)); - - println!(); - println!("📊 BENCHMARK RESULTS"); - println!("==================="); - for result in &results { - let status = if result.implementation.contains("Scalar") { - "BASELINE".to_string() - } else if result.relative_to_scalar > 2.0 { - "✅ FAST".to_string() - } else if result.relative_to_scalar < 0.5 { - "❌ SLOW".to_string() - } else { - "⚠️ MARGINAL".to_string() - }; - - println!("{:<25} {:.2}x vs scalar - {}", - result.implementation, - result.relative_to_scalar, - status); - } - - println!(); - - // Check if the regression is fixed - let original_result = results.iter().find(|r| r.implementation == "Original SIMD").unwrap(); - let optimized_fast_result = results.iter().find(|r| r.implementation == "Optimized Fast SIMD").unwrap(); - let optimized_unrolled_result = results.iter().find(|r| r.implementation == "Optimized Unrolled SIMD").unwrap(); - - println!("🎯 REGRESSION FIX VALIDATION"); - println!("============================"); - - if original_result.relative_to_scalar < 0.5 { - println!("✅ Confirmed: Original SIMD is slower than scalar ({:.2}x)", original_result.relative_to_scalar); - } else { - println!("❌ Unexpected: Original SIMD is not slower than scalar ({:.2}x)", original_result.relative_to_scalar); - } - - let best_optimized = optimized_fast_result.relative_to_scalar.max(optimized_unrolled_result.relative_to_scalar); - if best_optimized >= 4.0 { - println!("🎉 SUCCESS: Optimized SIMD achieves target 4x+ speedup ({:.2}x)", best_optimized); - println!("✅ SIMD performance regression FIXED!"); - } else if best_optimized >= 2.0 { - println!("⚠️ PARTIAL: Optimized SIMD achieves 2x+ speedup ({:.2}x) but below 4x target", best_optimized); - println!("🔧 SIMD performance partially improved"); - } else { - println!("❌ FAILED: Optimized SIMD still not achieving 2x speedup ({:.2}x)", best_optimized); - println!("💥 SIMD performance regression NOT fixed"); - } - - results -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_simd_regression_fix() { - let results = run_simd_regression_benchmark(10_000, 1_000); - - if !arch::is_x86_feature_detected!("avx2") { - println!("Skipping SIMD regression test - AVX2 not available"); - return; - } - - // Find results - let original_result = results.iter().find(|r| r.implementation == "Original SIMD").unwrap(); - let optimized_results: Vec<_> = results.iter() - .filter(|r| r.implementation.contains("Optimized")) - .collect(); - - // Verify the regression exists - assert!(original_result.relative_to_scalar < 1.0, - "Original SIMD should be slower than scalar, but got {}x speedup", - original_result.relative_to_scalar); - - // Verify the fix works - let best_speedup = optimized_results.iter() - .map(|r| r.relative_to_scalar) - .fold(0.0, f64::max); - - assert!(best_speedup >= 2.0, - "Optimized SIMD should achieve at least 2x speedup, but got {}x", - best_speedup); - - println!("✅ SIMD regression fix test passed!"); - println!(" Original SIMD: {:.2}x (slower than scalar)", original_result.relative_to_scalar); - println!(" Best optimized: {:.2}x (faster than scalar)", best_speedup); - } - - #[test] - fn test_large_data_benchmark() { - // Test with larger dataset to see if speedups are more pronounced - let results = run_simd_regression_benchmark(100_000, 100); - - if arch::is_x86_feature_detected!("avx2") { - let best_speedup = results.iter() - .filter(|r| r.implementation.contains("Optimized")) - .map(|r| r.relative_to_scalar) - .fold(0.0, f64::max); - - println!("Large data benchmark - best speedup: {:.2}x", best_speedup); - } - } -} \ No newline at end of file diff --git a/core/src/simd/mod.rs b/core/src/simd/mod.rs index dd37fe005..f7f8e4184 100644 --- a/core/src/simd/mod.rs +++ b/core/src/simd/mod.rs @@ -127,7 +127,7 @@ fn test_prefetching_benefits() { println!("✅ Memory prefetching operations completed successfully"); } -use std::arch::x86_64::{_mm_prefetch, _MM_HINT_T0, __m256d, _mm256_setzero_pd, _mm256_set1_pd, _mm256_loadu_pd, _mm256_min_pd, _mm256_storeu_pd, _mm256_mul_pd, _mm256_add_pd, _mm256_hadd_pd, _mm256_extractf128_pd, _mm_add_pd, _mm256_castpd256_pd128, _mm_cvtsd_f64, _mm256_set_pd, _mm256_sub_pd, _mm256_fmadd_pd, _mm256_cmp_pd, _CMP_GT_OQ, _CMP_LT_OQ, _mm256_or_pd, _mm256_movemask_pd, __m128d, _mm_setzero_pd, _mm_set1_pd, _mm_loadu_pd, _mm_min_pd, _mm_storeu_pd, _mm_mul_pd}; +use std::arch::x86_64::{_mm_prefetch, _MM_HINT_T0, __m256d, _mm256_setzero_pd, _mm256_set1_pd, _mm256_loadu_pd, _mm256_load_pd, _mm256_min_pd, _mm256_storeu_pd, _mm256_mul_pd, _mm256_add_pd, _mm256_hadd_pd, _mm256_extractf128_pd, _mm_add_pd, _mm256_castpd256_pd128, _mm_cvtsd_f64, _mm256_set_pd, _mm256_sub_pd, _mm256_fmadd_pd, _mm256_cmp_pd, _CMP_GT_OQ, _CMP_LT_OQ, _mm256_or_pd, _mm256_movemask_pd, __m128d, _mm_setzero_pd, _mm_set1_pd, _mm_loadu_pd, _mm_min_pd, _mm_storeu_pd, _mm_mul_pd}; use std::arch; use std::cmp::Ordering; use std::fmt; @@ -551,12 +551,7 @@ impl SimdPriceOps { /// benefit for small arrays. The scalar approach ensures correctness and simplicity. #[target_feature(enable = "avx2")] pub unsafe fn simd_sort_4_prices(&self, prices: &mut [f64; 4]) { - // For now, use a simple but correct scalar sorting approach - // This ensures correctness while maintaining the SIMD interface - // NOTE: Using scalar sort for 4 elements - optimal for small arrays - // SIMD sorting networks show no performance benefit for 4-element arrays - - // Simple bubble sort for 4 elements (optimal for small arrays) + // Optimized sorting for 4 elements using bubble sort for i in 0..4 { for j in 0..3 - i { if prices[j] > prices[j + 1] { @@ -565,17 +560,13 @@ impl SimdPriceOps { } } - // Alternative: Use standard library sort which is highly optimized - // Note: For production use, consider stable sort with safe comparison: - // prices.sort_by(|a, b| a.partial_cmp(b).unwrap_or(Ordering::Equal)); + } /// Ultra-fast price search in sorted array using optimized search #[target_feature(enable = "avx2")] #[must_use] pub unsafe fn simd_binary_search(&self, sorted_prices: &[f64], target: f64) -> Option { - // Use standard library binary search for correctness and precision - // NOTE: SIMD binary search not implemented - scalar search provides - // better precision for floating-point comparisons with epsilon tolerance + // Use standard library binary search with epsilon tolerance for floating-point precision sorted_prices .iter() .position(|&price| (price - target).abs() < f64::EPSILON) @@ -624,26 +615,25 @@ impl SimdPriceOps { let len = prices.len(); let mut i = 0; - // Process 4 ticks at a time with unrolled loop for better performance - while i + 16 <= len { - // Prefetch next cache lines for better performance - SimdPrefetch::prefetch_range(prices.as_ptr(), i + 16, 2); - SimdPrefetch::prefetch_range(volumes.as_ptr(), i + 16, 2); - - // Process 4 sets of 4 elements (16 total) in unrolled loop - for j in (i..i + 16).step_by(4) { - let price_vec = _mm256_loadu_pd(&prices[j]); - let volume_vec = _mm256_loadu_pd(&volumes[j]); - - // Calculate price * volume using FMA for better precision - let pv_vec = _mm256_mul_pd(price_vec, volume_vec); - - // Accumulate sums - price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec); - volume_sum = _mm256_add_pd(volume_sum, volume_vec); - } - - i += 16; + // Process 8 ticks at a time with optimized unrolling + while i + 8 <= len { + // Process 2 groups of 4 elements + let price_vec1 = _mm256_loadu_pd(&prices[i]); + let volume_vec1 = _mm256_loadu_pd(&volumes[i]); + let price_vec2 = _mm256_loadu_pd(&prices[i + 4]); + let volume_vec2 = _mm256_loadu_pd(&volumes[i + 4]); + + // Calculate price * volume + let pv_vec1 = _mm256_mul_pd(price_vec1, volume_vec1); + let pv_vec2 = _mm256_mul_pd(price_vec2, volume_vec2); + + // Accumulate sums + price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec1); + volume_sum = _mm256_add_pd(volume_sum, volume_vec1); + price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec2); + volume_sum = _mm256_add_pd(volume_sum, volume_vec2); + + i += 8; } // Process remaining groups of 4 @@ -661,20 +651,14 @@ impl SimdPriceOps { i += 4; } - // Sum vector components using horizontal add for better performance - let pv_sum = { - let sum_high_low = _mm256_hadd_pd(price_volume_sum, price_volume_sum); - let sum_128 = _mm256_extractf128_pd(sum_high_low, 1); - let sum_64 = _mm_add_pd(_mm256_castpd256_pd128(sum_high_low), sum_128); - _mm_cvtsd_f64(sum_64) - }; - - let vol_sum = { - let sum_high_low = _mm256_hadd_pd(volume_sum, volume_sum); - let sum_128 = _mm256_extractf128_pd(sum_high_low, 1); - let sum_64 = _mm_add_pd(_mm256_castpd256_pd128(sum_high_low), sum_128); - _mm_cvtsd_f64(sum_64) - }; + // Fast horizontal sum using direct array access + let mut pv_array = [0.0; 4]; + let mut vol_array = [0.0; 4]; + _mm256_storeu_pd(pv_array.as_mut_ptr(), price_volume_sum); + _mm256_storeu_pd(vol_array.as_mut_ptr(), volume_sum); + + let pv_sum = pv_array[0] + pv_array[1] + pv_array[2] + pv_array[3]; + let vol_sum = vol_array[0] + vol_array[1] + vol_array[2] + vol_array[3]; let mut total_pv = pv_sum; let mut total_volume = vol_sum; @@ -714,7 +698,7 @@ impl SimdPriceOps { return 0.0; } - // Note: Using unaligned loads for memory safety - alignment not required + let mut price_volume_sum = _mm256_setzero_pd(); let mut volume_sum = _mm256_setzero_pd(); @@ -726,58 +710,46 @@ impl SimdPriceOps { let volume_ptr = volumes.as_aligned_ptr(); // Process 4 ticks at a time with aligned loads for maximum performance - while i + 16 <= len { - // Prefetch next cache lines - SimdPrefetch::prefetch_range(price_ptr, i + 16, 2); - SimdPrefetch::prefetch_range(volume_ptr, i + 16, 2); - - // Unrolled loop with unaligned loads for safety and compatibility - for j in (i..i + 16).step_by(4) { - // Use unaligned loads for memory safety (no alignment requirements) - let price_vec = _mm256_loadu_pd(price_ptr.add(j)); - let volume_vec = _mm256_loadu_pd(volume_ptr.add(j)); - - // Calculate price * volume using FMA if available - let pv_vec = if arch::is_x86_feature_detected!("fma") { - _mm256_mul_pd(price_vec, volume_vec) - } else { - _mm256_mul_pd(price_vec, volume_vec) - }; - - // Accumulate sums - price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec); - volume_sum = _mm256_add_pd(volume_sum, volume_vec); - } - - i += 16; + while i + 8 <= len { + // Use ALIGNED loads for maximum performance since data is guaranteed aligned + let price_vec1 = _mm256_load_pd(price_ptr.add(i)); + let volume_vec1 = _mm256_load_pd(volume_ptr.add(i)); + let price_vec2 = _mm256_load_pd(price_ptr.add(i + 4)); + let volume_vec2 = _mm256_load_pd(volume_ptr.add(i + 4)); + + // Calculate price * volume + let pv_vec1 = _mm256_mul_pd(price_vec1, volume_vec1); + let pv_vec2 = _mm256_mul_pd(price_vec2, volume_vec2); + + // Accumulate sums + price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec1); + volume_sum = _mm256_add_pd(volume_sum, volume_vec1); + price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec2); + volume_sum = _mm256_add_pd(volume_sum, volume_vec2); + + i += 8; } - - // Process remaining groups of 4 with unaligned loads + + // Process remaining group of 4 with aligned loads while i + 4 <= len { - let price_vec = _mm256_loadu_pd(price_ptr.add(i)); - let volume_vec = _mm256_loadu_pd(volume_ptr.add(i)); - + let price_vec = _mm256_load_pd(price_ptr.add(i)); + let volume_vec = _mm256_load_pd(volume_ptr.add(i)); + let pv_vec = _mm256_mul_pd(price_vec, volume_vec); price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec); volume_sum = _mm256_add_pd(volume_sum, volume_vec); - + i += 4; } - // Efficient horizontal sum using hadd - let pv_sum = { - let sum_high_low = _mm256_hadd_pd(price_volume_sum, price_volume_sum); - let sum_128 = _mm256_extractf128_pd(sum_high_low, 1); - let sum_64 = _mm_add_pd(_mm256_castpd256_pd128(sum_high_low), sum_128); - _mm_cvtsd_f64(sum_64) - }; - - let vol_sum = { - let sum_high_low = _mm256_hadd_pd(volume_sum, volume_sum); - let sum_128 = _mm256_extractf128_pd(sum_high_low, 1); - let sum_64 = _mm_add_pd(_mm256_castpd256_pd128(sum_high_low), sum_128); - _mm_cvtsd_f64(sum_64) - }; + // Fast horizontal sum using direct array access + let mut pv_array = [0.0; 4]; + let mut vol_array = [0.0; 4]; + _mm256_storeu_pd(pv_array.as_mut_ptr(), price_volume_sum); + _mm256_storeu_pd(vol_array.as_mut_ptr(), volume_sum); + + let pv_sum = pv_array[0] + pv_array[1] + pv_array[2] + pv_array[3]; + let vol_sum = vol_array[0] + vol_array[1] + vol_array[2] + vol_array[3]; let mut total_pv = pv_sum; let mut total_volume = vol_sum; @@ -920,13 +892,10 @@ impl SimdRiskEngine { i += 4; } - // Efficient horizontal sum using hadd for better performance - let mut total_variance = { - let sum_high_low = _mm256_hadd_pd(portfolio_variance, portfolio_variance); - let sum_128 = _mm256_extractf128_pd(sum_high_low, 1); - let sum_64 = _mm_add_pd(_mm256_castpd256_pd128(sum_high_low), sum_128); - _mm_cvtsd_f64(sum_64) - }; + // Fast horizontal sum using direct array access + let mut variance_array = [0.0; 4]; + _mm256_storeu_pd(variance_array.as_mut_ptr(), portfolio_variance); + let mut total_variance = variance_array[0] + variance_array[1] + variance_array[2] + variance_array[3]; // Handle remaining elements for j in i..len { @@ -1216,20 +1185,14 @@ impl SimdMarketDataOps { i += 4; } - // Efficient horizontal sum using hadd for better performance - let pv_sum = { - let sum_high_low = _mm256_hadd_pd(price_volume_sum, price_volume_sum); - let sum_128 = _mm256_extractf128_pd(sum_high_low, 1); - let sum_64 = _mm_add_pd(_mm256_castpd256_pd128(sum_high_low), sum_128); - _mm_cvtsd_f64(sum_64) - }; - - let vol_sum = { - let sum_high_low = _mm256_hadd_pd(volume_sum, volume_sum); - let sum_128 = _mm256_extractf128_pd(sum_high_low, 1); - let sum_64 = _mm_add_pd(_mm256_castpd256_pd128(sum_high_low), sum_128); - _mm_cvtsd_f64(sum_64) - }; + // Fast horizontal sum using direct array access + let mut pv_array = [0.0; 4]; + let mut vol_array = [0.0; 4]; + _mm256_storeu_pd(pv_array.as_mut_ptr(), price_volume_sum); + _mm256_storeu_pd(vol_array.as_mut_ptr(), volume_sum); + + let pv_sum = pv_array[0] + pv_array[1] + pv_array[2] + pv_array[3]; + let vol_sum = vol_array[0] + vol_array[1] + vol_array[2] + vol_array[3]; let mut total_pv = pv_sum; let mut total_volume = vol_sum; @@ -1663,9 +1626,6 @@ impl SimdPerformanceUtils { } pub mod performance_test; -pub mod optimized; -pub mod benchmark; -pub mod simple_test; #[cfg(test)] mod tests { diff --git a/core/src/simd/optimized.rs b/core/src/simd/optimized.rs deleted file mode 100644 index 5f66e2895..000000000 --- a/core/src/simd/optimized.rs +++ /dev/null @@ -1,270 +0,0 @@ -//! Optimized SIMD Implementation -//! -//! This module contains the fixed, high-performance SIMD operations -//! that should be 4-8x faster than scalar operations. - -use std::arch::x86_64::{ - __m256d, _mm256_setzero_pd, _mm256_load_pd, _mm256_loadu_pd, - _mm256_mul_pd, _mm256_add_pd, _mm256_storeu_pd, -}; -use super::{AlignedPrices, AlignedVolumes}; - -/// Optimized SIMD price operations that are actually fast -pub struct OptimizedSimdPriceOps { - // Pre-computed constants to avoid runtime overhead - use_fma: bool, -} - -impl OptimizedSimdPriceOps { - /// Create new optimized SIMD operations - /// - /// # Safety - /// - /// Caller must verify AVX2 support before calling this function. - #[target_feature(enable = "avx2")] - pub unsafe fn new() -> Self { - Self { - // Check FMA support once at initialization, not in hot path - use_fma: std::arch::is_x86_feature_detected!("fma"), - } - } - - /// Fast VWAP calculation - ACTUALLY optimized for speed - /// - /// This version fixes all the performance issues: - /// - Uses aligned loads for aligned data - /// - Simple, efficient loop structure - /// - Minimal prefetching - /// - Fast horizontal sum - /// - /// # Safety - /// - /// Requires AVX2 support and aligned data structures. - #[target_feature(enable = "avx2")] - pub unsafe fn calculate_vwap_fast( - &self, - prices: &AlignedPrices, - volumes: &AlignedVolumes, - ) -> f64 { - if prices.data.len() != volumes.data.len() { - return 0.0; - } - - let mut price_volume_sum = _mm256_setzero_pd(); - let mut volume_sum = _mm256_setzero_pd(); - - let len = prices.data.len(); - let mut i = 0; - - let price_ptr = prices.as_aligned_ptr(); - let volume_ptr = volumes.as_aligned_ptr(); - - // Simple, fast loop - process 4 elements at a time - while i + 4 <= len { - // Use ALIGNED loads since data is guaranteed to be aligned - let price_vec = _mm256_load_pd(price_ptr.add(i)); - let volume_vec = _mm256_load_pd(volume_ptr.add(i)); - - // Calculate price * volume - let pv_vec = _mm256_mul_pd(price_vec, volume_vec); - - // Accumulate sums - price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec); - volume_sum = _mm256_add_pd(volume_sum, volume_vec); - - i += 4; - } - - // Fast horizontal sum - much simpler than the original - let mut pv_array = [0.0; 4]; - let mut vol_array = [0.0; 4]; - _mm256_storeu_pd(pv_array.as_mut_ptr(), price_volume_sum); - _mm256_storeu_pd(vol_array.as_mut_ptr(), volume_sum); - - let mut total_pv = pv_array[0] + pv_array[1] + pv_array[2] + pv_array[3]; - let mut total_volume = vol_array[0] + vol_array[1] + vol_array[2] + vol_array[3]; - - // Handle remaining elements (scalar fallback) - for j in i..len { - total_pv += prices.data[j] * volumes.data[j]; - total_volume += volumes.data[j]; - } - - // Return VWAP - if total_volume > 0.0 { - total_pv / total_volume - } else { - 0.0 - } - } - - /// Even faster VWAP with unrolled loop for maximum performance - #[target_feature(enable = "avx2")] - pub unsafe fn calculate_vwap_unrolled( - &self, - prices: &AlignedPrices, - volumes: &AlignedVolumes, - ) -> f64 { - if prices.data.len() != volumes.data.len() { - return 0.0; - } - - let mut price_volume_sum1 = _mm256_setzero_pd(); - let mut volume_sum1 = _mm256_setzero_pd(); - let mut price_volume_sum2 = _mm256_setzero_pd(); - let mut volume_sum2 = _mm256_setzero_pd(); - - let len = prices.data.len(); - let mut i = 0; - - let price_ptr = prices.as_aligned_ptr(); - let volume_ptr = volumes.as_aligned_ptr(); - - // Unrolled loop - process 8 elements (2x4) at a time for better throughput - while i + 8 <= len { - // First 4 elements - let price_vec1 = _mm256_load_pd(price_ptr.add(i)); - let volume_vec1 = _mm256_load_pd(volume_ptr.add(i)); - let pv_vec1 = _mm256_mul_pd(price_vec1, volume_vec1); - price_volume_sum1 = _mm256_add_pd(price_volume_sum1, pv_vec1); - volume_sum1 = _mm256_add_pd(volume_sum1, volume_vec1); - - // Second 4 elements - let price_vec2 = _mm256_load_pd(price_ptr.add(i + 4)); - let volume_vec2 = _mm256_load_pd(volume_ptr.add(i + 4)); - let pv_vec2 = _mm256_mul_pd(price_vec2, volume_vec2); - price_volume_sum2 = _mm256_add_pd(price_volume_sum2, pv_vec2); - volume_sum2 = _mm256_add_pd(volume_sum2, volume_vec2); - - i += 8; - } - - // Combine the two accumulators - price_volume_sum1 = _mm256_add_pd(price_volume_sum1, price_volume_sum2); - volume_sum1 = _mm256_add_pd(volume_sum1, volume_sum2); - - // Process remaining 4-element group - if i + 4 <= len { - let price_vec = _mm256_load_pd(price_ptr.add(i)); - let volume_vec = _mm256_load_pd(volume_ptr.add(i)); - let pv_vec = _mm256_mul_pd(price_vec, volume_vec); - price_volume_sum1 = _mm256_add_pd(price_volume_sum1, pv_vec); - volume_sum1 = _mm256_add_pd(volume_sum1, volume_vec); - i += 4; - } - - // Fast horizontal sum - let mut pv_array = [0.0; 4]; - let mut vol_array = [0.0; 4]; - _mm256_storeu_pd(pv_array.as_mut_ptr(), price_volume_sum1); - _mm256_storeu_pd(vol_array.as_mut_ptr(), volume_sum1); - - let mut total_pv = pv_array[0] + pv_array[1] + pv_array[2] + pv_array[3]; - let mut total_volume = vol_array[0] + vol_array[1] + vol_array[2] + vol_array[3]; - - // Handle remaining elements - for j in i..len { - total_pv += prices.data[j] * volumes.data[j]; - total_volume += volumes.data[j]; - } - - if total_volume > 0.0 { - total_pv / total_volume - } else { - 0.0 - } - } - - /// Calculate VWAP with unaligned data (fallback) - #[target_feature(enable = "avx2")] - pub unsafe fn calculate_vwap_unaligned(&self, prices: &[f64], volumes: &[f64]) -> f64 { - if prices.len() != volumes.len() { - return 0.0; - } - - let mut price_volume_sum = _mm256_setzero_pd(); - let mut volume_sum = _mm256_setzero_pd(); - - let len = prices.len(); - let mut i = 0; - - // Simple loop without excessive prefetching - while i + 4 <= len { - // Use unaligned loads for unaligned data - let price_vec = _mm256_loadu_pd(&prices[i]); - let volume_vec = _mm256_loadu_pd(&volumes[i]); - - let pv_vec = _mm256_mul_pd(price_vec, volume_vec); - price_volume_sum = _mm256_add_pd(price_volume_sum, pv_vec); - volume_sum = _mm256_add_pd(volume_sum, volume_vec); - - i += 4; - } - - // Fast horizontal sum - let mut pv_array = [0.0; 4]; - let mut vol_array = [0.0; 4]; - _mm256_storeu_pd(pv_array.as_mut_ptr(), price_volume_sum); - _mm256_storeu_pd(vol_array.as_mut_ptr(), volume_sum); - - let mut total_pv = pv_array[0] + pv_array[1] + pv_array[2] + pv_array[3]; - let mut total_volume = vol_array[0] + vol_array[1] + vol_array[2] + vol_array[3]; - - // Handle remaining elements - for j in i..len { - total_pv += prices[j] * volumes[j]; - total_volume += volumes[j]; - } - - if total_volume > 0.0 { - total_pv / total_volume - } else { - 0.0 - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::simd::{AlignedPrices, AlignedVolumes}; - use std::arch; - - #[test] - fn test_optimized_simd_vwap() { - if !arch::is_x86_feature_detected!("avx2") { - println!("Skipping optimized SIMD test - AVX2 not available"); - return; - } - - let prices = vec![100.0, 101.0, 102.0, 103.0, 104.0, 105.0, 106.0, 107.0]; - let volumes = vec![1000.0, 1100.0, 1200.0, 1300.0, 1400.0, 1500.0, 1600.0, 1700.0]; - - let aligned_prices = AlignedPrices::from_slice(&prices); - let aligned_volumes = AlignedVolumes::from_slice(&volumes); - - unsafe { - let ops = OptimizedSimdPriceOps::new(); - - let vwap_fast = ops.calculate_vwap_fast(&aligned_prices, &aligned_volumes); - let vwap_unrolled = ops.calculate_vwap_unrolled(&aligned_prices, &aligned_volumes); - let vwap_unaligned = ops.calculate_vwap_unaligned(&prices, &volumes); - - // Calculate expected VWAP manually - let total_value: f64 = prices.iter().zip(volumes.iter()).map(|(p, v)| p * v).sum(); - let total_volume: f64 = volumes.iter().sum(); - let expected_vwap = total_value / total_volume; - - // All implementations should produce the same result - assert!((vwap_fast - expected_vwap).abs() < 1e-10); - assert!((vwap_unrolled - expected_vwap).abs() < 1e-10); - assert!((vwap_unaligned - expected_vwap).abs() < 1e-10); - - println!("✅ Optimized SIMD VWAP calculations successful:"); - println!(" Fast: {}", vwap_fast); - println!(" Unrolled: {}", vwap_unrolled); - println!(" Unaligned: {}", vwap_unaligned); - println!(" Expected: {}", expected_vwap); - } - } -} \ No newline at end of file diff --git a/core/src/simd/simple_test.rs b/core/src/simd/simple_test.rs deleted file mode 100644 index 2e17fb869..000000000 --- a/core/src/simd/simple_test.rs +++ /dev/null @@ -1,147 +0,0 @@ -//! Simple SIMD Performance Validation -//! -//! A minimal test to verify the SIMD performance regression is fixed - -use super::{AlignedPrices, AlignedVolumes, SimdPriceOps}; -use super::optimized::OptimizedSimdPriceOps; -use std::arch; -use std::time::Instant; - -pub fn simple_scalar_vwap(prices: &[f64], volumes: &[f64]) -> f64 { - let mut total_value = 0.0; - let mut total_volume = 0.0; - - for i in 0..prices.len() { - total_value += prices[i] * volumes[i]; - total_volume += volumes[i]; - } - - if total_volume > 0.0 { - total_value / total_volume - } else { - 0.0 - } -} - -pub fn run_simple_simd_test() { - println!("🚀 Simple SIMD Performance Validation"); - println!("====================================="); - - if !arch::is_x86_feature_detected!("avx2") { - println!("❌ AVX2 not available - cannot test SIMD"); - return; - } - - // Generate test data - let data_size = 10_000; - let iterations = 1_000; - - let mut prices = Vec::with_capacity(data_size); - let mut volumes = Vec::with_capacity(data_size); - - for i in 0..data_size { - prices.push(100.0 + (i as f64) * 0.01); - volumes.push(1000.0 + (i as f64) * 0.1); - } - - let aligned_prices = AlignedPrices::from_slice(&prices); - let aligned_volumes = AlignedVolumes::from_slice(&volumes); - - println!("Data size: {} elements", data_size); - println!("Iterations: {}", iterations); - println!(); - - // Verify correctness first - let scalar_result = simple_scalar_vwap(&prices, &volumes); - - unsafe { - let original_ops = SimdPriceOps::new(); - let optimized_ops = OptimizedSimdPriceOps::new(); - - let original_result = original_ops.calculate_vwap_aligned(&aligned_prices, &aligned_volumes); - let optimized_result = optimized_ops.calculate_vwap_fast(&aligned_prices, &aligned_volumes); - - println!("Correctness Check:"); - println!(" Scalar: {:.6}", scalar_result); - println!(" Original: {:.6}", original_result); - println!(" Optimized: {:.6}", optimized_result); - - assert!((scalar_result - original_result).abs() < 1e-10, "Original SIMD incorrect"); - assert!((scalar_result - optimized_result).abs() < 1e-10, "Optimized SIMD incorrect"); - println!("✅ All results match"); - println!(); - - // Warmup - for _ in 0..10 { - let _ = simple_scalar_vwap(&prices, &volumes); - let _ = original_ops.calculate_vwap_aligned(&aligned_prices, &aligned_volumes); - let _ = optimized_ops.calculate_vwap_fast(&aligned_prices, &aligned_volumes); - } - - // Benchmark scalar - let start = Instant::now(); - for _ in 0..iterations { - let _ = simple_scalar_vwap(&prices, &volumes); - } - let scalar_time = start.elapsed().as_nanos(); - - // Benchmark original SIMD - let start = Instant::now(); - for _ in 0..iterations { - let _ = original_ops.calculate_vwap_aligned(&aligned_prices, &aligned_volumes); - } - let original_simd_time = start.elapsed().as_nanos(); - - // Benchmark optimized SIMD - let start = Instant::now(); - for _ in 0..iterations { - let _ = optimized_ops.calculate_vwap_fast(&aligned_prices, &aligned_volumes); - } - let optimized_simd_time = start.elapsed().as_nanos(); - - println!("Performance Results:"); - println!(" Scalar: {} ns", scalar_time); - println!(" Original: {} ns", original_simd_time); - println!(" Optimized: {} ns", optimized_simd_time); - println!(); - - let original_speedup = scalar_time as f64 / original_simd_time as f64; - let optimized_speedup = scalar_time as f64 / optimized_simd_time as f64; - - println!("Speedup vs Scalar:"); - println!(" Original SIMD: {:.2}x", original_speedup); - println!(" Optimized SIMD: {:.2}x", optimized_speedup); - println!(); - - // Validate results - if original_speedup < 1.0 { - println!("✅ Confirmed: Original SIMD is slower than scalar ({:.2}x)", original_speedup); - } else { - println!("⚠️ Unexpected: Original SIMD is not slower than scalar ({:.2}x)", original_speedup); - } - - if optimized_speedup >= 4.0 { - println!("🎉 SUCCESS: Optimized SIMD achieves target 4x+ speedup ({:.2}x)", optimized_speedup); - println!("✅ SIMD performance regression FIXED!"); - } else if optimized_speedup >= 2.0 { - println!("⚠️ PARTIAL: Optimized SIMD achieves 2x+ speedup ({:.2}x) but below 4x target", optimized_speedup); - println!("🔧 SIMD performance partially improved"); - } else if optimized_speedup >= 1.0 { - println!("🆗 IMPROVEMENT: Optimized SIMD is faster than scalar ({:.2}x) but below 2x", optimized_speedup); - println!("📈 SIMD performance improved but needs more work"); - } else { - println!("❌ FAILED: Optimized SIMD still slower than scalar ({:.2}x)", optimized_speedup); - println!("💥 SIMD performance regression NOT fixed"); - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_simple_simd_validation() { - run_simple_simd_test(); - } -} \ No newline at end of file diff --git a/crates/config/Cargo.toml b/crates/config/Cargo.toml new file mode 100644 index 000000000..680bbac13 --- /dev/null +++ b/crates/config/Cargo.toml @@ -0,0 +1,67 @@ +[package] +name = "config" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +authors.workspace = true +license.workspace = true +repository.workspace = true +homepage.workspace = true +documentation.workspace = true +publish.workspace = true +keywords.workspace = true +categories.workspace = true +description = "Centralized configuration management for Foxhunt HFT Trading System" + +[dependencies] +# Core async and utilities +tokio.workspace = true +serde = { workspace = true, features = ["derive"] } +serde_json.workspace = true +uuid.workspace = true +thiserror.workspace = true +anyhow.workspace = true +futures.workspace = true +async-trait.workspace = true +once_cell.workspace = true + +# Time handling +chrono.workspace = true + +# Configuration formats +toml.workspace = true +serde_yaml.workspace = true + +# Database for PostgreSQL config loader +sqlx = { workspace = true, features = ["postgres", "runtime-tokio-rustls", "chrono", "uuid", "json"] } + +# Logging and tracing +tracing.workspace = true + +# HashiCorp Vault integration +vaultrs = "0.7" + +# High-performance data structures +rustc-hash.workspace = true +dashmap.workspace = true +parking_lot.workspace = true + +# Networking +reqwest.workspace = true + +# Security +sha2.workspace = true +base64.workspace = true + +[dev-dependencies] +tokio-test.workspace = true +tempfile.workspace = true +test-case.workspace = true +wiremock.workspace = true +testcontainers.workspace = true + +[features] +default = ["postgres", "vault"] +postgres = [] +vault = [] +redis = [] \ No newline at end of file diff --git a/crates/config/src/error.rs b/crates/config/src/error.rs new file mode 100644 index 000000000..c73114d2c --- /dev/null +++ b/crates/config/src/error.rs @@ -0,0 +1,322 @@ +//! Error types for the configuration system + +use thiserror::Error; + +/// Result type for configuration operations +pub type ConfigResult = Result; + +/// Configuration system errors +#[derive(Error, Debug)] +pub enum ConfigError { + #[error("Database error: {message}")] + DatabaseError { message: String }, + + #[error("Vault error: {message}")] + VaultError { message: String }, + + #[error("Failed to read config file {path}: {source}")] + FileReadError { + path: String, + #[source] + source: std::io::Error, + }, + + #[error("Failed to write config file {path}: {source}")] + FileWriteError { + path: String, + #[source] + source: std::io::Error, + }, + + #[error("Failed to parse configuration: {source}")] + ParseError { + #[from] + source: serde_json::Error, + }, + + #[error("Failed to serialize configuration: {source}")] + SerializationError { + #[from] + source: toml::ser::Error, + }, + + #[error("Configuration validation failed: {message}")] + ValidationError { message: String }, + + #[error("Configuration key '{key}' not found in category '{category}'")] + KeyNotFound { category: String, key: String }, + + #[error("Invalid configuration type for key '{key}': expected {expected}, got {actual}")] + TypeError { + key: String, + expected: String, + actual: String, + }, + + #[error("Environment variable '{var}' not found")] + EnvironmentError { var: String }, + + #[error("Network error: {source}")] + NetworkError { + #[from] + source: reqwest::Error, + }, + + #[error("SQL database error: {source}")] + SqlError { + #[from] + source: sqlx::Error, + }, + + #[error("Vault client error: {source}")] + VaultClientError { + #[from] + source: vaultrs::error::ClientError, + }, + + #[error("Authentication failed: {message}")] + AuthenticationError { message: String }, + + #[error("Connection timeout after {timeout_ms}ms")] + TimeoutError { timeout_ms: u64 }, + + #[error("Circuit breaker is open: {message}")] + CircuitBreakerError { message: String }, + + #[error("Configuration system not initialized")] + NotInitialized, + + #[error("Configuration cache error: {message}")] + CacheError { message: String }, + + #[error("Hot-reload listener error: {message}")] + HotReloadError { message: String }, + + #[error("Unknown configuration error: {message}")] + Unknown { message: String }, +} + +impl ConfigError { + /// Create a database error + pub fn database>(message: S) -> Self { + ConfigError::DatabaseError { + message: message.into(), + } + } + + /// Create a vault error + pub fn vault>(message: S) -> Self { + ConfigError::VaultError { + message: message.into(), + } + } + + /// Create a validation error + pub fn validation>(message: S) -> Self { + ConfigError::ValidationError { + message: message.into(), + } + } + + /// Create a key not found error + pub fn key_not_found>(category: S, key: S) -> Self { + ConfigError::KeyNotFound { + category: category.into(), + key: key.into(), + } + } + + /// Create a type error + pub fn type_error>(key: S, expected: S, actual: S) -> Self { + ConfigError::TypeError { + key: key.into(), + expected: expected.into(), + actual: actual.into(), + } + } + + /// Create an environment error + pub fn environment>(var: S) -> Self { + ConfigError::EnvironmentError { var: var.into() } + } + + /// Create an authentication error + pub fn authentication>(message: S) -> Self { + ConfigError::AuthenticationError { + message: message.into(), + } + } + + /// Create a timeout error + pub fn timeout(timeout_ms: u64) -> Self { + ConfigError::TimeoutError { timeout_ms } + } + + /// Create a circuit breaker error + pub fn circuit_breaker>(message: S) -> Self { + ConfigError::CircuitBreakerError { + message: message.into(), + } + } + + /// Create a cache error + pub fn cache>(message: S) -> Self { + ConfigError::CacheError { + message: message.into(), + } + } + + /// Create a hot-reload error + pub fn hot_reload>(message: S) -> Self { + ConfigError::HotReloadError { + message: message.into(), + } + } + + /// Create an unknown error + pub fn unknown>(message: S) -> Self { + ConfigError::Unknown { + message: message.into(), + } + } + + /// Check if this error is retryable + pub fn is_retryable(&self) -> bool { + match self { + ConfigError::DatabaseError { .. } => true, + ConfigError::VaultError { .. } => true, + ConfigError::NetworkError { .. } => true, + ConfigError::TimeoutError { .. } => true, + ConfigError::CircuitBreakerError { .. } => false, + ConfigError::AuthenticationError { .. } => false, + ConfigError::ValidationError { .. } => false, + ConfigError::KeyNotFound { .. } => false, + ConfigError::TypeError { .. } => false, + ConfigError::EnvironmentError { .. } => false, + ConfigError::FileReadError { .. } => true, + ConfigError::FileWriteError { .. } => true, + ConfigError::ParseError { .. } => false, + ConfigError::SerializationError { .. } => false, + ConfigError::SqlError { .. } => true, + ConfigError::VaultClientError { .. } => true, + ConfigError::NotInitialized => false, + ConfigError::CacheError { .. } => false, + ConfigError::HotReloadError { .. } => true, + ConfigError::Unknown { .. } => false, + } + } + + /// Get error category for metrics and monitoring + pub fn category(&self) -> &'static str { + match self { + ConfigError::DatabaseError { .. } => "database", + ConfigError::VaultError { .. } => "vault", + ConfigError::FileReadError { .. } => "file_io", + ConfigError::FileWriteError { .. } => "file_io", + ConfigError::ParseError { .. } => "parsing", + ConfigError::SerializationError { .. } => "serialization", + ConfigError::ValidationError { .. } => "validation", + ConfigError::KeyNotFound { .. } => "not_found", + ConfigError::TypeError { .. } => "type", + ConfigError::EnvironmentError { .. } => "environment", + ConfigError::NetworkError { .. } => "network", + ConfigError::SqlError { .. } => "database", + ConfigError::VaultClientError { .. } => "vault", + ConfigError::AuthenticationError { .. } => "auth", + ConfigError::TimeoutError { .. } => "timeout", + ConfigError::CircuitBreakerError { .. } => "circuit_breaker", + ConfigError::NotInitialized => "initialization", + ConfigError::CacheError { .. } => "cache", + ConfigError::HotReloadError { .. } => "hot_reload", + ConfigError::Unknown { .. } => "unknown", + } + } +} + +// Convert from common error types +impl From for ConfigError { + fn from(err: std::io::Error) -> Self { + ConfigError::Unknown { + message: err.to_string(), + } + } +} + +impl From for ConfigError { + fn from(err: anyhow::Error) -> Self { + ConfigError::Unknown { + message: err.to_string(), + } + } +} + +impl From for ConfigError { + fn from(err: serde_yaml::Error) -> Self { + ConfigError::ParseError { + source: serde_json::Error::custom(err.to_string()), + } + } +} + +impl From for ConfigError { + fn from(err: toml::de::Error) -> Self { + ConfigError::ParseError { + source: serde_json::Error::custom(err.to_string()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_error_creation() { + let db_err = ConfigError::database("Connection failed"); + assert!(matches!(db_err, ConfigError::DatabaseError { .. })); + assert!(db_err.is_retryable()); + assert_eq!(db_err.category(), "database"); + + let vault_err = ConfigError::vault("Authentication failed"); + assert!(matches!(vault_err, ConfigError::VaultError { .. })); + assert!(vault_err.is_retryable()); + assert_eq!(vault_err.category(), "vault"); + + let validation_err = ConfigError::validation("Invalid value"); + assert!(matches!(validation_err, ConfigError::ValidationError { .. })); + assert!(!validation_err.is_retryable()); + assert_eq!(validation_err.category(), "validation"); + + let key_err = ConfigError::key_not_found("trading", "max_order_size"); + assert!(matches!(key_err, ConfigError::KeyNotFound { .. })); + assert!(!key_err.is_retryable()); + assert_eq!(key_err.category(), "not_found"); + + let type_err = ConfigError::type_error("max_order_size", "f64", "string"); + assert!(matches!(type_err, ConfigError::TypeError { .. })); + assert!(!type_err.is_retryable()); + assert_eq!(type_err.category(), "type"); + } + + #[test] + fn test_retryable_errors() { + // Retryable errors + assert!(ConfigError::database("test").is_retryable()); + assert!(ConfigError::vault("test").is_retryable()); + assert!(ConfigError::timeout(1000).is_retryable()); + + // Non-retryable errors + assert!(!ConfigError::validation("test").is_retryable()); + assert!(!ConfigError::key_not_found("cat", "key").is_retryable()); + assert!(!ConfigError::circuit_breaker("test").is_retryable()); + } + + #[test] + fn test_error_categories() { + assert_eq!(ConfigError::database("test").category(), "database"); + assert_eq!(ConfigError::vault("test").category(), "vault"); + assert_eq!(ConfigError::validation("test").category(), "validation"); + assert_eq!(ConfigError::authentication("test").category(), "auth"); + assert_eq!(ConfigError::timeout(1000).category(), "timeout"); + } +} \ No newline at end of file diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs new file mode 100644 index 000000000..077f73b19 --- /dev/null +++ b/crates/config/src/lib.rs @@ -0,0 +1,226 @@ +//! Centralized Configuration Management for Foxhunt HFT Trading System +//! +//! This crate provides a unified configuration system that consolidates all +//! configuration management across the Foxhunt trading system. Features include: +//! +//! - PostgreSQL-based configuration with hot-reload via NOTIFY/LISTEN +//! - HashiCorp Vault integration for secure credential management +//! - Unified configuration structures for all services +//! - Type-safe configuration getters with caching +//! - Environment-specific configuration support +//! - Configuration validation and health monitoring +//! +//! # Example Usage +//! +//! ```rust +//! use config::{ConfigManager, DatabaseConfig, VaultConfig}; +//! +//! # async fn example() -> anyhow::Result<()> { +//! // Initialize configuration manager +//! let db_config = DatabaseConfig::from_env()?; +//! let vault_config = VaultConfig::from_env()?; +//! let config_manager = ConfigManager::new(db_config, Some(vault_config)).await?; +//! +//! // Get trading configuration +//! let max_order_size = config_manager.get_trading_config("max_order_size").await?; +//! +//! // Subscribe to configuration changes +//! let mut changes = config_manager.subscribe_to_changes().await?; +//! while let Some((category, key)) = changes.recv().await { +//! println!("Configuration changed: {}.{}", category, key); +//! } +//! # Ok(()) +//! # } +//! ``` + +pub mod database; +pub mod error; +pub mod manager; +pub mod structures; +pub mod vault; + +pub use database::{DatabaseConfig, PostgresConfigLoader}; +pub use error::{ConfigError, ConfigResult}; +pub use manager::ConfigManager; +pub use structures::*; +pub use vault::{VaultConfig, VaultSecrets}; + +// Re-export commonly used types +pub use serde::{Deserialize, Serialize}; +pub use serde_json::Value as JsonValue; +pub use std::collections::HashMap; +pub use chrono::{DateTime, Utc}; + +/// Configuration categories supported by the system +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum ConfigCategory { + /// Trading engine parameters (order sizes, position limits) + Trading, + /// Risk management settings (VaR, drawdown limits, circuit breakers) + Risk, + /// Market data provider configurations + MarketData, + /// Machine learning model settings + MachineLearning, + /// Broker connection configurations + Brokers, + /// System performance tuning parameters + Performance, + /// Security and authentication settings + Security, + /// Environment-specific configurations + Environment, +} + +impl ConfigCategory { + /// Get the PostgreSQL table name for this category + pub fn table_name(&self) -> &'static str { + match self { + ConfigCategory::Trading => "trading_configs", + ConfigCategory::Risk => "risk_configs", + ConfigCategory::MarketData => "market_data_configs", + ConfigCategory::MachineLearning => "ml_configs", + ConfigCategory::Brokers => "broker_configs", + ConfigCategory::Performance => "performance_configs", + ConfigCategory::Security => "security_configs", + ConfigCategory::Environment => "environment_configs", + } + } + + /// Get the NOTIFY channel name for this category + pub fn notify_channel(&self) -> &'static str { + match self { + ConfigCategory::Trading => "config_trading_changes", + ConfigCategory::Risk => "config_risk_changes", + ConfigCategory::MarketData => "config_market_data_changes", + ConfigCategory::MachineLearning => "config_ml_changes", + ConfigCategory::Brokers => "config_broker_changes", + ConfigCategory::Performance => "config_performance_changes", + ConfigCategory::Security => "config_security_changes", + ConfigCategory::Environment => "config_environment_changes", + } + } + + /// Get the Vault path for this category + pub fn vault_path(&self) -> &'static str { + match self { + ConfigCategory::Trading => "foxhunt/trading", + ConfigCategory::Risk => "foxhunt/risk", + ConfigCategory::MarketData => "foxhunt/market-data", + ConfigCategory::MachineLearning => "foxhunt/ml", + ConfigCategory::Brokers => "foxhunt/brokers", + ConfigCategory::Performance => "foxhunt/performance", + ConfigCategory::Security => "foxhunt/security", + ConfigCategory::Environment => "foxhunt/environment", + } + } +} + +/// Configuration value with metadata +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigValue { + /// The configuration key + pub key: String, + /// The configuration value as JSON + pub value: JsonValue, + /// Configuration category + pub category: ConfigCategory, + /// Environment (development, staging, production) + pub environment: String, + /// When this configuration was last updated + pub updated_at: DateTime, + /// Configuration description/documentation + pub description: Option, + /// Whether this configuration is active + pub is_active: bool, + /// Configuration source (database, vault, environment, default) + pub source: ConfigSource, +} + +/// Configuration source +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum ConfigSource { + /// Loaded from PostgreSQL database + Database, + /// Loaded from HashiCorp Vault + Vault, + /// Loaded from environment variables + Environment, + /// Default value + Default, + /// Loaded from configuration file + File, +} + +/// Configuration change event +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigChange { + /// Configuration category that changed + pub category: ConfigCategory, + /// Configuration key that changed + pub key: String, + /// New configuration value + pub new_value: ConfigValue, + /// Previous configuration value (if any) + pub old_value: Option, + /// When the change occurred + pub changed_at: DateTime, + /// Change source + pub changed_by: String, +} + +/// Health status for configuration components +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigHealth { + /// Whether the component is healthy + pub is_healthy: bool, + /// Health check message + pub message: String, + /// Last successful operation timestamp + pub last_success: Option>, + /// Number of recent failures + pub failure_count: u64, + /// Component-specific metrics + pub metrics: HashMap, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_config_category_names() { + assert_eq!(ConfigCategory::Trading.table_name(), "trading_configs"); + assert_eq!(ConfigCategory::Risk.table_name(), "risk_configs"); + assert_eq!(ConfigCategory::MarketData.table_name(), "market_data_configs"); + assert_eq!(ConfigCategory::MachineLearning.table_name(), "ml_configs"); + assert_eq!(ConfigCategory::Brokers.table_name(), "broker_configs"); + assert_eq!(ConfigCategory::Performance.table_name(), "performance_configs"); + assert_eq!(ConfigCategory::Security.table_name(), "security_configs"); + assert_eq!(ConfigCategory::Environment.table_name(), "environment_configs"); + } + + #[test] + fn test_config_category_channels() { + assert_eq!(ConfigCategory::Trading.notify_channel(), "config_trading_changes"); + assert_eq!(ConfigCategory::Risk.notify_channel(), "config_risk_changes"); + assert_eq!(ConfigCategory::MarketData.notify_channel(), "config_market_data_changes"); + assert_eq!(ConfigCategory::MachineLearning.notify_channel(), "config_ml_changes"); + assert_eq!(ConfigCategory::Brokers.notify_channel(), "config_broker_changes"); + assert_eq!(ConfigCategory::Performance.notify_channel(), "config_performance_changes"); + assert_eq!(ConfigCategory::Security.notify_channel(), "config_security_changes"); + assert_eq!(ConfigCategory::Environment.notify_channel(), "config_environment_changes"); + } + + #[test] + fn test_config_category_vault_paths() { + assert_eq!(ConfigCategory::Trading.vault_path(), "foxhunt/trading"); + assert_eq!(ConfigCategory::Risk.vault_path(), "foxhunt/risk"); + assert_eq!(ConfigCategory::MarketData.vault_path(), "foxhunt/market-data"); + assert_eq!(ConfigCategory::MachineLearning.vault_path(), "foxhunt/ml"); + assert_eq!(ConfigCategory::Brokers.vault_path(), "foxhunt/brokers"); + assert_eq!(ConfigCategory::Performance.vault_path(), "foxhunt/performance"); + assert_eq!(ConfigCategory::Security.vault_path(), "foxhunt/security"); + assert_eq!(ConfigCategory::Environment.vault_path(), "foxhunt/environment"); + } +} \ No newline at end of file diff --git a/services/backtesting_service/src/strategy_engine.rs b/services/backtesting_service/src/strategy_engine.rs index 05a525553..c451d7ca3 100644 --- a/services/backtesting_service/src/strategy_engine.rs +++ b/services/backtesting_service/src/strategy_engine.rs @@ -268,7 +268,7 @@ pub struct StrategyEngine { } /// Trait for strategy execution -pub trait StrategyExecutor: Send + Sync { +pub trait StrategyExecutor: Send + Sync + std::fmt::Debug { /// Execute strategy for a given market data point fn execute( &self, diff --git a/storage/Cargo.toml b/storage/Cargo.toml new file mode 100644 index 000000000..f51b7acf5 --- /dev/null +++ b/storage/Cargo.toml @@ -0,0 +1,79 @@ +[package] +name = "storage" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +authors.workspace = true +license.workspace = true +repository.workspace = true +homepage.workspace = true +documentation.workspace = true +publish.workspace = true +keywords.workspace = true +categories.workspace = true +description = "Storage layer for Foxhunt HFT Trading System - S3, local file operations, and data archival" + +[dependencies] +# Core async and utilities +tokio = { workspace = true, features = ["rt-multi-thread", "fs", "sync", "time"] } +tokio-util = { workspace = true } +async-trait = { workspace = true } +futures = { workspace = true } +anyhow = { workspace = true } +thiserror = { workspace = true } +uuid = { workspace = true } + +# Serialization and time +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +chrono = { workspace = true, features = ["serde"] } +bincode = { workspace = true } + +# Compression and hashing +flate2 = "1.0" +sha2 = "0.10" + +# Logging +tracing = { workspace = true } + +# Collections and utilities +rustc-hash = { workspace = true } +indexmap = { workspace = true } + +# AWS S3 SDK +aws-config = { version = "1.1", features = ["behavior-version-latest"] } +aws-sdk-s3 = { version = "1.15", features = ["behavior-version-latest"] } +aws-types = "1.1" + +# Vault integration for secure credential management +vault = "0.7" + +# File system operations +fs2 = { workspace = true } +tempfile = { workspace = true } + +# Data structures and caching +dashmap = { workspace = true } +lru = "0.12" + +# Error handling and retry logic +backoff = "0.4" + +[dev-dependencies] +# Testing +tokio-test = { workspace = true } +tempfile = { workspace = true } +serial_test = { workspace = true } +wiremock = { workspace = true } + +[features] +default = ["s3", "vault-integration"] + +# S3 storage backend +s3 = ["aws-config", "aws-sdk-s3", "aws-types"] + +# Vault integration for secure credential management +vault-integration = ["vault"] + +# Local file operations only (no cloud dependencies) +local-only = [] \ No newline at end of file diff --git a/storage/src/error.rs b/storage/src/error.rs new file mode 100644 index 000000000..d68ab3cb9 --- /dev/null +++ b/storage/src/error.rs @@ -0,0 +1,286 @@ +//! Error types for storage operations + +use thiserror::Error; + +/// Storage-related errors +#[derive(Error, Debug, Clone)] +pub enum StorageError { + /// IO error during file operations + #[error("IO error: {message}")] + IoError { message: String }, + + /// Network error during remote storage operations + #[error("Network error: {message}")] + NetworkError { message: String }, + + /// Authentication/authorization error + #[error("Authentication error: {message}")] + AuthError { message: String }, + + /// Data not found at the specified path + #[error("Data not found at path: {path}")] + NotFound { path: String }, + + /// Permission denied for storage operation + #[error("Permission denied for operation on: {path}")] + PermissionDenied { path: String }, + + /// Storage quota exceeded + #[error("Storage quota exceeded (used: {used}, limit: {limit})")] + QuotaExceeded { used: u64, limit: u64 }, + + /// Data integrity check failed + #[error("Data integrity check failed for: {path} (expected: {expected}, actual: {actual})")] + IntegrityError { + path: String, + expected: String, + actual: String, + }, + + /// Serialization/deserialization error + #[error("Serialization error: {message}")] + SerializationError { message: String }, + + /// Compression/decompression error + #[error("Compression error: {message}")] + CompressionError { message: String }, + + /// Configuration error + #[error("Configuration error: {message}")] + ConfigError { message: String }, + + /// Timeout during storage operation + #[error("Operation timed out after {timeout_ms}ms")] + Timeout { timeout_ms: u64 }, + + /// Rate limit exceeded + #[error("Rate limit exceeded, retry after {retry_after_ms}ms")] + RateLimited { retry_after_ms: u64 }, + + /// Generic storage error + #[error("Storage error: {message}")] + Generic { message: String }, + + /// Vault-related error + #[cfg(feature = "vault-integration")] + #[error("Vault error: {0}")] + VaultError(#[from] crate::vault::VaultError), + + /// S3-specific error + #[cfg(feature = "s3")] + #[error("S3 error: {message}")] + S3Error { message: String }, +} + +impl StorageError { + /// Check if the error is retryable + pub fn is_retryable(&self) -> bool { + match self { + StorageError::NetworkError { .. } => true, + StorageError::Timeout { .. } => true, + StorageError::RateLimited { .. } => true, + StorageError::Generic { .. } => true, + #[cfg(feature = "s3")] + StorageError::S3Error { .. } => true, + _ => false, + } + } + + /// Get retry delay in milliseconds for retryable errors + pub fn retry_delay_ms(&self) -> Option { + match self { + StorageError::NetworkError { .. } => Some(100), + StorageError::Timeout { .. } => Some(200), + StorageError::RateLimited { retry_after_ms } => Some(*retry_after_ms), + StorageError::Generic { .. } => Some(100), + #[cfg(feature = "s3")] + StorageError::S3Error { .. } => Some(100), + _ => None, + } + } + + /// Check if error indicates a transient issue + pub fn is_transient(&self) -> bool { + match self { + StorageError::NetworkError { .. } => true, + StorageError::Timeout { .. } => true, + StorageError::RateLimited { .. } => true, + _ => false, + } + } + + /// Get error category for metrics and monitoring + pub fn category(&self) -> &'static str { + match self { + StorageError::IoError { .. } => "io", + StorageError::NetworkError { .. } => "network", + StorageError::AuthError { .. } => "auth", + StorageError::NotFound { .. } => "not_found", + StorageError::PermissionDenied { .. } => "permission", + StorageError::QuotaExceeded { .. } => "quota", + StorageError::IntegrityError { .. } => "integrity", + StorageError::SerializationError { .. } => "serialization", + StorageError::CompressionError { .. } => "compression", + StorageError::ConfigError { .. } => "config", + StorageError::Timeout { .. } => "timeout", + StorageError::RateLimited { .. } => "rate_limit", + StorageError::Generic { .. } => "generic", + #[cfg(feature = "vault-integration")] + StorageError::VaultError(_) => "vault", + #[cfg(feature = "s3")] + StorageError::S3Error { .. } => "s3", + } + } + + /// Create a sanitized error message safe for logging (removes sensitive info) + pub fn safe_message(&self) -> String { + match self { + StorageError::AuthError { .. } => { + "Authentication error (details masked for security)".to_string() + } + #[cfg(feature = "vault-integration")] + StorageError::VaultError(vault_err) => vault_err.safe_message(), + _ => self.to_string(), + } + } +} + +/// Result type for storage operations +pub type StorageResult = Result; + +// Conversion implementations for common error types + +impl From for StorageError { + fn from(err: std::io::Error) -> Self { + use std::io::ErrorKind; + match err.kind() { + ErrorKind::NotFound => StorageError::NotFound { + path: "unknown".to_string(), + }, + ErrorKind::PermissionDenied => StorageError::PermissionDenied { + path: "unknown".to_string(), + }, + ErrorKind::TimedOut => StorageError::Timeout { timeout_ms: 5000 }, + _ => StorageError::IoError { + message: err.to_string(), + }, + } + } +} + +impl From for StorageError { + fn from(err: serde_json::Error) -> Self { + StorageError::SerializationError { + message: err.to_string(), + } + } +} + +impl From for StorageError { + fn from(err: bincode::Error) -> Self { + StorageError::SerializationError { + message: err.to_string(), + } + } +} + +#[cfg(feature = "s3")] +impl From> for StorageError +where + E: std::error::Error + Send + Sync + 'static, +{ + fn from(err: aws_sdk_s3::error::SdkError) -> Self { + match err { + aws_sdk_s3::error::SdkError::TimeoutError(_) => StorageError::Timeout { + timeout_ms: 5000, + }, + aws_sdk_s3::error::SdkError::DispatchFailure(dispatch_err) => { + if dispatch_err.is_timeout() { + StorageError::Timeout { timeout_ms: 5000 } + } else if dispatch_err.is_io() { + StorageError::NetworkError { + message: dispatch_err.to_string(), + } + } else { + StorageError::S3Error { + message: dispatch_err.to_string(), + } + } + } + _ => StorageError::S3Error { + message: err.to_string(), + }, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_error_retryability() { + assert!(StorageError::NetworkError { + message: "test".to_string() + } + .is_retryable()); + + assert!(!StorageError::NotFound { + path: "test".to_string() + } + .is_retryable()); + + assert!(StorageError::Timeout { timeout_ms: 1000 }.is_retryable()); + } + + #[test] + fn test_error_categories() { + assert_eq!( + StorageError::IoError { + message: "test".to_string() + } + .category(), + "io" + ); + + assert_eq!( + StorageError::NetworkError { + message: "test".to_string() + } + .category(), + "network" + ); + + assert_eq!( + StorageError::NotFound { + path: "test".to_string() + } + .category(), + "not_found" + ); + } + + #[test] + fn test_error_transient() { + assert!(StorageError::NetworkError { + message: "test".to_string() + } + .is_transient()); + + assert!(!StorageError::ConfigError { + message: "test".to_string() + } + .is_transient()); + } + + #[test] + fn test_safe_message() { + let auth_error = StorageError::AuthError { + message: "sensitive auth details".to_string(), + }; + + let safe_msg = auth_error.safe_message(); + assert!(safe_msg.contains("masked")); + assert!(!safe_msg.contains("sensitive")); + } +} \ No newline at end of file diff --git a/storage/src/lib.rs b/storage/src/lib.rs new file mode 100644 index 000000000..3d9d80870 --- /dev/null +++ b/storage/src/lib.rs @@ -0,0 +1,231 @@ +//! Storage Library for Foxhunt HFT Trading System +//! +//! This crate provides comprehensive storage solutions for the HFT system including: +//! - S3 archival with lifecycle management and compression +//! - Local file operations with atomic writes and locking +//! - Vault integration for secure credential management +//! - Model storage and retrieval utilities +//! - Backup and disaster recovery operations +//! +//! # Features +//! +//! - **S3 Integration**: High-performance S3 operations with automatic retry, compression, and lifecycle policies +//! - **Vault Security**: Secure credential retrieval from HashiCorp Vault with circuit breakers +//! - **Local Storage**: Thread-safe local file operations with atomic writes and file locking +//! - **Data Integrity**: Checksums and verification for all storage operations +//! - **Performance Monitoring**: Built-in metrics and telemetry for storage operations + +#![warn(missing_docs)] +#![warn(clippy::unwrap_used)] +#![warn(clippy::expect_used)] + +pub mod s3; +pub mod local; +pub mod vault; +pub mod error; +pub mod config; +pub mod metrics; + +// Re-export commonly used types and traits +pub use error::{StorageError, StorageResult}; +pub use config::StorageConfig; + +#[cfg(feature = "s3")] +pub use s3::{S3Storage, S3StorageConfig, ArchivalDataType, ArchivalMetadata, ArchivalStats}; + +#[cfg(feature = "vault-integration")] +pub use vault::{VaultClient, VaultConfig, VaultCredentials}; + +pub use local::{LocalStorage, LocalStorageConfig, FileOperation}; + +/// Common storage trait for abstracting different storage backends +#[async_trait::async_trait] +pub trait Storage: Send + Sync { + /// Store data at the specified path + async fn store(&self, path: &str, data: &[u8]) -> StorageResult<()>; + + /// Retrieve data from the specified path + async fn retrieve(&self, path: &str) -> StorageResult>; + + /// Check if data exists at the specified path + async fn exists(&self, path: &str) -> StorageResult; + + /// Delete data at the specified path + async fn delete(&self, path: &str) -> StorageResult; + + /// List all paths with the given prefix + async fn list(&self, prefix: &str) -> StorageResult>; + + /// Get metadata for the data at the specified path + async fn metadata(&self, path: &str) -> StorageResult; +} + +/// Metadata for stored data +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct StorageMetadata { + /// Path where the data is stored + pub path: String, + /// Size of the data in bytes + pub size: u64, + /// Content type/MIME type + pub content_type: Option, + /// Last modified timestamp + pub last_modified: chrono::DateTime, + /// ETag or checksum for data integrity + pub etag: Option, + /// Custom metadata tags + pub tags: std::collections::HashMap, +} + +/// Storage provider enumeration for configuration +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub enum StorageProvider { + /// Local filesystem storage + Local(local::LocalStorageConfig), + /// AWS S3 storage + #[cfg(feature = "s3")] + S3(s3::S3StorageConfig), + /// Multi-tier storage (e.g., local cache + S3 archival) + MultiTier { + /// Primary fast storage + primary: Box, + /// Secondary archival storage + secondary: Box, + }, +} + +/// Factory for creating storage instances from configuration +pub struct StorageFactory; + +impl StorageFactory { + /// Create a storage instance from the provided configuration + pub async fn create(provider: StorageProvider) -> StorageResult> { + match provider { + StorageProvider::Local(config) => { + let storage = local::LocalStorage::new(config).await?; + Ok(Box::new(storage)) + } + #[cfg(feature = "s3")] + StorageProvider::S3(config) => { + let storage = s3::S3Storage::new(config).await?; + Ok(Box::new(storage)) + } + StorageProvider::MultiTier { primary, secondary } => { + let primary_storage = Self::create(*primary).await?; + let secondary_storage = Self::create(*secondary).await?; + let storage = MultiTierStorage::new(primary_storage, secondary_storage); + Ok(Box::new(storage)) + } + } + } +} + +/// Multi-tier storage implementation that uses primary storage for fast access +/// and secondary storage for archival/backup +pub struct MultiTierStorage { + primary: Box, + secondary: Box, +} + +impl MultiTierStorage { + /// Create a new multi-tier storage with primary and secondary backends + pub fn new(primary: Box, secondary: Box) -> Self { + Self { primary, secondary } + } +} + +#[async_trait::async_trait] +impl Storage for MultiTierStorage { + async fn store(&self, path: &str, data: &[u8]) -> StorageResult<()> { + // Store in primary first + self.primary.store(path, data).await?; + + // Asynchronously store in secondary (don't block on it) + let secondary_clone = &self.secondary; + let path_clone = path.to_string(); + let data_clone = data.to_vec(); + + tokio::spawn(async move { + if let Err(e) = secondary_clone.store(&path_clone, &data_clone).await { + tracing::warn!("Failed to store in secondary storage: {}", e); + } + }); + + Ok(()) + } + + async fn retrieve(&self, path: &str) -> StorageResult> { + // Try primary first + match self.primary.retrieve(path).await { + Ok(data) => Ok(data), + Err(_) => { + // Fallback to secondary + tracing::debug!("Primary storage failed, trying secondary for path: {}", path); + self.secondary.retrieve(path).await + } + } + } + + async fn exists(&self, path: &str) -> StorageResult { + // Check primary first, then secondary + if self.primary.exists(path).await.unwrap_or(false) { + Ok(true) + } else { + self.secondary.exists(path).await + } + } + + async fn delete(&self, path: &str) -> StorageResult { + // Delete from both storages + let primary_result = self.primary.delete(path).await.unwrap_or(false); + let secondary_result = self.secondary.delete(path).await.unwrap_or(false); + + Ok(primary_result || secondary_result) + } + + async fn list(&self, prefix: &str) -> StorageResult> { + // Combine results from both storages + let mut paths = std::collections::HashSet::new(); + + if let Ok(primary_paths) = self.primary.list(prefix).await { + paths.extend(primary_paths); + } + + if let Ok(secondary_paths) = self.secondary.list(prefix).await { + paths.extend(secondary_paths); + } + + Ok(paths.into_iter().collect()) + } + + async fn metadata(&self, path: &str) -> StorageResult { + // Try primary first, fallback to secondary + match self.primary.metadata(path).await { + Ok(metadata) => Ok(metadata), + Err(_) => self.secondary.metadata(path).await, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_storage_metadata_serialization() { + let metadata = StorageMetadata { + path: "test/path".to_string(), + size: 1024, + content_type: Some("application/json".to_string()), + last_modified: chrono::Utc::now(), + etag: Some("abc123".to_string()), + tags: std::collections::HashMap::new(), + }; + + let serialized = serde_json::to_string(&metadata).unwrap(); + let deserialized: StorageMetadata = serde_json::from_str(&serialized).unwrap(); + + assert_eq!(metadata.path, deserialized.path); + assert_eq!(metadata.size, deserialized.size); + } +} \ No newline at end of file diff --git a/storage/src/vault.rs b/storage/src/vault.rs new file mode 100644 index 000000000..793119e44 --- /dev/null +++ b/storage/src/vault.rs @@ -0,0 +1,480 @@ +//! Vault integration for secure credential management + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::sync::RwLock; +use tracing::{debug, info, warn, error}; + +/// Re-export commonly used Vault types +pub use vault::Client; + +/// Vault configuration for storage operations +#[derive(Debug, Clone)] +pub struct VaultConfig { + /// Vault server address + pub address: String, + /// Vault namespace (for Vault Enterprise) + pub namespace: Option, + /// AppRole role ID + pub role_id: String, + /// Path to secret ID file + pub secret_id_file: String, + /// Request timeout + pub timeout: Duration, + /// Enable TLS verification + pub verify_tls: bool, + /// CA certificate path (optional) + pub ca_cert_path: Option, +} + +impl Default for VaultConfig { + fn default() -> Self { + Self { + address: "https://vault.company.com:8200".to_string(), + namespace: None, + role_id: String::new(), + secret_id_file: "/opt/foxhunt/vault/secret-id".to_string(), + timeout: Duration::from_secs(5), + verify_tls: true, + ca_cert_path: None, + } + } +} + +impl VaultConfig { + /// Create configuration from environment variables + pub fn from_env() -> Result { + let address = std::env::var("VAULT_ADDR") + .unwrap_or_else(|_| "https://vault.company.com:8200".to_string()); + + let role_id = std::env::var("VAULT_ROLE_ID") + .map_err(|_| VaultError::ConfigurationError { + message: "VAULT_ROLE_ID environment variable not set".to_string(), + })?; + + let secret_id_file = std::env::var("VAULT_SECRET_ID_FILE") + .unwrap_or_else(|_| "/opt/foxhunt/vault/secret-id".to_string()); + + let namespace = std::env::var("VAULT_NAMESPACE").ok(); + + let timeout = std::env::var("VAULT_TIMEOUT") + .ok() + .and_then(|s| s.parse::().ok()) + .map(Duration::from_secs) + .unwrap_or(Duration::from_secs(5)); + + let verify_tls = std::env::var("VAULT_VERIFY_TLS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(true); + + let ca_cert_path = std::env::var("VAULT_CA_CERT").ok(); + + Ok(Self { + address, + namespace, + role_id, + secret_id_file, + timeout, + verify_tls, + ca_cert_path, + }) + } +} + +/// AWS credentials retrieved from Vault +#[derive(Debug, Clone)] +pub struct VaultCredentials { + /// AWS access key ID + pub access_key_id: String, + /// AWS secret access key + pub secret_access_key: String, + /// AWS session token (optional, for temporary credentials) + pub session_token: Option, + /// When these credentials expire + pub expires_at: Option, +} + +impl VaultCredentials { + /// Check if credentials are expired or will expire soon + pub fn needs_refresh(&self, buffer: Duration) -> bool { + if let Some(expires_at) = self.expires_at { + Instant::now() + buffer >= expires_at + } else { + false // Non-expiring credentials + } + } + + /// Create AWS credentials provider from Vault credentials + #[cfg(feature = "s3")] + pub fn to_aws_credentials(&self) -> aws_types::Credentials { + aws_types::Credentials::new( + &self.access_key_id, + &self.secret_access_key, + self.session_token.clone(), + self.expires_at.map(|instant| { + std::time::SystemTime::UNIX_EPOCH + Duration::from_secs( + instant.duration_since(Instant::now()).as_secs() + + std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + ) + }), + "vault", + ) + } +} + +/// Vault client for retrieving storage credentials +pub struct VaultClient { + client: Arc>>, + config: VaultConfig, + auth_token: Arc>>, + token_expires_at: Arc>>, + /// Cached credentials to avoid frequent Vault calls + credentials_cache: Arc>>, + /// Cache TTL for credentials + cache_ttl: Duration, +} + +impl VaultClient { + /// Create new Vault client for storage operations + pub async fn new(config: VaultConfig) -> Result { + let client = Self { + client: Arc::new(RwLock::new(None)), + config, + auth_token: Arc::new(RwLock::new(None)), + token_expires_at: Arc::new(RwLock::new(None)), + credentials_cache: Arc::new(RwLock::new(HashMap::new())), + cache_ttl: Duration::from_secs(300), // 5 minutes cache + }; + + client.connect().await?; + Ok(client) + } + + /// Connect and authenticate with Vault + async fn connect(&self) -> Result<(), VaultError> { + debug!("Connecting to Vault at {}", self.config.address); + + let mut client = Client::new(&self.config.address) + .map_err(|e| VaultError::ConnectionFailed { + message: format!("Failed to create Vault client: {}", e), + })?; + + if let Some(ref namespace) = self.config.namespace { + client.set_namespace(namespace); + debug!("Set Vault namespace: {}", namespace); + } + + // Authenticate with AppRole + self.authenticate_approle(&mut client).await?; + + let mut client_guard = self.client.write().await; + *client_guard = Some(client); + + info!("Successfully connected to Vault for storage operations"); + Ok(()) + } + + /// Authenticate using AppRole + async fn authenticate_approle(&self, client: &mut Client) -> Result<(), VaultError> { + debug!("Authenticating with Vault using AppRole"); + + let secret_id = tokio::fs::read_to_string(&self.config.secret_id_file) + .await + .map_err(|e| VaultError::ConfigurationError { + message: format!("Failed to read secret ID file {}: {}", self.config.secret_id_file, e), + })? + .trim() + .to_string(); + + let auth_data = serde_json::json!({ + "role_id": self.config.role_id, + "secret_id": secret_id + }); + + let response = client + .write("auth/approle/login", &auth_data) + .await + .map_err(|e| VaultError::AuthenticationFailed { + message: format!("AppRole authentication failed: {}", e), + })?; + + let auth_info = response.get("auth") + .and_then(|auth| auth.as_object()) + .ok_or_else(|| VaultError::AuthenticationFailed { + message: "No auth information in response".to_string(), + })?; + + let token = auth_info.get("client_token") + .and_then(|token| token.as_str()) + .ok_or_else(|| VaultError::AuthenticationFailed { + message: "No client token in response".to_string(), + })? + .to_string(); + + let lease_duration = auth_info.get("lease_duration") + .and_then(|duration| duration.as_u64()) + .unwrap_or(3600); + + client.set_token(&token); + + let mut token_guard = self.auth_token.write().await; + *token_guard = Some(token); + + let mut expiry_guard = self.token_expires_at.write().await; + *expiry_guard = Some(Instant::now() + Duration::from_secs(lease_duration)); + + info!("Successfully authenticated with Vault for storage, token expires in {}s", lease_duration); + Ok(()) + } + + /// Get AWS S3 credentials from Vault + pub async fn get_s3_credentials(&self, path: &str) -> Result { + // Check cache first + { + let cache = self.credentials_cache.read().await; + if let Some((creds, cached_at)) = cache.get(path) { + if cached_at.elapsed() < self.cache_ttl && !creds.needs_refresh(Duration::from_secs(60)) { + debug!("Returning cached S3 credentials for path: {}", path); + return Ok(creds.clone()); + } + } + } + + debug!("Retrieving S3 credentials from Vault path: {}", path); + + // Ensure we're authenticated + self.ensure_authenticated().await?; + + let client_guard = self.client.read().await; + let client = client_guard.as_ref() + .ok_or_else(|| VaultError::ConnectionFailed { + message: "No Vault client connection".to_string(), + })?; + + let response = client + .read(path) + .await + .map_err(|e| VaultError::ClientError { + message: format!("Failed to read S3 credentials from {}: {}", path, e), + })?; + + let data = response.get("data") + .and_then(|data| data.as_object()) + .ok_or_else(|| VaultError::InvalidSecretFormat { + path: path.to_string(), + message: "No data field in secret response".to_string(), + })?; + + let access_key_id = data.get("access_key_id") + .or_else(|| data.get("access_key")) + .and_then(|v| v.as_str()) + .ok_or_else(|| VaultError::InvalidSecretFormat { + path: path.to_string(), + message: "Missing access_key_id field".to_string(), + })? + .to_string(); + + let secret_access_key = data.get("secret_access_key") + .or_else(|| data.get("secret_key")) + .and_then(|v| v.as_str()) + .ok_or_else(|| VaultError::InvalidSecretFormat { + path: path.to_string(), + message: "Missing secret_access_key field".to_string(), + })? + .to_string(); + + let session_token = data.get("session_token") + .or_else(|| data.get("token")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + // Check for expiration information + let expires_at = data.get("ttl") + .or_else(|| data.get("lease_duration")) + .and_then(|v| v.as_u64()) + .map(|ttl| Instant::now() + Duration::from_secs(ttl)); + + let credentials = VaultCredentials { + access_key_id, + secret_access_key, + session_token, + expires_at, + }; + + // Cache the credentials + { + let mut cache = self.credentials_cache.write().await; + cache.insert(path.to_string(), (credentials.clone(), Instant::now())); + } + + info!("Successfully retrieved S3 credentials from Vault path: {}", path); + Ok(credentials) + } + + /// Check if authentication token needs renewal + async fn needs_token_renewal(&self) -> bool { + let expiry_guard = self.token_expires_at.read().await; + if let Some(expires_at) = *expiry_guard { + Instant::now() + Duration::from_secs(300) > expires_at + } else { + true + } + } + + /// Ensure we have a valid authentication token + async fn ensure_authenticated(&self) -> Result<(), VaultError> { + if self.needs_token_renewal().await { + debug!("Token needs renewal, re-authenticating"); + let mut client_guard = self.client.write().await; + if let Some(ref mut client) = *client_guard { + self.authenticate_approle(client).await?; + } else { + return Err(VaultError::ConnectionFailed { + message: "No Vault client connection".to_string(), + }); + } + } + Ok(()) + } + + /// Clear credentials cache (useful for testing or manual refresh) + pub async fn clear_cache(&self) { + let mut cache = self.credentials_cache.write().await; + cache.clear(); + debug!("Cleared Vault credentials cache"); + } + + /// Health check for Vault connection + pub async fn health_check(&self) -> Result { + let client_guard = self.client.read().await; + let client = client_guard.as_ref() + .ok_or_else(|| VaultError::ConnectionFailed { + message: "No Vault client connection".to_string(), + })?; + + client + .read("sys/health") + .await + .map_err(|e| VaultError::ConnectionFailed { + message: format!("Health check failed: {}", e), + })?; + + Ok(true) + } +} + +/// Vault-specific error types +#[derive(thiserror::Error, Debug, Clone)] +pub enum VaultError { + /// Authentication failed with Vault + #[error("Vault authentication failed: {message}")] + AuthenticationFailed { message: String }, + + /// Connection to Vault server failed + #[error("Failed to connect to Vault: {message}")] + ConnectionFailed { message: String }, + + /// Secret not found at specified path + #[error("Secret not found at path: {path}")] + SecretNotFound { path: String }, + + /// Invalid secret format or content + #[error("Invalid secret format for {path}: {message}")] + InvalidSecretFormat { path: String, message: String }, + + /// Configuration error + #[error("Vault configuration error: {message}")] + ConfigurationError { message: String }, + + /// Generic Vault client error + #[error("Vault client error: {message}")] + ClientError { message: String }, +} + +impl VaultError { + /// Check if error is retryable + pub fn is_retryable(&self) -> bool { + match self { + VaultError::ConnectionFailed { .. } => true, + VaultError::ClientError { .. } => true, + VaultError::AuthenticationFailed { .. } => false, + VaultError::SecretNotFound { .. } => false, + VaultError::InvalidSecretFormat { .. } => false, + VaultError::ConfigurationError { .. } => false, + } + } + + /// Get a sanitized error message safe for logging + pub fn safe_message(&self) -> String { + match self { + VaultError::AuthenticationFailed { .. } => { + "Vault authentication failed (details masked for security)".to_string() + } + VaultError::SecretNotFound { .. } => { + "Secret not found (path masked for security)".to_string() + } + VaultError::InvalidSecretFormat { .. } => { + "Invalid secret format (details masked for security)".to_string() + } + _ => self.to_string(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_vault_config_from_env() { + std::env::set_var("VAULT_ADDR", "https://test-vault:8200"); + std::env::set_var("VAULT_ROLE_ID", "test-role-id"); + std::env::set_var("VAULT_SECRET_ID_FILE", "/tmp/test-secret"); + + let config = VaultConfig::from_env().unwrap(); + assert_eq!(config.address, "https://test-vault:8200"); + assert_eq!(config.role_id, "test-role-id"); + assert_eq!(config.secret_id_file, "/tmp/test-secret"); + + // Cleanup + std::env::remove_var("VAULT_ADDR"); + std::env::remove_var("VAULT_ROLE_ID"); + std::env::remove_var("VAULT_SECRET_ID_FILE"); + } + + #[test] + fn test_credentials_expiration() { + let creds = VaultCredentials { + access_key_id: "test".to_string(), + secret_access_key: "test".to_string(), + session_token: None, + expires_at: Some(Instant::now() + Duration::from_secs(30)), + }; + + // Should not need refresh yet (30s remaining, 60s buffer) + assert!(!creds.needs_refresh(Duration::from_secs(60))); + + // Should need refresh (30s remaining, 10s buffer) + assert!(creds.needs_refresh(Duration::from_secs(10))); + } + + #[test] + fn test_vault_error_retryability() { + assert!(VaultError::ConnectionFailed { + message: "test".to_string() + }.is_retryable()); + + assert!(!VaultError::AuthenticationFailed { + message: "test".to_string() + }.is_retryable()); + + assert!(!VaultError::SecretNotFound { + path: "secret/test".to_string() + }.is_retryable()); + } +} \ No newline at end of file