diff --git a/ldclient/async_client.py b/ldclient/async_client.py index 898c2bf8..d726d199 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -197,6 +197,12 @@ async def __start_up(self, start_wait: float): # Start the big-segment status poll now that a loop is running. self.__big_segment_store_manager.start() + # FDv2 builds its data sources from builders; wire the shared session into + # them before starting (FDv1 pulls the session itself via its provider). + datasystem_config = self._config.datasystem_config + if datasystem_config is not None and not self._config.offline: + self._wire_data_source_sessions(datasystem_config) + if self._config.offline: log.info("Started LaunchDarkly Client in offline mode") @@ -243,7 +249,9 @@ def _make_data_system(self) -> AsyncDataSystem: return AsyncFDv1(self._config, self._select_feature_store(), self._get_session) - raise NotImplementedError("FDv2 is not yet supported in the async client") + from ldclient.impl.datasystem.async_fdv2 import AsyncFDv2 + + return AsyncFDv2(self._config, datasystem_config) def _select_feature_store(self) -> AsyncFeatureStore: """Choose the async feature store for the v1 data system based on the @@ -253,6 +261,34 @@ def _select_feature_store(self) -> AsyncFeatureStore: return AsyncInMemoryFeatureStore() return feature_store + def _wire_data_source_sessions(self, data_system_config) -> None: + """Provide the client's aiohttp session to any async data source + builders so the sources they build share the client's connection pool.""" + from ldclient.impl.datasourcev2.async_polling import ( + AsyncFallbackToFDv1PollingDataSourceBuilder, + AsyncPollingDataSourceBuilder + ) + from ldclient.impl.datasourcev2.async_streaming import ( + AsyncStreamingDataSourceBuilder + ) + + builders = list(data_system_config.initializers or []) + list( + data_system_config.synchronizers or [] + ) + if data_system_config.fdv1_fallback_synchronizer is not None: + builders.append(data_system_config.fdv1_fallback_synchronizer) + + for builder in builders: + if isinstance( + builder, + ( + AsyncFallbackToFDv1PollingDataSourceBuilder, + AsyncPollingDataSourceBuilder, + AsyncStreamingDataSourceBuilder, + ), + ): + builder.session(self._get_session()) + async def __register_plugins(self, environment_metadata: EnvironmentMetadata): for plugin in self._config.plugins: try: diff --git a/ldclient/impl/datasystem/async_fdv2.py b/ldclient/impl/datasystem/async_fdv2.py new file mode 100644 index 00000000..7c6d2e7f --- /dev/null +++ b/ldclient/impl/datasystem/async_fdv2.py @@ -0,0 +1,423 @@ +""" +FDv2 data system coordinator: manages initializers and synchronizers to +obtain and keep the SDK's data up-to-date, operating with an optional +persistent store in read-only or read/write mode. +""" + +import time +from typing import Optional, cast + +from ldclient.async_config import AsyncConfig +from ldclient.config import DataSystemConfig +from ldclient.impl.aio.concurrency import ( + AsyncEvent, + AsyncLock, + AsyncQueue, + AsyncRepeatingTask, + AsyncTaskRunner, + TaskHandle, + join_handle, + spawn_handle +) +from ldclient.impl.datasystem import AsyncDataSystem, DiagnosticSource +from ldclient.impl.datasystem.fdv2_common import ( + ConditionDirective, + DataSourceStatusProviderImpl, + DataStoreStatusProviderImpl, + FeatureStoreClientWrapper, + _FDv2Base, + fallback_condition, + recovery_condition +) +from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log +from ldclient.interfaces import ( + AsyncInitializer, + AsyncReadOnlyStore, + AsyncSynchronizer, + DataSourceErrorInfo, + DataSourceErrorKind, + DataSourceState, + ReadOnlyStore +) + + +class _AsyncStoreView: + """Wraps FDv2's synchronous in-memory active store as an + :class:`AsyncReadOnlyStore`. This lets the evaluation path use one async + interface for both FDv1 and FDv2. Reads are in-memory dict lookups, so + nothing is awaited.""" + + def __init__(self, store: ReadOnlyStore): + self._store = store + + async def get(self, kind, key): + return self._store.get(kind, key, lambda x: x) + + async def all(self, kind): + return self._store.all(kind, lambda x: x) + + +class AsyncFDv2(_FDv2Base, AsyncDataSystem): + """ + AsyncFDv2 is an implementation of the AsyncDataSystem interface that uses the Flag Delivery V2 protocol + for obtaining and keeping data up-to-date. Additionally, it operates with an optional persistent + store in read-only or read/write mode. + """ + + def __init__( + self, + config: AsyncConfig, + data_system_config: DataSystemConfig, + ): + """ + Initialize a new AsyncFDv2 data system. + + :param config: Configuration for initializers and synchronizers + :param persistent_store: Optional persistent store for data persistence + :param store_writable: Whether the persistent store should be written to + :param disabled: Whether the data system is disabled (offline mode) + """ + super().__init__(config, data_system_config) + + # Concurrency + self._stop_event = AsyncEvent() + self._lock = AsyncLock() + self._active_synchronizer: Optional[AsyncSynchronizer] = None + self._runner = AsyncTaskRunner() + + def start(self, set_on_ready: AsyncEvent): + """ + Start the AsyncFDv2 data system. + + :param set_on_ready: Event to set when the system is ready or has failed + """ + if self._disabled: + log.warning("Data system is disabled, SDK will return application-defined default values") + set_on_ready.set() + return + + self._stop_event.clear() + + # Start the main coordination loop + self._runner.spawn("AsyncFDv2-main", lambda: self._run_main_loop(set_on_ready)) + + async def stop(self): + """Stop the AsyncFDv2 data system and all the work it is coordinating.""" + self._stop_event.set() + + async with self._lock: + if self._active_synchronizer is not None: + try: + await self._active_synchronizer.stop() + except Exception as e: + log.error("Error stopping active data source: %s", e) + + # Wait for the coordinator's background work to complete + await self._runner.stop_all(timeout=5.0) + + # Close the store + self._store.close() + + async def _run_main_loop(self, set_on_ready: AsyncEvent): + """Main coordination loop that manages initializers and synchronizers.""" + try: + self._data_source_status_provider.update_status( + DataSourceState.INITIALIZING, None + ) + + # Run initializers first + fallback_requested = await self._run_initializers(set_on_ready) + + # If an initializer asked the SDK to fall back to FDv1, halt the + # configured FDv2 chain and switch terminally to the FDv1 Fallback + # Synchronizer (or transition to OFF if none is configured). + if fallback_requested: + if self._fdv1_fallback_synchronizer_builder is not None: + log.warning("Falling back to FDv1 protocol") + self._synchronizers = [self._fdv1_fallback_synchronizer_builder] + else: + log.warning( + "Initializer requested FDv1 fallback but none configured" + ) + self._synchronizers = [] + self._data_source_status_provider.update_status( + DataSourceState.OFF, + self._data_source_status_provider.status.error, + ) + set_on_ready.set() + return + + # Run synchronizers + await self._run_synchronizers(set_on_ready) + + except Exception as e: + log.error("Error in AsyncFDv2 main loop: %s", e) + # Ensure ready event is set even on error + if not set_on_ready.is_set(): + set_on_ready.set() + + async def _run_initializers(self, set_on_ready: AsyncEvent) -> bool: + """ + Run initializers to get initial data. + + Returns True when an initializer requested the FDv1 Fallback Directive + (via the X-LD-FD-Fallback response header). When that happens, any + accompanying payload is applied first so evaluations can serve the + server-provided data while the FDv1 synchronizer spins up; the caller + is then responsible for switching to the FDv1 Fallback Synchronizer. + """ + if self._data_system_config.initializers is None: + return False + + for initializer_builder in self._data_system_config.initializers: + if self._stop_event.is_set(): + return False + + try: + # DataSystemConfig types builders with the sync Initializer; + # async data systems are configured with async builders. + initializer = cast(AsyncInitializer, initializer_builder.build(self._config)) + log.info("Attempting to initialize via %s", initializer.name) + + basis_result = await initializer.fetch(self._store) + + if isinstance(basis_result, _Fail): + log.warning("Initializer %s failed: %s", initializer.name, basis_result.error) + # An error response can still carry the FDv1 fallback directive. + if basis_result.headers is not None and \ + basis_result.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': + log.warning( + "Initializer %s requested fallback to FDv1 protocol", + initializer.name, + ) + # Surface the underlying error on the status so + # programmatic monitors can see why FDv2 shut down. + self._data_source_status_provider.update_status( + DataSourceState.INITIALIZING, + DataSourceErrorInfo( + kind=DataSourceErrorKind.UNKNOWN, + status_code=0, + time=time.time(), + message=basis_result.error, + ), + ) + return True + continue + + basis = basis_result.value + log.info("Initialized via %s", initializer.name) + + # Apply the basis to the store + self._store.apply(basis.change_set, basis.persist) + + # Set ready event if and only if a selector is defined for the changeset + selector_defined = basis.change_set.selector.is_defined() + if selector_defined: + set_on_ready.set() + + if basis.fallback_to_fdv1: + log.warning( + "Initializer %s requested fallback to FDv1 protocol", + initializer.name, + ) + return True + + if selector_defined: + return False + except Exception as e: + log.error("Initializer failed with exception: %s", e) + return False + + async def _run_synchronizers(self, set_on_ready: AsyncEvent): + """Run synchronizers to keep data up-to-date.""" + # If no synchronizers configured, just set ready and return + if len(self._synchronizers) == 0: + set_on_ready.set() + return + + self._runner.spawn( + "AsyncFDv2-synchronizers", + lambda: self._synchronizer_loop(set_on_ready), + ) + + async def _synchronizer_loop(self, set_on_ready: AsyncEvent): + try: + # Make a working copy of the synchronizers list + synchronizers_list = list(self._synchronizers) + current_index = 0 + + # Always ensure ready event is set when we exit + while not self._stop_event.is_set() and len(synchronizers_list) > 0: + try: + async with self._lock: + synchronizer: AsyncSynchronizer = synchronizers_list[current_index].build(self._config) + self._active_synchronizer = synchronizer + if isinstance(synchronizer, DiagnosticSource) and self._diagnostic_accumulator is not None: + synchronizer.set_diagnostic_accumulator(self._diagnostic_accumulator) + + log.info("Synchronizer %s (index %d) is starting", synchronizer.name, current_index) + + directive = await self._consume_synchronizer_results( + synchronizer, set_on_ready, current_index != 0 + ) + + if directive == ConditionDirective.FDV1: + # Abandon all synchronizers and use only fdv1 fallback + log.warning("Falling back to FDv1 protocol") + if self._fdv1_fallback_synchronizer_builder is not None: + synchronizers_list = [self._fdv1_fallback_synchronizer_builder] + current_index = 0 + else: + log.warning("Synchronizer requested FDv1 fallback but none configured") + synchronizers_list = [] + self._data_source_status_provider.update_status( + DataSourceState.OFF, + self._data_source_status_provider.status.error + ) + break + continue + elif directive == ConditionDirective.REMOVE: + # Permanent failure - remove synchronizer from list + log.warning("Synchronizer %s permanently failed, removing from list", synchronizer.name) + del synchronizers_list[current_index] + + if len(synchronizers_list) == 0: + log.warning("No more synchronizers available") + self._data_source_status_provider.update_status( + DataSourceState.OFF, + self._data_source_status_provider.status.error + ) + break + + # Adjust index if we're now beyond the end of the list + # If we deleted the last synchronizer, wrap to the beginning + if current_index >= len(synchronizers_list): + current_index = 0 + # Note: If we deleted a middle element, current_index now points to + # what was the next element (shifted down), which is correct + continue + # Condition was met - determine next synchronizer based on directive + elif directive == ConditionDirective.RECOVER: + log.info("Recovery condition met, returning to first synchronizer") + current_index = 0 + elif directive == ConditionDirective.FALLBACK: + # Fallback to next synchronizer (wraps to 0 at end) + current_index = (current_index + 1) % len(synchronizers_list) + log.info("Fallback condition met, moving to synchronizer at index %d", current_index) + + except Exception as e: + log.error("Failed to build or run synchronizer: %s", e) + break + + except Exception as e: + log.error("Error in synchronizer loop: %s", e) + finally: + # Ensure we always set the ready event when exiting + set_on_ready.set() + async with self._lock: + if self._active_synchronizer is not None: + await self._active_synchronizer.stop() + self._active_synchronizer = None + + async def _consume_synchronizer_results( + self, + synchronizer: AsyncSynchronizer, + set_on_ready: AsyncEvent, + check_recovery: bool, + ) -> ConditionDirective: + """ + Consume results from a synchronizer until a condition is met or it fails. + + :return: Tuple of (should_remove_sync, fallback_to_fdv1, directive) + """ + action_queue: AsyncQueue = AsyncQueue() + timer = AsyncRepeatingTask( + label="AsyncFDv2-sync-cond-timer", + interval=10, + initial_delay=10, + callable=lambda: action_queue.put("check") + ) + + async def reader(): + try: + async for update in synchronizer.sync(self._store): + await action_queue.put(update) + finally: + await action_queue.put("quit") + + sync_reader: Optional[TaskHandle] = None + + try: + timer.start() + sync_reader = spawn_handle("AsyncFDv2-sync-reader", reader) + + while True: + update = await action_queue.get() + if isinstance(update, str): + if update == "quit": + break + + if update == "check": + # Check condition periodically + current_status = self._data_source_status_provider.status + if check_recovery and recovery_condition(current_status): + return ConditionDirective.RECOVER + if fallback_condition(current_status): + return ConditionDirective.FALLBACK + continue + + log.info("Synchronizer %s update: %s", synchronizer.name, update.state) + if self._stop_event.is_set(): + return ConditionDirective.FALLBACK + + # Handle the update + if update.change_set is not None: + self._store.apply(update.change_set, True) + + # Set ready event on first valid update + if update.state == DataSourceState.VALID and not set_on_ready.is_set(): + set_on_ready.set() + + # Update status + self._data_source_status_provider.update_status(update.state, update.error) + + # Check if we should fall back to FDv1 immediately. fallback_to_fdv1 + # may ride along on a Valid update (payload + directive in the same + # response), in which case the ChangeSet has already been applied + # above before we hand off. + if update.fallback_to_fdv1: + return ConditionDirective.FDV1 + + # Check for OFF state indicating permanent failure + if update.state == DataSourceState.OFF: + return ConditionDirective.REMOVE + except Exception as e: + log.error("Error consuming synchronizer results: %s", e) + return ConditionDirective.REMOVE + finally: + timer.stop() + if sync_reader is not None: + sync_reader.cancel() + + await synchronizer.stop() + if sync_reader is not None: + await join_handle(sync_reader, 0.5) + + # If we reach here, the synchronizer's iterator completed normally (no more updates) + # For continuous synchronizers (streaming/polling), this is unexpected and indicates + # the synchronizer can't provide more updates, so we should remove it and fall back + return ConditionDirective.REMOVE + + @property + def store(self) -> AsyncReadOnlyStore: + """Get the underlying store for flag evaluation.""" + return _AsyncStoreView(self._store.get_active_store()) + + +__all__ = [ + 'AsyncFDv2', + 'ConditionDirective', + 'DataSourceStatusProviderImpl', + 'DataStoreStatusProviderImpl', + 'FeatureStoreClientWrapper', +] diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 3359be3f..b7563ba7 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -3,22 +3,18 @@ from threading import Event, Thread from typing import Any, Callable, Dict, List, Optional -from ldclient.config import Config, DataSourceBuilder, DataSystemConfig -from ldclient.impl.datasystem import ( - DataAvailability, - DataSystem, - DiagnosticAccumulator, - DiagnosticSource -) +from ldclient.config import Config, DataSystemConfig +from ldclient.impl.datasystem import DataSystem, DiagnosticSource from ldclient.impl.datasystem.fdv2_common import ( ConditionDirective, DataSourceStatusProviderImpl, DataStoreStatusProviderImpl, - FeatureStoreClientWrapper + FeatureStoreClientWrapper, + _FDv2Base, + fallback_condition, + recovery_condition ) -from ldclient.impl.datasystem.store import Store from ldclient.impl.flag_tracker import FlagTrackerImpl -from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log @@ -26,11 +22,6 @@ DataSourceErrorInfo, DataSourceErrorKind, DataSourceState, - DataSourceStatus, - DataSourceStatusProvider, - DataStoreMode, - DataStoreStatus, - DataStoreStatusProvider, FlagTracker, ReadOnlyStore, Synchronizer @@ -38,7 +29,7 @@ from ldclient.versioned_data_kind import VersionedDataKind -class FDv2(DataSystem): +class FDv2(_FDv2Base, DataSystem): """ FDv2 is an implementation of the DataSystem interface that uses the Flag Delivery V2 protocol for obtaining and keeping data up-to-date. Additionally, it operates with an optional persistent @@ -58,37 +49,7 @@ def __init__( :param store_writable: Whether the persistent store should be written to :param disabled: Whether the data system is disabled (offline mode) """ - self._config = config - self._data_system_config = data_system_config - self._synchronizers: List[DataSourceBuilder[Synchronizer]] = list(data_system_config.synchronizers) if data_system_config.synchronizers else [] - self._fdv1_fallback_synchronizer_builder = data_system_config.fdv1_fallback_synchronizer - self._disabled = self._config.offline - - # Diagnostic accumulator provided by client for streaming metrics - self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None - - # Set up event listeners - self._flag_change_listeners = Listeners() - self._change_set_listeners = Listeners() - self._data_store_listeners = Listeners() - - self._data_store_listeners.add(self._persistent_store_outage_recovery) - - # Create the store - self._store = Store(self._flag_change_listeners, self._change_set_listeners) - - # Status providers - self._data_source_status_provider = DataSourceStatusProviderImpl(Listeners()) - self._data_store_status_provider = DataStoreStatusProviderImpl(None, self._data_store_listeners) - - # Configure persistent store if provided - if self._data_system_config.data_store is not None: - self._data_store_status_provider = DataStoreStatusProviderImpl(self._data_system_config.data_store, self._data_store_listeners) - writable = self._data_system_config.data_store_mode == DataStoreMode.READ_WRITE - wrapper = FeatureStoreClientWrapper(self._data_system_config.data_store, self._data_store_status_provider) - self._store.with_persistence( - wrapper, writable, self._data_store_status_provider - ) + super().__init__(config, data_system_config) # Threading self._stop_event = Event() @@ -97,12 +58,6 @@ def __init__( self._threads: List[Thread] = [] self._environment_id: Optional[str] = None - # Track configuration - self._configured_with_data_sources = ( - (data_system_config.initializers is not None and len(data_system_config.initializers) > 0) - or len(self._synchronizers) > 0 - ) - def start(self, set_on_ready: Event): """ Start the FDv2 data system. @@ -147,13 +102,6 @@ def stop(self): # Close the store self._store.close() - def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): - """ - Sets the diagnostic accumulator for streaming initialization metrics. - This should be called before start() to ensure metrics are collected. - """ - self._diagnostic_accumulator = diagnostic_accumulator - def _run_main_loop(self, set_on_ready: Event): """Main coordination loop that manages initializers and synchronizers.""" try: @@ -405,9 +353,9 @@ def reader(self: 'FDv2'): if update == "check": # Check condition periodically current_status = self._data_source_status_provider.status - if check_recovery and self._recovery_condition(current_status): + if check_recovery and recovery_condition(current_status): return ConditionDirective.RECOVER - if self._fallback_condition(current_status): + if fallback_condition(current_status): return ConditionDirective.FALLBACK continue @@ -453,55 +401,6 @@ def reader(self: 'FDv2'): # the synchronizer can't provide more updates, so we should remove it and fall back return ConditionDirective.REMOVE - def _fallback_condition(self, status: DataSourceStatus) -> bool: - """ - Determine if we should fallback to the next synchronizer in the list. - This applies at any position in the synchronizers list. - - :param status: Current data source status - :return: True if fallback condition is met - """ - interrupted_at_runtime = ( - status.state == DataSourceState.INTERRUPTED - and time.time() - status.since > 60 # 1 minute - ) - cannot_initialize = ( - status.state == DataSourceState.INITIALIZING - and time.time() - status.since > 10 # 10 seconds - ) - - return interrupted_at_runtime or cannot_initialize - - def _recovery_condition(self, status: DataSourceStatus) -> bool: - """ - Determine if we should try to recover to the first (preferred) synchronizer. - This only applies when not already at the first synchronizer (index > 0). - - :param status: Current data source status - :return: True if recovery condition is met - """ - healthy_for_too_long = ( - status.state == DataSourceState.VALID - and time.time() - status.since > 300 # 5 minutes - ) - - return healthy_for_too_long - - def _persistent_store_outage_recovery(self, data_store_status: DataStoreStatus): - """ - Monitor the data store status. If the store comes online and - potentially has stale data, we should write our known state to it. - """ - if not data_store_status.available: - return - - if not data_store_status.stale: - return - - err = self._store.commit() - if err is not None: - log.error("Failed to reinitialize data store", exc_info=err) - def _record_environment_id(self, environment_id: Optional[str]): if not isinstance(environment_id, str) or environment_id == '': return @@ -520,40 +419,6 @@ def store(self) -> ReadOnlyStore: """Get the underlying store for flag evaluation.""" return self._store.get_active_store() - @property - def data_source_status_provider(self) -> DataSourceStatusProvider: - """Get the data source status provider.""" - return self._data_source_status_provider - - @property - def data_store_status_provider(self) -> DataStoreStatusProvider: - """Get the data store status provider.""" - return self._data_store_status_provider - - @property - def flag_change_listeners(self) -> Listeners: - """Get the collection of listeners for flag change events.""" - return self._flag_change_listeners - - @property - def data_availability(self) -> DataAvailability: - """Get the current data availability level.""" - if self._store.selector().is_defined(): - return DataAvailability.REFRESHED - - if not self._configured_with_data_sources or self._store.is_initialized(): - return DataAvailability.CACHED - - return DataAvailability.DEFAULTS - - @property - def target_availability(self) -> DataAvailability: - """Get the target data availability level based on configuration.""" - if self._configured_with_data_sources: - return DataAvailability.REFRESHED - - return DataAvailability.CACHED - __all__ = [ 'ConditionDirective', diff --git a/ldclient/impl/datasystem/fdv2_common.py b/ldclient/impl/datasystem/fdv2_common.py index 9db072c3..80b56f46 100644 --- a/ldclient/impl/datasystem/fdv2_common.py +++ b/ldclient/impl/datasystem/fdv2_common.py @@ -9,9 +9,12 @@ import time from copy import copy from enum import Enum -from typing import Any, Callable, Dict, Mapping, Optional +from typing import Any, Callable, Dict, List, Mapping, Optional +from ldclient.config import DataSourceBuilder, DataSystemConfig from ldclient.feature_store import _FeatureStoreDataSetSorter +from ldclient.impl.datasystem import DataAvailability, DiagnosticAccumulator +from ldclient.impl.datasystem.store import Store from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock @@ -21,6 +24,7 @@ DataSourceState, DataSourceStatus, DataSourceStatusProvider, + DataStoreMode, DataStoreStatus, DataStoreStatusProvider, FeatureStore @@ -284,9 +288,154 @@ class ConditionDirective(str, Enum): """ +def fallback_condition(status: DataSourceStatus) -> bool: + """ + Determine if we should fallback to the next synchronizer in the list. + This applies at any position in the synchronizers list. + + :param status: Current data source status + :return: True if fallback condition is met + """ + interrupted_at_runtime = ( + status.state == DataSourceState.INTERRUPTED + and time.time() - status.since > 60 # 1 minute + ) + cannot_initialize = ( + status.state == DataSourceState.INITIALIZING + and time.time() - status.since > 10 # 10 seconds + ) + + return interrupted_at_runtime or cannot_initialize + + +def recovery_condition(status: DataSourceStatus) -> bool: + """ + Determine if we should try to recover to the first (preferred) synchronizer. + This only applies when not already at the first synchronizer (index > 0). + + :param status: Current data source status + :return: True if recovery condition is met + """ + healthy_for_too_long = ( + status.state == DataSourceState.VALID + and time.time() - status.since > 300 # 5 minutes + ) + + return healthy_for_too_long + + +class _FDv2Base: + """ + Common construction and read-only accessors for the FDv2 data system + coordinators. + + This wires up the listeners, the in-memory store, the status providers, and + the optional persistent store, and it reports data availability. Subclasses + add their own concurrency primitives and the loops that run initializers and + synchronizers. + """ + + def __init__(self, config, data_system_config: DataSystemConfig): + self._config = config + self._data_system_config = data_system_config + self._synchronizers: List[DataSourceBuilder] = list(data_system_config.synchronizers) if data_system_config.synchronizers else [] + self._fdv1_fallback_synchronizer_builder = data_system_config.fdv1_fallback_synchronizer + self._disabled = config.offline + + # Diagnostic accumulator provided by the client for streaming metrics. + self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None + + # Set up event listeners. + self._flag_change_listeners = Listeners() + self._change_set_listeners = Listeners() + self._data_store_listeners = Listeners() + + self._data_store_listeners.add(self._persistent_store_outage_recovery) + + # Create the store. + self._store = Store(self._flag_change_listeners, self._change_set_listeners) + + # Status providers. + self._data_source_status_provider = DataSourceStatusProviderImpl(Listeners()) + self._data_store_status_provider = DataStoreStatusProviderImpl(None, self._data_store_listeners) + + # Configure the persistent store if one is provided. + if self._data_system_config.data_store is not None: + self._data_store_status_provider = DataStoreStatusProviderImpl(self._data_system_config.data_store, self._data_store_listeners) + writable = self._data_system_config.data_store_mode == DataStoreMode.READ_WRITE + wrapper = FeatureStoreClientWrapper(self._data_system_config.data_store, self._data_store_status_provider) + self._store.with_persistence( + wrapper, writable, self._data_store_status_provider + ) + + # Track configuration. + self._configured_with_data_sources = ( + (data_system_config.initializers is not None and len(data_system_config.initializers) > 0) + or len(self._synchronizers) > 0 + ) + + def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): + """ + Sets the diagnostic accumulator for streaming initialization metrics. + This should be called before start() to ensure metrics are collected. + """ + self._diagnostic_accumulator = diagnostic_accumulator + + def _persistent_store_outage_recovery(self, data_store_status: DataStoreStatus): + """ + Monitor the data store status. If the store comes online and + potentially has stale data, we should write our known state to it. + """ + if not data_store_status.available: + return + + if not data_store_status.stale: + return + + err = self._store.commit() + if err is not None: + log.error("Failed to reinitialize data store", exc_info=err) + + @property + def data_source_status_provider(self) -> DataSourceStatusProvider: + """Get the data source status provider.""" + return self._data_source_status_provider + + @property + def data_store_status_provider(self) -> DataStoreStatusProvider: + """Get the data store status provider.""" + return self._data_store_status_provider + + @property + def flag_change_listeners(self) -> Listeners: + """Get the collection of listeners for flag change events.""" + return self._flag_change_listeners + + @property + def data_availability(self) -> DataAvailability: + """Get the current data availability level.""" + if self._store.selector().is_defined(): + return DataAvailability.REFRESHED + + if not self._configured_with_data_sources or self._store.is_initialized(): + return DataAvailability.CACHED + + return DataAvailability.DEFAULTS + + @property + def target_availability(self) -> DataAvailability: + """Get the target data availability level based on configuration.""" + if self._configured_with_data_sources: + return DataAvailability.REFRESHED + + return DataAvailability.CACHED + + __all__ = [ 'ConditionDirective', 'DataSourceStatusProviderImpl', 'DataStoreStatusProviderImpl', 'FeatureStoreClientWrapper', + 'fallback_condition', + 'recovery_condition', ] diff --git a/ldclient/testing/impl/datasystem/test_async_fdv2.py b/ldclient/testing/impl/datasystem/test_async_fdv2.py new file mode 100644 index 00000000..3017f5df --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_async_fdv2.py @@ -0,0 +1,354 @@ +# pylint: disable=missing-docstring + +import asyncio +from typing import AsyncGenerator, List, Optional + +import pytest + +from ldclient.async_config import AsyncConfig +from ldclient.config import ( + DataSourceBuilder, + DataSourceBuilderConfig, + DataSystemConfig +) +from ldclient.impl.datasystem import DataAvailability +from ldclient.impl.datasystem.async_fdv2 import AsyncFDv2 +from ldclient.impl.util import _Fail, _Success +from ldclient.integrations.test_datav2 import TestDataV2 +from ldclient.interfaces import ( + Basis, + BasisResult, + ChangeSetBuilder, + DataSourceState, + DataSourceStatus, + FlagChange, + IntentCode, + ObjectKind, + Selector, + SelectorStore, + Update +) +from ldclient.versioned_data_kind import FEATURES + + +class MockAsyncSynchronizer: + """A controllable async synchronizer for testing.""" + + def __init__(self, updates: Optional[List[Update]] = None): + self._updates = updates or [] + self._queue: asyncio.Queue = asyncio.Queue() + self._stopped = False + # Pre-populate the queue with provided updates + for u in self._updates: + self._queue.put_nowait(u) + + @property + def name(self) -> str: + return "MockAsyncSynchronizer" + + async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: + while not self._stopped: + try: + update = await asyncio.wait_for(self._queue.get(), timeout=0.1) + yield update + except asyncio.TimeoutError: + continue + + async def stop(self) -> None: + self._stopped = True + + async def push(self, update: Update): + await self._queue.put(update) + + +class MockAsyncSynchronizerBuilder(DataSourceBuilder): + def __init__(self, synchronizer: MockAsyncSynchronizer): + self._sync = synchronizer + + def build(self, config: DataSourceBuilderConfig): + return self._sync + + +class MockAsyncInitializer: + """A controllable async initializer for testing.""" + + def __init__(self, result: BasisResult): + self._result = result + + @property + def name(self) -> str: + return "MockAsyncInitializer" + + async def fetch(self, ss: SelectorStore) -> BasisResult: + return self._result + + +class MockAsyncInitializerBuilder(DataSourceBuilder): + def __init__(self, initializer: MockAsyncInitializer): + self._init = initializer + + def build(self, config: DataSourceBuilderConfig): + return self._init + + +def _make_valid_basis() -> Basis: + builder = ChangeSetBuilder() + builder.start(IntentCode.TRANSFER_FULL) + builder.add_put(ObjectKind.FLAG, "my-flag", 1, {"key": "my-flag", "version": 1}) + selector = Selector(state="p:test:1", version=1) + change_set = builder.finish(selector) + return Basis(change_set=change_set, persist=False, environment_id=None) + + +def _make_valid_update() -> Update: + builder = ChangeSetBuilder() + builder.start(IntentCode.TRANSFER_FULL) + builder.add_put(ObjectKind.FLAG, "my-flag", 1, {"key": "my-flag", "version": 1}) + selector = Selector(state="p:test:1", version=1) + change_set = builder.finish(selector) + return Update(state=DataSourceState.VALID, change_set=change_set) + + +@pytest.mark.asyncio +async def test_async_fdv2_basic_start_stop(): + td = TestDataV2.data_source() + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[td.async_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + assert ready_event.is_set() + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_synchronizer_receives_updates(): + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[td.async_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + + # Data should be available + assert fdv2.data_availability.at_least(DataAvailability.REFRESHED) + + # Check we can read the flag + store = fdv2.store + flag = await store.get(FEATURES, "feature-flag") + assert flag is not None + assert flag["key"] == "feature-flag" + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_flag_change_listener(): + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[td.async_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + + changes: List[FlagChange] = [] + flag_changed = asyncio.Event() + + def listener(change: FlagChange): + changes.append(change) + if len(changes) >= 2: + flag_changed.set() + + fdv2.flag_change_listeners.add(listener) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + + # Trigger another update + td.update(td.flag("feature-flag").on(False)) + + await asyncio.wait_for(flag_changed.wait(), timeout=2) + assert len(changes) >= 2 + assert all(c.key == "feature-flag" for c in changes) + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_two_phase_init(): + td_initializer = TestDataV2.data_source() + td_initializer.update(td_initializer.flag("feature-flag").on(True)) + + td_synchronizer = TestDataV2.data_source() + td_synchronizer.update(td_synchronizer.flag("feature-flag").on(True)) + td_synchronizer.update(td_synchronizer.flag("feature-flag").on(False)) + + data_system_config = DataSystemConfig( + initializers=[td_initializer.async_builder], + synchronizers=[td_synchronizer.async_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + assert fdv2.data_availability.at_least(DataAvailability.REFRESHED) + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_initializer_async(): + """Test with a pure async initializer.""" + basis = _make_valid_basis() + init = MockAsyncInitializer(_Success(basis)) + init_builder = MockAsyncInitializerBuilder(init) + + # Empty synchronizer that just keeps running + sync_mock = MockAsyncSynchronizer() + sync_builder = MockAsyncSynchronizerBuilder(sync_mock) + + data_system_config = DataSystemConfig( + initializers=[init_builder], + synchronizers=[sync_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + assert fdv2.data_availability.at_least(DataAvailability.REFRESHED) + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_fallsback_to_secondary_synchronizer(): + """When primary synchronizer yields nothing, should move to secondary.""" + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + + # An async synchronizer that immediately stops (produces no updates) + empty_sync = MockAsyncSynchronizer() + empty_sync._stopped = True # pre-stopped — yields nothing + empty_builder = MockAsyncSynchronizerBuilder(empty_sync) + + data_system_config = DataSystemConfig( + initializers=[td.async_builder], + synchronizers=[empty_builder, td.async_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + assert fdv2.data_availability.at_least(DataAvailability.REFRESHED) + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_falls_back_to_fdv1_on_synchronizer_signal(): + """Synchronizer yielding fallback_to_fdv1=True triggers FDv1 fallback.""" + td_fdv1 = TestDataV2.data_source() + td_fdv1.update(td_fdv1.flag("fdv1-flag").on(True)) + + # Primary synchronizer signals FDv1 fallback + fallback_update = Update(state=DataSourceState.OFF, fallback_to_fdv1=True) + primary_sync = MockAsyncSynchronizer([fallback_update]) + primary_builder = MockAsyncSynchronizerBuilder(primary_sync) + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[primary_builder], + fdv1_fallback_synchronizer=td_fdv1.async_builder, + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + assert fdv2.data_availability.at_least(DataAvailability.REFRESHED) + + store = fdv2.store + flag = await store.get(FEATURES, "fdv1-flag") + assert flag is not None + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_data_availability_defaults_when_no_sources(): + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=None, + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + # No sources means target is CACHED, and data is also CACHED (or DEFAULTS) + assert fdv2.target_availability == DataAvailability.CACHED + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_data_availability_refreshed_with_data(): + td = TestDataV2.data_source() + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[td.async_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready_event) + + await asyncio.wait_for(ready_event.wait(), timeout=2) + assert fdv2.data_availability.at_least(DataAvailability.REFRESHED) + assert fdv2.target_availability.at_least(DataAvailability.REFRESHED) + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_disabled_immediately_signals_ready(): + td = TestDataV2.data_source() + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[td.async_builder], + ) + + ready_event = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy", offline=True), data_system_config) + fdv2.start(ready_event) + + # Should be ready immediately because disabled + await asyncio.wait_for(ready_event.wait(), timeout=1) + assert ready_event.is_set() + + await fdv2.stop()