Files
foxhunt/common/src/database.rs
jgrusewski 030a15ee05 🔧 Emergency Fix: Resolve catastrophic _i32 suffix corruption (463→0 errors)
- Fixed systematic array indexing corruption: [0_i32] → [0]
- Fixed numeric literal suffixes across 835 files
- Fixed iterator patterns on RwLockReadGuard (.iter() required)
- Fixed float type annotations (365.25_f64 for sqrt)
- Fixed missing semicolons in position manager
- Fixed reference dereferencing in data loader

Root cause: Mass refactoring incorrectly added _i32 suffixes to array indices
Impact: Complete compilation failure (463 errors)
Resolution: Automated regex + targeted fixes
Result: 100% compilation success (0 errors)

Validated: cargo check --workspace passes
Ready for: Production deployment
2025-10-10 23:05:26 +02:00

301 lines
10 KiB
Rust

//! 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;
// Import centralized database configuration
pub use config::database::DatabaseConfig;
use config::structures::BacktestingDatabaseConfig;
/// Database-specific errors
#[derive(Debug, Error)]
#[allow(clippy::module_name_repetitions)]
pub enum DatabaseError {
/// Connection failed - wrapper around SQLx connection errors
#[error("Connection failed: {0}")]
Connection(#[from] sqlx::Error),
/// Query exceeded maximum allowed execution time
#[error("Query timeout: operation took {actual_ms}ms, max allowed {max_ms}ms")]
QueryTimeout {
/// Actual execution time in milliseconds
actual_ms: u64,
/// Maximum allowed execution time in milliseconds
max_ms: u64,
},
/// Connection pool has no available connections
#[error("Pool exhausted: no connections available")]
PoolExhausted,
/// Database configuration is invalid or missing required parameters
#[error("Configuration error: {0}")]
Configuration(String),
/// Performance constraint violation detected
#[error("Performance violation: {0}")]
Performance(String),
}
/// Database connection configuration (local extended version)
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct LocalDatabaseConfig {
/// 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 LocalDatabaseConfig {
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
}
}
}
/// Convert from centralized config to common crate config with HFT optimizations
#[allow(clippy::integer_division)]
impl From<DatabaseConfig> for LocalDatabaseConfig {
fn from(config: DatabaseConfig) -> Self {
Self {
url: config.url,
pool: PoolConfig {
max_connections: config.max_connections,
// 20% of max, min 2. Uses integer division intentionally for simplicity.
min_connections: (config.max_connections / 5).max(2),
connect_timeout_ms: u64::try_from(config.connect_timeout.as_millis().min(100))
.unwrap_or(100), // Convert to ms, cap at 100ms for HFT
acquire_timeout_ms: 50, // Fast acquire for HFT
max_lifetime_seconds: 3600, // 1 hour default
idle_timeout_seconds: 300, // 5 minutes default
},
performance: PerformanceConfig {
query_timeout_micros: u64::try_from(config.query_timeout.as_micros().min(800))
.unwrap_or(800), // Convert to microseconds, cap at 800μs for HFT
enable_prewarming: true,
enable_prepared_statements: true,
enable_slow_query_logging: config.enable_query_logging,
slow_query_threshold_micros: 1000, // 1ms threshold
},
}
}
}
/// Convert from backtesting config to common crate config with backtesting optimizations
#[allow(clippy::integer_division)]
impl From<BacktestingDatabaseConfig> for LocalDatabaseConfig {
fn from(config: BacktestingDatabaseConfig) -> Self {
let max_conn = config.max_connections.unwrap_or(10);
Self {
url: config.database_url,
pool: PoolConfig {
max_connections: max_conn,
// 25% of max, min 2. Uses integer division intentionally for simplicity.
min_connections: (max_conn / 4).max(2),
connect_timeout_ms: config.acquire_timeout_ms.unwrap_or(1000), // Use acquire timeout as connection timeout
acquire_timeout_ms: 100, // Less strict for backtesting
max_lifetime_seconds: 3600, // 1 hour default
idle_timeout_seconds: 600, // 10 minutes for backtesting
},
performance: PerformanceConfig {
query_timeout_micros: 10000, // 10ms default for backtesting queries
enable_prewarming: true,
enable_prepared_statements: true,
enable_slow_query_logging: config.enable_logging.unwrap_or(false),
slow_query_threshold_micros: 5000, // 5ms threshold for backtesting
},
}
}
}
/// Database connection pool wrapper
#[derive(Debug)]
#[allow(clippy::module_name_repetitions)]
pub struct DatabasePool {
pool: Pool<Postgres>,
config: LocalDatabaseConfig,
}
impl DatabasePool {
/// Create a new database connection pool
///
/// # Errors
///
/// Returns `DatabaseError` if:
/// - Connection URL is invalid
/// - Database connection fails
pub async fn new(config: LocalDatabaseConfig) -> Result<Self, DatabaseError> {
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<Postgres> {
&self.pool
}
/// Get current configuration
pub const fn config(&self) -> &LocalDatabaseConfig {
&self.config
}
/// Health check for the database connection
///
/// # Errors
///
/// Returns `DatabaseError` if:
/// - Database connection fails
/// - Query times out
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
#[allow(clippy::integer_division)]
pub fn pool_stats(&self) -> PoolStats {
PoolStats {
size: self.pool.size(),
idle: u32::try_from(self.pool.num_idle()).unwrap_or(0),
active: self.pool.size().saturating_sub(u32::try_from(self.pool.num_idle()).unwrap_or(0)),
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
#[allow(clippy::float_arithmetic)]
pub fn utilization_percentage(&self) -> f64 {
(f64::from(self.active) / f64::from(self.max_size)) * 100.0
}
/// Check if pool is healthy (not over-utilized)
pub fn is_healthy(&self) -> bool {
self.utilization_percentage() < 80.0
}
}