19742b4a5e
Complete systematic resolution of ML crate compilation errors through parallel agent deployment and comprehensive type system integration. Key Achievements: - ✅ Reduced ML errors from 83 to ZERO compilation errors - ✅ Successfully converted ML crate to use common::Price, common::Decimal - ✅ Fixed all type system conflicts and import issues - ✅ Achieved full workspace compilation success - ✅ Systematic parallel agent approach validated Technical Details: - Deployed 6+ specialized parallel agents using skydesk and zen tools - Fixed 114+ specific compilation errors systematically - Converted IntegerPrice → common::Price throughout - Resolved trait bounds, method resolution, and enum variant issues - Added proper type conversions and error handling Verification: - cargo check -p ml: ✅ SUCCESS (warnings only) - cargo check --workspace: ✅ SUCCESS (warnings only) 🤖 Generated with Claude Code (https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
1638 lines
61 KiB
Rust
1638 lines
61 KiB
Rust
//! PostgreSQL-based Configuration Database Module
|
|
//!
|
|
//! This module provides PostgreSQL-backed configuration storage with:
|
|
//! - Hot-reload via PostgreSQL NOTIFY/LISTEN using existing comprehensive schema
|
|
//! - In-memory caching with TTL for performance
|
|
//! - Type-safe configuration getters
|
|
//! - Integration with existing configuration tables and functions
|
|
//! - Configuration change notifications with atomic updates
|
|
|
|
use crate::schemas::{ModelConfig, ModelLoadRequest, ModelLoadResponse, ModelVersion};
|
|
use crate::{ConfigCategory, ConfigError, ConfigResult, ConfigSource, ConfigValue};
|
|
use anyhow::Context;
|
|
use chrono::Utc;
|
|
use serde::{Deserialize, Serialize};
|
|
use sqlx::{PgPool, Row};
|
|
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
use std::time::{Duration, Instant};
|
|
use tokio::sync::{mpsc, RwLock};
|
|
use tokio::time::interval;
|
|
use tracing::{debug, error, info, warn};
|
|
use uuid::Uuid;
|
|
|
|
/// Pool configuration for database connections
|
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
|
pub struct PoolConfig {
|
|
/// Database connection URL
|
|
pub database_url: String,
|
|
/// Minimum number of connections in the pool
|
|
pub min_connections: u32,
|
|
/// Maximum number of connections in the pool
|
|
pub max_connections: u32,
|
|
/// Maximum time to wait for a connection from the pool
|
|
pub acquire_timeout_secs: u64,
|
|
/// Maximum lifetime of a connection
|
|
pub max_lifetime_secs: u64,
|
|
/// Maximum idle time for a connection
|
|
pub idle_timeout_secs: u64,
|
|
/// Test connections before use
|
|
pub test_before_acquire: bool,
|
|
/// Enable connection health checks
|
|
pub health_check_enabled: bool,
|
|
/// Health check interval in seconds
|
|
pub health_check_interval_secs: u64,
|
|
}
|
|
|
|
impl Default for PoolConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
database_url: "postgresql://localhost:5432/database".to_string(),
|
|
min_connections: 5,
|
|
max_connections: 100,
|
|
acquire_timeout_secs: 30,
|
|
max_lifetime_secs: 1800, // 30 minutes
|
|
idle_timeout_secs: 600, // 10 minutes
|
|
test_before_acquire: true,
|
|
health_check_enabled: true,
|
|
health_check_interval_secs: 60,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Transaction configuration
|
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
|
pub struct TransactionConfig {
|
|
/// Default timeout for transactions in seconds
|
|
pub default_timeout_secs: u64,
|
|
/// Maximum number of savepoints allowed
|
|
pub max_savepoints: u32,
|
|
/// Enable automatic retry for serialization failures
|
|
pub enable_retry: bool,
|
|
/// Maximum number of retry attempts
|
|
pub max_retries: u32,
|
|
/// Base delay between retries in milliseconds
|
|
pub retry_delay_ms: u64,
|
|
}
|
|
|
|
impl Default for TransactionConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
default_timeout_secs: 30,
|
|
max_savepoints: 10,
|
|
enable_retry: true,
|
|
max_retries: 3,
|
|
retry_delay_ms: 100,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Database configuration for PostgreSQL connection with optimized pool settings
|
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
|
pub struct DatabaseConfig {
|
|
/// PostgreSQL connection URL (for backward compatibility)
|
|
pub url: String,
|
|
/// Pool configuration
|
|
pub pool: PoolConfig,
|
|
/// Transaction configuration
|
|
pub transaction: TransactionConfig,
|
|
/// Maximum number of connections in the pool (deprecated, use pool.max_connections)
|
|
pub max_connections: u32,
|
|
/// Connection timeout in seconds (deprecated, use pool.acquire_timeout_secs)
|
|
pub connect_timeout: u64,
|
|
/// Query timeout in seconds
|
|
pub query_timeout: u64,
|
|
/// Whether to validate schema on startup
|
|
pub validate_schema: bool,
|
|
/// Enable query logging
|
|
pub enable_query_logging: bool,
|
|
/// Enable metrics collection
|
|
pub enable_metrics: bool,
|
|
/// Application name for connection identification
|
|
pub application_name: String,
|
|
}
|
|
|
|
impl DatabaseConfig {
|
|
/// Create a new database configuration with the given URL
|
|
pub fn new(url: String) -> Self {
|
|
let mut pool_config = PoolConfig::default();
|
|
pool_config.database_url = url.clone();
|
|
|
|
Self {
|
|
url,
|
|
pool: pool_config,
|
|
transaction: TransactionConfig::default(),
|
|
max_connections: 20,
|
|
connect_timeout: 30,
|
|
query_timeout: 60,
|
|
validate_schema: true,
|
|
enable_query_logging: false,
|
|
enable_metrics: true,
|
|
application_name: "database-lib".to_string(),
|
|
}
|
|
}
|
|
|
|
/// Validate the database configuration
|
|
pub fn validate(&self) -> ConfigResult<()> {
|
|
if self.url.is_empty() {
|
|
return Err(ConfigError::ValidationError {
|
|
message: "Database URL cannot be empty".to_string(),
|
|
});
|
|
}
|
|
|
|
if self.application_name.is_empty() {
|
|
return Err(ConfigError::ValidationError {
|
|
message: "Application name cannot be empty".to_string(),
|
|
});
|
|
}
|
|
|
|
if self.max_connections == 0 {
|
|
return Err(ConfigError::ValidationError {
|
|
message: "Max connections must be greater than 0".to_string(),
|
|
});
|
|
}
|
|
|
|
if self.connect_timeout == 0 {
|
|
return Err(ConfigError::ValidationError {
|
|
message: "Connect timeout must be greater than 0".to_string(),
|
|
});
|
|
}
|
|
|
|
if self.query_timeout == 0 {
|
|
return Err(ConfigError::ValidationError {
|
|
message: "Query timeout must be greater than 0".to_string(),
|
|
});
|
|
}
|
|
|
|
// Validate URL format (basic check)
|
|
if !self.url.starts_with("postgresql://") && !self.url.starts_with("postgres://") {
|
|
return Err(ConfigError::ValidationError {
|
|
message: "Database URL must be a valid PostgreSQL connection string".to_string(),
|
|
});
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Create database configuration from environment variables
|
|
pub fn from_env() -> ConfigResult<Self> {
|
|
let mut url = std::env::var("DATABASE_URL").expect(
|
|
"CRITICAL SECURITY: DATABASE_URL environment variable must be set in production",
|
|
);
|
|
|
|
// SECURITY: Enforce TLS for production database connections
|
|
url = Self::enforce_database_security(&url)?;
|
|
let max_connections = std::env::var("DATABASE_MAX_CONNECTIONS")
|
|
.unwrap_or_else(|_| "20".to_string()) // Increased default for production
|
|
.parse()
|
|
.unwrap_or(20);
|
|
|
|
let connect_timeout = std::env::var("DATABASE_CONNECT_TIMEOUT")
|
|
.unwrap_or_else(|_| "30".to_string())
|
|
.parse()
|
|
.unwrap_or(30);
|
|
|
|
let query_timeout = std::env::var("DATABASE_QUERY_TIMEOUT")
|
|
.unwrap_or_else(|_| "60".to_string())
|
|
.parse()
|
|
.unwrap_or(60);
|
|
|
|
let validate_schema = std::env::var("DATABASE_VALIDATE_SCHEMA")
|
|
.unwrap_or_else(|_| "true".to_string())
|
|
.parse()
|
|
.unwrap_or(true);
|
|
|
|
let enable_query_logging = std::env::var("DATABASE_ENABLE_QUERY_LOGGING")
|
|
.unwrap_or_else(|_| "false".to_string())
|
|
.parse()
|
|
.unwrap_or(false);
|
|
|
|
let enable_metrics = std::env::var("DATABASE_ENABLE_METRICS")
|
|
.unwrap_or_else(|_| "true".to_string())
|
|
.parse()
|
|
.unwrap_or(true);
|
|
|
|
let application_name = std::env::var("DATABASE_APPLICATION_NAME")
|
|
.unwrap_or_else(|_| "foxhunt-config-loader".to_string());
|
|
|
|
// Create pool configuration
|
|
let mut pool_config = PoolConfig::default();
|
|
pool_config.database_url = url.clone();
|
|
pool_config.max_connections = max_connections;
|
|
pool_config.acquire_timeout_secs = connect_timeout;
|
|
|
|
// Create transaction configuration with defaults
|
|
let transaction_config = TransactionConfig::default();
|
|
|
|
Ok(Self {
|
|
url,
|
|
pool: pool_config,
|
|
transaction: transaction_config,
|
|
max_connections,
|
|
connect_timeout,
|
|
query_timeout,
|
|
validate_schema,
|
|
enable_query_logging,
|
|
enable_metrics,
|
|
application_name,
|
|
})
|
|
}
|
|
|
|
/// Enforce database security settings including TLS encryption
|
|
fn enforce_database_security(url: &str) -> ConfigResult<String> {
|
|
let mut secure_url = url.to_string();
|
|
|
|
// Check if this is a production environment
|
|
let is_production = std::env::var("FOXHUNT_ENV")
|
|
.unwrap_or_else(|_| "development".to_string()) == "production" ||
|
|
std::env::var("RUST_ENV")
|
|
.unwrap_or_else(|_| "development".to_string()) == "production";
|
|
|
|
// Parse the URL to check for security parameters
|
|
if secure_url.contains("sslmode=") {
|
|
// URL already has SSL configuration, validate it's secure
|
|
if is_production && (secure_url.contains("sslmode=disable") || secure_url.contains("sslmode=allow")) {
|
|
warn!("SECURITY WARNING: Insecure SSL mode detected in production DATABASE_URL");
|
|
// Force secure SSL mode in production
|
|
secure_url = secure_url.replace("sslmode=disable", "sslmode=require");
|
|
secure_url = secure_url.replace("sslmode=allow", "sslmode=require");
|
|
warn!("Forced SSL mode to 'require' for production security");
|
|
}
|
|
} else {
|
|
// No SSL configuration found, add secure defaults
|
|
let separator = if secure_url.contains('?') { "&" } else { "?" };
|
|
|
|
if is_production {
|
|
// Production: require SSL with certificate verification
|
|
secure_url.push_str(&format!("{}sslmode=require", separator));
|
|
info!("Added sslmode=require to production DATABASE_URL for security");
|
|
} else {
|
|
// Development: prefer SSL but allow fallback
|
|
secure_url.push_str(&format!("{}sslmode=prefer", separator));
|
|
info!("Added sslmode=prefer to development DATABASE_URL");
|
|
}
|
|
}
|
|
|
|
// Add additional security parameters if not present
|
|
if !secure_url.contains("connect_timeout=") {
|
|
let separator = if secure_url.contains('?') { "&" } else { "?" };
|
|
secure_url.push_str(&format!("{}connect_timeout=30", separator));
|
|
}
|
|
|
|
// Add application name for security monitoring if not present
|
|
if !secure_url.contains("application_name=") {
|
|
let separator = if secure_url.contains('?') { "&" } else { "?" };
|
|
let app_name = std::env::var("DATABASE_APPLICATION_NAME")
|
|
.unwrap_or_else(|_| "foxhunt-trading-service".to_string());
|
|
secure_url.push_str(&format!("{}application_name={}", separator, app_name));
|
|
}
|
|
|
|
if secure_url != url {
|
|
info!("Database URL enhanced with security parameters");
|
|
}
|
|
|
|
Ok(secure_url)
|
|
}
|
|
}
|
|
|
|
impl Default for DatabaseConfig {
|
|
fn default() -> Self {
|
|
let url = "postgresql://postgres:password@localhost/foxhunt".to_string();
|
|
let mut pool_config = PoolConfig::default();
|
|
pool_config.database_url = url.clone();
|
|
|
|
Self {
|
|
url,
|
|
pool: pool_config,
|
|
transaction: TransactionConfig::default(),
|
|
max_connections: 20, // Increased default pool size
|
|
connect_timeout: 30,
|
|
query_timeout: 60,
|
|
validate_schema: true,
|
|
enable_query_logging: false,
|
|
enable_metrics: true,
|
|
application_name: "foxhunt-config".to_string(),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Builder pattern methods for DatabaseConfig
|
|
impl DatabaseConfig {
|
|
/// Set application name
|
|
pub fn with_application_name(mut self, name: String) -> Self {
|
|
self.application_name = name;
|
|
self
|
|
}
|
|
|
|
/// Set query logging enabled/disabled
|
|
pub fn with_query_logging(mut self, enabled: bool) -> Self {
|
|
self.enable_query_logging = enabled;
|
|
self
|
|
}
|
|
|
|
/// Set metrics enabled/disabled
|
|
pub fn with_metrics(mut self, enabled: bool) -> Self {
|
|
self.enable_metrics = enabled;
|
|
self
|
|
}
|
|
|
|
/// Set maximum connections
|
|
pub fn with_max_connections(mut self, max: u32) -> Self {
|
|
self.max_connections = max;
|
|
self
|
|
}
|
|
|
|
/// Set connect timeout
|
|
pub fn with_connect_timeout(mut self, timeout: u64) -> Self {
|
|
self.connect_timeout = timeout;
|
|
self
|
|
}
|
|
|
|
/// Set query timeout
|
|
pub fn with_query_timeout(mut self, timeout: u64) -> Self {
|
|
self.query_timeout = timeout;
|
|
self
|
|
}
|
|
|
|
/// Set schema validation enabled/disabled
|
|
pub fn with_schema_validation(mut self, enabled: bool) -> Self {
|
|
self.validate_schema = enabled;
|
|
self
|
|
}
|
|
}
|
|
|
|
/// Cached configuration entry with TTL
|
|
#[derive(Debug, Clone)]
|
|
struct CachedConfig {
|
|
/// The configuration value
|
|
value: ConfigValue,
|
|
/// When this entry was cached
|
|
cached_at: Instant,
|
|
/// TTL for this entry
|
|
ttl: Duration,
|
|
/// Cache hit count for LRU eviction
|
|
hit_count: u64,
|
|
}
|
|
|
|
impl CachedConfig {
|
|
/// Check if this cached entry has expired
|
|
fn is_expired(&self) -> bool {
|
|
self.cached_at.elapsed() > self.ttl
|
|
}
|
|
|
|
/// Increment hit count and return the value
|
|
fn hit(&mut self) -> &ConfigValue {
|
|
self.hit_count += 1;
|
|
&self.value
|
|
}
|
|
}
|
|
|
|
/// PostgreSQL Configuration Loader with hot-reload support using existing comprehensive schema
|
|
pub struct PostgresConfigLoader {
|
|
/// PostgreSQL connection pool with optimized settings
|
|
pool: PgPool,
|
|
/// In-memory cache with TTL and hit tracking
|
|
cache: Arc<RwLock<HashMap<(ConfigCategory, String), CachedConfig>>>,
|
|
/// Default TTL for cached entries
|
|
default_ttl: Duration,
|
|
/// Channel for hot-reload notifications
|
|
reload_tx: mpsc::UnboundedSender<(ConfigCategory, String)>,
|
|
/// Receiver for hot-reload notifications (for external subscribers)
|
|
reload_rx: Arc<RwLock<Option<mpsc::UnboundedReceiver<(ConfigCategory, String)>>>>,
|
|
/// Environment for configuration loading
|
|
environment: String,
|
|
}
|
|
|
|
impl PostgresConfigLoader {
|
|
/// Create a new PostgreSQL configuration loader with connection pooling
|
|
pub async fn new(config: DatabaseConfig, default_ttl: Duration) -> ConfigResult<Self> {
|
|
// Build connection pool with proper configuration for HFT workloads
|
|
let pool = sqlx::postgres::PgPoolOptions::new()
|
|
.max_connections(config.max_connections)
|
|
.min_connections(2) // Always maintain minimum connections
|
|
// .connect_timeout(Duration::from_secs(config.connect_timeout)) // Method may not exist in this sqlx version
|
|
.acquire_timeout(Duration::from_secs(10)) // Timeout for acquiring connections
|
|
.idle_timeout(Some(Duration::from_secs(300))) // 5 minutes idle timeout
|
|
.max_lifetime(Some(Duration::from_secs(1800))) // 30 minutes max lifetime
|
|
.test_before_acquire(true) // Test connections before use
|
|
.after_connect(move |conn, _meta| {
|
|
let app_name = config.application_name.clone();
|
|
Box::pin(async move {
|
|
// Set application name and prepare connection for high performance
|
|
sqlx::query(&format!("SET application_name = '{}'", app_name))
|
|
.execute(&mut *conn)
|
|
.await?;
|
|
|
|
// Optimize for low latency
|
|
sqlx::query("SET statement_timeout = '30s'")
|
|
.execute(&mut *conn)
|
|
.await?;
|
|
|
|
sqlx::query("SET lock_timeout = '10s'")
|
|
.execute(&mut *conn)
|
|
.await?;
|
|
|
|
Ok(())
|
|
})
|
|
})
|
|
.connect(&config.url)
|
|
.await
|
|
.context("Failed to create PostgreSQL connection pool")?;
|
|
|
|
let environment =
|
|
std::env::var("FOXHUNT_ENV").unwrap_or_else(|_| "development".to_string());
|
|
|
|
let (reload_tx, reload_rx) = mpsc::unbounded_channel();
|
|
|
|
let loader = Self {
|
|
pool,
|
|
cache: Arc::new(RwLock::new(HashMap::new())),
|
|
default_ttl,
|
|
reload_tx,
|
|
reload_rx: Arc::new(RwLock::new(Some(reload_rx))),
|
|
environment,
|
|
};
|
|
|
|
// Validate configuration schema exists (uses existing comprehensive schema)
|
|
if config.validate_schema {
|
|
loader.validate_schema().await?;
|
|
}
|
|
|
|
// Start the hot-reload listener for the existing schema
|
|
loader.start_notify_listener().await?;
|
|
|
|
info!(
|
|
"PostgreSQL ConfigLoader initialized for environment '{}' with TTL {:?}, pool size: {}",
|
|
loader.environment, default_ttl, config.max_connections
|
|
);
|
|
|
|
Ok(loader)
|
|
}
|
|
|
|
/// Validate that the existing comprehensive configuration schema is available
|
|
async fn validate_schema(&self) -> ConfigResult<()> {
|
|
// Check if the main configuration tables exist (from migrations 007/008)
|
|
let tables_exist = sqlx::query_scalar::<_, bool>(
|
|
r#"
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM information_schema.tables
|
|
WHERE table_name IN ('config_settings', 'config_categories', 'config_history')
|
|
)
|
|
"#,
|
|
)
|
|
.fetch_one(&self.pool)
|
|
.await
|
|
.context("Failed to check if configuration tables exist")?;
|
|
|
|
if !tables_exist {
|
|
return Err(ConfigError::DatabaseError {
|
|
message: "Configuration tables not found. Please run migrations 007_configuration_schema.sql and 008_initial_config_data.sql".to_string(),
|
|
});
|
|
}
|
|
|
|
// Verify we can access the configuration functions
|
|
let functions_exist = sqlx::query_scalar::<_, bool>(
|
|
r#"
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM pg_proc
|
|
WHERE proname IN ('get_config_value', 'set_config_value', 'notify_config_change')
|
|
)
|
|
"#,
|
|
)
|
|
.fetch_one(&self.pool)
|
|
.await
|
|
.context("Failed to check if configuration functions exist")?;
|
|
|
|
if !functions_exist {
|
|
return Err(ConfigError::DatabaseError {
|
|
message: "Configuration functions not found. Please run migration 007_configuration_schema.sql".to_string(),
|
|
});
|
|
}
|
|
|
|
// Check for model configuration tables
|
|
let model_tables_exist = sqlx::query_scalar::<_, bool>(
|
|
"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'model_config')"
|
|
)
|
|
.fetch_one(&self.pool)
|
|
.await
|
|
.context("Failed to check if model_config table exists")?;
|
|
|
|
if !model_tables_exist {
|
|
warn!("Model configuration tables not found. Some model management features may not work.");
|
|
}
|
|
|
|
info!("Configuration schema validation successful");
|
|
Ok(())
|
|
}
|
|
|
|
/// Start the PostgreSQL NOTIFY listener for hot-reload using the existing schema
|
|
async fn start_notify_listener(&self) -> ConfigResult<()> {
|
|
let pool = self.pool.clone();
|
|
let reload_tx = self.reload_tx.clone();
|
|
let cache = self.cache.clone();
|
|
|
|
tokio::spawn(async move {
|
|
let mut retry_count = 0;
|
|
const MAX_RETRIES: u32 = 5;
|
|
|
|
loop {
|
|
let mut listener = match sqlx::postgres::PgListener::connect_with(&pool).await {
|
|
Ok(listener) => listener,
|
|
Err(e) => {
|
|
retry_count += 1;
|
|
if retry_count > MAX_RETRIES {
|
|
error!(
|
|
"Failed to create NOTIFY listener after {} retries: {}",
|
|
MAX_RETRIES, e
|
|
);
|
|
return;
|
|
}
|
|
warn!(
|
|
"Failed to create NOTIFY listener (retry {}): {}",
|
|
retry_count, e
|
|
);
|
|
tokio::time::sleep(Duration::from_secs(2_u64.pow(retry_count - 1))).await;
|
|
continue;
|
|
}
|
|
};
|
|
|
|
// Listen to the main configuration changes channel from the existing schema
|
|
if let Err(e) = listener.listen("foxhunt_config_changes").await {
|
|
error!("Failed to listen on foxhunt_config_changes channel: {}", e);
|
|
tokio::time::sleep(Duration::from_secs(1)).await;
|
|
continue;
|
|
}
|
|
|
|
// Also listen to model configuration changes (legacy support)
|
|
if let Err(e) = listener.listen("config_change").await {
|
|
warn!(
|
|
"Failed to listen on config_change channel (model configs): {}",
|
|
e
|
|
);
|
|
}
|
|
|
|
info!("NOTIFY listener started for configuration hot-reload on channels: foxhunt_config_changes, config_change");
|
|
retry_count = 0; // Reset retry count on successful connection
|
|
|
|
loop {
|
|
match listener.recv().await {
|
|
Ok(notification) => {
|
|
let channel = notification.channel();
|
|
let payload = notification.payload();
|
|
|
|
debug!("Received NOTIFY on channel {}: {}", channel, payload);
|
|
|
|
match channel {
|
|
"foxhunt_config_changes" => {
|
|
// Parse JSON payload from the existing schema notification function
|
|
if let Ok(parsed) =
|
|
serde_json::from_str::<serde_json::Value>(payload)
|
|
{
|
|
let config_key = parsed
|
|
.get("config_key")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("unknown")
|
|
.to_string();
|
|
let category_path = parsed
|
|
.get("category_path")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("system")
|
|
.to_string();
|
|
|
|
// Map category path to ConfigCategory
|
|
let category =
|
|
Self::map_category_path_to_enum(&category_path);
|
|
|
|
// Invalidate cache entry for atomic cache consistency
|
|
let cache_key = (category.clone(), config_key.clone());
|
|
{
|
|
let mut cache_guard = cache.write().await;
|
|
if cache_guard.remove(&cache_key).is_some() {
|
|
debug!(
|
|
"Invalidated cache entry for {}.{}",
|
|
category_path, config_key
|
|
);
|
|
}
|
|
}
|
|
|
|
// Send reload notification to subscribers
|
|
if let Err(e) = reload_tx.send((category, config_key)) {
|
|
error!("Failed to send reload notification: {}", e);
|
|
break;
|
|
}
|
|
} else {
|
|
warn!(
|
|
"Failed to parse foxhunt_config_changes payload: {}",
|
|
payload
|
|
);
|
|
}
|
|
}
|
|
"config_change" => {
|
|
// Handle model configuration changes (legacy format)
|
|
if let Ok(parsed) =
|
|
serde_json::from_str::<serde_json::Value>(payload)
|
|
{
|
|
let key = parsed
|
|
.get("key")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("unknown")
|
|
.to_string();
|
|
|
|
// For model configs, use MachineLearning category
|
|
let category = ConfigCategory::MachineLearning;
|
|
let cache_key = (category.clone(), key.clone());
|
|
|
|
// Invalidate cache entry for atomic cache consistency
|
|
{
|
|
let mut cache_guard = cache.write().await;
|
|
if cache_guard.remove(&cache_key).is_some() {
|
|
debug!(
|
|
"Invalidated model config cache entry for {}",
|
|
key
|
|
);
|
|
}
|
|
}
|
|
|
|
// Send reload notification to subscribers
|
|
if let Err(e) = reload_tx.send((category, key)) {
|
|
error!("Failed to send reload notification: {}", e);
|
|
break;
|
|
}
|
|
} else {
|
|
warn!("Failed to parse config_change payload: {}", payload);
|
|
}
|
|
}
|
|
_ => {
|
|
warn!("Unknown notification channel: {}", channel);
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
error!("Error receiving NOTIFY: {}", e);
|
|
break; // Break inner loop to reconnect
|
|
}
|
|
}
|
|
}
|
|
|
|
// Connection lost, wait before retrying
|
|
warn!("NOTIFY listener connection lost, retrying in 5 seconds...");
|
|
tokio::time::sleep(Duration::from_secs(5)).await;
|
|
}
|
|
});
|
|
|
|
// Start cache cleanup task with advanced LRU eviction
|
|
self.start_cache_cleanup().await;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Map category path from database to ConfigCategory enum
|
|
fn map_category_path_to_enum(category_path: &str) -> ConfigCategory {
|
|
match category_path.split('.').next().unwrap_or("system") {
|
|
"trading" => ConfigCategory::Trading,
|
|
"risk" => ConfigCategory::Risk,
|
|
"ml" => ConfigCategory::MachineLearning,
|
|
"performance" => ConfigCategory::Performance,
|
|
"security" => ConfigCategory::Security,
|
|
"storage" => ConfigCategory::Storage,
|
|
"database" | "system" => ConfigCategory::Environment,
|
|
_ => ConfigCategory::Environment,
|
|
}
|
|
}
|
|
|
|
/// Start background task to clean up expired cache entries with LRU eviction
|
|
async fn start_cache_cleanup(&self) {
|
|
let cache = self.cache.clone();
|
|
let cleanup_interval = self.default_ttl / 4; // Clean up 4x more frequently than TTL
|
|
|
|
tokio::spawn(async move {
|
|
let mut interval = interval(cleanup_interval);
|
|
|
|
loop {
|
|
interval.tick().await;
|
|
|
|
let mut cache_guard = cache.write().await;
|
|
let initial_size = cache_guard.len();
|
|
|
|
// Remove expired entries
|
|
cache_guard.retain(|_, cached| !cached.is_expired());
|
|
|
|
// If cache is too large, evict least recently used entries
|
|
const MAX_CACHE_SIZE: usize = 1000;
|
|
if cache_guard.len() > MAX_CACHE_SIZE {
|
|
let mut entries: Vec<_> = cache_guard
|
|
.iter()
|
|
.map(|(k, v)| (k.clone(), v.hit_count))
|
|
.collect();
|
|
entries.sort_by_key(|(_, hit_count)| *hit_count);
|
|
|
|
let to_remove = cache_guard.len() - (MAX_CACHE_SIZE * 3 / 4); // Remove 25% when over limit
|
|
for (key, _) in entries.into_iter().take(to_remove) {
|
|
cache_guard.remove(&key);
|
|
}
|
|
}
|
|
|
|
let final_size = cache_guard.len();
|
|
if initial_size != final_size {
|
|
debug!(
|
|
"Cache cleanup: removed {} entries (expired + LRU eviction), {} entries remaining",
|
|
initial_size - final_size,
|
|
final_size
|
|
);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Get a configuration value with caching using the existing comprehensive schema
|
|
pub async fn get_config<T>(
|
|
&self,
|
|
category: ConfigCategory,
|
|
key: &str,
|
|
) -> ConfigResult<Option<T>>
|
|
where
|
|
T: for<'de> Deserialize<'de>,
|
|
{
|
|
let cache_key = (category.clone(), key.to_string());
|
|
|
|
// Check cache first with hit tracking
|
|
{
|
|
let mut cache_guard = self.cache.write().await;
|
|
if let Some(cached) = cache_guard.get_mut(&cache_key) {
|
|
if !cached.is_expired() {
|
|
debug!(
|
|
"Cache hit for {}.{} (hits: {})",
|
|
category.table_name(),
|
|
key,
|
|
cached.hit_count + 1
|
|
);
|
|
let value = cached.hit().value.clone();
|
|
return Ok(Some(serde_json::from_value(value)?));
|
|
}
|
|
}
|
|
}
|
|
|
|
// Cache miss or expired - fetch from database using the existing comprehensive schema
|
|
debug!(
|
|
"Cache miss for {}.{}, fetching from database",
|
|
category.table_name(),
|
|
key
|
|
);
|
|
|
|
// Use the get_config_value function from the existing schema for proper inheritance
|
|
let sql = "SELECT get_config_value($1, $2) as config_value";
|
|
let row = sqlx::query(sql)
|
|
.bind(key)
|
|
.bind(&self.environment)
|
|
.fetch_optional(&self.pool)
|
|
.await
|
|
.with_context(|| format!("Failed to fetch config {}.{}", category.table_name(), key))?;
|
|
|
|
if let Some(row) = row {
|
|
let config_value_json: Option<serde_json::Value> = row.try_get("config_value")?;
|
|
|
|
if let Some(value) = config_value_json {
|
|
// Also get metadata from config_settings table for complete ConfigValue
|
|
let metadata_sql = r#"
|
|
SELECT config_key, category_path, description, updated_at, is_active, hot_reload
|
|
FROM config_settings
|
|
WHERE config_key = $1 AND environment = $2 AND is_active = TRUE
|
|
ORDER BY updated_at DESC LIMIT 1
|
|
"#;
|
|
|
|
let metadata_row = sqlx::query(metadata_sql)
|
|
.bind(key)
|
|
.bind(&self.environment)
|
|
.fetch_optional(&self.pool)
|
|
.await
|
|
.with_context(|| format!("Failed to fetch metadata for config {}", key))?;
|
|
|
|
let config_value = if let Some(meta) = metadata_row {
|
|
ConfigValue {
|
|
key: meta.try_get("config_key")?,
|
|
value: value.clone(),
|
|
category: category.clone(),
|
|
environment: self.environment.clone(),
|
|
updated_at: meta.try_get("updated_at")?,
|
|
description: meta.try_get("description")?,
|
|
is_active: meta.try_get("is_active")?,
|
|
source: ConfigSource::Database,
|
|
}
|
|
} else {
|
|
// Fallback for cases where metadata is not available
|
|
ConfigValue {
|
|
key: key.to_string(),
|
|
value: value.clone(),
|
|
category: category.clone(),
|
|
environment: self.environment.clone(),
|
|
updated_at: Utc::now(),
|
|
description: None,
|
|
is_active: true,
|
|
source: ConfigSource::Database,
|
|
}
|
|
};
|
|
|
|
// Cache the result with hit tracking
|
|
let cached = CachedConfig {
|
|
value: config_value.clone(),
|
|
cached_at: Instant::now(),
|
|
ttl: self.default_ttl,
|
|
hit_count: 0,
|
|
};
|
|
|
|
{
|
|
let mut cache_guard = self.cache.write().await;
|
|
cache_guard.insert(cache_key, cached);
|
|
}
|
|
|
|
Ok(Some(serde_json::from_value(value)?))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
/// Set a configuration value using the existing comprehensive schema with atomic updates
|
|
pub async fn set_config<T>(
|
|
&self,
|
|
category: ConfigCategory,
|
|
key: &str,
|
|
value: &T,
|
|
description: Option<&str>,
|
|
) -> ConfigResult<()>
|
|
where
|
|
T: Serialize,
|
|
{
|
|
self.set_config_for_environment(category, key, value, &self.environment, description)
|
|
.await
|
|
}
|
|
|
|
/// Set a configuration value for a specific environment using atomic updates
|
|
pub async fn set_config_for_environment<T>(
|
|
&self,
|
|
category: ConfigCategory,
|
|
key: &str,
|
|
value: &T,
|
|
environment: &str,
|
|
description: Option<&str>,
|
|
) -> ConfigResult<()>
|
|
where
|
|
T: Serialize,
|
|
{
|
|
let json_value = serde_json::to_value(value)?;
|
|
|
|
// Use the set_config_value function from the existing schema for proper history tracking
|
|
let sql = "SELECT set_config_value($1, $2, $3, $4, $5) as success";
|
|
|
|
let result = sqlx::query(sql)
|
|
.bind(key)
|
|
.bind(&json_value)
|
|
.bind(environment)
|
|
.bind("config-api") // changed_by
|
|
.bind(description.unwrap_or("Updated via ConfigLoader API")) // change_reason
|
|
.fetch_one(&self.pool)
|
|
.await
|
|
.with_context(|| format!("Failed to set config {} for {}", key, environment))?;
|
|
|
|
let success: Option<bool> = result.try_get("success")?;
|
|
|
|
if success == Some(true) {
|
|
// Invalidate cache entry atomically
|
|
let cache_key = (category, key.to_string());
|
|
{
|
|
let mut cache_guard = self.cache.write().await;
|
|
cache_guard.remove(&cache_key);
|
|
}
|
|
|
|
info!(
|
|
"Updated configuration {} for {} using schema function with atomic update",
|
|
key, environment
|
|
);
|
|
Ok(())
|
|
} else {
|
|
// If the function returns false, the setting doesn't exist, try to insert it
|
|
self.create_config_setting(category, key, value, environment, description)
|
|
.await
|
|
}
|
|
}
|
|
|
|
/// Create a new configuration setting in the comprehensive schema
|
|
async fn create_config_setting<T>(
|
|
&self,
|
|
category: ConfigCategory,
|
|
key: &str,
|
|
value: &T,
|
|
environment: &str,
|
|
description: Option<&str>,
|
|
) -> ConfigResult<()>
|
|
where
|
|
T: Serialize,
|
|
{
|
|
let json_value = serde_json::to_value(value)?;
|
|
|
|
// Map ConfigCategory to category path for the comprehensive schema
|
|
let category_path = match category {
|
|
ConfigCategory::Trading => "trading",
|
|
ConfigCategory::Risk => "risk",
|
|
ConfigCategory::MarketData => "trading.market_data",
|
|
ConfigCategory::MachineLearning => "ml",
|
|
ConfigCategory::Brokers => "trading.brokers",
|
|
ConfigCategory::Performance => "performance",
|
|
ConfigCategory::Security => "security",
|
|
ConfigCategory::Environment => "system",
|
|
ConfigCategory::Storage => "storage",
|
|
};
|
|
|
|
// Get category_id for the comprehensive schema
|
|
let category_id_sql = "SELECT id FROM config_categories WHERE category_path = $1 LIMIT 1";
|
|
let category_row = sqlx::query(category_id_sql)
|
|
.bind(category_path)
|
|
.fetch_optional(&self.pool)
|
|
.await
|
|
.context("Failed to fetch category ID")?;
|
|
|
|
let category_id: i32 = if let Some(row) = category_row {
|
|
row.try_get("id")?
|
|
} else {
|
|
return Err(ConfigError::ValidationError {
|
|
message: format!(
|
|
"Category '{}' not found in config_categories",
|
|
category_path
|
|
),
|
|
});
|
|
};
|
|
|
|
// Insert new configuration setting with atomic operation
|
|
let insert_sql = r#"
|
|
INSERT INTO config_settings
|
|
(config_key, category_id, category_path, config_value, value_type, environment, description, hot_reload, version)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, true, 1)
|
|
ON CONFLICT (config_key, environment)
|
|
DO UPDATE SET
|
|
config_value = $4,
|
|
description = $7,
|
|
updated_at = NOW(),
|
|
updated_by = CURRENT_USER,
|
|
version = config_settings.version + 1
|
|
"#;
|
|
|
|
// Determine value type for validation
|
|
let value_type = match &json_value {
|
|
serde_json::Value::String(_) => "string",
|
|
serde_json::Value::Number(_) => "number",
|
|
serde_json::Value::Bool(_) => "boolean",
|
|
serde_json::Value::Array(_) => "array",
|
|
serde_json::Value::Object(_) => "object",
|
|
serde_json::Value::Null => "null",
|
|
};
|
|
|
|
sqlx::query(insert_sql)
|
|
.bind(key)
|
|
.bind(category_id)
|
|
.bind(category_path)
|
|
.bind(&json_value)
|
|
.bind(value_type)
|
|
.bind(environment)
|
|
.bind(description.unwrap_or(""))
|
|
.execute(&self.pool)
|
|
.await
|
|
.with_context(|| {
|
|
format!(
|
|
"Failed to insert config setting {} for {}",
|
|
key, environment
|
|
)
|
|
})?;
|
|
|
|
info!(
|
|
"Created new configuration setting {} in category {} for {}",
|
|
key, category_path, environment
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
/// Get all configurations for a category using the comprehensive schema
|
|
pub async fn get_category_configs(
|
|
&self,
|
|
category: ConfigCategory,
|
|
) -> ConfigResult<Vec<ConfigValue>> {
|
|
self.get_category_configs_for_environment(category, &self.environment)
|
|
.await
|
|
}
|
|
|
|
/// Get all configurations for a category and environment using the comprehensive schema
|
|
pub async fn get_category_configs_for_environment(
|
|
&self,
|
|
category: ConfigCategory,
|
|
environment: &str,
|
|
) -> ConfigResult<Vec<ConfigValue>> {
|
|
let category_path = match category {
|
|
ConfigCategory::Trading => "trading%",
|
|
ConfigCategory::Risk => "risk%",
|
|
ConfigCategory::MarketData => "trading.market_data%",
|
|
ConfigCategory::MachineLearning => "ml%",
|
|
ConfigCategory::Brokers => "trading.brokers%",
|
|
ConfigCategory::Performance => "performance%",
|
|
ConfigCategory::Security => "security%",
|
|
ConfigCategory::Environment => "system%",
|
|
ConfigCategory::Storage => "storage%",
|
|
};
|
|
|
|
let sql = r#"
|
|
SELECT config_key, config_value, category_path, updated_at, description, is_active
|
|
FROM config_settings
|
|
WHERE category_path LIKE $1 AND environment = $2 AND is_active = TRUE
|
|
ORDER BY config_key
|
|
"#;
|
|
|
|
let rows = sqlx::query(sql)
|
|
.bind(category_path)
|
|
.bind(environment)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.with_context(|| {
|
|
format!(
|
|
"Failed to fetch configs for category path {}",
|
|
category_path
|
|
)
|
|
})?;
|
|
|
|
let mut configs = Vec::new();
|
|
for row in rows {
|
|
configs.push(ConfigValue {
|
|
key: row.try_get("config_key")?,
|
|
value: row.try_get("config_value")?,
|
|
category: category.clone(),
|
|
environment: environment.to_string(),
|
|
updated_at: row.try_get("updated_at")?,
|
|
description: row.try_get("description")?,
|
|
is_active: row.try_get("is_active")?,
|
|
source: ConfigSource::Database,
|
|
});
|
|
}
|
|
|
|
Ok(configs)
|
|
}
|
|
|
|
/// Subscribe to configuration changes (returns receiver for hot-reload notifications)
|
|
pub async fn subscribe_to_changes(
|
|
&self,
|
|
) -> ConfigResult<mpsc::UnboundedReceiver<(ConfigCategory, String)>> {
|
|
let mut reload_rx_guard = self.reload_rx.write().await;
|
|
reload_rx_guard
|
|
.take()
|
|
.ok_or_else(|| ConfigError::ValidationError {
|
|
message: "Configuration change subscription already taken".to_string(),
|
|
})
|
|
}
|
|
|
|
/// Get cache statistics including hit ratios
|
|
pub async fn cache_stats(&self) -> (usize, usize, u64, f64) {
|
|
let cache_guard = self.cache.read().await;
|
|
let total = cache_guard.len();
|
|
let expired = cache_guard.values().filter(|c| c.is_expired()).count();
|
|
let total_hits: u64 = cache_guard.values().map(|c| c.hit_count).sum();
|
|
let hit_ratio = if total > 0 {
|
|
total_hits as f64 / total as f64
|
|
} else {
|
|
0.0
|
|
};
|
|
(total, expired, total_hits, hit_ratio)
|
|
}
|
|
|
|
/// Clear the entire cache
|
|
pub async fn clear_cache(&self) {
|
|
let mut cache_guard = self.cache.write().await;
|
|
let size = cache_guard.len();
|
|
cache_guard.clear();
|
|
info!("Cleared {} entries from configuration cache", size);
|
|
}
|
|
|
|
/// Get current environment
|
|
pub fn environment(&self) -> &str {
|
|
&self.environment
|
|
}
|
|
|
|
/// Test database connection and schema
|
|
pub async fn test_connection(&self) -> ConfigResult<()> {
|
|
// Test basic connectivity
|
|
sqlx::query("SELECT 1")
|
|
.fetch_one(&self.pool)
|
|
.await
|
|
.context("Failed to test database connection")?;
|
|
|
|
// Test configuration schema functions
|
|
sqlx::query("SELECT get_config_value('system.test', $1)")
|
|
.bind(&self.environment)
|
|
.fetch_one(&self.pool)
|
|
.await
|
|
.context("Failed to test configuration schema functions")?;
|
|
|
|
info!("Database connection and schema test successful");
|
|
Ok(())
|
|
}
|
|
|
|
/// Get database connection pool for direct access
|
|
pub fn get_pool(&self) -> &PgPool {
|
|
&self.pool
|
|
}
|
|
}
|
|
|
|
/// Type-safe configuration getters for common configuration parameters
|
|
impl PostgresConfigLoader {
|
|
/// Get trading configuration
|
|
pub async fn get_trading_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::Trading, key).await
|
|
}
|
|
|
|
/// Get risk configuration
|
|
pub async fn get_risk_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::Risk, key).await
|
|
}
|
|
|
|
/// Get market data configuration
|
|
pub async fn get_market_data_config(
|
|
&self,
|
|
key: &str,
|
|
) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::MarketData, key).await
|
|
}
|
|
|
|
/// Get machine learning configuration
|
|
pub async fn get_ml_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::MachineLearning, key).await
|
|
}
|
|
|
|
/// Get broker configuration
|
|
pub async fn get_broker_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::Brokers, key).await
|
|
}
|
|
|
|
/// Get performance configuration
|
|
pub async fn get_performance_config(
|
|
&self,
|
|
key: &str,
|
|
) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::Performance, key).await
|
|
}
|
|
|
|
/// Get security configuration
|
|
pub async fn get_security_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::Security, key).await
|
|
}
|
|
|
|
/// Get environment configuration
|
|
pub async fn get_environment_config(
|
|
&self,
|
|
key: &str,
|
|
) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::Environment, key).await
|
|
}
|
|
|
|
/// Get storage configuration
|
|
pub async fn get_storage_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
|
|
self.get_config(ConfigCategory::Storage, key).await
|
|
}
|
|
}
|
|
|
|
/// Model management methods for configuration loader with atomic operations
|
|
impl PostgresConfigLoader {
|
|
/// Get model configuration by name
|
|
pub async fn get_model_config(&self, model_name: &str) -> ConfigResult<Option<ModelConfig>> {
|
|
let sql = r#"
|
|
SELECT id, name, version, s3_path, cache_path, metadata, is_active, created_at, updated_at
|
|
FROM model_config
|
|
WHERE name = $1 AND is_active = true
|
|
ORDER BY updated_at DESC
|
|
LIMIT 1
|
|
"#;
|
|
|
|
let row = sqlx::query(sql)
|
|
.bind(model_name)
|
|
.fetch_optional(&self.pool)
|
|
.await
|
|
.context("Failed to fetch model configuration")?;
|
|
|
|
if let Some(row) = row {
|
|
Ok(Some(ModelConfig {
|
|
id: row.try_get("id")?,
|
|
name: row.try_get("name")?,
|
|
version: row.try_get("version")?,
|
|
s3_path: row.try_get("s3_path")?,
|
|
cache_path: row.try_get("cache_path")?,
|
|
metadata: row.try_get("metadata")?,
|
|
is_active: row.try_get("is_active")?,
|
|
created_at: row.try_get("created_at")?,
|
|
updated_at: row.try_get("updated_at")?,
|
|
}))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
/// Get model configuration by name and version
|
|
pub async fn get_model_config_version(
|
|
&self,
|
|
model_name: &str,
|
|
version: &str,
|
|
) -> ConfigResult<Option<ModelConfig>> {
|
|
let sql = r#"
|
|
SELECT id, name, version, s3_path, cache_path, metadata, is_active, created_at, updated_at
|
|
FROM model_config
|
|
WHERE name = $1 AND version = $2 AND is_active = true
|
|
"#;
|
|
|
|
let row = sqlx::query(sql)
|
|
.bind(model_name)
|
|
.bind(version)
|
|
.fetch_optional(&self.pool)
|
|
.await
|
|
.context("Failed to fetch model configuration")?;
|
|
|
|
if let Some(row) = row {
|
|
Ok(Some(ModelConfig {
|
|
id: row.try_get("id")?,
|
|
name: row.try_get("name")?,
|
|
version: row.try_get("version")?,
|
|
s3_path: row.try_get("s3_path")?,
|
|
cache_path: row.try_get("cache_path")?,
|
|
metadata: row.try_get("metadata")?,
|
|
is_active: row.try_get("is_active")?,
|
|
created_at: row.try_get("created_at")?,
|
|
updated_at: row.try_get("updated_at")?,
|
|
}))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
/// List all model versions for a given model
|
|
pub async fn list_model_versions(
|
|
&self,
|
|
model_config_id: Uuid,
|
|
) -> ConfigResult<Vec<ModelVersion>> {
|
|
let sql = r#"
|
|
SELECT id, model_config_id, version, s3_path, cache_path, checksum, size_bytes,
|
|
performance_metrics, training_metadata, is_current, created_at, updated_at
|
|
FROM model_versions
|
|
WHERE model_config_id = $1
|
|
ORDER BY created_at DESC
|
|
"#;
|
|
|
|
let rows = sqlx::query(sql)
|
|
.bind(model_config_id)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to fetch model versions")?;
|
|
|
|
let mut versions = Vec::new();
|
|
for row in rows {
|
|
versions.push(ModelVersion {
|
|
id: row.try_get("id")?,
|
|
model_config_id: row.try_get("model_config_id")?,
|
|
version: row.try_get("version")?,
|
|
s3_path: row.try_get("s3_path")?,
|
|
cache_path: row.try_get("cache_path")?,
|
|
checksum: row.try_get("checksum")?,
|
|
size_bytes: row.try_get("size_bytes")?,
|
|
performance_metrics: row.try_get("performance_metrics")?,
|
|
training_metadata: row.try_get("training_metadata")?,
|
|
is_current: row.try_get("is_current")?,
|
|
created_at: row.try_get("created_at")?,
|
|
updated_at: row.try_get("updated_at")?,
|
|
});
|
|
}
|
|
|
|
Ok(versions)
|
|
}
|
|
|
|
/// Get all active model configurations
|
|
pub async fn list_active_models(&self) -> ConfigResult<Vec<ModelConfig>> {
|
|
let sql = r#"
|
|
SELECT id, name, version, s3_path, cache_path, metadata, is_active, created_at, updated_at
|
|
FROM model_config
|
|
WHERE is_active = true
|
|
ORDER BY name, updated_at DESC
|
|
"#;
|
|
|
|
let rows = sqlx::query(sql)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to fetch active models")?;
|
|
|
|
let mut models = Vec::new();
|
|
for row in rows {
|
|
models.push(ModelConfig {
|
|
id: row.try_get("id")?,
|
|
name: row.try_get("name")?,
|
|
version: row.try_get("version")?,
|
|
s3_path: row.try_get("s3_path")?,
|
|
cache_path: row.try_get("cache_path")?,
|
|
metadata: row.try_get("metadata")?,
|
|
is_active: row.try_get("is_active")?,
|
|
created_at: row.try_get("created_at")?,
|
|
updated_at: row.try_get("updated_at")?,
|
|
});
|
|
}
|
|
|
|
Ok(models)
|
|
}
|
|
|
|
/// Set model configuration active status with atomic update
|
|
pub async fn set_model_active(
|
|
&self,
|
|
model_name: &str,
|
|
version: &str,
|
|
is_active: bool,
|
|
) -> ConfigResult<()> {
|
|
let sql = r#"
|
|
UPDATE model_config
|
|
SET is_active = $3, updated_at = NOW()
|
|
WHERE name = $1 AND version = $2
|
|
"#;
|
|
|
|
let result = sqlx::query(sql)
|
|
.bind(model_name)
|
|
.bind(version)
|
|
.bind(is_active)
|
|
.execute(&self.pool)
|
|
.await
|
|
.context("Failed to update model active status")?;
|
|
|
|
if result.rows_affected() == 0 {
|
|
return Err(ConfigError::NotFound {
|
|
key: format!("{}-{}", model_name, version),
|
|
});
|
|
}
|
|
|
|
info!(
|
|
"Set model {}/{} active status to {} with atomic update",
|
|
model_name, version, is_active
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Create or update model configuration with atomic operation
|
|
pub async fn upsert_model_config(&self, config: &ModelConfig) -> ConfigResult<()> {
|
|
let sql = r#"
|
|
INSERT INTO model_config (id, name, version, s3_path, cache_path, metadata, is_active, updated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
|
|
ON CONFLICT (name, version)
|
|
DO UPDATE SET
|
|
s3_path = $4,
|
|
cache_path = $5,
|
|
metadata = $6,
|
|
is_active = $7,
|
|
updated_at = NOW()
|
|
"#;
|
|
|
|
sqlx::query(sql)
|
|
.bind(&config.id)
|
|
.bind(&config.name)
|
|
.bind(&config.version)
|
|
.bind(&config.s3_path)
|
|
.bind(&config.cache_path)
|
|
.bind(&config.metadata)
|
|
.bind(config.is_active)
|
|
.execute(&self.pool)
|
|
.await
|
|
.context("Failed to upsert model configuration")?;
|
|
|
|
info!(
|
|
"Upserted model configuration for {}/{} with atomic operation",
|
|
config.name, config.version
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
/// Create or update model version with atomic operation
|
|
pub async fn upsert_model_version(&self, version: &ModelVersion) -> ConfigResult<()> {
|
|
let sql = r#"
|
|
INSERT INTO model_versions
|
|
(id, model_config_id, version, s3_path, cache_path, checksum, size_bytes,
|
|
performance_metrics, training_metadata, is_current, updated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW())
|
|
ON CONFLICT (model_config_id, version)
|
|
DO UPDATE SET
|
|
s3_path = $4,
|
|
cache_path = $5,
|
|
checksum = $6,
|
|
size_bytes = $7,
|
|
performance_metrics = $8,
|
|
training_metadata = $9,
|
|
is_current = $10,
|
|
updated_at = NOW()
|
|
"#;
|
|
|
|
sqlx::query(sql)
|
|
.bind(&version.id)
|
|
.bind(&version.model_config_id)
|
|
.bind(&version.version)
|
|
.bind(&version.s3_path)
|
|
.bind(&version.cache_path)
|
|
.bind(&version.checksum)
|
|
.bind(version.size_bytes)
|
|
.bind(&version.performance_metrics)
|
|
.bind(&version.training_metadata)
|
|
.bind(version.is_current)
|
|
.execute(&self.pool)
|
|
.await
|
|
.context("Failed to upsert model version")?;
|
|
|
|
info!(
|
|
"Upserted model version {} for config {} with atomic operation",
|
|
version.version, version.model_config_id
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
/// Handle model load request with caching and performance optimizations
|
|
pub async fn handle_model_load_request(
|
|
&self,
|
|
request: &ModelLoadRequest,
|
|
) -> ConfigResult<ModelLoadResponse> {
|
|
let start_time = std::time::Instant::now();
|
|
|
|
let model_config = if let Some(version) = &request.version {
|
|
self.get_model_config_version(&request.model_name, version)
|
|
.await?
|
|
} else {
|
|
self.get_model_config(&request.model_name).await?
|
|
};
|
|
|
|
match model_config {
|
|
Some(config) if config.is_active => {
|
|
// Check if cache exists and request doesn't force reload
|
|
if !request.force_reload && config.cache_path.is_some() {
|
|
let cache_path = config.cache_path.clone();
|
|
return Ok(ModelLoadResponse {
|
|
success: true,
|
|
model_config: Some(config),
|
|
cache_path,
|
|
error: None,
|
|
load_time_ms: start_time.elapsed().as_millis() as u64,
|
|
});
|
|
}
|
|
|
|
// For cache_only requests, fail if no cache available
|
|
if request.cache_only && config.cache_path.is_none() {
|
|
return Ok(ModelLoadResponse {
|
|
success: false,
|
|
model_config: Some(config),
|
|
cache_path: None,
|
|
error: Some("Model not cached and cache_only requested".to_string()),
|
|
load_time_ms: start_time.elapsed().as_millis() as u64,
|
|
});
|
|
}
|
|
|
|
Ok(ModelLoadResponse {
|
|
success: true,
|
|
model_config: Some(config.clone()),
|
|
cache_path: config.cache_path,
|
|
error: None,
|
|
load_time_ms: start_time.elapsed().as_millis() as u64,
|
|
})
|
|
}
|
|
Some(_) => Ok(ModelLoadResponse {
|
|
success: false,
|
|
model_config: None,
|
|
cache_path: None,
|
|
error: Some(format!("Model {} is not active", request.model_name)),
|
|
load_time_ms: start_time.elapsed().as_millis() as u64,
|
|
}),
|
|
None => Ok(ModelLoadResponse {
|
|
success: false,
|
|
model_config: None,
|
|
cache_path: None,
|
|
error: Some(format!("Model {} not found", request.model_name)),
|
|
load_time_ms: start_time.elapsed().as_millis() as u64,
|
|
}),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use std::time::Duration;
|
|
|
|
#[test]
|
|
fn test_database_config_new() {
|
|
let url = "postgresql://localhost:5432/test".to_string();
|
|
let config = DatabaseConfig::new(url.clone());
|
|
assert_eq!(config.url, url);
|
|
assert_eq!(config.max_connections, 20);
|
|
assert_eq!(config.application_name, "database-lib");
|
|
}
|
|
|
|
#[test]
|
|
fn test_database_config_validation() {
|
|
let mut config = DatabaseConfig::default();
|
|
assert!(config.validate().is_ok());
|
|
|
|
// Test empty URL
|
|
config.url = String::new();
|
|
assert!(config.validate().is_err());
|
|
|
|
// Test empty application name
|
|
config.url = "postgresql://localhost/test".to_string();
|
|
config.application_name = String::new();
|
|
assert!(config.validate().is_err());
|
|
|
|
// Test zero connections
|
|
config.application_name = "test".to_string();
|
|
config.max_connections = 0;
|
|
assert!(config.validate().is_err());
|
|
|
|
// Test invalid URL format
|
|
config.max_connections = 10;
|
|
config.url = "invalid://url".to_string();
|
|
assert!(config.validate().is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn test_database_config_builder() {
|
|
let config = DatabaseConfig::default()
|
|
.with_application_name("test-app".to_string())
|
|
.with_query_logging(true)
|
|
.with_metrics(false)
|
|
.with_max_connections(50);
|
|
|
|
assert_eq!(config.application_name, "test-app");
|
|
assert!(config.enable_query_logging);
|
|
assert!(!config.enable_metrics);
|
|
assert_eq!(config.max_connections, 50);
|
|
}
|
|
|
|
#[test]
|
|
fn test_database_config_from_env() {
|
|
std::env::set_var("DATABASE_URL", "postgresql://test:test@localhost/test");
|
|
std::env::set_var("DATABASE_MAX_CONNECTIONS", "25");
|
|
std::env::set_var("DATABASE_CONNECT_TIMEOUT", "15");
|
|
|
|
let config = DatabaseConfig::from_env().unwrap();
|
|
assert_eq!(config.max_connections, 25);
|
|
assert_eq!(config.connect_timeout, 15);
|
|
|
|
std::env::remove_var("DATABASE_URL");
|
|
std::env::remove_var("DATABASE_MAX_CONNECTIONS");
|
|
std::env::remove_var("DATABASE_CONNECT_TIMEOUT");
|
|
}
|
|
|
|
#[test]
|
|
fn test_cached_config_expiry_and_hits() {
|
|
let config_value = ConfigValue {
|
|
key: "test".to_string(),
|
|
value: serde_json::json!("test_value"),
|
|
category: ConfigCategory::Trading,
|
|
environment: "test".to_string(),
|
|
updated_at: Utc::now(),
|
|
description: None,
|
|
is_active: true,
|
|
source: ConfigSource::Database,
|
|
};
|
|
|
|
let cached = CachedConfig {
|
|
value: config_value.clone(),
|
|
cached_at: Instant::now() - Duration::from_secs(10),
|
|
ttl: Duration::from_secs(5),
|
|
hit_count: 0,
|
|
};
|
|
|
|
assert!(cached.is_expired());
|
|
|
|
let mut fresh_cached = CachedConfig {
|
|
value: config_value,
|
|
cached_at: Instant::now(),
|
|
ttl: Duration::from_secs(60),
|
|
hit_count: 0,
|
|
};
|
|
|
|
assert!(!fresh_cached.is_expired());
|
|
|
|
// Test hit counting
|
|
fresh_cached.hit();
|
|
assert_eq!(fresh_cached.hit_count, 1);
|
|
fresh_cached.hit();
|
|
assert_eq!(fresh_cached.hit_count, 2);
|
|
}
|
|
|
|
#[test]
|
|
fn test_category_path_mapping() {
|
|
assert_eq!(
|
|
PostgresConfigLoader::map_category_path_to_enum("trading.order_management"),
|
|
ConfigCategory::Trading
|
|
);
|
|
assert_eq!(
|
|
PostgresConfigLoader::map_category_path_to_enum("ml.models"),
|
|
ConfigCategory::MachineLearning
|
|
);
|
|
assert_eq!(
|
|
PostgresConfigLoader::map_category_path_to_enum("system.logging"),
|
|
ConfigCategory::Environment
|
|
);
|
|
}
|
|
}
|