/home/jgrusewski/Work/foxhunt/common/src/database.rs

Source
//! 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)]
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
impl From<DatabaseConfig> for LocalDatabaseConfig {
    fn from(config: DatabaseConfig) -> Self {
        Self {
            url: config.url,
            pool: PoolConfig {
                max_connections: config.max_connections,
                min_connections: (config.max_connections / 5).max(2), // 20% of max, min 2
                connect_timeout_ms: config.connect_timeout.as_millis().min(100) as u64, // 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: config.query_timeout.as_micros().min(800) as u64, // 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
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,
                min_connections: (max_conn / 4).max(2), // 25% of max, min 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)]
pub struct DatabasePool {
    pool: Pool<Postgres>,
    config: LocalDatabaseConfig,
}

impl DatabasePool {
    /// Create a new database connection pool
    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
    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
    }
}

Coverage Report

Created: 2025-10-06 12:43

Line	Count	Source
1		//! Database connection utilities and configurations
2		//!
3		//! This module provides shared database connection management utilities
4		//! that can be used across all Foxhunt services.
5
6		use serde::{Deserialize, Serialize};
7		use sqlx::{Pool, Postgres};
8		use std::time::Duration;
9		use thiserror::Error;
10
11		// Import centralized database configuration
12		pub use config::database::DatabaseConfig;
13		use config::structures::BacktestingDatabaseConfig;
14
15		/// Database-specific errors
16		#[derive(Debug, Error)]
17		pub enum DatabaseError {
18		/// Connection failed - wrapper around SQLx connection errors
19		#[error("Connection failed: {0}")]
20		Connection(#[from] sqlx::Error),
21		/// Query exceeded maximum allowed execution time
22		#[error("Query timeout: operation took {actual_ms}ms, max allowed {max_ms}ms")]
23		QueryTimeout {
24		/// Actual execution time in milliseconds
25		actual_ms: u64,
26		/// Maximum allowed execution time in milliseconds
27		max_ms: u64,
28		},
29		/// Connection pool has no available connections
30		#[error("Pool exhausted: no connections available")]
31		PoolExhausted,
32		/// Database configuration is invalid or missing required parameters
33		#[error("Configuration error: {0}")]
34		Configuration(String),
35		/// Performance constraint violation detected
36		#[error("Performance violation: {0}")]
37		Performance(String),
38		}
39
40		/// Database connection configuration (local extended version)
41		#[derive(Debug, Clone, Deserialize, Serialize)]
42		pub struct LocalDatabaseConfig {
43		/// Database connection URL
44		pub url: String,
45		/// Pool configuration
46		pub pool: PoolConfig,
47		/// Performance settings
48		pub performance: PerformanceConfig,
49		}
50
51		/// Connection pool configuration
52		#[derive(Debug, Clone, Deserialize, Serialize)]
53		pub struct PoolConfig {
54		/// Maximum number of connections in the pool
55		pub max_connections: u32,
56		/// Minimum number of connections to maintain
57		pub min_connections: u32,
58		/// Connection timeout in milliseconds
59		pub connect_timeout_ms: u64,
60		/// Connection acquire timeout in milliseconds
61		pub acquire_timeout_ms: u64,
62		/// Maximum connection lifetime in seconds
63		pub max_lifetime_seconds: u64,
64		/// Idle timeout in seconds
65		pub idle_timeout_seconds: u64,
66		}
67
68		/// Performance configuration for HFT operations
69		#[derive(Debug, Clone, Deserialize, Serialize)]
70		pub struct PerformanceConfig {
71		/// Query timeout in microseconds for HFT operations
72		pub query_timeout_micros: u64,
73		/// Enable connection prewarming
74		pub enable_prewarming: bool,
75		/// Enable statement preparation
76		pub enable_prepared_statements: bool,
77		/// Enable query logging for slow queries
78		pub enable_slow_query_logging: bool,
79		/// Slow query threshold in microseconds
80		pub slow_query_threshold_micros: u64,
81		}
82
83		impl Default for LocalDatabaseConfig {
84	0	fn default() -> Self {
85	0	Self {
86	0	url: "postgresql://foxhunt:password@localhost:5432/foxhunt".to_owned(),
87	0	pool: PoolConfig::default(),
88	0	performance: PerformanceConfig::default(),
89	0	}
90	0	}
91		}
92
93		impl Default for PoolConfig {
94	0	fn default() -> Self {
95	0	Self {
96	0	max_connections: 50,
97	0	min_connections: 10,
98	0	connect_timeout_ms: 100,
99	0	acquire_timeout_ms: 50,
100	0	max_lifetime_seconds: 3600,
101	0	idle_timeout_seconds: 300,
102	0	}
103	0	}
104		}
105
106		impl Default for PerformanceConfig {
107	0	fn default() -> Self {
108	0	Self {
109	0	query_timeout_micros: 800, // <1ms for HFT operations
110	0	enable_prewarming: true,
111	0	enable_prepared_statements: true,
112	0	enable_slow_query_logging: true,
113	0	slow_query_threshold_micros: 1000, // Log queries >1ms
114	0	}
115	0	}
116		}
117
118		/// Convert from centralized config to common crate config with HFT optimizations
119		impl From<DatabaseConfig> for LocalDatabaseConfig {
120	0	fn from(config: DatabaseConfig) -> Self {
121	0	Self {
122	0	url: config.url,
123	0	pool: PoolConfig {
124	0	max_connections: config.max_connections,
125	0	min_connections: (config.max_connections / 5).max(2), // 20% of max, min 2
126	0	connect_timeout_ms: config.connect_timeout.as_millis().min(100) as u64, // Convert to ms, cap at 100ms for HFT
127	0	acquire_timeout_ms: 50, // Fast acquire for HFT
128	0	max_lifetime_seconds: 3600, // 1 hour default
129	0	idle_timeout_seconds: 300, // 5 minutes default
130	0	},
131	0	performance: PerformanceConfig {
132	0	query_timeout_micros: config.query_timeout.as_micros().min(800) as u64, // Convert to microseconds, cap at 800μs for HFT
133	0	enable_prewarming: true,
134	0	enable_prepared_statements: true,
135	0	enable_slow_query_logging: config.enable_query_logging,
136	0	slow_query_threshold_micros: 1000, // 1ms threshold
137	0	},
138	0	}
139	0	}
140		}
141
142		/// Convert from backtesting config to common crate config with backtesting optimizations
143		impl From<BacktestingDatabaseConfig> for LocalDatabaseConfig {
144	0	fn from(config: BacktestingDatabaseConfig) -> Self {
145	0	let max_conn = config.max_connections.unwrap_or(10);
146	0	Self {
147	0	url: config.database_url,
148	0	pool: PoolConfig {
149	0	max_connections: max_conn,
150	0	min_connections: (max_conn / 4).max(2), // 25% of max, min 2
151	0	connect_timeout_ms: config.acquire_timeout_ms.unwrap_or(1000), // Use acquire timeout as connection timeout
152	0	acquire_timeout_ms: 100, // Less strict for backtesting
153	0	max_lifetime_seconds: 3600, // 1 hour default
154	0	idle_timeout_seconds: 600, // 10 minutes for backtesting
155	0	},
156	0	performance: PerformanceConfig {
157	0	query_timeout_micros: 10000, // 10ms default for backtesting queries
158	0	enable_prewarming: true,
159	0	enable_prepared_statements: true,
160	0	enable_slow_query_logging: config.enable_logging.unwrap_or(false),
161	0	slow_query_threshold_micros: 5000, // 5ms threshold for backtesting
162	0	},
163	0	}
164	0	}
165		}
166
167		/// Database connection pool wrapper
168		#[derive(Debug)]
169		pub struct DatabasePool {
170		pool: Pool<Postgres>,
171		config: LocalDatabaseConfig,
172		}
173
174		impl DatabasePool {
175		/// Create a new database connection pool
176	0	pub async fn new(config: LocalDatabaseConfig) -> Result<Self, DatabaseError> {
177		use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
178
179		// Parse connection options
180	0	let mut connect_options: PgConnectOptions = config
181	0	.url
182	0	.parse()
183	0	.map_err(\|e\| DatabaseError::Configuration(format!("Invalid URL: {}", e)))?;
184
185		// Configure connection-level optimizations
186	0	connect_options = connect_options
187	0	.application_name("foxhunt-service")
188	0	.statement_cache_capacity(1000);
189
190		// Create connection pool with optimized settings
191	0	let pool = PgPoolOptions::new()
192	0	.max_connections(config.pool.max_connections)
193	0	.min_connections(config.pool.min_connections)
194	0	.acquire_timeout(Duration::from_millis(config.pool.acquire_timeout_ms))
195	0	.max_lifetime(Duration::from_secs(config.pool.max_lifetime_seconds))
196	0	.idle_timeout(Duration::from_secs(config.pool.idle_timeout_seconds))
197	0	.test_before_acquire(true)
198	0	.connect_with(connect_options)
199	0	.await
200	0	.map_err(DatabaseError::Connection)?;
201
202		// Pre-warm connections if enabled
203	0	if config.performance.enable_prewarming {
204	0	for _ in 0..config.pool.min_connections {
205	0	let _conn = pool.acquire().await.map_err(DatabaseError::Connection)?;
206	0	sqlx::query("SELECT 1")
207	0	.fetch_one(&pool)
208	0	.await
209	0	.map_err(DatabaseError::Connection)?;
210		}
211	0	}
212
213	0	Ok(Self { pool, config })
214	0	}
215
216		/// Get the underlying connection pool
217	0	pub const fn pool(&self) -> &Pool<Postgres> {
218	0	&self.pool
219	0	}
220
221		/// Get current configuration
222	0	pub const fn config(&self) -> &LocalDatabaseConfig {
223	0	&self.config
224	0	}
225
226		/// Health check for the database connection
227	0	pub async fn health_check(&self) -> Result<(), DatabaseError> {
228	0	let result = tokio::time::timeout(
229	0	Duration::from_millis(100),
230	0	sqlx::query("SELECT 1").fetch_one(&self.pool),
231	0	)
232	0	.await;
233
234	0	match result {
235	0	Ok(Ok(_)) => Ok(()),
236	0	Ok(Err(e)) => Err(DatabaseError::Connection(e)),
237	0	Err(_) => Err(DatabaseError::QueryTimeout {
238	0	actual_ms: 100,
239	0	max_ms: 100,
240	0	}),
241		}
242	0	}
243
244		/// Get connection pool statistics
245	0	pub fn pool_stats(&self) -> PoolStats {
246	0	PoolStats {
247	0	size: self.pool.size(),
248	0	idle: self.pool.num_idle() as u32,
249	0	active: self.pool.size() - self.pool.num_idle() as u32,
250	0	max_size: self.config.pool.max_connections,
251	0	}
252	0	}
253		}
254
255		/// Connection pool statistics
256		#[derive(Debug, Clone, Serialize, Deserialize)]
257		pub struct PoolStats {
258		/// Current pool size
259		pub size: u32,
260		/// Number of idle connections
261		pub idle: u32,
262		/// Number of active connections
263		pub active: u32,
264		/// Maximum pool size
265		pub max_size: u32,
266		}
267
268		impl PoolStats {
269		/// Calculate pool utilization percentage
270	0	pub fn utilization_percentage(&self) -> f64 {
271	0	(self.active as f64 / self.max_size as f64) * 100.0
272	0	}
273
274		/// Check if pool is healthy (not over-utilized)
275	0	pub fn is_healthy(&self) -> bool {
276	0	self.utilization_percentage() < 80.0
277	0	}
278		}