diff --git a/contract-tests/async_client_entity.py b/contract-tests/async_client_entity.py index b02dfd01..a979f401 100644 --- a/contract-tests/async_client_entity.py +++ b/contract-tests/async_client_entity.py @@ -104,8 +104,8 @@ async def start(self): await self._client.start(start_wait / 1000.0) self._listeners = AsyncListenerRegistry(self._client.flag_tracker) - def is_initializing(self) -> bool: - return self._client.is_initialized() if self._client else False + async def is_initializing(self) -> bool: + return await self._client.is_initialized() if self._client else False async def evaluate(self, params: dict) -> dict: response = {} diff --git a/contract-tests/async_service.py b/contract-tests/async_service.py index 54b4d759..5288c7ec 100644 --- a/contract-tests/async_service.py +++ b/contract-tests/async_service.py @@ -102,7 +102,7 @@ async def handle_create_client(request: aiohttp.web.Request) -> aiohttp.web.Resp await client.close() return aiohttp.web.Response(text=str(e), status=500) - if not client.is_initializing() and not options['configuration'].get('initCanFail', False): + if not await client.is_initializing() and not options['configuration'].get('initCanFail', False): await client.close() return aiohttp.web.Response(text='Failed to initialize', status=500) diff --git a/ldclient/async_client.py b/ldclient/async_client.py index 898c2bf8..abb9855e 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") @@ -221,7 +227,7 @@ async def __start_up(self, start_wait: float): log.info("Waiting up to " + str(start_wait) + " seconds for LaunchDarkly client to initialize...") await update_processor_ready.wait(start_wait) - if self.is_initialized() is True: + if await self.is_initialized() is True: log.info("Started LaunchDarkly Client: OK") else: log.warning("Initialization timeout exceeded for LaunchDarkly Client or an error occurred. " "Feature Flags may not yet be available.") @@ -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: @@ -343,18 +379,20 @@ def is_offline(self) -> bool: """Returns true if the client is in offline mode.""" return self._config.offline - def is_initialized(self) -> bool: + async def is_initialized(self) -> bool: """Returns true if the client has successfully connected to LaunchDarkly. If this returns false, it means that the client has not yet successfully connected to LaunchDarkly. It might still be in the process of starting up, or it might be attempting to reconnect after an unsuccessful attempt, or it might have received an unrecoverable error (such as an invalid SDK key) and given up. + + This is a coroutine because determining readiness may query a persistent store. """ if self.is_offline() or self._config.use_ldd: return True - return self._data_system.data_availability.at_least(DataAvailability.CACHED) + return (await self._data_system.data_availability()).at_least(DataAvailability.CACHED) async def flush(self): """Flushes all pending analytics events. @@ -455,8 +493,9 @@ async def _evaluate_internal(self, key: str, context: Context, default: Any, eve if self._config.offline: return EvaluationDetail(default, None, error_reason('CLIENT_NOT_READY')), None - if self._data_system.data_availability != DataAvailability.REFRESHED: - if self._data_system.data_availability == DataAvailability.CACHED: + availability = await self._data_system.data_availability() + if availability != DataAvailability.REFRESHED: + if availability == DataAvailability.CACHED: log.warning("Feature Flag evaluation attempted before client has initialized - using last known values from feature store for feature key: " + key) else: log.warning("Feature Flag evaluation attempted before client has initialized! Feature store unavailable - returning default: " + str(default) + " for feature key: " + key) @@ -526,8 +565,9 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState log.warning("all_flags_state() called, but client is in offline mode. Returning empty state") return FeatureFlagsState(False) - if self._data_system.data_availability != DataAvailability.REFRESHED: - if self._data_system.data_availability == DataAvailability.CACHED: + availability = await self._data_system.data_availability() + if availability != DataAvailability.REFRESHED: + if availability == DataAvailability.CACHED: log.warning("all_flags_state() called before client has finished initializing! Using last known values from feature store") else: log.warning("all_flags_state() called before client has finished initializing! Feature store unavailable - returning empty state") diff --git a/ldclient/async_feature_store.py b/ldclient/async_feature_store.py index 931729c5..c29041e3 100644 --- a/ldclient/async_feature_store.py +++ b/ldclient/async_feature_store.py @@ -88,6 +88,10 @@ def initialized(self) -> bool: """ """ return self._initialized + async def is_initialized(self) -> bool: + """ """ + return self._initialized + async def close(self) -> None: """ """ pass diff --git a/ldclient/async_feature_store_helpers.py b/ldclient/async_feature_store_helpers.py index 5a2bc876..94aa2756 100644 --- a/ldclient/async_feature_store_helpers.py +++ b/ldclient/async_feature_store_helpers.py @@ -37,6 +37,8 @@ class AsyncCachingStoreWrapper(_CachingStoreWrapperBase, DiagnosticDescription, event loop because its reads and writes never suspend between one another. """ + __INITED_CACHE_KEY__ = "$inited" + _core: AsyncFeatureStoreCore def __init__(self, core: AsyncFeatureStoreCore, cache_config: CacheConfig): @@ -95,12 +97,26 @@ async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: @property def initialized(self) -> bool: - """Returns whether ``init`` has completed in this process. + """Returns the store's last observed initialized state.""" + return self._inited - This property does not query the store: it is synchronous, but a persistent-store query is - a coroutine, so it reflects only whether this process has initialized the store. + async def is_initialized(self) -> bool: + """Queries the store's initialized state, updating :attr:`initialized`. + + Honors the cache: with caching off the store is queried on every call; + with a TTL it is queried once per interval; with an infinite TTL it is + queried once. Once the store reports initialized the state latches and + later calls return without I/O. """ - return self._inited + if self._inited: + return True + result = self._cache.get(AsyncCachingStoreWrapper.__INITED_CACHE_KEY__) + if result is None: + result = bool(await self._core.initialized_internal()) + self._cache[AsyncCachingStoreWrapper.__INITED_CACHE_KEY__] = result + if result: + self._inited = True + return result async def close(self) -> None: """Releases the cache and closes the underlying core if it supports it.""" diff --git a/ldclient/impl/datasystem/__init__.py b/ldclient/impl/datasystem/__init__.py index 1180e8e5..26be520d 100644 --- a/ldclient/impl/datasystem/__init__.py +++ b/ldclient/impl/datasystem/__init__.py @@ -208,11 +208,11 @@ def flag_change_listeners(self) -> Listeners: """ raise NotImplementedError - @property @abstractmethod - def data_availability(self) -> DataAvailability: + async def data_availability(self) -> DataAvailability: """ - Indicates what form of data is currently available. + Indicates what form of data is currently available, awaiting the store's + readiness so a persistent store populated by another process is recognized. """ raise NotImplementedError diff --git a/ldclient/impl/datasystem/async_fdv1.py b/ldclient/impl/datasystem/async_fdv1.py index 958facc5..96fbc660 100644 --- a/ldclient/impl/datasystem/async_fdv1.py +++ b/ldclient/impl/datasystem/async_fdv1.py @@ -143,18 +143,22 @@ def data_store_status_provider(self) -> DataStoreStatusProvider: def flag_change_listeners(self) -> Listeners: return self._flag_change_listeners - @property - def data_availability(self) -> DataAvailability: + async def data_availability(self) -> DataAvailability: if self._config.offline: return DataAvailability.DEFAULTS if self._update_processor is not None and self._update_processor.initialized(): return DataAvailability.REFRESHED - if self._store.initialized: - return DataAvailability.CACHED + # Awaits the store so a persistent store populated by another process is + # recognized. A persistent-store error is logged and reported as no data. + try: + ready = await self._store.is_initialized() + except Exception as e: + log.warning("Error checking persistent store readiness: %s", e) + return DataAvailability.DEFAULTS - return DataAvailability.DEFAULTS + return DataAvailability.CACHED if ready else DataAvailability.DEFAULTS @property def target_availability(self) -> DataAvailability: diff --git a/ldclient/impl/datasystem/async_fdv2.py b/ldclient/impl/datasystem/async_fdv2.py new file mode 100644 index 00000000..8749721e --- /dev/null +++ b/ldclient/impl/datasystem/async_fdv2.py @@ -0,0 +1,653 @@ +""" +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 asyncio +import inspect +import time +from typing import Any, Callable, Dict, List, Mapping, Optional + +from ldclient.async_config import AsyncConfig, AsyncDataSystemConfig +from ldclient.config import DataSourceBuilder +from ldclient.feature_store import _FeatureStoreDataSetSorter +from ldclient.impl.aio.concurrency import ( + AsyncEvent, + AsyncLock, + AsyncQueue, + AsyncRepeatingTask, + AsyncTaskRunner, + TaskHandle, + join_handle, + spawn_handle +) +from ldclient.impl.datasystem import ( + AsyncDataSystem, + DataAvailability, + DiagnosticSource +) +from ldclient.impl.datasystem.async_store import AsyncStore +from ldclient.impl.datasystem.fdv2_common import ( + ConditionDirective, + DataSourceStatusProviderImpl, + DataStoreStatusProviderImpl, + _FDv2Base, + fallback_condition, + recovery_condition +) +from ldclient.impl.datasystem.store import _decode +from ldclient.impl.listeners import Listeners +from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log +from ldclient.interfaces import ( + AsyncFeatureStore, + AsyncReadOnlyStore, + AsyncSynchronizer, + DataSourceErrorInfo, + DataSourceErrorKind, + DataSourceState, + DataStoreMode, + DataStoreStatus +) +from ldclient.versioned_data_kind import VersionedDataKind + + +class AsyncFeatureStoreClientWrapper(AsyncFeatureStore): + """Adds availability tracking around an async feature store. + + Every store operation runs through a wrapper that watches for failures. When + an operation fails, the wrapper marks the store unavailable and starts a + background task that polls the store's ``is_available`` method every half + second. When the store recovers, the wrapper reports the new status to the + sink and stops polling. + + The status sink is any callable that accepts a :class:`DataStoreStatus`. + """ + + def __init__(self, store: AsyncFeatureStore, status_sink: Callable[[DataStoreStatus], None]): + """Constructs an instance wrapping ``store``. + + :param store: the async feature store to wrap + :param status_sink: a callable that receives status updates + """ + self._store = store + self._status_sink = status_sink + self._monitoring_enabled = self.is_monitoring_enabled() + + self._last_available = True + self._poller: Optional[AsyncRepeatingTask] = None + self._closed = False + + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + await self._wrap(lambda: self._store.init(_FeatureStoreDataSetSorter.sort_all_collections(all_data))) + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + return await self._wrap(lambda: self._store.get(kind, key)) + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + return await self._wrap(lambda: self._store.all(kind)) + + async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: + return await self._wrap(lambda: self._store.upsert(kind, item)) + + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: + return await self._wrap(lambda: self._store.delete(kind, key, version)) + + @property + def initialized(self) -> bool: + return self._store.initialized + + async def is_initialized(self) -> bool: + """Queries the inner store's initialized state. + + Runs through the availability wrapper so a failed query marks the store + unavailable like any other operation. + """ + return await self._wrap(lambda: self._store.is_initialized()) + + def disable_cache(self) -> None: + """Disables the inner store's cache if it supports it.""" + inner_disable = getattr(self._store, "disable_cache", None) + if callable(inner_disable): + inner_disable() + + def is_monitoring_enabled(self) -> bool: + """Returns whether the inner store opts in to availability monitoring. + + Delegates to the store's own opt-in. A store that does not report + availability is not polled, so it is never marked unavailable with no + path back to recovery. + """ + store_check = getattr(self._store, "is_monitoring_enabled", None) + if not callable(store_check): + return False + return store_check() + + async def _wrap(self, fn: Callable): + try: + return await fn() + except BaseException: + if self._monitoring_enabled: + self._update_availability(False) + raise + + def _update_availability(self, available: bool) -> None: + if self._closed: + return + if available == self._last_available: + return + + self._last_available = available + poller_to_stop = None + task_to_start = None + + if available: + poller_to_stop = self._poller + self._poller = None + log.warning("Persistent store is available again") + else: + log.warning("Detected persistent store unavailability; updates will be cached until it recovers") + if self._poller is None: + task_to_start = AsyncRepeatingTask("ldclient.check-availability", 0.5, 0, self._check_availability) + self._poller = task_to_start + + self._status_sink(DataStoreStatus(available, True)) + + if poller_to_stop is not None: + poller_to_stop.stop() + + if task_to_start is not None: + task_to_start.start() + + async def _check_availability(self) -> None: + try: + if await self._store.is_available(): # type: ignore[attr-defined] + self._update_availability(True) + except BaseException as e: + log.error("Unexpected error from data store status function: %s", e) + + async def close(self) -> None: + """Stops the availability poller and closes the inner store. + + Does nothing on a later call, so closing more than once is safe. + """ + if self._closed: + return + self._closed = True + + poller_to_stop = self._poller + self._poller = None + if poller_to_stop is not None: + poller_to_stop.stop() + try: + await asyncio.wait_for(poller_to_stop.wait_stopped(), timeout=5) + except asyncio.TimeoutError: + log.warning("Timed out waiting for the persistent store availability poller to stop") + + close = getattr(self._store, "close", None) + if callable(close): + try: + result = close() + if inspect.isawaitable(result): + await result + except Exception as e: + log.warning("Error closing the persistent store: %s", e) + + +class _AsyncReadOnlyStoreView(AsyncReadOnlyStore): + """Async read-only view of the FDv2 data system store. + + Serves every read from the store's active store, so a held instance follows + the swap from the persistent store to the in-memory store once it has data. + The active store may be the synchronous in-memory store, whose reads return + a value directly, or the async persistent store, whose reads are awaitable; + the view awaits only when the read result is awaitable. Items that a custom + persistent store keeps as raw dicts are decoded into model objects; items + that are already models pass through unchanged. + """ + + def __init__(self, store: AsyncStore): + self._store = store + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + item = self._store.get_active_store().get(kind, key) + if inspect.isawaitable(item): + item = await item + return _decode(kind, item) + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + items = self._store.get_active_store().all(kind) + if inspect.isawaitable(items): + items = await items + return {key: _decode(kind, value) for key, value in items.items()} + + +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. + """ + + _store: AsyncStore + + def __init__( + self, + config: AsyncConfig, + data_system_config: AsyncDataSystemConfig, + ): + """ + Initialize a new AsyncFDv2 data system. + + :param config: the SDK configuration + :param data_system_config: the data system configuration — initializers, + synchronizers, and the optional persistent store + """ + super().__init__() + + self._config = config + self._data_system_config = data_system_config + self._synchronizers: List[DataSourceBuilder[AsyncSynchronizer]] = 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 + self._configured_with_data_sources = ( + (data_system_config.initializers is not None and len(data_system_config.initializers) > 0) + or len(self._synchronizers) > 0 + ) + + if data_system_config.data_store is not None: + # The provider only calls monitoring methods, which the async store also has. + self._data_store_status_provider = DataStoreStatusProviderImpl(data_system_config.data_store, self._data_store_listeners) # type: ignore[arg-type] + writable = data_system_config.data_store_mode == DataStoreMode.READ_WRITE + # The async wrapper reports status through a plain callable sink, so + # pass the provider's update method rather than the provider itself. + wrapper = AsyncFeatureStoreClientWrapper(data_system_config.data_store, self._data_store_status_provider.update_status) + self._store.with_async_persistence(wrapper, writable, self._data_store_status_provider) + + self._store_view = _AsyncReadOnlyStoreView(self._store) + + # Concurrency + self._stop_event = AsyncEvent() + self._lock = AsyncLock() + self._active_synchronizer: Optional[AsyncSynchronizer] = None + self._runner = AsyncTaskRunner() + + def _create_store(self, flag_change_listeners: Listeners, change_set_listeners: Listeners) -> AsyncStore: + return AsyncStore(flag_change_listeners, change_set_listeners) + + def _persistent_store_outage_recovery(self, data_store_status: DataStoreStatus) -> None: + """ + On store recovery, write the current data back to it. The commit runs on + the task runner (so stop() cancels it) and is skipped once stopping, so + it never writes to a store that stop() is closing. + """ + if self._stop_event.is_set(): + return + + if not data_store_status.available: + return + + if not data_store_status.stale: + return + + async def _commit() -> None: + err = await self._store.commit() + if err is not None: + log.error("Failed to reinitialize data store", exc_info=err) + + self._runner.spawn("AsyncFDv2-store-recovery", _commit) + + 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 + await 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: + initializer = 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 + await 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: the ConditionDirective describing how to proceed + """ + 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: + # Honor a stop request every iteration so a queue that always has + # an item ready cannot starve the check. + if self._stop_event.is_set(): + return ConditionDirective.FALLBACK + 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: + await 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 self._store_view + + async def data_availability(self) -> DataAvailability: # type: ignore[override] + """Reports what form of data is currently available, awaiting the store's + readiness so a persistent store populated by another process is recognized + before a synchronizer supplies a basis. A persistent-store error is treated + as no data: it is logged and reported as ``DEFAULTS`` rather than raised.""" + if self._store.selector().is_defined(): + return DataAvailability.REFRESHED + if not self._configured_with_data_sources: + return DataAvailability.CACHED + try: + ready = await self._store.is_ready() + except Exception as e: + log.warning("Error checking persistent store readiness: %s", e) + return DataAvailability.DEFAULTS + return DataAvailability.CACHED if ready else DataAvailability.DEFAULTS + + +__all__ = [ + 'AsyncFDv2', + 'AsyncFeatureStoreClientWrapper', + 'ConditionDirective', + 'DataSourceStatusProviderImpl', + 'DataStoreStatusProviderImpl', +] diff --git a/ldclient/impl/datasystem/async_store.py b/ldclient/impl/datasystem/async_store.py index ce2e7742..248726d1 100644 --- a/ldclient/impl/datasystem/async_store.py +++ b/ldclient/impl/datasystem/async_store.py @@ -169,42 +169,49 @@ def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]: return None async with self._async_persist_lock: - all_data: Optional[Collections] = None - with self._lock: - if self._should_persist(): - all_data = {} - for kind in [FEATURES, SEGMENTS]: - all_data[kind] = self._memory_store.all(kind, __mapping_from_kind(kind)) + try: + all_data: Optional[Collections] = None + with self._lock: + if self._should_persist(): + all_data = {} + for kind in [FEATURES, SEGMENTS]: + all_data[kind] = self._memory_store.all(kind, __mapping_from_kind(kind)) - if all_data is None: - return None + if all_data is None: + return None - try: await store.init(all_data) except Exception as e: return e return None - async def close(self) -> Optional[Exception]: - """ - Close the store and the async persistent store, if configured. - - Returns: - Exception if closing failed, None otherwise - """ + async def close(self) -> None: + """Close the store and the async persistent store, if configured.""" store = self._persistent_store if store is None: - return None + return try: await store.close() except Exception as e: - return e - return None + log.warning("Error closing the persistent store: %s", e) def get_data_store_status_provider(self) -> Optional[DataStoreStatusProvider]: """Get the data store status provider for the persistent store, if configured.""" with self._lock: return self._persistent_store_status_provider + async def is_ready(self) -> bool: + """Reports whether the active store holds usable data. + + Once the in-memory store is active its readiness is authoritative and no + query is made. While the persistent store is active, its readiness is + queried (awaiting the store), so a store populated by another process is + recognized. + """ + store = self._persistent_store + if store is None or self._active_store is self._memory_store: + return self._active_store.initialized + return await store.is_initialized() + __all__ = ["AsyncStore"] diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 8939312c..065799a1 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -4,17 +4,15 @@ from typing import Any, Callable, List, Optional from ldclient.config import Config, DataSourceBuilder, DataSystemConfig -from ldclient.impl.datasystem import ( - DataAvailability, - DataSystem, - DiagnosticAccumulator, - DiagnosticSource -) +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, _decode from ldclient.impl.listeners import Listeners @@ -25,11 +23,8 @@ DataSourceErrorInfo, DataSourceErrorKind, DataSourceState, - DataSourceStatus, - DataSourceStatusProvider, DataStoreMode, DataStoreStatus, - DataStoreStatusProvider, ReadOnlyStore, Synchronizer ) @@ -61,13 +56,15 @@ def initialized(self) -> bool: return self._store.is_initialized() -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 store in read-only or read/write mode. """ + _store: Store + def __init__( self, config: Config, @@ -76,43 +73,27 @@ def __init__( """ Initialize a new FDv2 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) + :param config: the SDK configuration + :param data_system_config: the data system configuration — initializers, + synchronizers, and the optional persistent store """ + super().__init__() + 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) - self._store_view = _ReadOnlyStoreView(self._store) + self._disabled = config.offline + self._configured_with_data_sources = ( + (data_system_config.initializers is not None and len(data_system_config.initializers) > 0) + or len(self._synchronizers) > 0 + ) - # 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 - ) + if data_system_config.data_store is not None: + self._data_store_status_provider = DataStoreStatusProviderImpl(data_system_config.data_store, self._data_store_listeners) + writable = data_system_config.data_store_mode == DataStoreMode.READ_WRITE + wrapper = FeatureStoreClientWrapper(data_system_config.data_store, self._data_store_status_provider) + self._store.with_persistence(wrapper, writable, self._data_store_status_provider) # Threading self._stop_event = Event() @@ -121,11 +102,25 @@ 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 - ) + self._store_view = _ReadOnlyStoreView(self._store) + + def _create_store(self, flag_change_listeners: Listeners, change_set_listeners: Listeners) -> Store: + return Store(flag_change_listeners, change_set_listeners) + + def _persistent_store_outage_recovery(self, data_store_status: DataStoreStatus) -> None: + """ + 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 start(self, set_on_ready: Event): """ @@ -171,13 +166,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: @@ -429,9 +417,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 @@ -477,55 +465,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 @@ -544,40 +483,6 @@ def store(self) -> ReadOnlyStore: """Get the underlying store for flag evaluation.""" return self._store_view - @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..3bdaade4 100644 --- a/ldclient/impl/datasystem/fdv2_common.py +++ b/ldclient/impl/datasystem/fdv2_common.py @@ -12,6 +12,8 @@ from typing import Any, Callable, Dict, Mapping, Optional from ldclient.feature_store import _FeatureStoreDataSetSorter +from ldclient.impl.datasystem import DataAvailability, DiagnosticAccumulator +from ldclient.impl.datasystem.store import _StoreBase from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock @@ -284,9 +286,137 @@ 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 store, and the status providers, and it + reports data availability. Following the same split as + :class:`ldclient.impl.datasystem.store._StoreBase`, this base holds only the + shared, store-agnostic wiring. Subclasses own the config, supply the store + through :meth:`_create_store`, configure the optional persistent store, react + to persistent-store recovery through + :meth:`_persistent_store_outage_recovery`, and add their own concurrency + primitives and the loops that run initializers and synchronizers. + """ + + # Set by subclasses from their config; read by ``data_availability``. + _configured_with_data_sources: bool + + def __init__(self) -> None: + # 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; the subclass supplies the concrete type. + self._store = self._create_store(self._flag_change_listeners, self._change_set_listeners) + + # Status providers. A child that has a persistent store replaces the + # data store provider with one that wraps it. + self._data_source_status_provider = DataSourceStatusProviderImpl(Listeners()) + self._data_store_status_provider = DataStoreStatusProviderImpl(None, self._data_store_listeners) + + def _create_store(self, flag_change_listeners: Listeners, change_set_listeners: Listeners) -> _StoreBase: + """Create and return the coordinator's store.""" + raise NotImplementedError + + def _persistent_store_outage_recovery(self, data_store_status: DataStoreStatus) -> None: + """ + Monitor the data store status. If the store comes online and potentially + has stale data, write the known state back to it. + """ + raise NotImplementedError + + 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 + + @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/interfaces.py b/ldclient/interfaces.py index 202eb82c..ff28659e 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -408,7 +408,16 @@ async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: @abstractmethod def initialized(self) -> bool: """ - Returns whether the store has been initialized yet or not. + Returns the store's last observed initialized state without querying it. + """ + + @abstractmethod + async def is_initialized(self) -> bool: + """ + Queries whether the store has been initialized, awaiting the store if a query is required. + + A persistent store may have been populated by another process, so this can require I/O. + Implementations should latch a positive result: once the store is initialized it stays so. """ async def close(self) -> None: diff --git a/ldclient/testing/impl/datasystem/test_async_config.py b/ldclient/testing/impl/datasystem/test_async_config.py index 1f010c45..143a7b06 100644 --- a/ldclient/testing/impl/datasystem/test_async_config.py +++ b/ldclient/testing/impl/datasystem/test_async_config.py @@ -35,6 +35,9 @@ async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: def initialized(self) -> bool: return True + async def is_initialized(self) -> bool: + return True + def test_async_data_system_config_defaults(): cfg = AsyncDataSystemConfig() 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..8421368d --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_async_fdv2.py @@ -0,0 +1,780 @@ +# pylint: disable=missing-docstring + +import asyncio +from typing import Any, AsyncGenerator, Dict, List, Mapping, 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, + AsyncFeatureStoreClientWrapper +) +from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, _Success +from ldclient.integrations.test_datav2 import TestDataV2 +from ldclient.interfaces import ( + AsyncFeatureStore, + Basis, + BasisResult, + ChangeSetBuilder, + DataSourceState, + DataSourceStatus, + DataStoreStatus, + FlagChange, + IntentCode, + ObjectKind, + Selector, + SelectorStore, + Update +) +from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind + + +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 (await 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 (await 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 (await 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 (await 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 (await 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 (await 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() + + +class ListAsyncSynchronizer: + """An async synchronizer that yields a fixed list of updates and then + completes. A completed iterator signals a permanent failure to the + coordinator, so the coordinator moves on to the next synchronizer.""" + + def __init__(self, name: str, updates: List[Update]): + self._name = name + self._updates = updates + self.sync_called = False + + @property + def name(self) -> str: + return self._name + + async def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: + self.sync_called = True + for update in self._updates: + yield update + + async def stop(self) -> None: + pass + + +class RecordingAsyncInitializer: + """An async initializer that returns a fixed result and counts calls.""" + + def __init__(self, name: str, result: BasisResult): + self._name = name + self._result = result + self.call_count = 0 + + @property + def name(self) -> str: + return self._name + + async def fetch(self, ss: SelectorStore) -> BasisResult: + self.call_count += 1 + return self._result + + +async def _status_reaches(fdv2: AsyncFDv2, state: DataSourceState, timeout: float = 2.0): + deadline = 0 + while deadline < int(timeout / 0.01): + if fdv2.data_source_status_provider.status.state == state: + return + await asyncio.sleep(0.01) + deadline += 1 + + +@pytest.mark.asyncio +async def test_async_fdv2_both_synchronizers_fail_transitions_to_off(): + """Both synchronizers complete without data -> data source goes OFF.""" + primary = ListAsyncSynchronizer("primary", []) + secondary = ListAsyncSynchronizer("secondary", []) + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[ + MockAsyncSynchronizerBuilder(primary), # type: ignore[arg-type] + MockAsyncSynchronizerBuilder(secondary), # type: ignore[arg-type] + ], + ) + + 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) + await _status_reaches(fdv2, DataSourceState.OFF) + + assert fdv2.data_source_status_provider.status.state == DataSourceState.OFF + assert primary.sync_called + assert secondary.sync_called + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_initializer_header_fallback_engages_fdv1(): + """An initializer error carrying X-LD-FD-Fallback engages the FDv1 + fallback synchronizer, and the configured FDv2 synchronizer must not run.""" + init = RecordingAsyncInitializer( + "hdr-fallback", + _Fail(error="boom", exception=None, headers={_LD_FD_FALLBACK_HEADER: 'true'}), + ) + + # This FDv2 synchronizer must never run because we fell back during init. + fdv2_sync = ListAsyncSynchronizer("fdv2-should-not-run", []) + + td_fdv1 = TestDataV2.data_source() + td_fdv1.update(td_fdv1.flag("fdv1-flag").on(True)) + + data_system_config = DataSystemConfig( + initializers=[MockAsyncInitializerBuilder(init)], # type: ignore[arg-type] + synchronizers=[MockAsyncSynchronizerBuilder(fdv2_sync)], # type: ignore[arg-type] + 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) + + flag = await fdv2.store.get(FEATURES, "fdv1-flag") + assert flag is not None + assert fdv2_sync.sync_called is False + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_initializer_header_fallback_without_fdv1_transitions_to_off(): + """An initializer signals FDv1 fallback but no FDv1 synchronizer is + configured -> the data source transitions to OFF.""" + init = RecordingAsyncInitializer( + "hdr-fallback-no-fdv1", + _Fail(error="boom", exception=None, headers={_LD_FD_FALLBACK_HEADER: 'true'}), + ) + + fdv2_sync = ListAsyncSynchronizer("fdv2-should-not-run", []) + + data_system_config = DataSystemConfig( + initializers=[MockAsyncInitializerBuilder(init)], # type: ignore[arg-type] + synchronizers=[MockAsyncSynchronizerBuilder(fdv2_sync)], # type: ignore[arg-type] + fdv1_fallback_synchronizer=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) + await _status_reaches(fdv2, DataSourceState.OFF) + + assert fdv2.data_source_status_provider.status.state == DataSourceState.OFF + assert fdv2_sync.sync_called is False + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_interrupted_without_header_falls_back_to_secondary(): + """An INTERRUPTED update without the fallback header moves to the next + synchronizer, not to FDv1.""" + primary = ListAsyncSynchronizer( + "primary", + [Update(state=DataSourceState.INTERRUPTED, fallback_to_fdv1=False)], + ) + secondary = ListAsyncSynchronizer( + "secondary", + [Update(state=DataSourceState.VALID, fallback_to_fdv1=False)], + ) + + td_fdv1 = TestDataV2.data_source() + td_fdv1.update(td_fdv1.flag("fdv1-should-not-appear").on(True)) + + data_system_config = DataSystemConfig( + initializers=None, + synchronizers=[ + MockAsyncSynchronizerBuilder(primary), # type: ignore[arg-type] + MockAsyncSynchronizerBuilder(secondary), # type: ignore[arg-type] + ], + 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 primary.sync_called + assert secondary.sync_called + # FDv1 must not have been engaged. + assert await fdv2.store.get(FEATURES, "fdv1-should-not-appear") is None + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_initializers_run_until_success(): + """Initializers run in order until one succeeds; a failing initializer is + skipped and the next one is tried.""" + fail_init = RecordingAsyncInitializer("fail", _Fail(error="boom", exception=None)) + success_init = RecordingAsyncInitializer("ok", _Success(_make_valid_basis())) + + # An empty synchronizer that keeps running after initialization. + sync_mock = MockAsyncSynchronizer() + + data_system_config = DataSystemConfig( + initializers=[ + MockAsyncInitializerBuilder(fail_init), # type: ignore[arg-type] + MockAsyncInitializerBuilder(success_init), # type: ignore[arg-type] + ], + synchronizers=[MockAsyncSynchronizerBuilder(sync_mock)], + ) + + 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 fail_init.call_count == 1 + assert success_init.call_count == 1 + assert await fdv2.store.get(FEATURES, "my-flag") is not None + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_fdv2_initializers_stop_on_first_success(): + """Once an initializer returns a basis with a defined selector, the + remaining initializers are skipped.""" + first = RecordingAsyncInitializer("first", _Success(_make_valid_basis())) + second = RecordingAsyncInitializer("second", _Success(_make_valid_basis())) + + data_system_config = DataSystemConfig( + initializers=[ + MockAsyncInitializerBuilder(first), # type: ignore[arg-type] + MockAsyncInitializerBuilder(second), # type: ignore[arg-type] + ], + 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) + + assert first.call_count == 1 + assert second.call_count == 0 + + await fdv2.stop() + + +class FakeAsyncStore(AsyncFeatureStore): + """An async store whose operations can be made to fail on demand.""" + + def __init__(self): + self._data: Dict[VersionedDataKind, Dict[str, dict]] = {FEATURES: {}, SEGMENTS: {}} + self._inited = False + self._available = True + self.fail = False + self.init_calls: List[Mapping] = [] + self.closed = False + + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + if self.fail: + raise RuntimeError("store down") + self.init_calls.append(all_data) + self._data = {FEATURES: dict(all_data.get(FEATURES, {})), SEGMENTS: dict(all_data.get(SEGMENTS, {}))} + self._inited = True + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + if self.fail: + raise RuntimeError("store down") + return self._data.get(kind, {}).get(key) + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + if self.fail: + raise RuntimeError("store down") + return dict(self._data.get(kind, {})) + + async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: + if self.fail: + raise RuntimeError("store down") + self._data[kind][item["key"]] = item + return True + + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: + return await self.upsert(kind, {"key": key, "version": version, "deleted": True}) + + @property + def initialized(self) -> bool: + return self._inited + + async def is_initialized(self) -> bool: + if self.fail: + raise RuntimeError("store down") + return self._inited + + async def is_available(self) -> bool: + return self._available + + def is_monitoring_enabled(self) -> bool: + return True + + async def close(self) -> None: + self.closed = True + + +class StoreWithoutAvailability(AsyncFeatureStore): + async def init(self, all_data): + pass + + async def get(self, kind, key): + return None + + async def all(self, kind): + return {} + + async def upsert(self, kind, item): + return True + + async def delete(self, kind, key, version): + return True + + @property + def initialized(self) -> bool: + return True + + async def is_initialized(self) -> bool: + return True + + +class StoreWithAvailabilityNoOptIn(StoreWithoutAvailability): + """Reports availability but does not opt in to monitoring. + + Models a custom store that provides is_available yet omits + is_monitoring_enabled, which must not turn on availability polling. + """ + + async def is_available(self) -> bool: + return True + + +@pytest.mark.asyncio +async def test_is_monitoring_enabled_true_when_store_opts_in(): + wrapper = AsyncFeatureStoreClientWrapper(FakeAsyncStore(), lambda _s: None) + assert wrapper.is_monitoring_enabled() is True + + +@pytest.mark.asyncio +async def test_is_monitoring_enabled_false_without_is_available(): + wrapper = AsyncFeatureStoreClientWrapper(StoreWithoutAvailability(), lambda _s: None) + assert wrapper.is_monitoring_enabled() is False + + +@pytest.mark.asyncio +async def test_is_monitoring_enabled_false_when_store_does_not_opt_in(): + # A store with is_available but no is_monitoring_enabled must not be polled. + wrapper = AsyncFeatureStoreClientWrapper(StoreWithAvailabilityNoOptIn(), lambda _s: None) + assert wrapper.is_monitoring_enabled() is False + + +@pytest.mark.asyncio +async def test_init_sorts_and_delegates(): + store = FakeAsyncStore() + wrapper = AsyncFeatureStoreClientWrapper(store, lambda _s: None) + await wrapper.init({FEATURES: {}, SEGMENTS: {}}) + assert len(store.init_calls) == 1 + assert wrapper.initialized is True + + +@pytest.mark.asyncio +async def test_failure_marks_unavailable_polls_and_recovers(): + store = FakeAsyncStore() + statuses: List[DataStoreStatus] = [] + wrapper = AsyncFeatureStoreClientWrapper(store, lambda s: statuses.append(s)) + + # Make the next operation fail. + store.fail = True + store._available = False + + with pytest.raises(RuntimeError): + await wrapper.get(FEATURES, "flag-a") + + # The wrapper reported unavailability and started a poller. + assert len(statuses) == 1 + assert statuses[0].available is False + + # Bring the store back; the poller (0.5s interval) should notice and recover. + store.fail = False + store._available = True + + for _ in range(40): + await asyncio.sleep(0.05) + if len(statuses) >= 2: + break + + assert len(statuses) == 2 + assert statuses[1].available is True + + await wrapper.close() + + +@pytest.mark.asyncio +async def test_close_stops_poller_and_closes_inner(): + store = FakeAsyncStore() + statuses: List[DataStoreStatus] = [] + wrapper = AsyncFeatureStoreClientWrapper(store, lambda s: statuses.append(s)) + + # Trigger an outage so a poller is running. + store.fail = True + store._available = False + with pytest.raises(RuntimeError): + await wrapper.all(FEATURES) + + assert wrapper._poller is not None + + await wrapper.close() + + assert wrapper._poller is None + assert store.closed is True + + +@pytest.mark.asyncio +async def test_successful_ops_pass_through(): + store = FakeAsyncStore() + wrapper = AsyncFeatureStoreClientWrapper(store, lambda _s: None) + + await wrapper.upsert(FEATURES, {"key": "flag-a", "version": 1}) + got = await wrapper.get(FEATURES, "flag-a") + assert got is not None and got["key"] == "flag-a" + allf = await wrapper.all(FEATURES) + assert "flag-a" in allf diff --git a/ldclient/testing/impl/datasystem/test_async_fdv2_persistence.py b/ldclient/testing/impl/datasystem/test_async_fdv2_persistence.py new file mode 100644 index 00000000..2d5cad5e --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_async_fdv2_persistence.py @@ -0,0 +1,490 @@ +# pylint: disable=missing-docstring + +""" +Integration tests for ``AsyncFDv2`` wired to an async persistent store. + +These drive the async data system end-to-end: the coordinator creates an +``AsyncStore`` and persists through it. +""" + +import asyncio +import logging +from typing import Any, Dict, List, Mapping, Optional + +import pytest + +from ldclient.async_config import AsyncConfig, AsyncDataSystemConfig +from ldclient.impl.datasystem import DataAvailability +from ldclient.impl.datasystem.async_fdv2 import ( + AsyncFDv2, + _AsyncReadOnlyStoreView +) +from ldclient.impl.datasystem.async_store import AsyncStore +from ldclient.impl.listeners import Listeners +from ldclient.integrations.test_datav2 import TestDataV2 +from ldclient.interfaces import ( + AsyncFeatureStore, + DataStoreMode, + DataStoreStatus, + FlagChange +) +from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind + + +class StubAsyncFeatureStore(AsyncFeatureStore): + """An async feature store stub that records operations and lets tests + inspect state. Availability can be toggled to exercise recovery paths.""" + + def __init__( + self, + initial_data: Optional[Dict[VersionedDataKind, Dict[str, dict]]] = None, + ): + self._data: Dict[VersionedDataKind, Dict[str, dict]] = { + FEATURES: {}, + SEGMENTS: {}, + } + self._initialized = False + self._available = True + + self.init_called_count = 0 + self.upsert_calls: List[tuple] = [] + self.delete_calls: List[tuple] = [] + self.closed = False + + # Controls for the recovery-path tests. + self.fail_init = False # raise from init() when True + self.init_gate: Optional[asyncio.Event] = None # if set, init() waits on it + self.init_after_close = False # set if init() ever runs after close() + + if initial_data: + self._data = { + FEATURES: dict(initial_data.get(FEATURES, {})), + SEGMENTS: dict(initial_data.get(SEGMENTS, {})), + } + self._initialized = True + + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + if self.init_gate is not None: + await self.init_gate.wait() + if self.closed: + self.init_after_close = True + if self.fail_init: + raise RuntimeError("store down") + self.init_called_count += 1 + self._data = { + FEATURES: dict(all_data.get(FEATURES, {})), + SEGMENTS: dict(all_data.get(SEGMENTS, {})), + } + self._initialized = True + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + return self._data.get(kind, {}).get(key) + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + return dict(self._data.get(kind, {})) + + async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: + self.upsert_calls.append((kind, item.get("key"), item.get("version"))) + key = item["key"] + existing = self._data.get(kind, {}).get(key) + if not existing or existing.get("version", 0) < item.get("version", 0): + self._data[kind][key] = item + return True + return False + + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: + self.delete_calls.append((kind, key, version)) + return await self.upsert(kind, {"key": key, "version": version, "deleted": True}) + + @property + def initialized(self) -> bool: + return self._initialized + + async def is_initialized(self) -> bool: + return self._initialized + + async def is_available(self) -> bool: + return self._available + + async def close(self) -> None: + self.closed = True + + def snapshot(self) -> Dict[VersionedDataKind, Dict[str, dict]]: + return {FEATURES: dict(self._data[FEATURES]), SEGMENTS: dict(self._data[SEGMENTS])} + + def reset_operation_tracking(self): + self.init_called_count = 0 + self.upsert_calls = [] + self.delete_calls = [] + + +def _flag_dict(key: str, version: int, on: bool = True) -> dict: + return { + "key": key, + "version": version, + "on": on, + "variations": [True, False], + "fallthrough": {"variation": 0}, + } + + +async def _wait_for(event: asyncio.Event, timeout: float = 2.0): + await asyncio.wait_for(event.wait(), timeout=timeout) + + +@pytest.mark.asyncio +async def test_async_persistent_store_read_write_mode(): + persistent_store = StubAsyncFeatureStore() + + td = TestDataV2.data_source() + td.update(td.flag("new-flag").on(True)) + + data_system_config = AsyncDataSystemConfig( + data_store_mode=DataStoreMode.READ_WRITE, + data_store=persistent_store, + initializers=None, + synchronizers=[td.async_builder], + ) + + ready = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + + # The coordinator must build an AsyncStore -- not a sync Store. + assert isinstance(fdv2._store, AsyncStore) + + fdv2.start(ready) + await _wait_for(ready) + assert (await fdv2.data_availability()).at_least(DataAvailability.REFRESHED) + + # A full transfer persists through init(), and the new flag lands in the store. + assert persistent_store.init_called_count >= 1 + assert "new-flag" in persistent_store.snapshot()[FEATURES] + + await fdv2.stop() + assert persistent_store.closed is True + + +@pytest.mark.asyncio +async def test_async_persistent_store_read_only_mode(): + initial = {FEATURES: {"existing-flag": _flag_dict("existing-flag", 1)}, SEGMENTS: {}} + persistent_store = StubAsyncFeatureStore(initial) + persistent_store.reset_operation_tracking() + + td = TestDataV2.data_source() + td.update(td.flag("new-flag").on(True)) + + data_system_config = AsyncDataSystemConfig( + data_store_mode=DataStoreMode.READ_ONLY, + data_store=persistent_store, + initializers=None, + synchronizers=[td.async_builder], + ) + + ready = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready) + await _wait_for(ready) + assert (await fdv2.data_availability()).at_least(DataAvailability.REFRESHED) + + # READ_ONLY: nothing is written back to the persistent store. + assert persistent_store.init_called_count == 0 + assert len(persistent_store.upsert_calls) == 0 + + # In-memory now serves reads. + flag = await fdv2.store.get(FEATURES, "new-flag") + assert flag is not None and flag["key"] == "new-flag" + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_persistent_store_delta_updates_read_write(): + persistent_store = StubAsyncFeatureStore() + + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + + data_system_config = AsyncDataSystemConfig( + data_store_mode=DataStoreMode.READ_WRITE, + data_store=persistent_store, + initializers=None, + synchronizers=[td.async_builder], + ) + + ready = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + + flag_changed = asyncio.Event() + change_count = 0 + + def listener(_change: FlagChange): + nonlocal change_count + change_count += 1 + if change_count == 2: # first from initial sync, second from our update + flag_changed.set() + + fdv2.flag_change_listeners.add(listener) + fdv2.start(ready) + await _wait_for(ready) + + persistent_store.reset_operation_tracking() + + # A delta update. + td.update(td.flag("feature-flag").on(False)) + await _wait_for(flag_changed) + + # The delta persists via upsert. + assert any(call[1] == "feature-flag" for call in persistent_store.upsert_calls) + assert persistent_store.snapshot()[FEATURES]["feature-flag"]["on"] is False + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_persistent_store_outage_recovery_flushes_on_recovery(): + persistent_store = StubAsyncFeatureStore() + + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + + data_system_config = AsyncDataSystemConfig( + data_store_mode=DataStoreMode.READ_WRITE, + data_store=persistent_store, + initializers=None, + synchronizers=[td.async_builder], + ) + + ready = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + + new_flag_applied = asyncio.Event() + + def listener(change: FlagChange): + if change.key == "new-flag": + new_flag_applied.set() + + fdv2.flag_change_listeners.add(listener) + fdv2.start(ready) + await _wait_for(ready) + + assert "feature-flag" in persistent_store.snapshot()[FEATURES] + persistent_store.reset_operation_tracking() + + # A runtime update lands in memory (and the store, since READ_WRITE). + td.update(td.flag("new-flag").on(False)) + await _wait_for(new_flag_applied) + + persistent_store.reset_operation_tracking() + + # Store comes back online with stale data -> the coordinator schedules a commit. + fdv2._persistent_store_outage_recovery(DataStoreStatus(available=True, stale=True)) + + # The commit is scheduled with asyncio.ensure_future; let it run. + for _ in range(20): + await asyncio.sleep(0.01) + if persistent_store.init_called_count > 0: + break + + assert persistent_store.init_called_count > 0, "Store should have been reinitialized" + snapshot = persistent_store.snapshot() + assert "feature-flag" in snapshot[FEATURES] + assert "new-flag" in snapshot[FEATURES] + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_persistent_store_outage_recovery_no_flush_when_not_stale(): + persistent_store = StubAsyncFeatureStore() + + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + + data_system_config = AsyncDataSystemConfig( + data_store_mode=DataStoreMode.READ_WRITE, + data_store=persistent_store, + initializers=None, + synchronizers=[td.async_builder], + ) + + ready = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready) + await _wait_for(ready) + + persistent_store.reset_operation_tracking() + fdv2._persistent_store_outage_recovery(DataStoreStatus(available=True, stale=False)) + + # Give any (erroneously) scheduled task a chance to run. + await asyncio.sleep(0.05) + assert persistent_store.init_called_count == 0 + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_persistent_store_outage_recovery_no_flush_when_unavailable(): + persistent_store = StubAsyncFeatureStore() + + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + + data_system_config = AsyncDataSystemConfig( + data_store_mode=DataStoreMode.READ_WRITE, + data_store=persistent_store, + initializers=None, + synchronizers=[td.async_builder], + ) + + ready = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), data_system_config) + fdv2.start(ready) + await _wait_for(ready) + + persistent_store.reset_operation_tracking() + fdv2._persistent_store_outage_recovery(DataStoreStatus(available=False, stale=True)) + + await asyncio.sleep(0.05) + assert persistent_store.init_called_count == 0 + + await fdv2.stop() + + +@pytest.mark.asyncio +async def test_async_store_view_reads_through_async_persistent_store(): + """Before the memory store has data, the active store is the async + persistent store, so the view must await its get/all and decode dicts.""" + raw = _flag_dict("flag-a", 1) + inner = StubAsyncFeatureStore({FEATURES: {"flag-a": raw}, SEGMENTS: {}}) + + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(inner, True, None) + + view = _AsyncReadOnlyStoreView(store) + + # Active store is the async persistent store -> get()/all() are awaitables. + got = await view.get(FEATURES, "flag-a") + assert got == FEATURES.decode(raw) + assert not isinstance(got, dict) # dict decoded into a model + + all_flags = await view.all(FEATURES) + assert set(all_flags.keys()) == {"flag-a"} + assert all_flags["flag-a"] == FEATURES.decode(raw) + + assert await view.get(FEATURES, "missing") is None + + +@pytest.mark.asyncio +async def test_async_store_view_reads_from_in_memory_after_swap(): + """After a full apply the active store is the sync in-memory store, whose + reads are not awaitable; the view's isawaitable gate must handle that too.""" + from ldclient.interfaces import ( + Change, + ChangeSet, + ChangeType, + IntentCode, + ObjectKind, + Selector + ) + + inner = StubAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(inner, True, None) + view = _AsyncReadOnlyStoreView(store) + + changeset = ChangeSet( + intent_code=IntentCode.TRANSFER_FULL, + changes=[Change(action=ChangeType.PUT, kind=ObjectKind.FLAG, key="flag-a", version=1, object=_flag_dict("flag-a", 1))], + selector=Selector.no_selector(), + ) + await store.apply(changeset, True) + + # Active store is now the in-memory store (non-awaitable reads). + got = await view.get(FEATURES, "flag-a") + assert got == FEATURES.decode(_flag_dict("flag-a", 1)) + assert not isinstance(got, dict) + + +async def _started_fdv2_with_store(store): + td = TestDataV2.data_source() + td.update(td.flag("feature-flag").on(True)) + cfg = AsyncDataSystemConfig( + data_store_mode=DataStoreMode.READ_WRITE, + data_store=store, + initializers=None, + synchronizers=[td.async_builder], + ) + ready = asyncio.Event() + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="dummy"), cfg) + fdv2.start(ready) + await _wait_for(ready) + return fdv2 + + +@pytest.mark.asyncio +async def test_recovery_skipped_and_no_write_after_stop(): + """After stop() has closed the store, a recovery firing is skipped: it + schedules no task and never writes to the closed store.""" + store = StubAsyncFeatureStore() + fdv2 = await _started_fdv2_with_store(store) + + await fdv2.stop() + assert store.closed is True + store.reset_operation_tracking() + + fdv2._persistent_store_outage_recovery(DataStoreStatus(available=True, stale=True)) + await asyncio.sleep(0.05) + + assert store.init_called_count == 0 + assert store.init_after_close is False + assert not any(t.get_name() == "AsyncFDv2-store-recovery" for t in fdv2._runner._tasks) + + +@pytest.mark.asyncio +async def test_recovery_task_spawned_before_stop_is_drained_by_stop(): + """A recovery commit in flight is owned by the coordinator's runner, so + stop() cancels and awaits it -- it does not leak.""" + store = StubAsyncFeatureStore() + fdv2 = await _started_fdv2_with_store(store) + + # A long-running commit so the recovery task is reliably in flight. + started = asyncio.Event() + + async def slow_commit(): + started.set() + await asyncio.sleep(30) + return None + + fdv2._store.commit = slow_commit # type: ignore[assignment] + + fdv2._persistent_store_outage_recovery(DataStoreStatus(available=True, stale=True)) + await asyncio.wait_for(started.wait(), timeout=2) + + assert any(t.get_name() == "AsyncFDv2-store-recovery" for t in fdv2._runner._tasks) + + await fdv2.stop() + assert not any(t.get_name() == "AsyncFDv2-store-recovery" for t in fdv2._runner._tasks) + + +@pytest.mark.asyncio +async def test_recovery_error_is_logged_not_swallowed(caplog): + """A failing recovery commit is logged and does not crash the coordinator + or leak an unretrieved task exception.""" + store = StubAsyncFeatureStore() + fdv2 = await _started_fdv2_with_store(store) + + store.reset_operation_tracking() + store.fail_init = True + store._available = False # keep the wrapper's availability poller from flapping + + with caplog.at_level(logging.ERROR): + fdv2._persistent_store_outage_recovery(DataStoreStatus(available=True, stale=True)) + await asyncio.sleep(0.1) + + assert any("Failed to reinitialize data store" in r.getMessage() for r in caplog.records) + + # The coordinator is still usable and shuts down cleanly. + flag = await fdv2.store.get(FEATURES, "feature-flag") + assert flag is not None + store.fail_init = False + await fdv2.stop() diff --git a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py index 7879b263..dcfab21b 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py @@ -1,17 +1,26 @@ # pylint: disable=missing-docstring +import sys from typing import Any, Dict, List, Mapping, Optional +from unittest.mock import patch import pytest +from ldclient.async_config import AsyncConfig, AsyncDataSystemConfig +from ldclient.async_feature_store_helpers import AsyncCachingStoreWrapper +from ldclient.feature_store import CacheConfig +from ldclient.impl.datasystem import DataAvailability +from ldclient.impl.datasystem.async_fdv2 import AsyncFDv2 from ldclient.impl.datasystem.async_store import AsyncStore from ldclient.impl.datasystem.store import Store from ldclient.impl.listeners import Listeners from ldclient.interfaces import ( AsyncFeatureStore, + AsyncFeatureStoreCore, Change, ChangeSet, ChangeType, + DataStoreMode, IntentCode, ObjectKind, Selector @@ -60,6 +69,9 @@ async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: def initialized(self) -> bool: return self._inited + async def is_initialized(self) -> bool: + return self._inited + async def close(self) -> None: self.closed = True @@ -195,17 +207,215 @@ async def init(self, all_data): assert str(err) == "boom" +@pytest.mark.asyncio +async def test_commit_returns_error_when_snapshot_encode_raises(): + """If encoding the memory snapshot raises, commit() returns the exception + rather than raising it.""" + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + # Populate memory so the snapshot iterates a flag and calls FEATURES.encode. + await store.apply(_full_changeset("flag-a", 1, True), True) + async_store.init_called_count = 0 + + with patch.object(FEATURES, "encode", side_effect=RuntimeError("encode boom")): + err = await store.commit() + + assert isinstance(err, RuntimeError) + assert str(err) == "encode boom" + # The failure happened during the snapshot, so the store was never written. + assert async_store.init_called_count == 0 + + @pytest.mark.asyncio async def test_close_closes_async_store(): async_store = FakeAsyncFeatureStore() store = AsyncStore(Listeners(), Listeners()) store.with_async_persistence(async_store, True, None) - err = await store.close() - assert err is None + await store.close() assert async_store.closed is True +@pytest.mark.asyncio +async def test_close_logs_and_swallows_store_error(caplog): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + with patch.object(async_store, "close", side_effect=RuntimeError("close boom")): + # A close error is logged and swallowed, never raised. + await store.close() + + assert any( + "Error closing the persistent store" in record.message + for record in caplog.records + if record.levelname == "WARNING" + ) + + +class FakeAsyncCore(AsyncFeatureStoreCore): + """An async core whose data and initialized flag can be set out of band. + + Setting ``inited`` and ``data`` directly, without going through ``init``, + models another process populating the store. ``query_count`` records how + often ``initialized_internal`` runs so caching behavior can be asserted. + """ + + def __init__(self): + self.data: Dict[VersionedDataKind, Dict[str, dict]] = {FEATURES: {}, SEGMENTS: {}} + self.inited = False + self.query_count = 0 + + async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + self.data = {FEATURES: dict(all_data.get(FEATURES, {})), SEGMENTS: dict(all_data.get(SEGMENTS, {}))} + self.inited = True + + async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[dict]: + return self.data.get(kind, {}).get(key) + + async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, dict]: + return dict(self.data.get(kind, {})) + + async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict: + self.data[kind][item["key"]] = item + return item + + async def initialized_internal(self) -> bool: + self.query_count += 1 + return self.inited + + +@pytest.mark.asyncio +async def test_wrapper_is_initialized_reflects_external_init_and_latches(): + core = FakeAsyncCore() + wrapper = AsyncCachingStoreWrapper(core, CacheConfig.disabled()) + + assert await wrapper.is_initialized() is False + assert wrapper.initialized is False + + # Another process initializes the store. + core.inited = True + assert await wrapper.is_initialized() is True + assert wrapper.initialized is True + + # The state has latched, so a later loss of the init key does not flip it back. + core.inited = False + assert await wrapper.is_initialized() is True + + +@pytest.mark.asyncio +async def test_wrapper_is_initialized_queries_every_call_when_cache_off(): + core = FakeAsyncCore() + wrapper = AsyncCachingStoreWrapper(core, CacheConfig.disabled()) + + assert await wrapper.is_initialized() is False + assert await wrapper.is_initialized() is False + assert core.query_count == 2 + + core.inited = True + assert await wrapper.is_initialized() is True + assert core.query_count == 3 + + # Latched: no more queries. + assert await wrapper.is_initialized() is True + assert core.query_count == 3 + + +@pytest.mark.asyncio +async def test_wrapper_is_initialized_infinite_cache_never_reflects_later_init(): + core = FakeAsyncCore() + wrapper = AsyncCachingStoreWrapper(core, CacheConfig(expiration=sys.maxsize)) + + assert await wrapper.is_initialized() is False + assert core.query_count == 1 + + # The False result is cached forever, so a later init is not observed. + core.inited = True + assert await wrapper.is_initialized() is False + assert core.query_count == 1 + + +@pytest.mark.asyncio +async def test_store_is_ready_gates_reads_and_stops_once_memory_active(): + core = FakeAsyncCore() + wrapper = AsyncCachingStoreWrapper(core, CacheConfig.disabled()) + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(wrapper, False, None) + + assert await store.is_ready() is False + + # Another process initializes the store; the gate then reports ready. + core.inited = True + assert await store.is_ready() is True + + # Once a basis arrives the memory store is active; the persistent store is no + # longer queried. + await store.apply(_full_changeset("flag-a", 1, True), True) + assert store.get_active_store() is store._memory_store + queries_before = core.query_count + assert await store.is_ready() is True + assert core.query_count == queries_before + + +@pytest.mark.asyncio +async def test_fdv2_gate_serves_store_with_data_source_after_external_init(): + core = FakeAsyncCore() + core.data[FEATURES]["flag-a"] = _flag("flag-a", 1, True) + wrapper = AsyncCachingStoreWrapper(core, CacheConfig.disabled()) + + class _NeverBuiltSyncBuilder: + pass + + ds_config = AsyncDataSystemConfig( + synchronizers=[_NeverBuiltSyncBuilder()], # type: ignore[list-item] + data_store=wrapper, + data_store_mode=DataStoreMode.READ_ONLY, + ) + fdv2 = AsyncFDv2(AsyncConfig(sdk_key="fake", send_events=False), ds_config) + + # A synchronizer is configured but no basis has arrived, so before the store + # reports initialized the gate withholds the store's data. + assert await fdv2.data_availability() == DataAvailability.DEFAULTS + + # Another process initializes the store; the gate now serves its data, and the + # store read agrees (no is_initialized-vs-evaluation divergence). + core.inited = True + assert await fdv2.data_availability() == DataAvailability.CACHED + flag = await fdv2.store.get(FEATURES, "flag-a") + assert flag is not None + assert flag.key == "flag-a" + + +@pytest.mark.asyncio +async def test_user_store_missing_readiness_check_is_a_typed_error(): + """A user-supplied AsyncFeatureStore that omits the readiness check cannot be + instantiated, so the gap surfaces as a typed error rather than silent DEFAULTS.""" + class StoreWithoutReadiness(AsyncFeatureStore): + async def get(self, kind, key): + return None + + async def all(self, kind): + return {} + + async def init(self, all_data): + pass + + async def upsert(self, kind, item): + return True + + async def delete(self, kind, key, version): + return True + + @property + def initialized(self) -> bool: + return True + + with pytest.raises(TypeError): + StoreWithoutReadiness() # type: ignore[abstract] + + def test_sync_apply_still_persists_synchronously(): """The default sync path is unchanged: it persists inline to a sync store.""" from ldclient.testing.impl.datasystem.test_fdv2_persistence import ( diff --git a/ldclient/testing/test_async_client.py b/ldclient/testing/test_async_client.py index 944b0b3a..2e8ae161 100644 --- a/ldclient/testing/test_async_client.py +++ b/ldclient/testing/test_async_client.py @@ -109,7 +109,7 @@ async def test_close_is_idempotent(): async def test_context_manager(): """async with AsyncLDClient(config) as client: starts and closes the client.""" async with AsyncLDClient(_offline_config()) as client: - assert client.is_initialized() + assert await client.is_initialized() assert client._closed is True diff --git a/ldclient/testing/test_sync_async_parity.py b/ldclient/testing/test_sync_async_parity.py index 3d5dde76..845fc20d 100644 --- a/ldclient/testing/test_sync_async_parity.py +++ b/ldclient/testing/test_sync_async_parity.py @@ -41,7 +41,9 @@ def _public_surface(cls) -> set: pytest.param( InMemoryFeatureStore, AsyncInMemoryFeatureStore, set(), - {"close"}, + # is_initialized is an awaitable readiness check; the sync store exposes + # readiness through the synchronous ``initialized`` property instead. + {"close", "is_initialized"}, id="feature_store", ), pytest.param(Evaluator, AsyncEvaluator, set(), set(), id="evaluator"),