diff --git a/config/src/compliance_config.rs b/config/src/compliance_config.rs index 55a789010..ca8bf112d 100644 --- a/config/src/compliance_config.rs +++ b/config/src/compliance_config.rs @@ -13,7 +13,7 @@ use std::collections::HashMap; #[cfg(feature = "postgres")] use std::sync::Arc; #[cfg(feature = "postgres")] -use std::time::Duration; +use std::time::{Duration, Instant}; #[cfg(feature = "postgres")] use tokio::sync::RwLock; #[cfg(feature = "postgres")] @@ -34,6 +34,8 @@ pub struct PostgresComplianceRuleLoader { listener: Arc>>, /// Cached compliance rules by rule_id rules_cache: Arc>>, + /// Timestamp of last full cache refresh + cache_loaded_at: Arc>>, /// Cache timeout duration cache_timeout: Duration, } @@ -93,6 +95,7 @@ impl PostgresComplianceRuleLoader { pool, listener: Arc::new(RwLock::new(None)), rules_cache: Arc::new(RwLock::new(HashMap::new())), + cache_loaded_at: Arc::new(RwLock::new(None)), cache_timeout: Duration::from_secs(300), // 5 minutes }) } @@ -111,6 +114,7 @@ impl PostgresComplianceRuleLoader { pool, listener: Arc::new(RwLock::new(None)), rules_cache: Arc::new(RwLock::new(HashMap::new())), + cache_loaded_at: Arc::new(RwLock::new(None)), cache_timeout: Duration::from_secs(300), } } @@ -240,11 +244,14 @@ impl PostgresComplianceRuleLoader { .fetch_all(&self.pool) .await?; - // Update cache + // Update cache and refresh timestamp let mut cache = self.rules_cache.write().await; + cache.clear(); for rule in &rules { cache.insert(rule.rule_id.clone(), rule.clone()); } + drop(cache); + *self.cache_loaded_at.write().await = Some(Instant::now()); info!("Loaded {} active compliance rules", rules.len()); @@ -297,11 +304,16 @@ impl PostgresComplianceRuleLoader { /// # Errors /// Returns error if the operation fails pub async fn get_rule(&self, rule_id: &str) -> ConfigResult> { - // Check cache first + // Check cache first, but only if it hasn't expired { - let cache = self.rules_cache.read().await; - if let Some(rule) = cache.get(rule_id) { - return Ok(Some(rule.clone())); + let loaded_at = self.cache_loaded_at.read().await; + if let Some(at) = *loaded_at { + if at.elapsed() < self.cache_timeout { + let cache = self.rules_cache.read().await; + if let Some(rule) = cache.get(rule_id) { + return Ok(Some(rule.clone())); + } + } } } @@ -316,12 +328,17 @@ impl PostgresComplianceRuleLoader { .fetch_optional(&self.pool) .await?; - // Update cache if found + // Update cache if found and mark refresh time if let Some(ref rule_data) = rule { self.rules_cache .write() .await .insert(rule_id.to_owned(), rule_data.clone()); + // If this is the first load, initialize the cache timestamp + let mut loaded_at = self.cache_loaded_at.write().await; + if loaded_at.is_none() { + *loaded_at = Some(Instant::now()); + } } Ok(rule) @@ -370,6 +387,7 @@ impl PostgresComplianceRuleLoader { /// Clears the rule cache (forces reload on next access) pub async fn clear_cache(&self) { self.rules_cache.write().await.clear(); + *self.cache_loaded_at.write().await = None; info!("Compliance rule cache cleared"); } diff --git a/config/src/database.rs b/config/src/database.rs index b5deb9679..849660d64 100644 --- a/config/src/database.rs +++ b/config/src/database.rs @@ -7,6 +7,8 @@ use serde::{Deserialize, Serialize}; use std::time::Duration; +#[cfg(feature = "postgres")] +use std::time::Instant; #[cfg(feature = "postgres")] use sqlx::Row; @@ -188,6 +190,8 @@ pub struct PostgresSymbolConfigLoader { cache_timeout: Duration, /// PostgreSQL listener for configuration changes listener: Option, + /// Per-symbol cache: symbol -> (loaded_at, config) + symbol_cache: std::collections::HashMap, } #[cfg(feature = "postgres")] @@ -203,26 +207,36 @@ impl PostgresSymbolConfigLoader { pool, cache_timeout: Duration::from_secs(300), // 5 minutes listener: None, + symbol_cache: std::collections::HashMap::new(), }) } /// Creates a new loader with an existing connection pool. - pub const fn with_pool(pool: sqlx::PgPool) -> Self { + pub fn with_pool(pool: sqlx::PgPool) -> Self { Self { pool, cache_timeout: Duration::from_secs(300), listener: None, + symbol_cache: std::collections::HashMap::new(), } } - /// Loads a symbol configuration by symbol name. + /// Loads a symbol configuration by symbol name, using the in-memory cache + /// when the entry is fresher than `cache_timeout`. /// /// # Errors /// Returns error if the operation fails pub async fn load_symbol_config( - &self, + &mut self, symbol: &str, ) -> Result, sqlx::Error> { + // Return cached entry if fresh + if let Some((loaded_at, config)) = self.symbol_cache.get(symbol) { + if loaded_at.elapsed() < self.cache_timeout { + return Ok(Some(config.clone())); + } + } + // Simplified implementation using basic sqlx::query instead of macros let query = " SELECT @@ -289,6 +303,8 @@ impl PostgresSymbolConfigLoader { } } + // Store in cache + self.symbol_cache.insert(symbol.to_owned(), (Instant::now(), config.clone())); Ok(Some(config)) } else { Ok(None) @@ -442,6 +458,8 @@ pub struct PostgresAssetClassificationLoader { cache_timeout: Duration, /// PostgreSQL listener for configuration changes listener: Option, + /// Cached asset configurations with load timestamp + config_cache: Option<(Instant, Vec)>, } #[cfg(feature = "postgres")] @@ -457,25 +475,35 @@ impl PostgresAssetClassificationLoader { pool, cache_timeout: Duration::from_secs(300), // 5 minutes listener: None, + config_cache: None, }) } /// Creates a new loader with an existing connection pool. - pub const fn with_pool(pool: sqlx::PgPool) -> Self { + pub fn with_pool(pool: sqlx::PgPool) -> Self { Self { pool, cache_timeout: Duration::from_secs(300), listener: None, + config_cache: None, } } - /// Loads all active asset configurations ordered by priority. + /// Loads all active asset configurations ordered by priority, using + /// the in-memory cache when fresher than `cache_timeout`. /// /// # Errors /// Returns error if the operation fails pub async fn load_asset_configurations( - &self, + &mut self, ) -> Result, sqlx::Error> { + // Return cached if fresh + if let Some((loaded_at, cached)) = &self.config_cache { + if loaded_at.elapsed() < self.cache_timeout { + return Ok(cached.clone()); + } + } + let query = " SELECT id, @@ -504,6 +532,9 @@ impl PostgresAssetClassificationLoader { } } + // Store in cache + self.config_cache = Some((Instant::now(), configs.clone())); + Ok(configs) } @@ -916,8 +947,10 @@ impl PostgresAssetClassificationLoader { pub struct PostgresConfigLoader { /// Database connection pool pool: sqlx::PgPool, - /// Configuration cache timeout + /// Configuration cache timeout (callers use via `cache_timeout()`) cache_timeout: Duration, + /// Strategy config cache: strategy_id -> (loaded_at, config JSON) + strategy_cache: std::collections::HashMap, } #[cfg(feature = "postgres")] @@ -932,20 +965,26 @@ impl PostgresConfigLoader { Ok(Self { pool, cache_timeout: Duration::from_secs(300), // 5 minutes + strategy_cache: std::collections::HashMap::new(), }) } /// Creates a new loader with an existing pool - pub const fn with_pool(pool: sqlx::PgPool) -> Self { + pub fn with_pool(pool: sqlx::PgPool) -> Self { Self { pool, cache_timeout: Duration::from_secs(300), + strategy_cache: std::collections::HashMap::new(), } } /// Returns a reference to the connection pool pub const fn pool(&self) -> &sqlx::PgPool { &self.pool } + /// Returns the configured cache timeout duration + pub const fn cache_timeout(&self) -> Duration { + self.cache_timeout + } // ============================================================================ // ADAPTIVE STRATEGY CONFIGURATION METHODS // ============================================================================ @@ -979,9 +1018,16 @@ impl PostgresConfigLoader { /// # Errors /// Returns error if the operation fails pub async fn get_adaptive_strategy_config( - &self, + &mut self, strategy_id: &str, ) -> Result, sqlx::Error> { + // Return cached strategy if fresh + if let Some((loaded_at, cached)) = self.strategy_cache.get(strategy_id) { + if loaded_at.elapsed() < self.cache_timeout { + return Ok(Some(cached.clone())); + } + } + // Query main configuration let row = sqlx::query( r#" @@ -1124,6 +1170,9 @@ impl PostgresConfigLoader { "updated_at": row.try_get::, _>("updated_at")?, }); + // Store in cache + self.strategy_cache.insert(strategy_id.to_owned(), (Instant::now(), config.clone())); + Ok(Some(config)) } diff --git a/config/src/manager.rs b/config/src/manager.rs index 86195fc8f..427f6f659 100644 --- a/config/src/manager.rs +++ b/config/src/manager.rs @@ -177,7 +177,7 @@ impl ConfigManager { &self, database_pool: sqlx::PgPool, ) -> Result<(), Box> { - let loader = crate::database::PostgresAssetClassificationLoader::with_pool(database_pool); + let mut loader = crate::database::PostgresAssetClassificationLoader::with_pool(database_pool); let configs = loader.load_asset_configurations().await?; let mut manager = crate::asset_classification::AssetClassificationManager::new();