diff --git a/crates/superposition_provider/src/data_source.rs b/crates/superposition_provider/src/data_source.rs index 76475a225..6fbf00649 100644 --- a/crates/superposition_provider/src/data_source.rs +++ b/crates/superposition_provider/src/data_source.rs @@ -107,6 +107,18 @@ pub trait SuperpositionDataSource: Send + Sync { .await } + /// Fetch the full resolved configuration only if the source changed. + /// + /// The default implementation delegates to `fetch_config`; data sources + /// with cheaper change detection can override this without changing normal + /// refresh behavior. + async fn fetch_config_if_modified( + &self, + if_modified_since: Option>, + ) -> Result> { + self.fetch_config(if_modified_since).await + } + /// Fetch a resolved configuration filtered by the given context and key prefixes. async fn fetch_filtered_config( &self, diff --git a/crates/superposition_provider/src/data_source/file.rs b/crates/superposition_provider/src/data_source/file.rs index 2adfdbad9..eb396c16c 100644 --- a/crates/superposition_provider/src/data_source/file.rs +++ b/crates/superposition_provider/src/data_source/file.rs @@ -47,20 +47,36 @@ impl FileDataSource { watcher: Mutex::new(None), }) } -} -#[async_trait] -impl SuperpositionDataSource for FileDataSource { - async fn fetch_filtered_config( + async fn last_modified_at(&self) -> Result> { + let metadata = tokio::fs::metadata(&self.file_path).await.map_err(|e| { + SuperpositionError::ConfigError(format!( + "Failed to read metadata for config file {:?}: {}", + self.file_path, e + )) + })?; + + metadata.modified().map(DateTime::::from).map_err(|e| { + SuperpositionError::ConfigError(format!( + "Failed to read modified time for config file {:?}: {}", + self.file_path, e + )) + }) + } + + fn is_not_modified( + last_modified_at: DateTime, + if_modified_since: Option>, + ) -> bool { + if_modified_since.is_some_and(|modified_since| last_modified_at <= modified_since) + } + + async fn read_config( &self, context: Option>, prefix_filter: Option>, - if_modified_since: Option>, + fetched_at: DateTime, ) -> Result> { - if if_modified_since.is_some() { - log::debug!("FileDataSource: ignoring if_modified_since, always reading fresh from file"); - } - let now = Utc::now(); let content = tokio::fs::read_to_string(&self.file_path) .await .map_err(|e| { @@ -89,9 +105,38 @@ impl SuperpositionDataSource for FileDataSource { Ok(FetchResponse::Data(ConfigData { data: config, - fetched_at: now, + fetched_at, })) } +} + +#[async_trait] +impl SuperpositionDataSource for FileDataSource { + async fn fetch_filtered_config( + &self, + context: Option>, + prefix_filter: Option>, + if_modified_since: Option>, + ) -> Result> { + if if_modified_since.is_some() { + log::debug!("FileDataSource: ignoring if_modified_since, always reading fresh from file"); + } + + self.read_config(context, prefix_filter, Utc::now()).await + } + + async fn fetch_config_if_modified( + &self, + if_modified_since: Option>, + ) -> Result> { + let last_modified_at = self.last_modified_at().await?; + if Self::is_not_modified(last_modified_at, if_modified_since) { + log::debug!("FileDataSource: config file not modified"); + return Ok(FetchResponse::NotModified); + } + + self.read_config(None, None, last_modified_at).await + } async fn fetch_active_experiments( &self, @@ -197,3 +242,136 @@ impl SuperpositionDataSource for FileDataSource { Ok(()) } } + +#[cfg(test)] +mod tests { + use std::path::{Path, PathBuf}; + + use tokio::time::{sleep, Duration}; + + use super::*; + + fn config_content(timeout: i64) -> String { + format!( + r#"[default-configs] +timeout = {{ value = {timeout}, schema = {{ type = "integer" }} }} + +[dimensions] +os = {{ position = 1, schema = {{ type = "string" }} }} + +[[overrides]] +_context_ = {{ os = "linux" }} +timeout = 45 +"# + ) + } + + fn temp_config_path() -> PathBuf { + std::env::temp_dir().join(format!( + "superposition-provider-file-source-{}.toml", + uuid::Uuid::new_v4() + )) + } + + async fn write_config(path: &Path, timeout: i64) { + tokio::fs::write(path, config_content(timeout)) + .await + .unwrap(); + } + + async fn rewrite_config_after( + source: &FileDataSource, + path: &Path, + timeout: i64, + previous_modified_at: DateTime, + ) { + for _ in 0..100 { + sleep(Duration::from_millis(20)).await; + write_config(path, timeout).await; + + if source.last_modified_at().await.unwrap() > previous_modified_at { + return; + } + } + + panic!("config file modified time did not advance"); + } + + #[tokio::test] + async fn fetch_config_keeps_reading_fresh_when_if_modified_since_is_present() { + let path = temp_config_path(); + write_config(&path, 30).await; + + let source = FileDataSource::new(path.clone()).unwrap(); + let first_fetch = source.fetch_config(None).await.unwrap(); + let fetched_at = match first_fetch { + FetchResponse::Data(data) => { + assert_eq!(data.data.default_configs.get("timeout"), Some(&30.into())); + data.fetched_at + } + FetchResponse::NotModified => panic!("initial fetch should return data"), + }; + + let second_fetch = source.fetch_config(Some(fetched_at)).await.unwrap(); + match second_fetch { + FetchResponse::Data(data) => { + assert_eq!(data.data.default_configs.get("timeout"), Some(&30.into())); + } + FetchResponse::NotModified => { + panic!("normal fetch should keep reading fresh") + } + } + + let _ = tokio::fs::remove_file(path).await; + } + + #[tokio::test] + async fn fetch_config_if_modified_returns_not_modified_when_file_has_not_changed() { + let path = temp_config_path(); + write_config(&path, 30).await; + + let source = FileDataSource::new(path.clone()).unwrap(); + let first_fetch = source.fetch_config(None).await.unwrap(); + let fetched_at = match first_fetch { + FetchResponse::Data(data) => data.fetched_at, + FetchResponse::NotModified => panic!("initial fetch should return data"), + }; + + let second_fetch = source + .fetch_config_if_modified(Some(fetched_at)) + .await + .unwrap(); + assert!(second_fetch.is_not_modified()); + + let _ = tokio::fs::remove_file(path).await; + } + + #[tokio::test] + async fn fetch_config_if_modified_returns_data_after_file_changes() { + let path = temp_config_path(); + write_config(&path, 30).await; + + let source = FileDataSource::new(path.clone()).unwrap(); + let first_fetch = source.fetch_config(None).await.unwrap(); + let fetched_at = match first_fetch { + FetchResponse::Data(data) => data.fetched_at, + FetchResponse::NotModified => panic!("initial fetch should return data"), + }; + + rewrite_config_after(&source, &path, 60, fetched_at).await; + + let changed_fetch = source + .fetch_config_if_modified(Some(fetched_at)) + .await + .unwrap(); + match changed_fetch { + FetchResponse::Data(data) => { + assert_eq!(data.data.default_configs.get("timeout"), Some(&60.into())); + assert!(data.fetched_at > fetched_at); + } + FetchResponse::NotModified => panic!("changed file should return data"), + } + + let _ = tokio::fs::remove_file(path).await; + } +} diff --git a/crates/superposition_provider/src/local_provider.rs b/crates/superposition_provider/src/local_provider.rs index cbba4c0d1..251817c40 100644 --- a/crates/superposition_provider/src/local_provider.rs +++ b/crates/superposition_provider/src/local_provider.rs @@ -209,6 +209,10 @@ impl LocalResolutionProvider { self.do_refresh().await } + pub async fn refresh_if_changed(&self) -> Result<()> { + self.do_refresh_if_changed().await + } + pub async fn close_provider(&self) -> Result<()> { // Abort background task { @@ -336,6 +340,38 @@ impl LocalResolutionProvider { } } + async fn do_refresh_if_changed(&self) -> Result<()> { + let last_fetched_at = { + self.cached_config + .read() + .await + .as_ref() + .map(|data| data.fetched_at) + }; + + match self.primary.fetch_config_if_modified(last_fetched_at).await { + Ok(FetchResponse::Data(data)) => { + let mut cached = self.cached_config.write().await; + *cached = Some(data); + log::debug!( + "LocalResolutionProvider: config refreshed from primary after change check" + ); + Ok(()) + } + Ok(FetchResponse::NotModified) => { + log::debug!("LocalResolutionProvider: config not modified"); + Ok(()) + } + Err(e) => { + log::warn!( + "LocalResolutionProvider: config refresh-if-changed failed, keeping last known good: {}", + e + ); + Err(e) + } + } + } + async fn start_polling(&self, interval: u64) -> JoinHandle<()> { let provider = self.clone(); tokio::spawn(async move { @@ -728,3 +764,117 @@ impl SuperpositionDataSource for LocalResolutionProvider { self.close_provider().await } } + +#[cfg(test)] +mod tests { + use std::path::{Path, PathBuf}; + + use open_feature::EvaluationContext; + use serde_json::Value; + use tokio::time::{sleep, Duration}; + + use super::*; + use crate::data_source::file::FileDataSource; + use crate::traits::AllFeatureProvider; + + fn config_content(timeout: i64) -> String { + format!( + r#"[default-configs] +timeout = {{ value = {timeout}, schema = {{ type = "integer" }} }} + +[dimensions] +os = {{ position = 1, schema = {{ type = "string" }} }} + +[[overrides]] +_context_ = {{ os = "linux" }} +timeout = 45 +"# + ) + } + + fn temp_config_path() -> PathBuf { + std::env::temp_dir().join(format!( + "superposition-provider-manual-refresh-{}.toml", + uuid::Uuid::new_v4() + )) + } + + async fn write_config(path: &Path, timeout: i64) { + tokio::fs::write(path, config_content(timeout)) + .await + .unwrap(); + } + + async fn rewrite_config_after( + path: &Path, + timeout: i64, + previous_modified_at: DateTime, + ) { + for _ in 0..100 { + sleep(Duration::from_millis(20)).await; + write_config(path, timeout).await; + + let modified_at = DateTime::::from( + tokio::fs::metadata(path).await.unwrap().modified().unwrap(), + ); + if modified_at > previous_modified_at { + return; + } + } + + panic!("config file modified time did not advance"); + } + + async fn cached_config_fetched_at( + provider: &LocalResolutionProvider, + ) -> DateTime { + provider + .cached_config + .read() + .await + .as_ref() + .map(|data| data.fetched_at) + .unwrap() + } + + async fn resolved_timeout(provider: &LocalResolutionProvider) -> Value { + provider + .resolve_all_features(EvaluationContext::default()) + .await + .unwrap() + .remove("timeout") + .unwrap() + } + + #[tokio::test] + async fn refresh_if_changed_updates_file_provider_only_when_file_changes() { + let path = temp_config_path(); + write_config(&path, 30).await; + + let provider = LocalResolutionProvider::new( + Box::new(FileDataSource::new(path.clone()).unwrap()), + None, + RefreshStrategy::Manual, + ); + provider.init(EvaluationContext::default()).await.unwrap(); + + assert_eq!(resolved_timeout(&provider).await, Value::from(30)); + let initial_fetched_at = cached_config_fetched_at(&provider).await; + + rewrite_config_after(&path, 60, initial_fetched_at).await; + provider.refresh_if_changed().await.unwrap(); + + assert_eq!(resolved_timeout(&provider).await, Value::from(60)); + let changed_fetched_at = cached_config_fetched_at(&provider).await; + assert!(changed_fetched_at > initial_fetched_at); + + provider.refresh_if_changed().await.unwrap(); + assert_eq!( + cached_config_fetched_at(&provider).await, + changed_fetched_at + ); + assert_eq!(resolved_timeout(&provider).await, Value::from(60)); + + let _ = tokio::fs::remove_file(path).await; + } +}