diff --git a/crates/superposition_provider/examples/local_file_example.rs b/crates/superposition_provider/examples/local_file_example.rs index be48b560c..c39fcd7d8 100644 --- a/crates/superposition_provider/examples/local_file_example.rs +++ b/crates/superposition_provider/examples/local_file_example.rs @@ -17,10 +17,7 @@ async fn main() { let provider = LocalResolutionProvider::new( Box::new(file_source), None, - RefreshStrategy::OnDemand(OnDemandStrategy { - ttl: 60, - ..Default::default() - }), + RefreshStrategy::OnDemand(OnDemandStrategy::new(60_000, None, None)), ); provider.init(EvaluationContext::default()).await.unwrap(); diff --git a/crates/superposition_provider/examples/local_http_example.rs b/crates/superposition_provider/examples/local_http_example.rs index 2c687be6a..e9c2e9774 100644 --- a/crates/superposition_provider/examples/local_http_example.rs +++ b/crates/superposition_provider/examples/local_http_example.rs @@ -20,10 +20,7 @@ async fn main() { let provider = LocalResolutionProvider::new( Box::new(http_source), None, - RefreshStrategy::Polling(PollingStrategy { - interval: 30, - timeout: Some(10), - }), + RefreshStrategy::Polling(PollingStrategy::new(30_000, Some(10_000))), ); provider.init(EvaluationContext::default()).await.unwrap(); diff --git a/crates/superposition_provider/examples/local_with_fallback_example.rs b/crates/superposition_provider/examples/local_with_fallback_example.rs index 74684571d..b2198a554 100644 --- a/crates/superposition_provider/examples/local_with_fallback_example.rs +++ b/crates/superposition_provider/examples/local_with_fallback_example.rs @@ -26,10 +26,7 @@ async fn main() { let provider = LocalResolutionProvider::new( Box::new(http_source), Some(Box::new(file_source)), - RefreshStrategy::Polling(PollingStrategy { - interval: 10, - timeout: Some(10), - }), + RefreshStrategy::Polling(PollingStrategy::new(10_000, Some(10_000))), ); // Register with OpenFeature and create a client diff --git a/crates/superposition_provider/examples/polling_example.rs b/crates/superposition_provider/examples/polling_example.rs index e51123bc8..58ac8424c 100644 --- a/crates/superposition_provider/examples/polling_example.rs +++ b/crates/superposition_provider/examples/polling_example.rs @@ -38,14 +38,16 @@ async fn main() { let token = env_or("SUPERPOSITION_TOKEN", "token"); let org_id = env_or("SUPERPOSITION_ORG_ID", "localorg"); let workspace = env_or("SUPERPOSITION_WORKSPACE", "dev"); - let poll_interval: u64 = env_or("POLL_INTERVAL", "10").parse().unwrap_or(10); + let poll_interval_ms: u64 = env_or("POLL_INTERVAL_MS", "10000") + .parse() + .unwrap_or(10_000); let print_interval: u64 = env_or("PRINT_INTERVAL", "5").parse().unwrap_or(5); let config_key = env_or("CONFIG_KEY", "max_connections"); println!("=== Superposition Polling Example ==="); println!("Endpoint: {}", endpoint); println!("Org / Workspace: {} / {}", org_id, workspace); - println!("Poll interval: {}s", poll_interval); + println!("Poll interval: {}ms", poll_interval_ms); println!("Print interval: {}s", print_interval); println!("Watching key: {}", config_key); println!(); @@ -57,10 +59,7 @@ async fn main() { let provider = LocalResolutionProvider::new( Box::new(http_source), None, - RefreshStrategy::Polling(PollingStrategy { - interval: poll_interval, - timeout: Some(10), - }), + RefreshStrategy::Polling(PollingStrategy::new(poll_interval_ms, Some(10_000))), ); // Register with OpenFeature and create a client diff --git a/crates/superposition_provider/src/client.rs b/crates/superposition_provider/src/client.rs index d9220a312..23d8a2bad 100644 --- a/crates/superposition_provider/src/client.rs +++ b/crates/superposition_provider/src/client.rs @@ -78,21 +78,21 @@ impl CacConfig { match &self.options.refresh_strategy { RefreshStrategy::Polling(polling_strategy) => { info!( - "Using PollingStrategy: interval={}s, timeout={}s", - polling_strategy.interval, - polling_strategy.timeout.unwrap_or(30) + "Using PollingStrategy: interval={}ms, timeout={}ms", + polling_strategy.interval_ms(), + polling_strategy.timeout_ms().unwrap_or(30_000) ); - let task_token = self.start_polling(polling_strategy.interval).await; + let task_token = self.start_polling(polling_strategy.interval_ms()).await; let mut polling_task_cancellation_token = self.polling_task_cancellation_token.write().await; *polling_task_cancellation_token = Some(task_token); } RefreshStrategy::OnDemand(on_demand_strategy) => { info!( - "Using OnDemandStrategy: ttl={}s, use_stale_on_error={}, timeout={}s", - on_demand_strategy.ttl, - on_demand_strategy.use_stale_on_error.unwrap_or(false), - on_demand_strategy.timeout.unwrap_or(30) + "Using OnDemandStrategy: ttl={}ms, use_stale_on_error={}, timeout={}ms", + on_demand_strategy.ttl_ms(), + on_demand_strategy.use_stale_on_error(), + on_demand_strategy.timeout_ms().unwrap_or(30_000) ); } RefreshStrategy::Watch(_) => { @@ -106,7 +106,7 @@ impl CacConfig { Ok(()) } - async fn start_polling(&self, interval: u64) -> CancellationToken { + async fn start_polling(&self, interval_ms: u64) -> CancellationToken { let superposition_options = self.superposition_options.clone(); let cached_config = self.cached_config.clone(); let last_updated = self.last_updated.clone(); @@ -132,7 +132,7 @@ impl CacConfig { error!("Polling error: {}", e); } } - sleep(Duration::from_secs(interval)).await; + sleep(Duration::from_millis(interval_ms)).await; } } => {} } @@ -141,14 +141,14 @@ impl CacConfig { cancellation_token } - pub async fn on_demand_config(&self, ttl: u64, use_stale: bool) -> Result { + pub async fn on_demand_config(&self, ttl_ms: u64, use_stale: bool) -> Result { let now = chrono::Utc::now(); let last_updated; { last_updated = self.last_updated.read().await; } let should_refresh = match *last_updated { - Some(last) => (now - last).num_seconds() > ttl as i64, + Some(last) => (now - last).num_milliseconds() > ttl_ms as i64, None => true, }; @@ -357,18 +357,18 @@ impl ExperimentationConfig { match &self.options.refresh_strategy { RefreshStrategy::Polling(polling_strategy) => { info!( - "Using PollingStrategy for experiments: interval={}s", - polling_strategy.interval + "Using PollingStrategy for experiments: interval={}ms", + polling_strategy.interval_ms() ); - let task_token = self.start_polling(polling_strategy.interval).await; + let task_token = self.start_polling(polling_strategy.interval_ms()).await; let mut polling_task_cancellation_token = self.polling_task_cancellation_token.write().await; *polling_task_cancellation_token = Some(task_token); } RefreshStrategy::OnDemand(on_demand_strategy) => { info!( - "Using OnDemandStrategy for experiments: ttl={}s", - on_demand_strategy.ttl + "Using OnDemandStrategy for experiments: ttl={}ms", + on_demand_strategy.ttl_ms() ); } RefreshStrategy::Watch(_) => { @@ -382,7 +382,7 @@ impl ExperimentationConfig { Ok(()) } - async fn start_polling(&self, interval: u64) -> CancellationToken { + async fn start_polling(&self, interval_ms: u64) -> CancellationToken { let superposition_options = self.superposition_options.clone(); let cached_experiments = self.cached_experiments.clone(); let cached_experiment_groups = self.cached_experiment_groups.clone(); @@ -421,7 +421,7 @@ impl ExperimentationConfig { } _ => {} } - sleep(Duration::from_secs(interval)).await; + sleep(Duration::from_millis(interval_ms)).await; } } => {} } @@ -432,14 +432,14 @@ impl ExperimentationConfig { pub async fn on_demand_config( &self, - ttl: u64, + ttl_ms: u64, use_stale: bool, ) -> Result { let now = chrono::Utc::now(); let last_updated = self.last_updated.read().await; let should_refresh = match *last_updated { - Some(last) => (now - last).num_seconds() > ttl as i64, + Some(last) => (now - last).num_milliseconds() > ttl_ms as i64, None => true, }; diff --git a/crates/superposition_provider/src/conversions.rs b/crates/superposition_provider/src/conversions.rs index a3c6c17b4..330dd57ab 100644 --- a/crates/superposition_provider/src/conversions.rs +++ b/crates/superposition_provider/src/conversions.rs @@ -192,33 +192,45 @@ pub fn evaluation_context_to_map(context: EvaluationContext) -> Map Result { match value { - Value::Object(map) => { - let mut fields = HashMap::new(); - for (k, v) in map { - let open_feature_value = value_to_openfeature_value(v)?; - fields.insert(k, open_feature_value); - } - // StructValue is just a struct with a fields HashMap, not a complex conversion - Ok(open_feature::StructValue { fields }) - } - Value::Array(list) => { - let mut fields = HashMap::new(); - for (index, item) in list.into_iter().enumerate() { - let open_feature_value = value_to_openfeature_value(item)?; - fields.insert(index.to_string(), open_feature_value); - } - Ok(open_feature::StructValue { fields }) - } + Value::Object(map) => Ok(open_feature::StructValue { + fields: object_fields(map)?, + }), _ => Err(SuperpositionError::ConfigError(format!( - "Cannot convert {:?} to StructValue - flag must be an object/array", + "Cannot convert {:?} to StructValue - flag must be an object", value ))), } } +/// Convert a JSON object's entries to OpenFeature fields, dropping null-valued keys. +/// +/// `open_feature::Value` has no null variant. Dropping the key leaves the rest of the +/// object readable; erroring would make one null field render the entire flag +/// unresolvable, which is what this used to do. +fn object_fields(map: Map) -> Result> { + let mut fields = HashMap::new(); + for (k, v) in map { + if v.is_null() { + log::debug!( + "Dropping null-valued field '{k}': OpenFeature has no null representation" + ); + continue; + } + fields.insert(k, value_to_openfeature_value(v)?); + } + Ok(fields) +} + /// Convert serde_json Value to OpenFeature Value pub fn value_to_openfeature_value(value: Value) -> Result { match value { @@ -241,18 +253,70 @@ pub fn value_to_openfeature_value(value: Value) -> Result { .map(value_to_openfeature_value) .collect::>>()?, )), - Value::Object(map) => { - let fields = map - .into_iter() - .map(|(k, v)| Ok((k, value_to_openfeature_value(v)?))) - .collect::>>()?; - - // Create StructValue directly with fields HashMap - let struct_value = open_feature::StructValue { fields }; - Ok(open_feature::Value::Struct(struct_value)) - } + Value::Object(map) => Ok(open_feature::Value::Struct(open_feature::StructValue { + fields: object_fields(map)?, + })), + // Reachable only for a null *array element*: object fields are filtered out by + // `object_fields` first. Dropping it here would shift every later index, so this + // stays an error. Value::Null => Err(SuperpositionError::ConfigError( "Cannot convert null to OpenFeature value".to_string(), )), } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn an_object_flag_keeps_its_field_types() { + let fields = value_to_struct(json!({"tier": "gold", "credits": 5, "ratio": 1.5})) + .expect("an object converts") + .fields; + + assert_eq!(fields["tier"], open_feature::Value::String("gold".into())); + assert_eq!(fields["credits"], open_feature::Value::Int(5)); + assert_eq!(fields["ratio"], open_feature::Value::Float(1.5)); + } + + #[test] + fn a_null_field_is_dropped_rather_than_failing_the_whole_flag() { + let fields = value_to_struct(json!({"tier": "gold", "promo": null})) + .expect("a null field does not fail the flag") + .fields; + + assert_eq!(fields["tier"], open_feature::Value::String("gold".into())); + assert!(!fields.contains_key("promo")); + } + + #[test] + fn a_top_level_array_is_a_type_mismatch_not_an_index_keyed_map() { + // This used to succeed, fabricating StructValue { "0": .., "1": .. } — a value the + // flag never held. Java and Python return the array itself; `StructValue` cannot. + assert!(value_to_struct(json!([1, 2, 3])).is_err()); + } + + #[test] + fn an_array_nested_in_an_object_needs_no_special_handling() { + let fields = value_to_struct(json!({"tags": ["a", "b"]})) + .expect("a nested array converts") + .fields; + + assert_eq!( + fields["tags"], + open_feature::Value::Array(vec![ + open_feature::Value::String("a".into()), + open_feature::Value::String("b".into()), + ]) + ); + } + + #[test] + fn a_primitive_is_not_an_object() { + assert!(value_to_struct(json!("Rupee")).is_err()); + assert!(value_to_struct(json!(true)).is_err()); + assert!(value_to_struct(json!(10)).is_err()); + } +} diff --git a/crates/superposition_provider/src/data_source/file.rs b/crates/superposition_provider/src/data_source/file.rs index 2adfdbad9..c7c1e4e36 100644 --- a/crates/superposition_provider/src/data_source/file.rs +++ b/crates/superposition_provider/src/data_source/file.rs @@ -64,7 +64,7 @@ impl SuperpositionDataSource for FileDataSource { let content = tokio::fs::read_to_string(&self.file_path) .await .map_err(|e| { - SuperpositionError::ConfigError(format!( + SuperpositionError::DataSourceError(format!( "Failed to read config file {:?}: {}", self.file_path, e )) @@ -75,7 +75,7 @@ impl SuperpositionDataSource for FileDataSource { _ => TomlFormat::parse_config, }; let mut config = parser(&content).map_err(|e| { - SuperpositionError::ConfigError(format!( + SuperpositionError::DataSourceError(format!( "Failed to parse {} config: {}", self.file_format.to_uppercase(), e @@ -97,7 +97,7 @@ impl SuperpositionDataSource for FileDataSource { &self, _if_modified_since: Option>, ) -> Result> { - Err(SuperpositionError::ConfigError( + Err(SuperpositionError::DataSourceError( "Experiments not supported by FileDataSource".into(), )) } @@ -108,7 +108,7 @@ impl SuperpositionDataSource for FileDataSource { _prefix_filter: Option>, _if_modified_since: Option>, ) -> Result> { - Err(SuperpositionError::ConfigError( + Err(SuperpositionError::DataSourceError( "Experiments not supported by FileDataSource".into(), )) } @@ -119,7 +119,7 @@ impl SuperpositionDataSource for FileDataSource { _prefix_filter: Option>, _if_modified_since: Option>, ) -> Result> { - Err(SuperpositionError::ConfigError( + Err(SuperpositionError::DataSourceError( "Experiments not supported by FileDataSource".into(), )) } @@ -131,7 +131,7 @@ impl SuperpositionDataSource for FileDataSource { fn watch(&self) -> Result> { // Acquire both locks upfront to prevent concurrent watcher creation let mut watcher_guard = self.watcher.lock().map_err(|e| { - SuperpositionError::ConfigError(format!( + SuperpositionError::DataSourceError(format!( "Failed to lock watcher mutex: {}", e )) @@ -159,7 +159,7 @@ impl SuperpositionDataSource for FileDataSource { }, ) .map_err(|e| { - SuperpositionError::ConfigError(format!( + SuperpositionError::DataSourceError(format!( "Failed to create file watcher: {}", e )) @@ -168,7 +168,7 @@ impl SuperpositionDataSource for FileDataSource { watcher .watch(&self.file_path, notify::RecursiveMode::NonRecursive) .map_err(|e| { - SuperpositionError::ConfigError(format!( + SuperpositionError::DataSourceError(format!( "Failed to watch file {:?}: {}", self.file_path, e )) @@ -187,7 +187,7 @@ impl SuperpositionDataSource for FileDataSource { async fn close(&self) -> Result<()> { let mut guard = self.watcher.lock().map_err(|e| { - SuperpositionError::ConfigError(format!( + SuperpositionError::DataSourceError(format!( "Failed to lock watcher mutex: {}", e )) diff --git a/crates/superposition_provider/src/data_source/http.rs b/crates/superposition_provider/src/data_source/http.rs index 310ceba3e..73a5811a9 100644 --- a/crates/superposition_provider/src/data_source/http.rs +++ b/crates/superposition_provider/src/data_source/http.rs @@ -1,7 +1,6 @@ use async_trait::async_trait; use chrono::{DateTime, TimeZone, Utc}; use serde_json::{Map, Value}; -use superposition_sdk::error::SdkError; use superposition_sdk::types::DimensionMatchStrategy; use superposition_sdk::{Client, Config as SdkConfig}; @@ -85,7 +84,7 @@ impl HttpDataSource { }) .map(FetchResponse::Data) } - Err(SdkError::ResponseError(r)) if r.raw().status().as_u16() == 304 => { + Err(e) if e.raw_response().map(|r| r.status().as_u16()) == Some(304) => { Ok(FetchResponse::NotModified) } Err(e) => Err(SuperpositionError::NetworkError(format!( @@ -148,7 +147,7 @@ impl SuperpositionDataSource for HttpDataSource { }) .map(FetchResponse::Data) } - Err(SdkError::ResponseError(r)) if r.raw().status().as_u16() == 304 => { + Err(e) if e.raw_response().map(|r| r.status().as_u16()) == Some(304) => { Ok(FetchResponse::NotModified) } Err(e) => Err(SuperpositionError::NetworkError(format!( diff --git a/crates/superposition_provider/src/lib.rs b/crates/superposition_provider/src/lib.rs index 7a27ea5da..9006de0cc 100644 --- a/crates/superposition_provider/src/lib.rs +++ b/crates/superposition_provider/src/lib.rs @@ -37,7 +37,6 @@ mod tests { let config_options = ConfigurationOptions::new( RefreshStrategy::OnDemand(OnDemandStrategy::default()), None, - None, ); let cac_config = CacConfig::new(superposition_options, config_options); diff --git a/crates/superposition_provider/src/local_provider.rs b/crates/superposition_provider/src/local_provider.rs index cbba4c0d1..86f4f6ed9 100644 --- a/crates/superposition_provider/src/local_provider.rs +++ b/crates/superposition_provider/src/local_provider.rs @@ -31,6 +31,8 @@ pub struct LocalResolutionProviderInner { refresh_strategy: RefreshStrategy, cached_config: RwLock>, cached_experiments: RwLock>, + config_checked_at: RwLock>>, + experiments_checked_at: RwLock>>, background_task: RwLock>>, metadata: ProviderMetadata, status: RwLock, @@ -52,6 +54,8 @@ impl LocalResolutionProvider { refresh_strategy, cached_config: RwLock::new(None), cached_experiments: RwLock::new(None), + config_checked_at: RwLock::new(None), + experiments_checked_at: RwLock::new(None), background_task: RwLock::new(None), metadata: ProviderMetadata { name: "LocalResolutionProvider".to_string(), @@ -62,6 +66,22 @@ impl LocalResolutionProvider { } pub async fn init(&self, context: EvaluationContext) -> Result<()> { + // Single-shot: a provider is initialized once and then served. Re-initializing a live + // provider would strand the running background task (it holds a strong Arc, so it would + // poll forever), so it is refused. A fresh (NotReady) or previously-failed (Error) + // provider proceeds. The claim happens under the status lock so it is atomic. + { + let mut status = self.status.write().await; + if matches!(*status, ProviderStatus::Ready | ProviderStatus::STALE) { + log::warn!( + "LocalResolutionProvider already initialized; ignoring init(). \ + Providers are single-shot — build a new instance." + ); + return Ok(()); + } + *status = ProviderStatus::NotReady; + } + // Fetch initial config from primary, fall back if needed let config_data = match self.primary.fetch_config(None).await { Ok(data) => { @@ -97,6 +117,7 @@ impl LocalResolutionProvider { let mut cached = self.cached_config.write().await; *cached = config_data.into_data(); } + *self.config_checked_at.write().await = Some(Utc::now()); // Fetch experiments best-effort: try primary, else fallback let exp_data = if self.primary.supports_experiments() { @@ -144,22 +165,25 @@ impl LocalResolutionProvider { let mut cached = self.cached_experiments.write().await; *cached = Some(data); } + if self.primary.supports_experiments() { + *self.experiments_checked_at.write().await = Some(Utc::now()); + } // Start refresh strategy match &self.refresh_strategy { RefreshStrategy::Polling(polling_strategy) => { log::info!( - "LocalResolutionProvider: starting polling with interval={}s", - polling_strategy.interval + "LocalResolutionProvider: starting polling with interval={}ms", + polling_strategy.interval_ms() ); - let task = self.start_polling(polling_strategy.interval).await; + let task = self.start_polling(polling_strategy.interval_ms()).await; let mut background_task = self.background_task.write().await; *background_task = Some(task); } RefreshStrategy::OnDemand(on_demand_strategy) => { log::info!( - "LocalResolutionProvider: using OnDemand strategy with ttl={}s", - on_demand_strategy.ttl + "LocalResolutionProvider: using OnDemand strategy with ttl={}ms", + on_demand_strategy.ttl_ms() ); } RefreshStrategy::Watch(watch_strategy) => { @@ -243,6 +267,8 @@ impl LocalResolutionProvider { let mut cached = self.cached_experiments.write().await; *cached = None; } + *self.config_checked_at.write().await = None; + *self.experiments_checked_at.write().await = None; { let mut global_context = self.global_context.write().await; @@ -258,7 +284,74 @@ impl LocalResolutionProvider { Ok(()) } + /// The timeout the configured strategy puts on a single refresh, if any. + fn refresh_timeout(&self) -> Option { + match &self.refresh_strategy { + RefreshStrategy::Polling(polling) => polling.timeout_ms(), + RefreshStrategy::OnDemand(on_demand) => on_demand.timeout_ms(), + RefreshStrategy::Watch(_) | RefreshStrategy::Manual => None, + } + .map(Duration::from_millis) + } + + /// Runs a refresh, bounded by the strategy's timeout. Without this a data source that never + /// answers would stall the poller — or, under OnDemand, the evaluating thread — indefinitely. + /// + /// Every refresh path funnels through here, so this is also where staleness is recorded. async fn do_refresh(&self) -> Result<()> { + let result = match self.refresh_timeout() { + Some(timeout) => { + match tokio::time::timeout(timeout, self.refresh_once()).await { + Ok(result) => result, + Err(_) => { + log::warn!( + "LocalResolutionProvider: refresh timed out after {}ms, keeping last known good", + timeout.as_millis() + ); + Err(SuperpositionError::RefreshError(format!( + "Refresh timed out after {}ms", + timeout.as_millis() + ))) + } + } + } + None => self.refresh_once().await, + }; + + self.record_refresh_outcome(result.is_ok()).await; + result + } + + /// A refresh that fails while the cache is still served leaves the provider STALE: the flags + /// are frozen at their last known good values, and this is the only signal a consumer has that + /// they stopped tracking the source of truth. The next successful refresh clears it. + /// + /// Only meaningful from Ready. A failure during init is an Error — there is no good data to be + /// stale — and a provider that has been shut down stays NotReady. + /// + /// Evaluation is unaffected: the SDK's client never consults provider status, so a STALE + /// provider keeps resolving flags from its cache. + /// + /// Unlike the Java and Python clients, this cannot notify anyone. `open_feature` 0.2.5 has no + /// provider-event API, so the only way to observe staleness is to hold the provider and call + /// [`FeatureProvider::status`]. Java emits PROVIDER_STALE and Python emits its equivalent, so + /// consumers there can subscribe. Revisit when the crate grows events. + async fn record_refresh_outcome(&self, succeeded: bool) { + let mut status = self.status.write().await; + match (&*status, succeeded) { + (ProviderStatus::Ready, false) => { + log::warn!("LocalResolutionProvider: refresh failed, serving stale data"); + *status = ProviderStatus::STALE; + } + (ProviderStatus::STALE, true) => { + log::info!("LocalResolutionProvider: refresh recovered, no longer stale"); + *status = ProviderStatus::Ready; + } + _ => {} + } + } + + async fn refresh_once(&self) -> Result<()> { // Fetch config from primary; keep last known good on failure let last_fetched_at = { self.cached_config @@ -273,10 +366,14 @@ impl LocalResolutionProvider { Ok(FetchResponse::Data(data)) => { let mut cached = self.cached_config.write().await; *cached = Some(data); + *self.config_checked_at.write().await = Some(Utc::now()); log::debug!("LocalResolutionProvider: config refreshed from primary"); Ok(()) } Ok(FetchResponse::NotModified) => { + // A 304 is a successful check: the cache is confirmed current, so the TTL + // clock restarts. Without this the next evaluation would ask again at once. + *self.config_checked_at.write().await = Some(Utc::now()); log::debug!("LocalResolutionProvider: config not modified"); Ok(()) } @@ -309,6 +406,8 @@ impl LocalResolutionProvider { if let Some(data) = exp_resp.into_data() { *cached = Some(data); } + // Data or NotModified — both are a successful check, so the TTL restarts. + *self.experiments_checked_at.write().await = Some(Utc::now()); log::debug!( "LocalResolutionProvider: experiments refreshed from primary" ); @@ -336,11 +435,11 @@ impl LocalResolutionProvider { } } - async fn start_polling(&self, interval: u64) -> JoinHandle<()> { + async fn start_polling(&self, interval_ms: u64) -> JoinHandle<()> { let provider = self.clone(); tokio::spawn(async move { loop { - sleep(Duration::from_secs(interval)).await; + sleep(Duration::from_millis(interval_ms)).await; let _ = provider.do_refresh().await; } }) @@ -375,28 +474,28 @@ impl LocalResolutionProvider { async fn ensure_fresh_data(&self) -> Result<()> { if let RefreshStrategy::OnDemand(on_demand) = &self.refresh_strategy { - let ttl = on_demand.ttl; - let use_stale_on_error = on_demand.use_stale_on_error.unwrap_or_default(); + let ttl = on_demand.ttl_ms(); + let use_stale_on_error = on_demand.use_stale_on_error(); let is_elapsed = |cached_at: DateTime| { - (chrono::Utc::now() - cached_at).num_seconds() > ttl as i64 + (chrono::Utc::now() - cached_at).num_milliseconds() > ttl as i64 }; - let should_refresh_config = { - let cached = self.cached_config.read().await; - cached - .as_ref() - .map(|data| is_elapsed(data.fetched_at)) - .unwrap_or(true) - }; + // Never checked, or last checked before the TTL window opened. + let should_refresh_config = self + .config_checked_at + .read() + .await + .map(is_elapsed) + .unwrap_or(true); - let should_refresh_experiments = { - let cached = self.cached_experiments.read().await; - cached - .as_ref() - .map(|data| is_elapsed(data.fetched_at)) - .unwrap_or(true) - }; + let should_refresh_experiments = self.primary.supports_experiments() + && self + .experiments_checked_at + .read() + .await + .map(is_elapsed) + .unwrap_or(true); if should_refresh_config || should_refresh_experiments { log::debug!("LocalResolutionProvider: TTL expired, refreshing on-demand"); @@ -480,8 +579,8 @@ impl LocalResolutionProvider { e )) }), - None => Err(SuperpositionError::ConfigError( - "No cached config available".into(), + None => Err(SuperpositionError::ProviderError( + "Provider not initialized: no cached config available".into(), )), } } @@ -530,10 +629,8 @@ impl FeatureExperimentMeta for LocalResolutionProvider { impl FeatureProvider for LocalResolutionProvider { async fn initialize(&mut self, context: &EvaluationContext) { log::info!("Initializing LocalResolutionProvider..."); - { - let mut status = self.status.write().await; - *status = ProviderStatus::NotReady; - } + // init() owns the status transition: it refuses a re-init of a live provider and only + // then moves to NotReady. Resetting the status here would defeat that guard. if (self.init(context.clone()).await).is_err() { let mut status = self.status.write().await; *status = ProviderStatus::Error; @@ -622,7 +719,7 @@ impl SuperpositionDataSource for LocalResolutionProvider { match cached.as_ref() { Some(data) => data.clone(), None => { - return Err(SuperpositionError::ConfigError( + return Err(SuperpositionError::DataSourceError( "No cached config available".into(), )) } @@ -640,7 +737,7 @@ impl SuperpositionDataSource for LocalResolutionProvider { if_modified_since: Option>, ) -> Result> { if !self.supports_experiments() { - return Err(SuperpositionError::ConfigError( + return Err(SuperpositionError::DataSourceError( "Experiments not supported by this provider".into(), )); } @@ -650,7 +747,7 @@ impl SuperpositionDataSource for LocalResolutionProvider { let cached = self.cached_experiments.read().await; match cached.clone() { Some(data) => Ok(FetchResponse::Data(data)), - None => Err(SuperpositionError::ConfigError( + None => Err(SuperpositionError::DataSourceError( "No cached experiments available".into(), )), } @@ -663,7 +760,7 @@ impl SuperpositionDataSource for LocalResolutionProvider { if_modified_since: Option>, ) -> Result> { if !self.supports_experiments() { - return Err(SuperpositionError::ConfigError( + return Err(SuperpositionError::DataSourceError( "Experiments not supported by this provider".into(), )); } @@ -695,7 +792,7 @@ impl SuperpositionDataSource for LocalResolutionProvider { if_modified_since: Option>, ) -> Result> { if !self.supports_experiments() { - return Err(SuperpositionError::ConfigError( + return Err(SuperpositionError::DataSourceError( "Experiments not supported by this provider".into(), )); } diff --git a/crates/superposition_provider/src/provider.rs b/crates/superposition_provider/src/provider.rs index 66e62105f..8235441a5 100644 --- a/crates/superposition_provider/src/provider.rs +++ b/crates/superposition_provider/src/provider.rs @@ -39,7 +39,6 @@ impl SuperpositionProvider { ); let cac_options = ConfigurationOptions::new( provider_options.refresh_strategy, - provider_options.evaluation_cache, provider_options.fallback_config.clone(), ); diff --git a/crates/superposition_provider/src/remote_provider.rs b/crates/superposition_provider/src/remote_provider.rs index ba4ead7f0..b6d5a4c14 100644 --- a/crates/superposition_provider/src/remote_provider.rs +++ b/crates/superposition_provider/src/remote_provider.rs @@ -118,11 +118,6 @@ impl FeatureExperimentMeta for SuperpositionAPIProvider { prefix_filter: Option>, ) -> Result> { let (query_data, targeting_key) = self.get_merged_context(context).await; - let Some(targeting_key) = targeting_key else { - return Err(SuperpositionError::ProviderError( - "Missing targeting key in evaluation context".to_string(), - )); - }; let applicable_variants = self .client @@ -130,7 +125,7 @@ impl FeatureExperimentMeta for SuperpositionAPIProvider { .workspace_id(&self.options.workspace_id) .org_id(&self.options.org_id) .set_context(Some(query_data)) - .identifier(targeting_key) + .identifier(targeting_key.unwrap_or_default()) .set_prefix(prefix_filter) .send() .await diff --git a/crates/superposition_provider/src/traits.rs b/crates/superposition_provider/src/traits.rs index 2879e6b9e..13f2b298e 100644 --- a/crates/superposition_provider/src/traits.rs +++ b/crates/superposition_provider/src/traits.rs @@ -43,6 +43,18 @@ pub trait AllFeatureProvider: Send + Sync { prefix_filter: Option>, ) -> Result>; + /// Resolve a flag and extract it as `T`. + /// + /// Error reasons need no handling here: the failure paths return `EvaluationError`, which has + /// no `reason` field, and the SDK stamps `EvaluationReason::Error` on the details it builds + /// from it. The Java and Python providers return the details object themselves, so they set it + /// explicitly — same outcome, different seam. + /// + /// TODO: successful resolutions leave `reason` unset. Reporting it accurately (STATIC for a + /// default-config value, TARGETING_MATCH for a context override, SPLIT for an experiment + /// variant) needs `eval_config` in `superposition_core` to say, per key, where the value came + /// from. Until it does, guessing would be worse than saying nothing — a flag no experiment + /// touched would still be labelled SPLIT. The same TODO applies to the Java and Python clients. async fn resolve_typed( &self, flag_key: &str, @@ -129,4 +141,30 @@ pub trait AllFeatureProvider: Send + Sync { }) .await } + + /// Resolve a flag whose value is a JSON array. + /// + /// `resolve_struct` cannot return one: OpenFeature models an object flag as a + /// `StructValue`, which has no array form, so a top-level array is a TypeMismatch there. + /// (An array *nested inside* an object flag is fine and needs no special handling.) + /// This is the typed way to read one; the alternative is `resolve_all_features`, which + /// hands back the raw `serde_json::Value`. + /// + /// The Java and Python clients return top-level arrays from their object accessor + /// directly, because their SDKs' object type admits one. This method exists to close + /// that gap, not to add a capability the other clients lack. + async fn resolve_array( + &self, + flag_key: &str, + evaluation_context: EvaluationContext, + ) -> EvaluationResult>> { + self.resolve_typed(flag_key, evaluation_context, "array", |v| match v { + Value::Array(items) => items + .into_iter() + .map(|item| conversions::value_to_openfeature_value(item).ok()) + .collect(), + _ => None, + }) + .await + } } diff --git a/crates/superposition_provider/src/types.rs b/crates/superposition_provider/src/types.rs index 292e2b8a6..eb2791bdb 100644 --- a/crates/superposition_provider/src/types.rs +++ b/crates/superposition_provider/src/types.rs @@ -1,4 +1,3 @@ -use serde::{Deserialize, Serialize}; use serde_json::Value; use thiserror::Error; @@ -12,11 +11,15 @@ pub enum SuperpositionError { SerializationError(String), #[error("Provider error: {0}")] ProviderError(String), + #[error("Data source error: {0}")] + DataSourceError(String), + #[error("Refresh error: {0}")] + RefreshError(String), } pub type Result = std::result::Result; -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone)] pub struct SuperpositionOptions { pub endpoint: String, pub token: String, @@ -40,73 +43,149 @@ impl SuperpositionOptions { } } -/// Cache configuration -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CacheOptions { - pub ttl: Option, - pub size: Option, +/// Polling strategy configuration. +/// +/// Durations are milliseconds. The seconds-based `interval` and `timeout` fields still work and are +/// deprecated: a `_milliseconds` field wins when set, otherwise the old field is read as seconds. +/// +/// `Default` deliberately leaves the `_milliseconds` fields as `None`, so that the common +/// `PollingStrategy { interval: 30, ..Default::default() }` still means 30 seconds. Defaulting them +/// to `Some(..)` would let the default silently override the caller's seconds value. +#[derive(Debug, Clone)] +pub struct PollingStrategy { + #[deprecated(note = "seconds-based; use `interval_milliseconds`")] + pub interval: u64, + /// How often to refresh, in milliseconds. Wins over `interval` when set. + pub interval_milliseconds: Option, + #[deprecated(note = "seconds-based; use `timeout_milliseconds`")] + pub timeout: Option, + /// How long a single refresh may take before it is abandoned, in milliseconds. + /// Wins over `timeout` when set. + pub timeout_milliseconds: Option, } -impl Default for CacheOptions { - fn default() -> Self { +impl PollingStrategy { + /// Build a polling strategy from millisecond durations. + pub fn new(interval_milliseconds: u64, timeout_milliseconds: Option) -> Self { + #[allow(deprecated)] Self { - ttl: Some(300), // 5 minutes - size: Some(1000), + // Left unset: the accessors read the millisecond fields. Mirroring a value back would + // round 5_500ms down to 5s. + interval: 0, + timeout: None, + interval_milliseconds: Some(interval_milliseconds), + timeout_milliseconds, } } -} - -/// Evaluation cache configuration -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct EvaluationCacheOptions { - pub ttl: Option, - pub size: Option, -} -impl Default for EvaluationCacheOptions { - fn default() -> Self { - Self { - ttl: Some(60), // 1 minute - size: Some(500), - } + /// The refresh interval in milliseconds, falling back to the deprecated seconds field. + pub fn interval_ms(&self) -> u64 { + #[allow(deprecated)] + self.interval_milliseconds + .unwrap_or_else(|| self.interval.saturating_mul(1000)) } -} -/// Polling strategy configuration -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct PollingStrategy { - pub interval: u64, // seconds - pub timeout: Option, + /// The refresh timeout in milliseconds, falling back to the deprecated seconds field. + pub fn timeout_ms(&self) -> Option { + #[allow(deprecated)] + self.timeout_milliseconds + .or_else(|| self.timeout.map(|secs| secs.saturating_mul(1000))) + } } impl Default for PollingStrategy { fn default() -> Self { + // Expressed in the deprecated fields on purpose — see the note on the struct. + #[allow(deprecated)] Self { interval: 60, // 1 minute timeout: Some(30), + interval_milliseconds: None, + timeout_milliseconds: None, } } } -#[derive(Debug, Clone, Serialize, Deserialize)] +/// On-demand strategy configuration. Durations are milliseconds; the seconds-based `ttl` and +/// `timeout` fields still work and are deprecated. See [`PollingStrategy`] for how the two interact. +#[derive(Debug, Clone)] pub struct OnDemandStrategy { - pub ttl: u64, // seconds + #[deprecated(note = "seconds-based; use `ttl_milliseconds`")] + pub ttl: u64, + /// How long cached data stays fresh, in milliseconds. Wins over `ttl` when set. + pub ttl_milliseconds: Option, + #[deprecated(note = "seconds-based; use `timeout_milliseconds`")] pub timeout: Option, + /// How long a single refresh may take before it is abandoned, in milliseconds. + /// Wins over `timeout` when set. + pub timeout_milliseconds: Option, pub use_stale_on_error: Option, } +impl OnDemandStrategy { + /// Build an on-demand strategy from millisecond durations. + pub fn new( + ttl_milliseconds: u64, + timeout_milliseconds: Option, + use_stale_on_error: Option, + ) -> Self { + #[allow(deprecated)] + Self { + // Left unset: the accessors read the millisecond fields. Mirroring a value back would + // round 5_500ms down to 5s. + ttl: 0, + timeout: None, + ttl_milliseconds: Some(ttl_milliseconds), + timeout_milliseconds, + use_stale_on_error, + } + } + + /// The cache TTL in milliseconds, falling back to the deprecated seconds field. + pub fn ttl_ms(&self) -> u64 { + #[allow(deprecated)] + self.ttl_milliseconds + .unwrap_or_else(|| self.ttl.saturating_mul(1000)) + } + + /// The refresh timeout in milliseconds, falling back to the deprecated seconds field. + pub fn timeout_ms(&self) -> Option { + #[allow(deprecated)] + self.timeout_milliseconds + .or_else(|| self.timeout.map(|secs| secs.saturating_mul(1000))) + } + + /// Whether to serve stale data when a refresh fails. + /// + /// Read through this rather than off the field: unset means "unspecified", not "off". A call + /// site reaching for `unwrap_or_default()` reads it as `false`, which contradicts both + /// [`Default`] and the Java and Python clients. + pub fn use_stale_on_error(&self) -> bool { + self.use_stale_on_error + .unwrap_or(DEFAULT_USE_STALE_ON_ERROR) + } +} + +/// The single place this default is written. [`Default`] and +/// [`OnDemandStrategy::use_stale_on_error`] both read it, so they cannot drift apart. +const DEFAULT_USE_STALE_ON_ERROR: bool = true; + impl Default for OnDemandStrategy { fn default() -> Self { + // Expressed in the deprecated fields on purpose — see the note on PollingStrategy. + #[allow(deprecated)] Self { ttl: 300, // 5 minutes timeout: Some(30), - use_stale_on_error: Some(true), + ttl_milliseconds: None, + timeout_milliseconds: None, + use_stale_on_error: Some(DEFAULT_USE_STALE_ON_ERROR), } } } /// Configuration for the watch refresh strategy. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone)] pub struct WatchStrategy { /// Debounce duration in milliseconds (default: 500). pub debounce_ms: Option, @@ -125,7 +204,7 @@ pub struct WatchStream { pub receiver: tokio::sync::broadcast::Receiver<()>, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone)] pub enum RefreshStrategy { Polling(PollingStrategy), OnDemand(OnDemandStrategy), @@ -139,66 +218,43 @@ impl Default for RefreshStrategy { } } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone)] pub struct ConfigurationOptions { pub fallback_config: Option>, - pub evaluation_cache: Option, pub refresh_strategy: RefreshStrategy, } impl ConfigurationOptions { pub fn new( refresh_strategy: RefreshStrategy, - evaluation_cache: Option, fallback_config: Option>, ) -> Self { Self { fallback_config, - evaluation_cache, refresh_strategy, } } } /// Experimentation options -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone)] pub struct ExperimentationOptions { pub refresh_strategy: RefreshStrategy, - pub evaluation_cache: Option, - pub default_toss: Option, } impl ExperimentationOptions { pub fn new(refresh_strategy: RefreshStrategy) -> Self { - Self { - refresh_strategy, - evaluation_cache: Some(EvaluationCacheOptions::default()), - default_toss: None, - } - } - - pub fn with_evaluation_cache( - mut self, - evaluation_cache: EvaluationCacheOptions, - ) -> Self { - self.evaluation_cache = Some(evaluation_cache); - self - } - - pub fn with_default_toss(mut self, default_toss: u32) -> Self { - self.default_toss = Some(default_toss); - self + Self { refresh_strategy } } } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone)] pub struct SuperpositionProviderOptions { pub endpoint: String, pub token: String, pub org_id: String, pub workspace_id: String, pub fallback_config: Option>, - pub evaluation_cache: Option, pub refresh_strategy: RefreshStrategy, pub experimentation_options: Option, } @@ -211,7 +267,6 @@ impl SuperpositionProviderOptions { org_id: String, workspace_id: String, fallback_config: Option>, - evaluation_cache: Option, refresh_strategy: RefreshStrategy, experimentation_options: Option, ) -> Self { @@ -221,9 +276,58 @@ impl SuperpositionProviderOptions { org_id, workspace_id, fallback_config, - evaluation_cache, refresh_strategy, experimentation_options, } } } + +#[cfg(test)] +mod tests { + use super::*; + + /// The regression this guards: `Default` must not fill in the millisecond fields, or it would + /// silently override a caller who set only the deprecated seconds field. + #[test] + #[allow(deprecated)] + fn a_seconds_field_with_default_still_means_seconds() { + let polling = PollingStrategy { + interval: 30, // 30 seconds + ..Default::default() + }; + assert_eq!(polling.interval_ms(), 30_000); + assert_eq!(polling.timeout_ms(), Some(30_000)); // default 30s + + let on_demand = OnDemandStrategy { + ttl: 60, // 60 seconds + ..Default::default() + }; + assert_eq!(on_demand.ttl_ms(), 60_000); + } + + #[test] + fn millisecond_fields_win_over_the_deprecated_ones() { + let polling = PollingStrategy { + interval_milliseconds: Some(1_500), + ..Default::default() + }; + assert_eq!(polling.interval_ms(), 1_500); + + assert_eq!( + PollingStrategy::new(5_500, Some(2_500)).interval_ms(), + 5_500 + ); + assert_eq!( + PollingStrategy::new(5_500, Some(2_500)).timeout_ms(), + Some(2_500) + ); + } + + #[test] + fn defaults_are_unchanged() { + assert_eq!(PollingStrategy::default().interval_ms(), 60_000); + assert_eq!(PollingStrategy::default().timeout_ms(), Some(30_000)); + assert_eq!(OnDemandStrategy::default().ttl_ms(), 300_000); + assert_eq!(OnDemandStrategy::default().timeout_ms(), Some(30_000)); + } +}