From 1455dc1e808a66501226a4b509362935cc061063 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Tue, 4 Aug 2026 17:28:57 -0700 Subject: [PATCH 01/13] feat: Add FDv2 data system orchestrator --- .../src/fdv2/data_system.rs | 525 ++++++++++++++++++ launchdarkly-server-sdk/src/fdv2/mod.rs | 1 + 2 files changed, 526 insertions(+) create mode 100644 launchdarkly-server-sdk/src/fdv2/data_system.rs diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs new file mode 100644 index 00000000..44d2ec5f --- /dev/null +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -0,0 +1,525 @@ +use std::sync::Arc; + +use futures::FutureExt; +use parking_lot::RwLock; +use tokio::sync::broadcast; + +use crate::data_system::DataSystem; +use crate::stores::store::{DataStore, InMemoryDataStore, TransactionalDataStore}; + +use super::model::Selector; +use super::source::{FDv2SourceResult, Initializer, Synchronizer}; + +/// Produces a fresh initializer each time the orchestrator starts a run. +pub(crate) trait InitializerFactory: Send + Sync { + fn create(&self) -> Box; +} + +/// Produces a fresh synchronizer each time the orchestrator starts a run. +pub(crate) trait SynchronizerFactory: Send + Sync { + fn create(&self) -> Box; +} + +/// FDv2 orchestrator: owns the memory store and keeps it populated by running +/// initializers to obtain a basis, then synchronizers for ongoing changes. +pub(crate) struct FDv2DataSystem { + initializer_factories: Vec>, + synchronizer_factories: Vec>, + store: Arc>, +} + +impl FDv2DataSystem { + pub(crate) fn new( + initializer_factories: Vec>, + synchronizer_factories: Vec>, + ) -> Self { + Self { + initializer_factories, + synchronizer_factories, + store: Arc::new(RwLock::new(InMemoryDataStore::new())), + } + } +} + +impl DataSystem for FDv2DataSystem { + fn start( + &self, + init_complete: Arc, + shutdown_receiver: broadcast::Receiver<()>, + ) { + let initializers = self + .initializer_factories + .iter() + .map(|f| f.create()) + .collect(); + let synchronizers = self + .synchronizer_factories + .iter() + .map(|f| f.create()) + .collect(); + let store = self.store.clone(); + + tokio::spawn(run( + initializers, + synchronizers, + store, + init_complete, + shutdown_receiver, + )); + } + + fn store(&self) -> Arc> { + self.store.clone() + } +} + +async fn run( + initializers: Vec>, + synchronizers: Vec>, + store: Arc>, + init_complete: Arc, + mut shutdown_receiver: broadcast::Receiver<()>, +) { + let mut selector: Selector = None; + let mut initialized = false; + + // Initializer phase: try each in order until one yields a basis. + for mut initializer in initializers { + let mut shutdown = Box::pin(shutdown_receiver.recv()).fuse(); + futures::select! { + _ = shutdown => return, + event = initializer.run().fuse() => { + if let FDv2SourceResult::ChangeSet(change_set) = event.result { + selector = change_set.selector.clone(); + store.write().apply(change_set); + init_complete(true); + initialized = true; + break; + } + } + } + } + + // Synchronizer phase: use each in order, advancing on a terminal result. + for mut synchronizer in synchronizers { + loop { + let mut shutdown = Box::pin(shutdown_receiver.recv()).fuse(); + let result = futures::select! { + _ = shutdown => return, + event = synchronizer.next(selector.clone()).fuse() => event.result, + }; + match result { + FDv2SourceResult::ChangeSet(change_set) => { + selector = change_set.selector.clone(); + store.write().apply(change_set); + if !initialized { + init_complete(true); + initialized = true; + } + } + FDv2SourceResult::Interrupted(_) => continue, + FDv2SourceResult::TerminalError(_) + | FDv2SourceResult::Shutdown + | FDv2SourceResult::Goodbye { .. } => break, + } + } + } + + // Every source exhausted without ever obtaining a basis. + if !initialized { + init_complete(false); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::VecDeque; + use std::sync::Mutex; + + use futures::future::BoxFuture; + use launchdarkly_server_sdk_evaluation::Store; + + use super::super::model::ChangeSetKind; + use super::super::source::{ErrorInfo, ErrorKind, FDv2SourceEvent}; + use crate::stores::change_set::{ChangeSet, ItemChange}; + use crate::stores::store_types::StorageItem; + use crate::test_common::basic_flag; + + type Selectors = Arc>>; + type InitCalls = Arc>>; + + fn changeset(kind: ChangeSetKind, key: &str, selector: Selector) -> FDv2SourceResult { + FDv2SourceResult::ChangeSet(ChangeSet { + kind, + changes: vec![ItemChange::Flag { + key: key.to_string(), + item: StorageItem::Item(basic_flag(key)), + }], + selector, + }) + } + + fn interrupted() -> FDv2SourceResult { + FDv2SourceResult::Interrupted(ErrorInfo { + kind: ErrorKind::Unknown, + message: "test".into(), + }) + } + + fn terminal() -> FDv2SourceResult { + FDv2SourceResult::TerminalError(ErrorInfo { + kind: ErrorKind::Unknown, + message: "test".into(), + }) + } + + fn event(result: FDv2SourceResult) -> FDv2SourceEvent { + FDv2SourceEvent { + result, + fdv1_fallback: None, + } + } + + struct MockInitializer { + results: VecDeque, + } + + impl Initializer for MockInitializer { + fn run(&mut self) -> BoxFuture<'_, FDv2SourceEvent> { + let result = self + .results + .pop_front() + .unwrap_or(FDv2SourceResult::Shutdown); + Box::pin(async move { event(result) }) + } + + fn name(&self) -> &str { + "mock-initializer" + } + } + + struct MockSynchronizer { + results: VecDeque, + selectors_seen: Selectors, + hang: bool, + } + + impl Synchronizer for MockSynchronizer { + fn next(&mut self, selector: Selector) -> BoxFuture<'_, FDv2SourceEvent> { + self.selectors_seen.lock().unwrap().push(selector); + if self.hang { + return Box::pin(std::future::pending()); + } + // Once scripted results run out, end the run instead of spinning. + let result = self + .results + .pop_front() + .unwrap_or(FDv2SourceResult::Shutdown); + Box::pin(async move { event(result) }) + } + + fn name(&self) -> &str { + "mock-synchronizer" + } + } + + struct MockInitializerFactory { + results: Mutex>, + } + + impl InitializerFactory for MockInitializerFactory { + fn create(&self) -> Box { + let results = std::mem::take(&mut *self.results.lock().unwrap()); + Box::new(MockInitializer { + results: results.into(), + }) + } + } + + struct MockSynchronizerFactory { + results: Mutex>, + selectors_seen: Selectors, + } + + impl SynchronizerFactory for MockSynchronizerFactory { + fn create(&self) -> Box { + let results = std::mem::take(&mut *self.results.lock().unwrap()); + Box::new(MockSynchronizer { + results: results.into(), + selectors_seen: self.selectors_seen.clone(), + hang: false, + }) + } + } + + fn recording_init_complete() -> (Arc, InitCalls) { + let calls: InitCalls = Arc::new(Mutex::new(Vec::new())); + let sink = calls.clone(); + let cb: Arc = + Arc::new(move |success| sink.lock().unwrap().push(success)); + (cb, calls) + } + + #[tokio::test] + async fn start_applies_basis_and_exposes_it_via_store_handle() { + // A data system whose sole initializer yields one full basis. + let system = FDv2DataSystem::new( + vec![Box::new(MockInitializerFactory { + results: Mutex::new(vec![changeset( + ChangeSetKind::Full, + "f1", + Some("s1".into()), + )]), + })], + vec![Box::new(MockSynchronizerFactory { + results: Mutex::new(vec![]), + selectors_seen: Arc::new(Mutex::new(Vec::new())), + })], + ); + + // Record init-complete calls and wake the test when one arrives. + let calls: InitCalls = Arc::new(Mutex::new(Vec::new())); + let notify = Arc::new(tokio::sync::Notify::new()); + let sink = calls.clone(); + let waker = notify.clone(); + let init_complete: Arc = Arc::new(move |success| { + sink.lock().unwrap().push(success); + waker.notify_one(); + }); + let (shutdown_tx, shutdown_rx) = broadcast::channel(1); + + // Start the system and wait for initialization to finish. + system.start(init_complete, shutdown_rx); + notify.notified().await; + + // The basis was applied and is readable through the store() handle. + assert_eq!(*calls.lock().unwrap(), vec![true]); + assert!(system.store().read().flag("f1").is_some()); + drop(shutdown_tx); + } + + #[tokio::test] + async fn initializer_basis_signals_once_and_propagates_selector() { + // Store and recorders shared with the mock sources. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); + let (init_complete, calls) = recording_init_complete(); + + // Initializer delivers a full basis carrying selector "sel-1". + let initializers: Vec> = vec![Box::new(MockInitializer { + results: VecDeque::from(vec![changeset( + ChangeSetKind::Full, + "init-flag", + Some("sel-1".into()), + )]), + })]; + + // Synchronizer delivers a partial change carrying selector "sel-2". + let synchronizers: Vec> = vec![Box::new(MockSynchronizer { + results: VecDeque::from(vec![changeset( + ChangeSetKind::Partial, + "sync-flag", + Some("sel-2".into()), + )]), + selectors_seen: selectors_seen.clone(), + hang: false, + })]; + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + synchronizers, + store.clone(), + init_complete, + shutdown_rx, + ) + .await; + + // init_complete fired exactly once despite two successful applies. + assert_eq!(*calls.lock().unwrap(), vec![true]); + + // Both the basis flag and the later partial change are in the store. + assert!(store.read().flag("init-flag").is_some()); + assert!(store.read().flag("sync-flag").is_some()); + + // The synchronizer's first call received the basis selector. + assert_eq!(selectors_seen.lock().unwrap()[0], Some("sel-1".into())); + } + + #[tokio::test] + async fn failed_initializers_let_synchronizer_provide_the_basis() { + // Store and init-complete recorder. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + + // Both initializers fail without producing a basis. + let initializers: Vec> = vec![ + Box::new(MockInitializer { + results: VecDeque::from(vec![interrupted()]), + }), + Box::new(MockInitializer { + results: VecDeque::from(vec![terminal()]), + }), + ]; + + // The synchronizer then delivers the basis. + let synchronizers: Vec> = vec![Box::new(MockSynchronizer { + results: VecDeque::from(vec![changeset( + ChangeSetKind::Full, + "sync-flag", + Some("s".into()), + )]), + selectors_seen: Arc::new(Mutex::new(Vec::new())), + hang: false, + })]; + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + synchronizers, + store.clone(), + init_complete, + shutdown_rx, + ) + .await; + + // Initialization succeeded via the synchronizer. + assert_eq!(*calls.lock().unwrap(), vec![true]); + assert!(store.read().flag("sync-flag").is_some()); + } + + #[tokio::test] + async fn exhausting_all_sources_signals_failure_once() { + // Store and init-complete recorder. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + + // The initializer fails. + let initializers: Vec> = vec![Box::new(MockInitializer { + results: VecDeque::from(vec![terminal()]), + })]; + + // The synchronizer only ever reports an interruption, never a basis. + let synchronizers: Vec> = vec![Box::new(MockSynchronizer { + results: VecDeque::from(vec![interrupted()]), + selectors_seen: Arc::new(Mutex::new(Vec::new())), + hang: false, + })]; + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + synchronizers, + store, + init_complete, + shutdown_rx, + ) + .await; + + // Failure was reported exactly once. + assert_eq!(*calls.lock().unwrap(), vec![false]); + } + + #[tokio::test] + async fn synchronizer_terminal_error_advances_to_next() { + // Store and init-complete recorder; no initializers. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializers: Vec> = vec![]; + + // First synchronizer fails terminally; the second provides the basis. + let synchronizers: Vec> = vec![ + Box::new(MockSynchronizer { + results: VecDeque::from(vec![terminal()]), + selectors_seen: Arc::new(Mutex::new(Vec::new())), + hang: false, + }), + Box::new(MockSynchronizer { + results: VecDeque::from(vec![changeset( + ChangeSetKind::Full, + "from-second", + Some("s".into()), + )]), + selectors_seen: Arc::new(Mutex::new(Vec::new())), + hang: false, + }), + ]; + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + synchronizers, + store.clone(), + init_complete, + shutdown_rx, + ) + .await; + + // The run advanced past the terminal source and applied the second's basis. + assert_eq!(*calls.lock().unwrap(), vec![true]); + assert!(store.read().flag("from-second").is_some()); + } + + #[tokio::test] + async fn synchronizer_interrupted_retries_same_source() { + // Store and init-complete recorder; no initializers. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializers: Vec> = vec![]; + + // A single synchronizer: an interruption, then a basis on the retry. + let synchronizers: Vec> = vec![Box::new(MockSynchronizer { + results: VecDeque::from(vec![ + interrupted(), + changeset(ChangeSetKind::Full, "after-retry", Some("s".into())), + ]), + selectors_seen: Arc::new(Mutex::new(Vec::new())), + hang: false, + })]; + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + synchronizers, + store.clone(), + init_complete, + shutdown_rx, + ) + .await; + + // The retry on the same synchronizer delivered the basis. + assert_eq!(*calls.lock().unwrap(), vec![true]); + assert!(store.read().flag("after-retry").is_some()); + } + + #[tokio::test] + async fn shutdown_ends_run_without_signaling() { + // Store and init-complete recorder; no initializers. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializers: Vec> = vec![]; + + // A synchronizer whose next() never resolves. + let synchronizers: Vec> = vec![Box::new(MockSynchronizer { + results: VecDeque::new(), + selectors_seen: Arc::new(Mutex::new(Vec::new())), + hang: true, + })]; + let (shutdown_tx, shutdown_rx) = broadcast::channel(1); + + // Drive the run on a task, then signal shutdown. + let handle = tokio::spawn(run( + initializers, + synchronizers, + store, + init_complete, + shutdown_rx, + )); + shutdown_tx.send(()).unwrap(); + handle.await.unwrap(); + + // The run returned before any initialization signal. + assert!(calls.lock().unwrap().is_empty()); + } +} diff --git a/launchdarkly-server-sdk/src/fdv2/mod.rs b/launchdarkly-server-sdk/src/fdv2/mod.rs index 40c0ce7c..44627015 100644 --- a/launchdarkly-server-sdk/src/fdv2/mod.rs +++ b/launchdarkly-server-sdk/src/fdv2/mod.rs @@ -1,3 +1,4 @@ +mod data_system; pub mod model; mod polling; mod protocol; From 7e7ef4ce4679c17999bf0d6548dc71b665bcb5b2 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Wed, 5 Aug 2026 12:22:38 -0700 Subject: [PATCH 02/13] feat: Add FDv2 synchronizer rotation and fallback/recovery timers --- .../src/fdv2/data_system.rs | 524 +++++++++++++++--- 1 file changed, 436 insertions(+), 88 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index 44d2ec5f..13eaf645 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -1,8 +1,10 @@ use std::sync::Arc; +use std::time::Duration; use futures::FutureExt; use parking_lot::RwLock; use tokio::sync::broadcast; +use tokio::time::{sleep_until, Instant}; use crate::data_system::DataSystem; use crate::stores::store::{DataStore, InMemoryDataStore, TransactionalDataStore}; @@ -24,14 +26,14 @@ pub(crate) trait SynchronizerFactory: Send + Sync { /// initializers to obtain a basis, then synchronizers for ongoing changes. pub(crate) struct FDv2DataSystem { initializer_factories: Vec>, - synchronizer_factories: Vec>, + synchronizer_factories: Vec>, store: Arc>, } impl FDv2DataSystem { pub(crate) fn new( initializer_factories: Vec>, - synchronizer_factories: Vec>, + synchronizer_factories: Vec>, ) -> Self { Self { initializer_factories, @@ -52,16 +54,12 @@ impl DataSystem for FDv2DataSystem { .iter() .map(|f| f.create()) .collect(); - let synchronizers = self - .synchronizer_factories - .iter() - .map(|f| f.create()) - .collect(); + let source_manager = SourceManager::new(self.synchronizer_factories.clone()); let store = self.store.clone(); tokio::spawn(run( initializers, - synchronizers, + source_manager, store, init_complete, shutdown_receiver, @@ -73,9 +71,98 @@ impl DataSystem for FDv2DataSystem { } } +/// Per-factory availability used by the synchronizer rotation. +#[derive(Clone, Copy, PartialEq, Eq)] +enum SourceState { + Available, + Blocked, +} + +/// Owns the synchronizer factories and tracks which one is currently active. +struct SourceManager { + factories: Vec>, + states: Vec, + /// Iteration cursor; `None` restarts the search from the prime. + synchronizer_index: Option, + /// Index of the most recently returned factory, for blocking and prime checks. + current_factory_index: Option, +} + +impl SourceManager { + fn new(factories: Vec>) -> Self { + let states = vec![SourceState::Available; factories.len()]; + Self { + factories, + states, + synchronizer_index: None, + current_factory_index: None, + } + } + + /// Builds the next available synchronizer, advancing cyclically past the + /// active one and skipping blocked factories. `None` when all are blocked. + fn next_synchronizer(&mut self) -> Option> { + let n = self.factories.len(); + if n == 0 { + self.current_factory_index = None; + return None; + } + let mut i = self.synchronizer_index.map_or(0, |c| (c + 1) % n); + for _ in 0..n { + if self.states[i] == SourceState::Available { + self.synchronizer_index = Some(i); + self.current_factory_index = Some(i); + return Some(self.factories[i].create()); + } + i = (i + 1) % n; + } + self.current_factory_index = None; + None + } + + /// Marks the active factory blocked, used on a terminal error. + fn block_current(&mut self) { + if let Some(i) = self.current_factory_index { + self.states[i] = SourceState::Blocked; + } + } + + /// Makes the next `next_synchronizer` restart from the prime, used on recovery. + fn reset_source_index(&mut self) { + self.synchronizer_index = None; + } + + /// Whether the active factory is the first available one. + fn is_prime(&self) -> bool { + let first = self + .states + .iter() + .position(|s| *s == SourceState::Available); + matches!((first, self.current_factory_index), (Some(f), Some(c)) if f == c) + } + + fn available_count(&self) -> usize { + self.states + .iter() + .filter(|s| **s == SourceState::Available) + .count() + } +} + +const FALLBACK_TIMEOUT: Duration = Duration::from_secs(120); +const RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); + +/// Sleeps until `at`, or never when `None` (an inactive timer arm). +async fn deadline(at: Option) { + match at { + Some(t) => sleep_until(t).await, + None => std::future::pending::<()>().await, + } +} + async fn run( initializers: Vec>, - synchronizers: Vec>, + mut source_manager: SourceManager, store: Arc>, init_complete: Arc, mut shutdown_receiver: broadcast::Receiver<()>, @@ -100,32 +187,62 @@ async fn run( } } - // Synchronizer phase: use each in order, advancing on a terminal result. - for mut synchronizer in synchronizers { + // Synchronizer phase: rotate through synchronizers as the timers fire. + let mut current = source_manager.next_synchronizer(); + while let Some(mut active) = current { + let has_fallback = source_manager.available_count() > 1; + let has_recovery = has_fallback && !source_manager.is_prime(); + let mut fallback_at: Option = None; + let recovery_at = has_recovery.then(|| Instant::now() + RECOVERY_TIMEOUT); + + // Drive the active synchronizer, racing its events against the timers. loop { let mut shutdown = Box::pin(shutdown_receiver.recv()).fuse(); - let result = futures::select! { + let mut fallback = Box::pin(deadline(fallback_at)).fuse(); + let mut recovery = Box::pin(deadline(recovery_at)).fuse(); + let mut next = active.next(selector.clone()).fuse(); + futures::select! { _ = shutdown => return, - event = synchronizer.next(selector.clone()).fuse() => event.result, - }; - match result { - FDv2SourceResult::ChangeSet(change_set) => { - selector = change_set.selector.clone(); - store.write().apply(change_set); - if !initialized { - init_complete(true); - initialized = true; - } + // Fall back to the next synchronizer. + _ = fallback => break, + // Recover to the prime. + _ = recovery => { + source_manager.reset_source_index(); + break; } - FDv2SourceResult::Interrupted(_) => continue, - FDv2SourceResult::TerminalError(_) - | FDv2SourceResult::Shutdown - | FDv2SourceResult::Goodbye { .. } => break, + event = next => match event.result { + FDv2SourceResult::ChangeSet(change_set) => { + selector = change_set.selector.clone(); + store.write().apply(change_set); + if !initialized { + init_complete(true); + initialized = true; + } + // Good data clears the fallback countdown. + fallback_at = None; + } + // Sustained interruption starts the fallback countdown. + FDv2SourceResult::Interrupted(_) => { + if has_fallback && fallback_at.is_none() { + fallback_at = Some(Instant::now() + FALLBACK_TIMEOUT); + } + } + // Handled internally by the synchronizer. + FDv2SourceResult::Goodbye { .. } => {} + FDv2SourceResult::TerminalError(_) => { + // Dead source: drop it and advance. + source_manager.block_current(); + break; + } + FDv2SourceResult::Shutdown => return, + }, } } + + current = source_manager.next_synchronizer(); } - // Every source exhausted without ever obtaining a basis. + // Every source blocked without ever obtaining a basis. if !initialized { init_complete(false); } @@ -135,6 +252,7 @@ async fn run( mod tests { use super::*; use std::collections::VecDeque; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Mutex; use futures::future::BoxFuture; @@ -208,15 +326,12 @@ mod tests { impl Synchronizer for MockSynchronizer { fn next(&mut self, selector: Selector) -> BoxFuture<'_, FDv2SourceEvent> { self.selectors_seen.lock().unwrap().push(selector); - if self.hang { - return Box::pin(std::future::pending()); + match self.results.pop_front() { + Some(result) => Box::pin(async move { event(result) }), + // Out of scripted results: idle if hang, otherwise end the run. + None if self.hang => Box::pin(std::future::pending()), + None => Box::pin(async move { event(FDv2SourceResult::Shutdown) }), } - // Once scripted results run out, end the run instead of spinning. - let result = self - .results - .pop_front() - .unwrap_or(FDv2SourceResult::Shutdown); - Box::pin(async move { event(result) }) } fn name(&self) -> &str { @@ -240,6 +355,7 @@ mod tests { struct MockSynchronizerFactory { results: Mutex>, selectors_seen: Selectors, + hang: bool, } impl SynchronizerFactory for MockSynchronizerFactory { @@ -248,7 +364,54 @@ mod tests { Box::new(MockSynchronizer { results: results.into(), selectors_seen: self.selectors_seen.clone(), - hang: false, + hang: self.hang, + }) + } + } + + /// A selector recorder that ignores what it captures. + fn no_selectors() -> Selectors { + Arc::new(Mutex::new(Vec::new())) + } + + /// A single-synchronizer factory scripted with the given results. + fn sync_factory( + results: Vec, + selectors_seen: Selectors, + hang: bool, + ) -> Arc { + Arc::new(MockSynchronizerFactory { + results: Mutex::new(results), + selectors_seen, + hang, + }) + } + + /// A prime factory that is down on first build and recovers on rebuild. + struct RecoveringPrimeFactory { + builds: AtomicUsize, + } + + impl SynchronizerFactory for RecoveringPrimeFactory { + fn create(&self) -> Box { + let (results, hang) = if self.builds.fetch_add(1, Ordering::SeqCst) == 0 { + // Down: interrupt, then idle so the fallback timer fires. + (vec![interrupted()], true) + } else { + // Recovered: deliver a delta once the prime is active again. + ( + vec![changeset( + ChangeSetKind::Partial, + "prime-recovered", + Some("s".into()), + )], + false, + ) + }; + Box::new(MockSynchronizer { + results: results.into(), + selectors_seen: no_selectors(), + hang, }) } } @@ -272,10 +435,7 @@ mod tests { Some("s1".into()), )]), })], - vec![Box::new(MockSynchronizerFactory { - results: Mutex::new(vec![]), - selectors_seen: Arc::new(Mutex::new(Vec::new())), - })], + vec![sync_factory(vec![], no_selectors(), false)], ); // Record init-complete calls and wake the test when one arrives. @@ -316,20 +476,20 @@ mod tests { })]; // Synchronizer delivers a partial change carrying selector "sel-2". - let synchronizers: Vec> = vec![Box::new(MockSynchronizer { - results: VecDeque::from(vec![changeset( + let source_manager = SourceManager::new(vec![sync_factory( + vec![changeset( ChangeSetKind::Partial, "sync-flag", Some("sel-2".into()), - )]), - selectors_seen: selectors_seen.clone(), - hang: false, - })]; + )], + selectors_seen.clone(), + false, + )]); let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( initializers, - synchronizers, + source_manager, store.clone(), init_complete, shutdown_rx, @@ -364,20 +524,20 @@ mod tests { ]; // The synchronizer then delivers the basis. - let synchronizers: Vec> = vec![Box::new(MockSynchronizer { - results: VecDeque::from(vec![changeset( + let source_manager = SourceManager::new(vec![sync_factory( + vec![changeset( ChangeSetKind::Full, "sync-flag", Some("s".into()), - )]), - selectors_seen: Arc::new(Mutex::new(Vec::new())), - hang: false, - })]; + )], + no_selectors(), + false, + )]); let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( initializers, - synchronizers, + source_manager, store.clone(), init_complete, shutdown_rx, @@ -400,17 +560,17 @@ mod tests { results: VecDeque::from(vec![terminal()]), })]; - // The synchronizer only ever reports an interruption, never a basis. - let synchronizers: Vec> = vec![Box::new(MockSynchronizer { - results: VecDeque::from(vec![interrupted()]), - selectors_seen: Arc::new(Mutex::new(Vec::new())), - hang: false, - })]; + // The synchronizer reports an interruption, then fails terminally. + let source_manager = SourceManager::new(vec![sync_factory( + vec![interrupted(), terminal()], + no_selectors(), + false, + )]); let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( initializers, - synchronizers, + source_manager, store, init_complete, shutdown_rx, @@ -429,27 +589,23 @@ mod tests { let initializers: Vec> = vec![]; // First synchronizer fails terminally; the second provides the basis. - let synchronizers: Vec> = vec![ - Box::new(MockSynchronizer { - results: VecDeque::from(vec![terminal()]), - selectors_seen: Arc::new(Mutex::new(Vec::new())), - hang: false, - }), - Box::new(MockSynchronizer { - results: VecDeque::from(vec![changeset( + let source_manager = SourceManager::new(vec![ + sync_factory(vec![terminal()], no_selectors(), false), + sync_factory( + vec![changeset( ChangeSetKind::Full, "from-second", Some("s".into()), - )]), - selectors_seen: Arc::new(Mutex::new(Vec::new())), - hang: false, - }), - ]; + )], + no_selectors(), + false, + ), + ]); let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( initializers, - synchronizers, + source_manager, store.clone(), init_complete, shutdown_rx, @@ -469,19 +625,19 @@ mod tests { let initializers: Vec> = vec![]; // A single synchronizer: an interruption, then a basis on the retry. - let synchronizers: Vec> = vec![Box::new(MockSynchronizer { - results: VecDeque::from(vec![ + let source_manager = SourceManager::new(vec![sync_factory( + vec![ interrupted(), changeset(ChangeSetKind::Full, "after-retry", Some("s".into())), - ]), - selectors_seen: Arc::new(Mutex::new(Vec::new())), - hang: false, - })]; + ], + no_selectors(), + false, + )]); let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( initializers, - synchronizers, + source_manager, store.clone(), init_complete, shutdown_rx, @@ -501,17 +657,13 @@ mod tests { let initializers: Vec> = vec![]; // A synchronizer whose next() never resolves. - let synchronizers: Vec> = vec![Box::new(MockSynchronizer { - results: VecDeque::new(), - selectors_seen: Arc::new(Mutex::new(Vec::new())), - hang: true, - })]; + let source_manager = SourceManager::new(vec![sync_factory(vec![], no_selectors(), true)]); let (shutdown_tx, shutdown_rx) = broadcast::channel(1); // Drive the run on a task, then signal shutdown. let handle = tokio::spawn(run( initializers, - synchronizers, + source_manager, store, init_complete, shutdown_rx, @@ -522,4 +674,200 @@ mod tests { // The run returned before any initialization signal. assert!(calls.lock().unwrap().is_empty()); } + + #[test] + fn rotates_cyclically_skips_blocked_and_exhausts() { + let mut sources = SourceManager::new(vec![ + sync_factory(vec![], no_selectors(), false), + sync_factory(vec![], no_selectors(), false), + sync_factory(vec![], no_selectors(), false), + ]); + + // Advances cyclically from the prime. + sources.next_synchronizer(); + assert_eq!(sources.current_factory_index, Some(0)); + sources.next_synchronizer(); + assert_eq!(sources.current_factory_index, Some(1)); + + // Blocking the active factory drops it from the rotation. + sources.block_current(); + sources.next_synchronizer(); + assert_eq!(sources.current_factory_index, Some(2)); + + // The next pass wraps around and skips the blocked factory. + sources.next_synchronizer(); + assert_eq!(sources.current_factory_index, Some(0)); + + // With every factory blocked there is nothing left to return. + sources.block_current(); + sources.next_synchronizer(); + sources.block_current(); + assert!(sources.next_synchronizer().is_none()); + } + + #[test] + fn reset_source_index_returns_to_prime() { + let mut sources = SourceManager::new(vec![ + sync_factory(vec![], no_selectors(), false), + sync_factory(vec![], no_selectors(), false), + ]); + + sources.next_synchronizer(); + sources.next_synchronizer(); + assert_eq!(sources.current_factory_index, Some(1)); + + // Recovery rewinds so the next search starts from the prime. + sources.reset_source_index(); + sources.next_synchronizer(); + assert_eq!(sources.current_factory_index, Some(0)); + } + + #[test] + fn is_prime_and_available_count_track_state() { + let mut sources = SourceManager::new(vec![ + sync_factory(vec![], no_selectors(), false), + sync_factory(vec![], no_selectors(), false), + ]); + assert_eq!(sources.available_count(), 2); + + sources.next_synchronizer(); + assert!(sources.is_prime()); + sources.next_synchronizer(); + assert!(!sources.is_prime()); + + sources.block_current(); + assert_eq!(sources.available_count(), 1); + } + + #[tokio::test(start_paused = true)] + async fn fallback_fires_after_sustained_interruption() { + // No initializers; the prime interrupts then idles. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializers: Vec> = vec![]; + + // Prime only ever interrupts; the fallback stands by with a basis. + let source_manager = SourceManager::new(vec![ + sync_factory(vec![interrupted()], no_selectors(), true), + sync_factory( + vec![changeset( + ChangeSetKind::Full, + "from-fallback", + Some("s".into()), + )], + no_selectors(), + false, + ), + ]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + // Paused time auto-advances past the fallback timeout while the prime idles. + let handle = tokio::spawn(run( + initializers, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + )); + handle.await.unwrap(); + + // The run fell back to the second synchronizer and applied its basis. + assert_eq!(*calls.lock().unwrap(), vec![true]); + assert!(store.read().flag("from-fallback").is_some()); + } + + #[tokio::test(start_paused = true)] + async fn changeset_cancels_the_fallback_timer() { + // Notify on init so the test can act once the prime's basis lands. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let calls: InitCalls = Arc::new(Mutex::new(Vec::new())); + let notify = Arc::new(tokio::sync::Notify::new()); + let sink = calls.clone(); + let waker = notify.clone(); + let init_complete: Arc = Arc::new(move |success| { + sink.lock().unwrap().push(success); + waker.notify_one(); + }); + let initializers: Vec> = vec![]; + + // Prime interrupts (arming the timer), delivers a basis (clearing it), then idles. + let source_manager = SourceManager::new(vec![ + sync_factory( + vec![ + interrupted(), + changeset(ChangeSetKind::Full, "from-prime", Some("s".into())), + ], + no_selectors(), + true, + ), + sync_factory( + vec![changeset( + ChangeSetKind::Full, + "from-fallback", + Some("s".into()), + )], + no_selectors(), + false, + ), + ]); + let (shutdown_tx, shutdown_rx) = broadcast::channel(1); + + let handle = tokio::spawn(run( + initializers, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + )); + + // Wait for the basis, let well past the fallback timeout elapse, then stop. + notify.notified().await; + tokio::time::advance(FALLBACK_TIMEOUT * 2).await; + shutdown_tx.send(()).unwrap(); + handle.await.unwrap(); + + // The basis kept the run on the prime; it never fell back. + assert!(store.read().flag("from-prime").is_some()); + assert!(store.read().flag("from-fallback").is_none()); + } + + #[tokio::test(start_paused = true)] + async fn recovery_fires_and_returns_to_the_prime() { + // No initializers; the prime is down, so the run falls back then recovers. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializers: Vec> = vec![]; + + // Prime recovers on rebuild; the fallback supplies a basis then idles. + let source_manager = SourceManager::new(vec![ + Arc::new(RecoveringPrimeFactory { + builds: AtomicUsize::new(0), + }) as Arc, + sync_factory( + vec![changeset( + ChangeSetKind::Full, + "from-fallback", + Some("s".into()), + )], + no_selectors(), + true, + ), + ]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + // Paused time auto-advances through the fallback then recovery timeouts. + let handle = tokio::spawn(run( + initializers, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + )); + handle.await.unwrap(); + + // Fell back to the fallback's basis, then recovered to the prime's delta. + assert_eq!(*calls.lock().unwrap(), vec![true]); + assert!(store.read().flag("from-fallback").is_some()); + assert!(store.read().flag("prime-recovered").is_some()); + } } From c822ce207b0eadc657984d2a08b6bbfa1cd9d7cb Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Wed, 5 Aug 2026 12:55:34 -0700 Subject: [PATCH 03/13] feat: Log FDv2 synchronizer status transitions --- launchdarkly-server-sdk/src/fdv2/data_system.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index 13eaf645..ee517f7f 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -194,6 +194,7 @@ async fn run( let has_recovery = has_fallback && !source_manager.is_prime(); let mut fallback_at: Option = None; let recovery_at = has_recovery.then(|| Instant::now() + RECOVERY_TIMEOUT); + let mut interrupted_logged = false; // Drive the active synchronizer, racing its events against the timers. loop { @@ -220,16 +221,22 @@ async fn run( } // Good data clears the fallback countdown. fallback_at = None; + interrupted_logged = false; } // Sustained interruption starts the fallback countdown. - FDv2SourceResult::Interrupted(_) => { + FDv2SourceResult::Interrupted(error) => { + if !interrupted_logged { + info!("FDv2 synchronizer interrupted: {}", error.message); + interrupted_logged = true; + } if has_fallback && fallback_at.is_none() { fallback_at = Some(Instant::now() + FALLBACK_TIMEOUT); } } // Handled internally by the synchronizer. FDv2SourceResult::Goodbye { .. } => {} - FDv2SourceResult::TerminalError(_) => { + FDv2SourceResult::TerminalError(error) => { + warn!("FDv2 synchronizer terminal error: {}", error.message); // Dead source: drop it and advance. source_manager.block_current(); break; From f1c940e49896c2715df61d167d0e91e2941938b4 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Wed, 5 Aug 2026 15:15:30 -0700 Subject: [PATCH 04/13] refactor: Rename RecoveringPrimeFactory to DownThenDataFactory --- launchdarkly-server-sdk/src/fdv2/data_system.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index ee517f7f..845e2dd3 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -394,18 +394,18 @@ mod tests { }) } - /// A prime factory that is down on first build and recovers on rebuild. - struct RecoveringPrimeFactory { + /// A factory that is down on its first build and delivers data on rebuild. + struct DownThenDataFactory { builds: AtomicUsize, } - impl SynchronizerFactory for RecoveringPrimeFactory { + impl SynchronizerFactory for DownThenDataFactory { fn create(&self) -> Box { let (results, hang) = if self.builds.fetch_add(1, Ordering::SeqCst) == 0 { // Down: interrupt, then idle so the fallback timer fires. (vec![interrupted()], true) } else { - // Recovered: deliver a delta once the prime is active again. + // Rebuilt: deliver a delta. ( vec![changeset( ChangeSetKind::Partial, @@ -847,7 +847,7 @@ mod tests { // Prime recovers on rebuild; the fallback supplies a basis then idles. let source_manager = SourceManager::new(vec![ - Arc::new(RecoveringPrimeFactory { + Arc::new(DownThenDataFactory { builds: AtomicUsize::new(0), }) as Arc, sync_factory( From 49c1f3cbe5061020508971a0f559d882b2888674 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Thu, 6 Aug 2026 13:56:15 -0700 Subject: [PATCH 05/13] feat: Make FDv2 fallback and recovery timeouts configurable --- .../src/fdv2/data_system.rs | 40 ++++++++++++++++--- 1 file changed, 35 insertions(+), 5 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index 845e2dd3..cc3321bd 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -27,6 +27,8 @@ pub(crate) trait SynchronizerFactory: Send + Sync { pub(crate) struct FDv2DataSystem { initializer_factories: Vec>, synchronizer_factories: Vec>, + fallback_timeout: Duration, + recovery_timeout: Duration, store: Arc>, } @@ -34,10 +36,14 @@ impl FDv2DataSystem { pub(crate) fn new( initializer_factories: Vec>, synchronizer_factories: Vec>, + fallback_timeout: Duration, + recovery_timeout: Duration, ) -> Self { Self { initializer_factories, synchronizer_factories, + fallback_timeout, + recovery_timeout, store: Arc::new(RwLock::new(InMemoryDataStore::new())), } } @@ -63,6 +69,8 @@ impl DataSystem for FDv2DataSystem { store, init_complete, shutdown_receiver, + self.fallback_timeout, + self.recovery_timeout, )); } @@ -149,9 +157,6 @@ impl SourceManager { } } -const FALLBACK_TIMEOUT: Duration = Duration::from_secs(120); -const RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); - /// Sleeps until `at`, or never when `None` (an inactive timer arm). async fn deadline(at: Option) { match at { @@ -166,6 +171,8 @@ async fn run( store: Arc>, init_complete: Arc, mut shutdown_receiver: broadcast::Receiver<()>, + fallback_timeout: Duration, + recovery_timeout: Duration, ) { let mut selector: Selector = None; let mut initialized = false; @@ -193,7 +200,7 @@ async fn run( let has_fallback = source_manager.available_count() > 1; let has_recovery = has_fallback && !source_manager.is_prime(); let mut fallback_at: Option = None; - let recovery_at = has_recovery.then(|| Instant::now() + RECOVERY_TIMEOUT); + let recovery_at = has_recovery.then(|| Instant::now() + recovery_timeout); let mut interrupted_logged = false; // Drive the active synchronizer, racing its events against the timers. @@ -230,7 +237,7 @@ async fn run( interrupted_logged = true; } if has_fallback && fallback_at.is_none() { - fallback_at = Some(Instant::now() + FALLBACK_TIMEOUT); + fallback_at = Some(Instant::now() + fallback_timeout); } } // Handled internally by the synchronizer. @@ -271,6 +278,9 @@ mod tests { use crate::stores::store_types::StorageItem; use crate::test_common::basic_flag; + const FALLBACK_TIMEOUT: Duration = Duration::from_secs(120); + const RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); + type Selectors = Arc>>; type InitCalls = Arc>>; @@ -443,6 +453,8 @@ mod tests { )]), })], vec![sync_factory(vec![], no_selectors(), false)], + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, ); // Record init-complete calls and wake the test when one arrives. @@ -500,6 +512,8 @@ mod tests { store.clone(), init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, ) .await; @@ -548,6 +562,8 @@ mod tests { store.clone(), init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, ) .await; @@ -581,6 +597,8 @@ mod tests { store, init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, ) .await; @@ -616,6 +634,8 @@ mod tests { store.clone(), init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, ) .await; @@ -648,6 +668,8 @@ mod tests { store.clone(), init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, ) .await; @@ -674,6 +696,8 @@ mod tests { store, init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, )); shutdown_tx.send(()).unwrap(); handle.await.unwrap(); @@ -775,6 +799,8 @@ mod tests { store.clone(), init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, )); handle.await.unwrap(); @@ -825,6 +851,8 @@ mod tests { store.clone(), init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, )); // Wait for the basis, let well past the fallback timeout elapse, then stop. @@ -869,6 +897,8 @@ mod tests { store.clone(), init_complete, shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, )); handle.await.unwrap(); From de3aef7611b380d60e9a48f87fa350e40602c792 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Sun, 9 Aug 2026 18:10:51 -0700 Subject: [PATCH 06/13] chore: Remove the FDv2 transactional-store dead-code allows now that they are consumed --- launchdarkly-server-sdk/src/stores/change_set.rs | 1 - launchdarkly-server-sdk/src/stores/store.rs | 1 - 2 files changed, 2 deletions(-) diff --git a/launchdarkly-server-sdk/src/stores/change_set.rs b/launchdarkly-server-sdk/src/stores/change_set.rs index 0981380b..51de5102 100644 --- a/launchdarkly-server-sdk/src/stores/change_set.rs +++ b/launchdarkly-server-sdk/src/stores/change_set.rs @@ -20,6 +20,5 @@ pub(crate) enum ItemChange { pub(crate) struct ChangeSet { pub(crate) kind: ChangeSetKind, pub(crate) changes: Vec, - #[allow(dead_code)] // Read by the orchestrator in a later phase. pub(crate) selector: Selector, } diff --git a/launchdarkly-server-sdk/src/stores/store.rs b/launchdarkly-server-sdk/src/stores/store.rs index 5201a65f..1c51bb33 100644 --- a/launchdarkly-server-sdk/src/stores/store.rs +++ b/launchdarkly-server-sdk/src/stores/store.rs @@ -34,7 +34,6 @@ pub trait DataStore: Store + Send + Sync { } /// Trait for a data store that accepts atomic batch updates from FDv2 delivery. -#[allow(dead_code)] // Consumed by the FDv2 orchestrator in a later phase. pub(crate) trait TransactionalDataStore: Send + Sync { /// Apply the batch atomically per the change set's kind. fn apply(&mut self, change_set: ChangeSet); From c30482ceec4c90c6ab135a55d09b8ddc567f37e0 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Sun, 9 Aug 2026 19:28:44 -0700 Subject: [PATCH 07/13] chore: Include the source name in FDv2 orchestrator log messages --- launchdarkly-server-sdk/src/fdv2/data_system.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index cc3321bd..8af62e81 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -179,6 +179,7 @@ async fn run( // Initializer phase: try each in order until one yields a basis. for mut initializer in initializers { + let name = initializer.name().to_string(); let mut shutdown = Box::pin(shutdown_receiver.recv()).fuse(); futures::select! { _ = shutdown => return, @@ -190,6 +191,7 @@ async fn run( initialized = true; break; } + debug!("{name} did not provide a basis"); } } } @@ -197,6 +199,7 @@ async fn run( // Synchronizer phase: rotate through synchronizers as the timers fire. let mut current = source_manager.next_synchronizer(); while let Some(mut active) = current { + let name = active.name().to_string(); let has_fallback = source_manager.available_count() > 1; let has_recovery = has_fallback && !source_manager.is_prime(); let mut fallback_at: Option = None; @@ -233,7 +236,7 @@ async fn run( // Sustained interruption starts the fallback countdown. FDv2SourceResult::Interrupted(error) => { if !interrupted_logged { - info!("FDv2 synchronizer interrupted: {}", error.message); + info!("{name} interrupted: {}", error.message); interrupted_logged = true; } if has_fallback && fallback_at.is_none() { @@ -241,9 +244,9 @@ async fn run( } } // Handled internally by the synchronizer. - FDv2SourceResult::Goodbye { .. } => {} + FDv2SourceResult::Goodbye => {} FDv2SourceResult::TerminalError(error) => { - warn!("FDv2 synchronizer terminal error: {}", error.message); + warn!("{name} terminal error: {}", error.message); // Dead source: drop it and advance. source_manager.block_current(); break; From 1909d4b66d6f0542ae88d70b744e8e4f943dd9f2 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Sun, 16 Aug 2026 22:46:34 -0700 Subject: [PATCH 08/13] fix: Ignore no-change changesets in the FDv2 orchestrator --- .../src/fdv2/data_system.rs | 63 +++++++++++++++---- 1 file changed, 51 insertions(+), 12 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index 8af62e81..03deebfb 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -9,7 +9,7 @@ use tokio::time::{sleep_until, Instant}; use crate::data_system::DataSystem; use crate::stores::store::{DataStore, InMemoryDataStore, TransactionalDataStore}; -use super::model::Selector; +use super::model::{ChangeSetKind, Selector}; use super::source::{FDv2SourceResult, Initializer, Synchronizer}; /// Produces a fresh initializer each time the orchestrator starts a run. @@ -185,11 +185,13 @@ async fn run( _ = shutdown => return, event = initializer.run().fuse() => { if let FDv2SourceResult::ChangeSet(change_set) = event.result { - selector = change_set.selector.clone(); - store.write().apply(change_set); - init_complete(true); - initialized = true; - break; + if !matches!(change_set.kind, ChangeSetKind::None) { + selector = change_set.selector.clone(); + store.write().apply(change_set); + init_complete(true); + initialized = true; + break; + } } debug!("{name} did not provide a basis"); } @@ -223,13 +225,15 @@ async fn run( } event = next => match event.result { FDv2SourceResult::ChangeSet(change_set) => { - selector = change_set.selector.clone(); - store.write().apply(change_set); - if !initialized { - init_complete(true); - initialized = true; + if !matches!(change_set.kind, ChangeSetKind::None) { + selector = change_set.selector.clone(); + store.write().apply(change_set); + if !initialized { + init_complete(true); + initialized = true; + } } - // Good data clears the fallback countdown. + // A successful response clears the countdown. fallback_at = None; interrupted_logged = false; } @@ -531,6 +535,41 @@ mod tests { assert_eq!(selectors_seen.lock().unwrap()[0], Some("sel-1".into())); } + #[tokio::test] + async fn none_changeset_does_not_clobber_the_selector() { + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); + let (init_complete, _calls) = recording_init_complete(); + let initializers: Vec> = vec![]; + + // A full basis carrying selector "s1", then a "no changes" (None) changeset. + let source_manager = SourceManager::new(vec![sync_factory( + vec![ + changeset(ChangeSetKind::Full, "flag", Some("s1".into())), + changeset(ChangeSetKind::None, "flag", None), + ], + selectors_seen.clone(), + false, + )]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + source_manager, + store, + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // Exactly three requests, and the one issued after the None changeset still + // carries "s1" -- the None must not reset the selector to None. + let seen = selectors_seen.lock().unwrap(); + assert_eq!(*seen, vec![None, Some("s1".into()), Some("s1".into())]); + } + #[tokio::test] async fn failed_initializers_let_synchronizer_provide_the_basis() { // Store and init-complete recorder. From e30283098f06835d815e1ae3c451e6409f3c02c9 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Wed, 19 Aug 2026 10:41:39 -0700 Subject: [PATCH 09/13] fix: Treat an FDv2 change set as a basis only when it has a selector --- .../src/fdv2/data_system.rs | 113 ++++++++++++++++-- 1 file changed, 105 insertions(+), 8 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index 03deebfb..128475a5 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -184,16 +184,25 @@ async fn run( futures::select! { _ = shutdown => return, event = initializer.run().fuse() => { - if let FDv2SourceResult::ChangeSet(change_set) = event.result { - if !matches!(change_set.kind, ChangeSetKind::None) { - selector = change_set.selector.clone(); + match event.result { + FDv2SourceResult::ChangeSet(change_set) + if !matches!(change_set.kind, ChangeSetKind::None) => + { + let got_basis = change_set.selector.is_some(); + if got_basis { + selector = change_set.selector.clone(); + } store.write().apply(change_set); - init_complete(true); - initialized = true; - break; + if !initialized { + init_complete(true); + initialized = true; + } + if got_basis { + break; + } } + _ => debug!("{name} did not provide a basis"), } - debug!("{name} did not provide a basis"); } } } @@ -226,7 +235,9 @@ async fn run( event = next => match event.result { FDv2SourceResult::ChangeSet(change_set) => { if !matches!(change_set.kind, ChangeSetKind::None) { - selector = change_set.selector.clone(); + if change_set.selector.is_some() { + selector = change_set.selector.clone(); + } store.write().apply(change_set); if !initialized { init_complete(true); @@ -535,6 +546,58 @@ mod tests { assert_eq!(selectors_seen.lock().unwrap()[0], Some("sel-1".into())); } + #[tokio::test] + async fn initializer_payload_without_basis_continues_to_next_initializer() { + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); + let (init_complete, calls) = recording_init_complete(); + + let initializers: Vec> = vec![ + // The first initializer delivers a payload with no basis. + Box::new(MockInitializer { + results: VecDeque::from(vec![changeset( + ChangeSetKind::Full, + "no-basis-flag", + None, + )]), + }), + // The second is a partial basis that merges without clearing the first payload. + Box::new(MockInitializer { + results: VecDeque::from(vec![changeset( + ChangeSetKind::Partial, + "basis-flag", + Some("sel-2".into()), + )]), + }), + ]; + + // The synchronizer only records the selector it is started with. + let source_manager = + SourceManager::new(vec![sync_factory(vec![], selectors_seen.clone(), false)]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // The no-basis payload survives and the second initializer's basis was applied. + assert!(store.read().flag("no-basis-flag").is_some()); + assert!(store.read().flag("basis-flag").is_some()); + + // The synchronizer is started from the second initializer's basis selector. + assert_eq!(selectors_seen.lock().unwrap()[0], Some("sel-2".into())); + + // init_complete fired exactly once. + assert_eq!(*calls.lock().unwrap(), vec![true]); + } + #[tokio::test] async fn none_changeset_does_not_clobber_the_selector() { let store = Arc::new(RwLock::new(InMemoryDataStore::new())); @@ -570,6 +633,40 @@ mod tests { assert_eq!(*seen, vec![None, Some("s1".into()), Some("s1".into())]); } + #[tokio::test] + async fn selectorless_change_does_not_clobber_the_selector() { + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); + let (init_complete, _calls) = recording_init_complete(); + let initializers: Vec> = vec![]; + + // A full basis carrying selector "s1", then a partial change with no selector. + let source_manager = SourceManager::new(vec![sync_factory( + vec![ + changeset(ChangeSetKind::Full, "flag", Some("s1".into())), + changeset(ChangeSetKind::Partial, "flag", None), + ], + selectors_seen.clone(), + false, + )]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializers, + source_manager, + store, + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // The request after the selectorless change still carries "s1". + let seen = selectors_seen.lock().unwrap(); + assert_eq!(*seen, vec![None, Some("s1".into()), Some("s1".into())]); + } + #[tokio::test] async fn failed_initializers_let_synchronizer_provide_the_basis() { // Store and init-complete recorder. From bfd5b918f79d136f119286739bb5fc4706af2c3d Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Wed, 19 Aug 2026 10:47:21 -0700 Subject: [PATCH 10/13] refactor: Simplify the FDv2 is_prime check --- launchdarkly-server-sdk/src/fdv2/data_system.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index 128475a5..0773415d 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -146,7 +146,7 @@ impl SourceManager { .states .iter() .position(|s| *s == SourceState::Available); - matches!((first, self.current_factory_index), (Some(f), Some(c)) if f == c) + first == self.current_factory_index && first.is_some() } fn available_count(&self) -> usize { From abf04632057f42b1db52bd1bcdf238d254036807 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Wed, 19 Aug 2026 14:09:36 -0700 Subject: [PATCH 11/13] refactor: Build FDv2 initializers lazily as the phase iterates --- .../src/fdv2/data_system.rs | 120 ++++++++---------- 1 file changed, 54 insertions(+), 66 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index 0773415d..ad48dd8e 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -25,7 +25,7 @@ pub(crate) trait SynchronizerFactory: Send + Sync { /// FDv2 orchestrator: owns the memory store and keeps it populated by running /// initializers to obtain a basis, then synchronizers for ongoing changes. pub(crate) struct FDv2DataSystem { - initializer_factories: Vec>, + initializer_factories: Vec>, synchronizer_factories: Vec>, fallback_timeout: Duration, recovery_timeout: Duration, @@ -34,7 +34,7 @@ pub(crate) struct FDv2DataSystem { impl FDv2DataSystem { pub(crate) fn new( - initializer_factories: Vec>, + initializer_factories: Vec>, synchronizer_factories: Vec>, fallback_timeout: Duration, recovery_timeout: Duration, @@ -55,16 +55,12 @@ impl DataSystem for FDv2DataSystem { init_complete: Arc, shutdown_receiver: broadcast::Receiver<()>, ) { - let initializers = self - .initializer_factories - .iter() - .map(|f| f.create()) - .collect(); + let initializer_factories = self.initializer_factories.clone(); let source_manager = SourceManager::new(self.synchronizer_factories.clone()); let store = self.store.clone(); tokio::spawn(run( - initializers, + initializer_factories, source_manager, store, init_complete, @@ -166,7 +162,7 @@ async fn deadline(at: Option) { } async fn run( - initializers: Vec>, + initializer_factories: Vec>, mut source_manager: SourceManager, store: Arc>, init_complete: Arc, @@ -178,7 +174,8 @@ async fn run( let mut initialized = false; // Initializer phase: try each in order until one yields a basis. - for mut initializer in initializers { + for factory in initializer_factories { + let mut initializer = factory.create(); let name = initializer.name().to_string(); let mut shutdown = Box::pin(shutdown_receiver.recv()).fuse(); futures::select! { @@ -387,6 +384,13 @@ mod tests { } } + /// A single-initializer factory scripted with the given results. + fn init_factory(results: Vec) -> Arc { + Arc::new(MockInitializerFactory { + results: Mutex::new(results), + }) + } + struct MockSynchronizerFactory { results: Mutex>, selectors_seen: Selectors, @@ -463,13 +467,11 @@ mod tests { async fn start_applies_basis_and_exposes_it_via_store_handle() { // A data system whose sole initializer yields one full basis. let system = FDv2DataSystem::new( - vec![Box::new(MockInitializerFactory { - results: Mutex::new(vec![changeset( - ChangeSetKind::Full, - "f1", - Some("s1".into()), - )]), - })], + vec![init_factory(vec![changeset( + ChangeSetKind::Full, + "f1", + Some("s1".into()), + )])], vec![sync_factory(vec![], no_selectors(), false)], FALLBACK_TIMEOUT, RECOVERY_TIMEOUT, @@ -504,13 +506,12 @@ mod tests { let (init_complete, calls) = recording_init_complete(); // Initializer delivers a full basis carrying selector "sel-1". - let initializers: Vec> = vec![Box::new(MockInitializer { - results: VecDeque::from(vec![changeset( + let initializer_factories: Vec> = + vec![init_factory(vec![changeset( ChangeSetKind::Full, "init-flag", Some("sel-1".into()), - )]), - })]; + )])]; // Synchronizer delivers a partial change carrying selector "sel-2". let source_manager = SourceManager::new(vec![sync_factory( @@ -525,7 +526,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, @@ -552,23 +553,15 @@ mod tests { let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); let (init_complete, calls) = recording_init_complete(); - let initializers: Vec> = vec![ + let initializer_factories: Vec> = vec![ // The first initializer delivers a payload with no basis. - Box::new(MockInitializer { - results: VecDeque::from(vec![changeset( - ChangeSetKind::Full, - "no-basis-flag", - None, - )]), - }), + init_factory(vec![changeset(ChangeSetKind::Full, "no-basis-flag", None)]), // The second is a partial basis that merges without clearing the first payload. - Box::new(MockInitializer { - results: VecDeque::from(vec![changeset( - ChangeSetKind::Partial, - "basis-flag", - Some("sel-2".into()), - )]), - }), + init_factory(vec![changeset( + ChangeSetKind::Partial, + "basis-flag", + Some("sel-2".into()), + )]), ]; // The synchronizer only records the selector it is started with. @@ -577,7 +570,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, @@ -603,7 +596,7 @@ mod tests { let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); let (init_complete, _calls) = recording_init_complete(); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // A full basis carrying selector "s1", then a "no changes" (None) changeset. let source_manager = SourceManager::new(vec![sync_factory( @@ -617,7 +610,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store, init_complete, @@ -638,7 +631,7 @@ mod tests { let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); let (init_complete, _calls) = recording_init_complete(); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // A full basis carrying selector "s1", then a partial change with no selector. let source_manager = SourceManager::new(vec![sync_factory( @@ -652,7 +645,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store, init_complete, @@ -674,13 +667,9 @@ mod tests { let (init_complete, calls) = recording_init_complete(); // Both initializers fail without producing a basis. - let initializers: Vec> = vec![ - Box::new(MockInitializer { - results: VecDeque::from(vec![interrupted()]), - }), - Box::new(MockInitializer { - results: VecDeque::from(vec![terminal()]), - }), + let initializer_factories: Vec> = vec![ + init_factory(vec![interrupted()]), + init_factory(vec![terminal()]), ]; // The synchronizer then delivers the basis. @@ -696,7 +685,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, @@ -718,9 +707,8 @@ mod tests { let (init_complete, calls) = recording_init_complete(); // The initializer fails. - let initializers: Vec> = vec![Box::new(MockInitializer { - results: VecDeque::from(vec![terminal()]), - })]; + let initializer_factories: Vec> = + vec![init_factory(vec![terminal()])]; // The synchronizer reports an interruption, then fails terminally. let source_manager = SourceManager::new(vec![sync_factory( @@ -731,7 +719,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store, init_complete, @@ -750,7 +738,7 @@ mod tests { // Store and init-complete recorder; no initializers. let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let (init_complete, calls) = recording_init_complete(); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // First synchronizer fails terminally; the second provides the basis. let source_manager = SourceManager::new(vec![ @@ -768,7 +756,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, @@ -788,7 +776,7 @@ mod tests { // Store and init-complete recorder; no initializers. let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let (init_complete, calls) = recording_init_complete(); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // A single synchronizer: an interruption, then a basis on the retry. let source_manager = SourceManager::new(vec![sync_factory( @@ -802,7 +790,7 @@ mod tests { let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, @@ -822,7 +810,7 @@ mod tests { // Store and init-complete recorder; no initializers. let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let (init_complete, calls) = recording_init_complete(); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // A synchronizer whose next() never resolves. let source_manager = SourceManager::new(vec![sync_factory(vec![], no_selectors(), true)]); @@ -830,7 +818,7 @@ mod tests { // Drive the run on a task, then signal shutdown. let handle = tokio::spawn(run( - initializers, + initializer_factories, source_manager, store, init_complete, @@ -914,7 +902,7 @@ mod tests { // No initializers; the prime interrupts then idles. let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let (init_complete, calls) = recording_init_complete(); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // Prime only ever interrupts; the fallback stands by with a basis. let source_manager = SourceManager::new(vec![ @@ -933,7 +921,7 @@ mod tests { // Paused time auto-advances past the fallback timeout while the prime idles. let handle = tokio::spawn(run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, @@ -960,7 +948,7 @@ mod tests { sink.lock().unwrap().push(success); waker.notify_one(); }); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // Prime interrupts (arming the timer), delivers a basis (clearing it), then idles. let source_manager = SourceManager::new(vec![ @@ -985,7 +973,7 @@ mod tests { let (shutdown_tx, shutdown_rx) = broadcast::channel(1); let handle = tokio::spawn(run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, @@ -1010,7 +998,7 @@ mod tests { // No initializers; the prime is down, so the run falls back then recovers. let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let (init_complete, calls) = recording_init_complete(); - let initializers: Vec> = vec![]; + let initializer_factories: Vec> = vec![]; // Prime recovers on rebuild; the fallback supplies a basis then idles. let source_manager = SourceManager::new(vec![ @@ -1031,7 +1019,7 @@ mod tests { // Paused time auto-advances through the fallback then recovery timeouts. let handle = tokio::spawn(run( - initializers, + initializer_factories, source_manager, store.clone(), init_complete, From 9f1f16c4d2995e31f48bc9f5db0022c00073aaa0 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Mon, 24 Aug 2026 10:07:40 -0700 Subject: [PATCH 12/13] fix: Signal FDv2 init on a full payload and update the selector on data changes --- .../src/fdv2/data_system.rs | 132 +++++++++++++++--- 1 file changed, 115 insertions(+), 17 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index ad48dd8e..d05f767a 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -173,7 +173,7 @@ async fn run( let mut selector: Selector = None; let mut initialized = false; - // Initializer phase: try each in order until one yields a basis. + // Initializer phase: try each until one yields a selector. for factory in initializer_factories { let mut initializer = factory.create(); let name = initializer.name().to_string(); @@ -182,19 +182,18 @@ async fn run( _ = shutdown => return, event = initializer.run().fuse() => { match event.result { - FDv2SourceResult::ChangeSet(change_set) - if !matches!(change_set.kind, ChangeSetKind::None) => - { - let got_basis = change_set.selector.is_some(); - if got_basis { + FDv2SourceResult::ChangeSet(change_set) => { + let is_full = matches!(change_set.kind, ChangeSetKind::Full); + let has_selector = change_set.selector.is_some(); + if !matches!(change_set.kind, ChangeSetKind::None) { selector = change_set.selector.clone(); } store.write().apply(change_set); - if !initialized { + if is_full && !initialized { init_complete(true); initialized = true; } - if got_basis { + if has_selector { break; } } @@ -231,12 +230,11 @@ async fn run( } event = next => match event.result { FDv2SourceResult::ChangeSet(change_set) => { + let is_full = matches!(change_set.kind, ChangeSetKind::Full); if !matches!(change_set.kind, ChangeSetKind::None) { - if change_set.selector.is_some() { - selector = change_set.selector.clone(); - } + selector = change_set.selector.clone(); store.write().apply(change_set); - if !initialized { + if is_full && !initialized { init_complete(true); initialized = true; } @@ -271,7 +269,7 @@ async fn run( current = source_manager.next_synchronizer(); } - // Every source blocked without ever obtaining a basis. + // Every source blocked without ever obtaining a full payload. if !initialized { init_complete(false); } @@ -627,13 +625,13 @@ mod tests { } #[tokio::test] - async fn selectorless_change_does_not_clobber_the_selector() { + async fn selectorless_change_clears_the_selector() { let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); let (init_complete, _calls) = recording_init_complete(); let initializer_factories: Vec> = vec![]; - // A full basis carrying selector "s1", then a partial change with no selector. + // A full payload carrying selector "s1", then a partial change with no selector. let source_manager = SourceManager::new(vec![sync_factory( vec![ changeset(ChangeSetKind::Full, "flag", Some("s1".into())), @@ -655,9 +653,109 @@ mod tests { ) .await; - // The request after the selectorless change still carries "s1". + // The selectorless change advanced state, so the stale selector is dropped. let seen = selectors_seen.lock().unwrap(); - assert_eq!(*seen, vec![None, Some("s1".into()), Some("s1".into())]); + assert_eq!(*seen, vec![None, Some("s1".into()), None]); + } + + #[tokio::test] + async fn initializer_delta_does_not_initialize() { + // A spec-compliant backend never returns a delta to an initializer. + // Here the sole initializer returns one anyway, which is not a full payload. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializer_factories: Vec> = + vec![init_factory(vec![changeset( + ChangeSetKind::Partial, + "delta-flag", + Some("s1".into()), + )])]; + + // The synchronizer then fails terminally, exhausting every source. + let source_manager = + SourceManager::new(vec![sync_factory(vec![terminal()], no_selectors(), false)]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializer_factories, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // The delta applied, but a delta is not a full payload, so failure is signaled. + assert!(store.read().flag("delta-flag").is_some()); + assert_eq!(*calls.lock().unwrap(), vec![false]); + } + + #[tokio::test] + async fn synchronizer_delta_does_not_initialize() { + // No initializers, so the store starts uninitialized. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializer_factories: Vec> = vec![]; + + // A spec-compliant backend never sends a delta before a full payload. + // Here the synchronizer delivers one first anyway, then fails terminally. + let source_manager = SourceManager::new(vec![sync_factory( + vec![ + changeset(ChangeSetKind::Partial, "delta-flag", Some("s1".into())), + terminal(), + ], + no_selectors(), + false, + )]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializer_factories, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // The delta applied, but a delta is not a full payload, so failure is signaled. + assert!(store.read().flag("delta-flag").is_some()); + assert_eq!(*calls.lock().unwrap(), vec![false]); + } + + #[tokio::test] + async fn selectorless_full_initializes() { + // No initializers, so the synchronizer provides the full payload. + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let (init_complete, calls) = recording_init_complete(); + let initializer_factories: Vec> = vec![]; + + // A full payload with no selector still initializes, as the FDv1 adapter produces. + let source_manager = SourceManager::new(vec![sync_factory( + vec![changeset(ChangeSetKind::Full, "full-flag", None)], + no_selectors(), + false, + )]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializer_factories, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // A selectorless full still initializes. + assert_eq!(*calls.lock().unwrap(), vec![true]); + assert!(store.read().flag("full-flag").is_some()); } #[tokio::test] From f8ccaf35e4b817c3ccf1a8609e7cd4cdf29496a8 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Tue, 25 Aug 2026 15:14:49 -0700 Subject: [PATCH 13/13] fix: Signal FDv2 init on a basis and defer a selector-less full --- .../src/fdv2/data_system.rs | 118 +++++++++++++++--- 1 file changed, 103 insertions(+), 15 deletions(-) diff --git a/launchdarkly-server-sdk/src/fdv2/data_system.rs b/launchdarkly-server-sdk/src/fdv2/data_system.rs index d05f767a..34cc0d04 100644 --- a/launchdarkly-server-sdk/src/fdv2/data_system.rs +++ b/launchdarkly-server-sdk/src/fdv2/data_system.rs @@ -172,8 +172,10 @@ async fn run( ) { let mut selector: Selector = None; let mut initialized = false; + // Whether an initializer produced a full payload. + let mut got_full = false; - // Initializer phase: try each until one yields a selector. + // Initializer phase: try each until one yields a basis. for factory in initializer_factories { let mut initializer = factory.create(); let name = initializer.name().to_string(); @@ -184,17 +186,16 @@ async fn run( match event.result { FDv2SourceResult::ChangeSet(change_set) => { let is_full = matches!(change_set.kind, ChangeSetKind::Full); - let has_selector = change_set.selector.is_some(); + let has_basis = is_full && change_set.selector.is_some(); if !matches!(change_set.kind, ChangeSetKind::None) { selector = change_set.selector.clone(); } store.write().apply(change_set); - if is_full && !initialized { - init_complete(true); - initialized = true; + if is_full { + got_full = true; } - if has_selector { - break; + if has_basis { + break; // transition to synchronizer phase } } _ => debug!("{name} did not provide a basis"), @@ -203,6 +204,11 @@ async fn run( } } + if got_full && !initialized { + init_complete(true); + initialized = true; + } + // Synchronizer phase: rotate through synchronizers as the timers fire. let mut current = source_manager.next_synchronizer(); while let Some(mut active) = current { @@ -546,18 +552,18 @@ mod tests { } #[tokio::test] - async fn initializer_payload_without_basis_continues_to_next_initializer() { + async fn selectorless_full_continues_to_next_initializer() { let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); let (init_complete, calls) = recording_init_complete(); let initializer_factories: Vec> = vec![ - // The first initializer delivers a payload with no basis. + // The first initializer delivers a full payload with no selector. init_factory(vec![changeset(ChangeSetKind::Full, "no-basis-flag", None)]), - // The second is a partial basis that merges without clearing the first payload. + // The second delivers a delta carrying a selector, so it merges over the first. init_factory(vec![changeset( ChangeSetKind::Partial, - "basis-flag", + "merged-flag", Some("sel-2".into()), )]), ]; @@ -578,17 +584,99 @@ mod tests { ) .await; - // The no-basis payload survives and the second initializer's basis was applied. + // The selector-less payload survives and the delta merged over it. assert!(store.read().flag("no-basis-flag").is_some()); - assert!(store.read().flag("basis-flag").is_some()); + assert!(store.read().flag("merged-flag").is_some()); - // The synchronizer is started from the second initializer's basis selector. + // The selector-less payload did not stop the initializers, so the synchronizer + // starts from the second initializer's selector. assert_eq!(selectors_seen.lock().unwrap()[0], Some("sel-2".into())); // init_complete fired exactly once. assert_eq!(*calls.lock().unwrap(), vec![true]); } + #[tokio::test] + async fn initializer_selectorless_full_defers_signal() { + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + + // Capture whether the second initializer's flag is present when the signal fires. + let flag_at_signal = Arc::new(Mutex::new(None)); + let probe = store.clone(); + let sink = flag_at_signal.clone(); + let init_complete: Arc = Arc::new(move |_| { + *sink.lock().unwrap() = Some(probe.read().flag("from-second").is_some()); + }); + + // Neither initializer produces a basis. + let initializer_factories: Vec> = vec![ + init_factory(vec![changeset(ChangeSetKind::Full, "from-first", None)]), + init_factory(vec![changeset(ChangeSetKind::Full, "from-second", None)]), + ]; + + let source_manager = SourceManager::new(vec![sync_factory(vec![], no_selectors(), false)]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializer_factories, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // The signal waited until the initializers were exhausted, so the second + // initializer's payload was already applied when it fired. + assert_eq!(*flag_at_signal.lock().unwrap(), Some(true)); + } + + #[tokio::test] + async fn basis_stops_later_initializers() { + let store = Arc::new(RwLock::new(InMemoryDataStore::new())); + let selectors_seen: Selectors = Arc::new(Mutex::new(Vec::new())); + let (init_complete, calls) = recording_init_complete(); + + // The first initializer yields a basis; the second would apply another flag. + let initializer_factories: Vec> = vec![ + init_factory(vec![changeset( + ChangeSetKind::Full, + "from-first", + Some("s1".into()), + )]), + init_factory(vec![changeset( + ChangeSetKind::Full, + "from-second", + Some("s2".into()), + )]), + ]; + + let source_manager = + SourceManager::new(vec![sync_factory(vec![], selectors_seen.clone(), false)]); + let (_shutdown_tx, shutdown_rx) = broadcast::channel(1); + + run( + initializer_factories, + source_manager, + store.clone(), + init_complete, + shutdown_rx, + FALLBACK_TIMEOUT, + RECOVERY_TIMEOUT, + ) + .await; + + // The basis ended the initializer phase, so the second initializer never ran. + assert!(store.read().flag("from-first").is_some()); + assert!(store.read().flag("from-second").is_none()); + assert_eq!(*calls.lock().unwrap(), vec![true]); + + // The synchronizer starts from the basis selector. + assert_eq!(selectors_seen.lock().unwrap()[0], Some("s1".into())); + } + #[tokio::test] async fn none_changeset_does_not_clobber_the_selector() { let store = Arc::new(RwLock::new(InMemoryDataStore::new())); @@ -728,7 +816,7 @@ mod tests { } #[tokio::test] - async fn selectorless_full_initializes() { + async fn synchronizer_selectorless_full_initializes() { // No initializers, so the synchronizer provides the full payload. let store = Arc::new(RwLock::new(InMemoryDataStore::new())); let (init_complete, calls) = recording_init_complete();