From 344bab6f3fe6bb1adb17f92f19d34781b048c6ed Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 18 Aug 2026 13:17:01 -0500 Subject: [PATCH 1/6] feat: Add read-only store views and async persistence foundation for the data system --- ldclient/async_config.py | 46 ++- ldclient/client.py | 14 +- ldclient/impl/datasourcev2/async_polling.py | 7 +- ldclient/impl/datasourcev2/async_streaming.py | 2 +- ldclient/impl/datastore/async_status.py | 144 ++++++++ ldclient/impl/datasystem/async_fdv1.py | 26 +- ldclient/impl/datasystem/async_store.py | 217 ++++++++++++ ldclient/impl/datasystem/fdv1.py | 31 +- ldclient/impl/datasystem/fdv2.py | 31 +- ldclient/impl/datasystem/store.py | 329 +++++++++++------- .../impl/datastore/test_async_status.py | 164 +++++++++ .../impl/datasystem/test_async_config.py | 71 ++++ .../impl/datasystem/test_async_store_view.py | 71 ++++ .../datasystem/test_fdv2_async_persistence.py | 224 ++++++++++++ .../impl/datasystem/test_store_view.py | 154 ++++++++ ldclient/testing/test_ldclient_evaluation.py | 43 +++ 16 files changed, 1430 insertions(+), 144 deletions(-) create mode 100644 ldclient/impl/datastore/async_status.py create mode 100644 ldclient/impl/datasystem/async_store.py create mode 100644 ldclient/testing/impl/datastore/test_async_status.py create mode 100644 ldclient/testing/impl/datasystem/test_async_config.py create mode 100644 ldclient/testing/impl/datasystem/test_async_store_view.py create mode 100644 ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py create mode 100644 ldclient/testing/impl/datasystem/test_store_view.py diff --git a/ldclient/async_config.py b/ldclient/async_config.py index 2af0a1fb..64cc1072 100644 --- a/ldclient/async_config.py +++ b/ldclient/async_config.py @@ -11,14 +11,16 @@ from typing import Callable, List, Optional, Set from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from dataclasses import dataclass + from ldclient.config import ( DEFAULT_BASE_URI, DEFAULT_EVENTS_URI, DEFAULT_STREAM_URI, GET_LATEST_FEATURES_PATH, STREAM_FLAGS_PATH, + DataSourceBuilder, DataSourceBuilderConfig, - DataSystemConfig, HTTPConfig, PrivateAttributesConfig ) @@ -34,7 +36,10 @@ AsyncDataSourceUpdateSink, AsyncEventProcessor, AsyncFeatureStore, - AsyncUpdateProcessor + AsyncInitializer, + AsyncSynchronizer, + AsyncUpdateProcessor, + DataStoreMode ) from ldclient.plugin import AsyncPlugin @@ -91,6 +96,39 @@ def stale_after(self) -> float: return self.__stale_after +@dataclass(frozen=True) +class AsyncDataSystemConfig: + """Configuration for the async SDK's data acquisition strategy. + + This mirrors :class:`ldclient.config.DataSystemConfig` for the async client. + Its data sources are async builders and its data store is an async store. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. + """ + + initializers: Optional[List[DataSourceBuilder[AsyncInitializer]]] = None + """The initializers for the data system.""" + + synchronizers: Optional[List[DataSourceBuilder[AsyncSynchronizer]]] = None + """ + The synchronizers for the data system, ordered by preference. + The first synchronizer is the most preferred, with subsequent synchronizers + serving as fallbacks in order of decreasing preference. + """ + + data_store_mode: DataStoreMode = DataStoreMode.READ_WRITE + """The data store mode specifies the mode in which the persistent store will operate, if present.""" + + data_store: Optional[AsyncFeatureStore] = None + """The (optional) async persistent data store instance.""" + + fdv1_fallback_synchronizer: Optional[DataSourceBuilder[AsyncSynchronizer]] = None + """An optional fallback synchronizer that will read from FDv1""" + + class AsyncConfig(DataSourceBuilderConfig, PrivateAttributesConfig): """Advanced configuration options for the async SDK client. @@ -138,7 +176,7 @@ def __init__( enable_event_compression: bool = False, omit_anonymous_contexts: bool = False, payload_filter_key: Optional[str] = None, - datasystem_config: Optional[DataSystemConfig] = None, + datasystem_config: Optional[AsyncDataSystemConfig] = None, ): """ :param sdk_key: The SDK key for your LaunchDarkly account. This is always required. @@ -466,7 +504,7 @@ def data_source_update_sink(self) -> Optional[AsyncDataSourceUpdateSink]: return self._data_source_update_sink @property - def datasystem_config(self) -> Optional[DataSystemConfig]: + def datasystem_config(self) -> Optional[AsyncDataSystemConfig]: """ Configuration for the upcoming enhanced data system design. This is experimental and should not be set without direction from LaunchDarkly diff --git a/ldclient/client.py b/ldclient/client.py index 1bda5577..48eee401 100644 --- a/ldclient/client.py +++ b/ldclient/client.py @@ -165,14 +165,6 @@ def is_monitoring_enabled(self) -> bool: return monitoring_enabled() -def _get_store_item(store, kind: VersionedDataKind, key: str) -> Any: - # This decorator around store.get provides backward compatibility with any custom data - # store implementation that might still be returning a dict, instead of our data model - # classes like FeatureFlag. - item = store.get(kind, key, lambda x: x) - return kind.decode(item) if isinstance(item, dict) else item - - class LDClient: """The LaunchDarkly SDK client object. @@ -258,8 +250,8 @@ def __start_up(self, start_wait: float): self.__big_segment_store_manager = big_segment_store_manager self._evaluator = Evaluator( - lambda key: _get_store_item(self._data_system.store, FEATURES, key), - lambda key: _get_store_item(self._data_system.store, SEGMENTS, key), + lambda key: self._data_system.store.get(FEATURES, key), + lambda key: self._data_system.store.get(SEGMENTS, key), lambda key: big_segment_store_manager.get_user_membership(key), log, ) @@ -544,7 +536,7 @@ def _evaluate_internal(self, key: str, context: Context, default: Any, event_fac return EvaluationDetail(default, None, error_reason('USER_NOT_SPECIFIED')), None try: - flag = _get_store_item(self._data_system.store, FEATURES, key) + flag = self._data_system.store.get(FEATURES, key) except Exception as e: log.error("Unexpected error while retrieving feature flag \"%s\": %s" % (key, repr(e))) log.debug(traceback.format_exc()) diff --git a/ldclient/impl/datasourcev2/async_polling.py b/ldclient/impl/datasourcev2/async_polling.py index 5444120a..e93b6972 100644 --- a/ldclient/impl/datasourcev2/async_polling.py +++ b/ldclient/impl/datasourcev2/async_polling.py @@ -242,9 +242,12 @@ async def close(self) -> None: await self._http.close() -class AsyncPollingDataSourceBuilder(DataSourceBuilder): +class AsyncPollingDataSourceBuilder(DataSourceBuilder[AsyncPollingDataSource]): """ Builder for a AsyncPollingDataSource. + + The built polling data source implements both :class:`AsyncInitializer` and + :class:`AsyncSynchronizer`, so this builder can be used in either role. """ def __init__(self): @@ -298,7 +301,7 @@ def build(self, config: DataSourceBuilderConfig) -> AsyncPollingDataSource: ) -class AsyncFallbackToFDv1PollingDataSourceBuilder(DataSourceBuilder): +class AsyncFallbackToFDv1PollingDataSourceBuilder(DataSourceBuilder[AsyncPollingDataSource]): """ Builder for a AsyncPollingDataSource that falls back to Flag Delivery v1. """ diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index 42013ea6..36e7d09a 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -284,7 +284,7 @@ async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[O return (decision.update, decision.should_continue) -class AsyncStreamingDataSourceBuilder(DataSourceBuilder): +class AsyncStreamingDataSourceBuilder(DataSourceBuilder[AsyncStreamingDataSource]): """ Builder for a AsyncStreamingDataSource. """ diff --git a/ldclient/impl/datastore/async_status.py b/ldclient/impl/datastore/async_status.py new file mode 100644 index 00000000..2b15b49c --- /dev/null +++ b/ldclient/impl/datastore/async_status.py @@ -0,0 +1,144 @@ +""" +Async persistent-store availability tracking. + +This module provides :class:`AsyncFeatureStoreClientWrapper`, the async analog of +``ldclient.impl.datasystem.fdv2_common.FeatureStoreClientWrapper``. It wraps an +async feature store, sorts collections on ``init``, and watches the store for +outages so that recovery can be reported to a status sink. + +The wrapper reports status through a generic callable sink, so it carries no +dependency on the FDv2 data system. +""" + +import inspect +from typing import Any, Callable, Dict, Mapping, Optional + +from ldclient.feature_store import _FeatureStoreDataSetSorter +from ldclient.impl.aio.concurrency import AsyncRepeatingTask +from ldclient.impl.util import log +from ldclient.interfaces import AsyncFeatureStore, 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 + + 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 supports availability checks. + + Availability polling requires the store to provide an ``is_available`` + method so the wrapper can detect recovery. + """ + return callable(getattr(self._store, "is_available", None)) + + 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.""" + poller_to_stop = None + if not self._closed: + self._closed = True + poller_to_stop = self._poller + self._poller = None + + if poller_to_stop is not None: + poller_to_stop.stop() + await poller_to_stop.wait_stopped() + + close = getattr(self._store, "close", None) + if callable(close): + result = close() + if inspect.isawaitable(result): + await result + + +__all__ = ["AsyncFeatureStoreClientWrapper"] diff --git a/ldclient/impl/datasystem/async_fdv1.py b/ldclient/impl/datasystem/async_fdv1.py index 8fc84f92..1b18355c 100644 --- a/ldclient/impl/datasystem/async_fdv1.py +++ b/ldclient/impl/datasystem/async_fdv1.py @@ -1,4 +1,4 @@ -from typing import Any, Callable, Optional +from typing import Any, Callable, Dict, Optional from ldclient.async_config import AsyncConfig from ldclient.impl.aio.concurrency import AsyncEvent @@ -21,6 +21,7 @@ DataAvailability, DiagnosticAccumulator ) +from ldclient.impl.datasystem.store import _decode from ldclient.impl.listeners import Listeners from ldclient.impl.stubs import AsyncNullUpdateProcessor from ldclient.impl.util import log @@ -31,6 +32,26 @@ DataSourceStatusProvider, DataStoreStatusProvider ) +from ldclient.versioned_data_kind import VersionedDataKind + + +class _AsyncReadOnlyFeatureStoreView(AsyncReadOnlyStore): + """Exposes only async ``get``/``all`` over a stable async store, decoding dict items. + + The wrapped store is always async and never swaps, so it is read directly. + Items stored as dicts are decoded into model objects; items already decoded + are returned unchanged. + """ + + def __init__(self, store: AsyncReadOnlyStore): + self._store = store + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + return _decode(kind, await self._store.get(kind, key)) + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + result = await self._store.all(kind) + return {key: _decode(kind, item) for key, item in result.items()} class AsyncFDv1(AsyncDataSystem): @@ -45,6 +66,7 @@ class AsyncFDv1(AsyncDataSystem): def __init__(self, config: AsyncConfig, store: AsyncFeatureStore, session_provider: Callable[[], Any]): self._config = config self._store = store + self._store_view = _AsyncReadOnlyFeatureStoreView(store) # The client creates the aiohttp session lazily inside the loop; the data # source resolves it here when it builds its network processor at start(). self._session_provider = session_provider @@ -100,7 +122,7 @@ async def stop(self): @property def store(self) -> AsyncReadOnlyStore: - return self._store + return self._store_view def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): """ diff --git a/ldclient/impl/datasystem/async_store.py b/ldclient/impl/datasystem/async_store.py new file mode 100644 index 00000000..4a0f1fea --- /dev/null +++ b/ldclient/impl/datasystem/async_store.py @@ -0,0 +1,217 @@ +""" +Async persist sibling of the FDv2 store. + +:class:`AsyncStore` shares the store-agnostic engine of +:class:`ldclient.impl.datasystem.store._StoreBase` with the synchronous +:class:`ldclient.impl.datasystem.store.Store`, but owns an async persistent +store and writes to it through awaited I/O run outside the synchronous lock. +""" + +from typing import Any, Callable, Dict, Optional + +from ldclient.impl.aio.concurrency import AsyncLock +from ldclient.impl.datasystem.store import Collections, _StoreBase +from ldclient.impl.listeners import Listeners +from ldclient.impl.model.entity import ModelEntity +from ldclient.impl.util import log +from ldclient.interfaces import ( + AsyncFeatureStore, + ChangeSet, + DataStoreStatusProvider, + IntentCode +) +from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind + + +class AsyncStore(_StoreBase): + """ + AsyncStore is a dual-mode persistent/in-memory store that persists asynchronously. + + It behaves like :class:`ldclient.impl.datasystem.store.Store` for in-memory + reads and change notification, but writes to an async persistent store + through awaited I/O run outside the synchronous lock. + """ + + def __init__( + self, + flag_change_listeners: Listeners, + change_set_listeners: Listeners, + ): + super().__init__(flag_change_listeners, change_set_listeners) + + self._persistent_store: Optional[AsyncFeatureStore] = None + self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None + self._persistent_store_writable = False + + # True if the data in the memory store may be persisted to the persistent store + self._persist = False + + # Serializes async store writes; held only across the awaited I/O, never with self._lock. + self._async_persist_lock = AsyncLock() + + def with_async_persistence( + self, + async_store: AsyncFeatureStore, + writable: bool, + status_provider: Optional[DataStoreStatusProvider] = None, + ) -> "AsyncStore": + """ + Configure the store with an async persistent store for read-only or read-write access. + + Args: + async_store: The async persistent store implementation + writable: Whether the persistent store should be written to + status_provider: Optional status provider for the persistent store + + Returns: + Self for method chaining + """ + with self._lock: + self._persistent_store = async_store + self._persistent_store_writable = writable + self._persistent_store_status_provider = status_provider + + # Initially use persistent store as active until memory store has data + self._active_store = async_store # type: ignore[assignment] + + return self + + def _should_persist(self) -> bool: + """Returns whether data should be persisted to the persistent store.""" + return ( + self._persist + and self._persistent_store is not None + and self._persistent_store_writable + ) + + def _on_memory_store_active(self) -> None: + # In-memory store is now authoritative. Replace the persistent-store + # cache with a no-op so we don't hold a duplicate copy of every flag. + # Done before the persist step so the wrapper's init can skip its decode + # loop now that the cache is disabled. + if self._persistent_store is not None and hasattr( + self._persistent_store, "disable_cache" + ): + try: + self._persistent_store.disable_cache() # type: ignore[attr-defined] + except Exception as e: + log.warning("Failed to disable persistent store cache: %s", e) + + def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]: + self._persist = persist + return collections if self._should_persist() else None + + def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]: + self._persist = persist + return collections if self._should_persist() else None + + async def apply_async(self, change_set: ChangeSet, persist: bool) -> None: + """ + Apply a changeset to the store using the async persist path. + + The in-memory update and change-set notification run under the + synchronous lock with no awaits inside it. The persistent-store write is + awaited afterwards, outside the lock, serialized by the async persist lock. + The in-memory store is authoritative, so change listeners fire before the + awaited store write completes. + + Args: + change_set: The changeset to apply + persist: Whether the changes should be persisted to the persistent store + """ + collections = self._changes_to_store_data(change_set.changes) + + pending: Optional[Collections] = None + is_full = False + + with self._lock: + try: + if change_set.intent_code == IntentCode.TRANSFER_FULL: + pending = self._set_basis(collections, change_set.selector, persist) + is_full = True + elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: + pending = self._apply_delta(collections, change_set.selector, persist) + elif change_set.intent_code == IntentCode.TRANSFER_NONE: + return + + self._change_set_listeners.notify(change_set) + except Exception as e: + log.error("Store: couldn't apply changeset: %s", str(e)) + return + + if pending is None: + return + + store = self._persistent_store + if store is None: + return + + async with self._async_persist_lock: + if is_full: + await store.init(pending) + else: + for kind in pending: + kind_data = pending[kind] + for key in kind_data: + await store.upsert(kind, kind_data[key]) + + async def commit_async(self) -> Optional[Exception]: + """ + Persist the data in the memory store to the async persistent store, if configured. + + The memory read happens under the synchronous lock; the store write is + awaited afterwards, serialized by the async persist lock. + + Returns: + Exception if the commit failed, None otherwise + """ + def __mapping_from_kind(kind: VersionedDataKind) -> Callable[[Dict[str, ModelEntity]], Dict[str, Dict[str, Any]]]: + def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]: + return {k: kind.encode(v) for k, v in data.items()} + + return __mapping + + 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 + + store = self._persistent_store + if store is None: + return None + + async with self._async_persist_lock: + try: + await store.init(all_data) + except Exception as e: + return e + return None + + async def close_async(self) -> Optional[Exception]: + """ + Close the store and the async persistent store, if configured. + + Returns: + Exception if closing failed, None otherwise + """ + store = self._persistent_store + if store is None: + return None + try: + await store.close() + except Exception as e: + return e + return None + + 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 + + +__all__ = ["AsyncStore"] diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index f6067e9d..a3771849 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -1,5 +1,5 @@ from threading import Event -from typing import Optional +from typing import Any, Callable, Optional from ldclient.config import Config from ldclient.impl.datasource.feature_requester import FeatureRequesterImpl @@ -18,6 +18,7 @@ DataSystem, DiagnosticAccumulator ) +from ldclient.impl.datasystem.store import _decode from ldclient.impl.listeners import Listeners from ldclient.impl.stubs import NullUpdateProcessor from ldclient.interfaces import ( @@ -27,10 +28,35 @@ ReadOnlyStore, UpdateProcessor ) +from ldclient.versioned_data_kind import VersionedDataKind # Delayed import inside __init__ to avoid circular dependency with ldclient.client +class _ReadOnlyFeatureStoreView(ReadOnlyStore): + """Exposes only ``get``/``all`` over a feature store, decoding dict items. + + The wrapped store is stable and never swaps, so it is read directly. Items + stored as dicts are decoded into model objects; items already decoded are + returned unchanged, then the caller's ``callback`` is applied. + """ + + def __init__(self, store: FeatureStore): + self._store = store + + def get(self, kind: VersionedDataKind, key: str, callback: Callable[[Any], Any] = lambda x: x) -> Any: + item = self._store.get(kind, key, lambda x: x) + return callback(_decode(kind, item)) + + def all(self, kind: VersionedDataKind, callback: Callable[[Any], Any] = lambda x: x) -> Any: + items = self._store.all(kind, lambda x: x) + return callback({key: _decode(kind, value) for key, value in items.items()}) + + @property + def initialized(self) -> bool: + return self._store.initialized + + class FDv1(DataSystem): """ FDv1 wires the existing v1 data source and store behavior behind the @@ -51,6 +77,7 @@ def __init__(self, config: Config): self._store_wrapper: FeatureStore = _FeatureStoreClientWrapper( self._config.feature_store, self._data_store_update_sink ) + self._store_view = _ReadOnlyFeatureStoreView(self._store_wrapper) self._data_store_status_provider_impl = DataStoreStatusProviderImpl( self._store_wrapper, self._data_store_update_sink ) @@ -94,7 +121,7 @@ def stop(self): @property def store(self) -> ReadOnlyStore: - return self._store_wrapper + return self._store_view @property def environment_id(self) -> Optional[str]: diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index a316c3c3..b5d0fed7 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -16,7 +16,7 @@ DataStoreStatusProviderImpl, FeatureStoreClientWrapper ) -from ldclient.impl.datasystem.store import Store +from ldclient.impl.datasystem.store import Store, _decode from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock @@ -35,6 +35,32 @@ ) +class _ReadOnlyStoreView(ReadOnlyStore): + """Exposes only ``get``/``all`` over a store, decoding dict items. + + Resolves the active store on each read rather than at construction, so a held + instance follows the active-store swap: reads hit the persistent store before + the in-memory store has data, and the in-memory store afterwards. Items stored + as dicts are decoded into model objects; items already decoded are returned + unchanged, then the caller's ``callback`` is applied. + """ + + def __init__(self, store: Store): + self._store = store + + def get(self, kind: VersionedDataKind, key: str, callback: Callable[[Any], Any] = lambda x: x) -> Any: + item = self._store.get_active_store().get(kind, key, lambda x: x) + return callback(_decode(kind, item)) + + def all(self, kind: VersionedDataKind, callback: Callable[[Any], Any] = lambda x: x) -> Any: + items = self._store.get_active_store().all(kind, lambda x: x) + return callback({key: _decode(kind, value) for key, value in items.items()}) + + @property + def initialized(self) -> bool: + return self._store.is_initialized() + + class FDv2(DataSystem): """ FDv2 is an implementation of the DataSystem interface that uses the Flag Delivery V2 protocol @@ -73,6 +99,7 @@ def __init__( # Create the store self._store = Store(self._flag_change_listeners, self._change_set_listeners) + self._store_view = _ReadOnlyStoreView(self._store) # Status providers self._data_source_status_provider = DataSourceStatusProviderImpl(Listeners()) @@ -515,7 +542,7 @@ def environment_id(self) -> Optional[str]: @property def store(self) -> ReadOnlyStore: """Get the underlying store for flag evaluation.""" - return self._store.get_active_store() + return self._store_view @property def data_source_status_provider(self) -> DataSourceStatusProvider: diff --git a/ldclient/impl/datasystem/store.py b/ldclient/impl/datasystem/store.py index 0c31912a..01903153 100644 --- a/ldclient/impl/datasystem/store.py +++ b/ldclient/impl/datasystem/store.py @@ -32,6 +32,11 @@ Collections = Dict[VersionedDataKind, Dict[str, dict]] +def _decode(kind: VersionedDataKind, item: Any) -> Any: + """Decode a dict item into a model object; return non-dict items unchanged.""" + return kind.decode(item) if isinstance(item, dict) else item + + class InMemoryFeatureStore(ReadOnlyStore): """ The default feature store implementation, which holds all data in a @@ -146,15 +151,20 @@ def initialized(self) -> bool: return self._initialized -class Store: +class _StoreBase: """ - Store is a dual-mode persistent/in-memory store that serves requests for - data from the evaluation algorithm. - - At any given moment one of two stores is active: in-memory, or persistent. - Once the in-memory store has data (either from initializers or a - synchronizer), the persistent store is no longer read from. From that point - forward, calls to get data will serve from the memory store. + Shared, store-agnostic engine for the FDv2 store. + + The store is dual-mode: at any moment one of two stores is active, in-memory + or persistent. Once the in-memory store has data (from an initializer or a + synchronizer), reads serve from memory and the persistent store is no longer + read from. + + This base holds the in-memory store, dependency tracking, listeners, the + active-store swap, and the changeset-to-memory apply. It never references a + persistent store: subclasses own their concretely-typed store and supply the + persist step through the ``_stage_persist_full``/``_stage_persist_delta`` + hooks and the ``_on_memory_store_active`` hook. """ def __init__( @@ -163,16 +173,12 @@ def __init__( change_set_listeners: Listeners, ): """ - Initialize a new Store. + Initialize a new store. Args: flag_change_listeners: Listeners for flag change events change_set_listeners: Listeners for changeset events """ - self._persistent_store: Optional[FeatureStore] = None - self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None - self._persistent_store_writable = False - # Source of truth for flag evaluations once initialized self._memory_store = InMemoryFeatureStore() @@ -183,9 +189,6 @@ def __init__( self._flag_change_listeners = flag_change_listeners self._change_set_listeners = change_set_listeners - # True if the data in the memory store may be persisted to the persistent store - self._persist = False - # Points to the active store. Swapped upon initialization. self._active_store: ReadOnlyStore = self._memory_store @@ -195,87 +198,52 @@ def __init__( # Thread synchronization self._lock = threading.RLock() - def with_persistence( - self, - persistent_store: FeatureStore, - writable: bool, - status_provider: Optional[DataStoreStatusProvider] = None, - ) -> "Store": - """ - Configure the store with a persistent store for read-only or read-write access. - - Args: - persistent_store: The persistent store implementation - writable: Whether the persistent store should be written to - status_provider: Optional status provider for the persistent store - - Returns: - Self for method chaining - """ - with self._lock: - self._persistent_store = persistent_store - self._persistent_store_writable = writable - self._persistent_store_status_provider = status_provider - - # Initially use persistent store as active until memory store has data - self._active_store = persistent_store - - return self - def selector(self) -> Selector: """Returns the current selector.""" with self._lock: return self._selector - def close(self) -> Optional[Exception]: - """Close the store and any persistent store if configured.""" - with self._lock: - if self._persistent_store is not None: - try: - # Most FeatureStore implementations don't have close methods - # but we'll try to call it if it exists - if hasattr(self._persistent_store, "close"): - self._persistent_store.close() - except Exception as e: - return e - return None - - def apply(self, change_set: ChangeSet, persist: bool) -> None: + def _on_memory_store_active(self) -> None: """ - Apply a changeset to the store. + Called after the memory store becomes the active store. Subclasses use + this to react to the persistent store no longer being read from. The + default does nothing. + """ + pass - Args: - change_set: The changeset to apply - persist: Whether the changes should be persisted to the persistent store + def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]: """ - collections = self._changes_to_store_data(change_set.changes) + Persist a full data set. Subclasses supply the persist step. - with self._lock: - try: - if change_set.intent_code == IntentCode.TRANSFER_FULL: - self._set_basis(collections, change_set.selector, persist) - elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: - self._apply_delta(collections, change_set.selector, persist) - elif change_set.intent_code == IntentCode.TRANSFER_NONE: - # No-op, no changes to apply - return + A subclass either writes the data synchronously and returns None, or + returns the collections for the caller to persist afterwards. The + subclass owns the decision of whether to persist at all. + """ + raise NotImplementedError - # Notify changeset listeners - self._change_set_listeners.notify(change_set) + def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]: + """ + Persist a delta update. Subclasses supply the persist step. - except Exception as e: - # Log error but don't re-raise - matches Go behavior - log.error("Store: couldn't apply changeset: %s", str(e)) + A subclass either writes the data synchronously and returns None, or + returns the collections for the caller to persist afterwards. The + subclass owns the decision of whether to persist at all. + """ + raise NotImplementedError def _set_basis( self, collections: Collections, selector: Selector, persist: bool - ) -> None: + ) -> Optional[Collections]: """ Set the basis of the store. Any existing data is discarded. Args: - change_set: The changeset containing the new basis data + collections: The new basis data + selector: The selector identifying the data persist: Whether to persist the data to the persistent store + + Returns: + The collections the subclass deferred for later persistence, or None. """ # Take snapshot for change detection if we have flag listeners old_data: Optional[Collections] = None @@ -286,33 +254,23 @@ def _set_basis( ok = self._memory_store.set_basis(collections) if ok is False: - return + return None # Update dependency tracker self._reset_dependency_tracker(collections) # Update state - self._persist = persist self._selector = selector if selector is not None else Selector.no_selector() # Switch to memory store as active self._active_store = self._memory_store - # In-memory store is now authoritative. Replace the persistent-store - # cache with a no-op so we don't hold a duplicate copy of every flag. - # Done before persistent_store.init() below so the wrapper's init can - # skip its decode loop now that the cache is disabled. - if self._persistent_store is not None and hasattr( - self._persistent_store, "disable_cache" - ): - try: - self._persistent_store.disable_cache() # type: ignore[attr-defined] - except Exception as e: - log.warning("Failed to disable persistent store cache: %s", e) + # In-memory store is now authoritative. The subclass reacts here (e.g. by + # disabling the persistent-store cache) before the persist step below. + self._on_memory_store_active() - # Persist to persistent store if configured and writable - if self._should_persist(): - self._persistent_store.init(collections) # type: ignore + # Persist through the subclass hook + pending = self._stage_persist_full(collections, persist) # Send change events if we had listeners if old_data is not None: @@ -321,19 +279,25 @@ def _set_basis( ) self._send_change_events(affected_items) + return pending + def _apply_delta( self, collections: Collections, selector: Selector, persist: bool - ) -> None: + ) -> Optional[Collections]: """ Apply a delta update to the store. Args: - change_set: The changeset containing the delta changes + collections: The delta changes + selector: The selector identifying the data persist: Whether to persist the changes to the persistent store + + Returns: + The collections the subclass deferred for later persistence, or None. """ ok = self._memory_store.apply_delta(collections) if ok is False: - return + return None has_listeners = self._flag_change_listeners.has_listeners() affected_items: Set[KindAndKey] = set() @@ -351,27 +315,15 @@ def _apply_delta( ) # Update state - self._persist = persist self._selector = selector if selector is not None else Selector.no_selector() - if self._should_persist(): - for kind in collections: - kind_data: Dict[str, dict] = collections[kind] - for i in kind_data: - item = kind_data[i] - self._persistent_store.upsert(kind, item) # type: ignore + pending = self._stage_persist_delta(collections, persist) # Send change events if affected_items: self._send_change_events(affected_items) - def _should_persist(self) -> bool: - """Returns whether data should be persisted to the persistent store.""" - return ( - self._persist - and self._persistent_store is not None - and self._persistent_store_writable - ) + return pending def _changes_to_store_data(self, changes: List[Change]) -> Collections: """ @@ -441,6 +393,136 @@ def _compute_changed_items_for_full_data_set( return affected_items + def get_active_store(self) -> ReadOnlyStore: + """Get the currently active store for reading data.""" + with self._lock: + return self._active_store + + def is_initialized(self) -> bool: + """Check if the active store is initialized.""" + return self.get_active_store().initialized + + +class Store(_StoreBase): + """ + Store is a dual-mode persistent/in-memory store that persists synchronously. + + At any given moment one of two stores is active: in-memory, or persistent. + Once the in-memory store has data (either from initializers or a + synchronizer), the persistent store is no longer read from. From that point + forward, calls to get data will serve from the memory store. + """ + + def __init__( + self, + flag_change_listeners: Listeners, + change_set_listeners: Listeners, + ): + super().__init__(flag_change_listeners, change_set_listeners) + + self._persistent_store: Optional[FeatureStore] = None + self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None + self._persistent_store_writable = False + + # True if the data in the memory store may be persisted to the persistent store + self._persist = False + + def with_persistence( + self, + persistent_store: FeatureStore, + writable: bool, + status_provider: Optional[DataStoreStatusProvider] = None, + ) -> "Store": + """ + Configure the store with a persistent store for read-only or read-write access. + + Args: + persistent_store: The persistent store implementation + writable: Whether the persistent store should be written to + status_provider: Optional status provider for the persistent store + + Returns: + Self for method chaining + """ + with self._lock: + self._persistent_store = persistent_store + self._persistent_store_writable = writable + self._persistent_store_status_provider = status_provider + + # Initially use persistent store as active until memory store has data + self._active_store = persistent_store + + return self + + def apply(self, change_set: ChangeSet, persist: bool) -> None: + """ + Apply a changeset to the store. + + Args: + change_set: The changeset to apply + persist: Whether the changes should be persisted to the persistent store + """ + collections = self._changes_to_store_data(change_set.changes) + + with self._lock: + try: + if change_set.intent_code == IntentCode.TRANSFER_FULL: + self._set_basis(collections, change_set.selector, persist) + elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: + self._apply_delta(collections, change_set.selector, persist) + elif change_set.intent_code == IntentCode.TRANSFER_NONE: + # No-op, no changes to apply + return + + # Notify changeset listeners + self._change_set_listeners.notify(change_set) + + except Exception as e: + # Log error but don't re-raise - matches Go behavior + log.error("Store: couldn't apply changeset: %s", str(e)) + + def _should_persist(self) -> bool: + """Returns whether data should be persisted to the persistent store.""" + return ( + self._persist + and self._persistent_store is not None + and self._persistent_store_writable + ) + + def _on_memory_store_active(self) -> None: + # In-memory store is now authoritative. Replace the persistent-store + # cache with a no-op so we don't hold a duplicate copy of every flag. + # Done before the persist step so the wrapper's init can skip its decode + # loop now that the cache is disabled. + if self._persistent_store is not None and hasattr( + self._persistent_store, "disable_cache" + ): + try: + self._persistent_store.disable_cache() # type: ignore[attr-defined] + except Exception as e: + log.warning("Failed to disable persistent store cache: %s", e) + + def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]: + self._persist = persist + if not self._should_persist(): + return None + store = self._persistent_store + assert store is not None + store.init(collections) + return None + + def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]: + self._persist = persist + if not self._should_persist(): + return None + store = self._persistent_store + assert store is not None + for kind in collections: + kind_data = collections[kind] + for key in kind_data: + store.upsert(kind, kind_data[key]) + return None + def commit(self) -> Optional[Exception]: """ Commit persists the data in the memory store to the persistent store, if configured. @@ -456,24 +538,31 @@ def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]: with self._lock: if self._should_persist(): + store = self._persistent_store + assert store is not None try: # Get all data from memory store and write to persistent store - all_data = {} + all_data: Collections = {} for kind in [FEATURES, SEGMENTS]: all_data[kind] = self._memory_store.all(kind, __mapping_from_kind(kind)) - self._persistent_store.init(all_data) # type: ignore + store.init(all_data) except Exception as e: return e return None - def get_active_store(self) -> ReadOnlyStore: - """Get the currently active store for reading data.""" + def close(self) -> Optional[Exception]: + """Close the store and any persistent store if configured.""" with self._lock: - return self._active_store - - def is_initialized(self) -> bool: - """Check if the active store is initialized.""" - return self.get_active_store().initialized + if self._persistent_store is not None: + try: + # Most FeatureStore implementations don't have close methods + # but we'll try to call it if it exists + close = getattr(self._persistent_store, "close", None) + if callable(close): + close() + except Exception as e: + return e + return None def get_data_store_status_provider(self) -> Optional[DataStoreStatusProvider]: """Get the data store status provider for the persistent store, if configured.""" diff --git a/ldclient/testing/impl/datastore/test_async_status.py b/ldclient/testing/impl/datastore/test_async_status.py new file mode 100644 index 00000000..d353d281 --- /dev/null +++ b/ldclient/testing/impl/datastore/test_async_status.py @@ -0,0 +1,164 @@ +# pylint: disable=missing-docstring + +import asyncio +from typing import Any, Dict, List, Mapping, Optional + +import pytest + +from ldclient.impl.datastore.async_status import AsyncFeatureStoreClientWrapper +from ldclient.interfaces import AsyncFeatureStore, DataStoreStatus +from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind + + +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_available(self) -> bool: + return self._available + + 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 + + +@pytest.mark.asyncio +async def test_is_monitoring_enabled_true_when_store_has_is_available(): + 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_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_config.py b/ldclient/testing/impl/datasystem/test_async_config.py new file mode 100644 index 00000000..1f010c45 --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_async_config.py @@ -0,0 +1,71 @@ +# pylint: disable=missing-docstring + +from typing import Any, Dict, Mapping, Optional + +from ldclient.async_config import AsyncDataSystemConfig +from ldclient.config import DataSystemConfig +from ldclient.impl.datasourcev2.async_polling import ( + AsyncFallbackToFDv1PollingDataSourceBuilder, + AsyncPollingDataSourceBuilder +) +from ldclient.impl.datasourcev2.async_streaming import ( + AsyncStreamingDataSourceBuilder +) +from ldclient.interfaces import AsyncFeatureStore, DataStoreMode +from ldclient.versioned_data_kind import VersionedDataKind + + +class FakeAsyncStore(AsyncFeatureStore): + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + pass + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + return None + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + return {} + + async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: + return True + + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: + return True + + @property + def initialized(self) -> bool: + return True + + +def test_async_data_system_config_defaults(): + cfg = AsyncDataSystemConfig() + assert cfg.initializers is None + assert cfg.synchronizers is None + assert cfg.data_store is None + assert cfg.fdv1_fallback_synchronizer is None + # Reuses the shared DataStoreMode enum with the same default as the sync config. + assert cfg.data_store_mode is DataStoreMode.READ_WRITE + assert cfg.data_store_mode is DataSystemConfig.data_store_mode + + +def test_async_data_system_config_accepts_async_builders_and_store(): + store = FakeAsyncStore() + cfg = AsyncDataSystemConfig( + initializers=[AsyncPollingDataSourceBuilder()], + synchronizers=[AsyncStreamingDataSourceBuilder(), AsyncPollingDataSourceBuilder()], + data_store_mode=DataStoreMode.READ_ONLY, + data_store=store, + fdv1_fallback_synchronizer=AsyncFallbackToFDv1PollingDataSourceBuilder(), + ) + + assert cfg.initializers is not None and len(cfg.initializers) == 1 + assert cfg.synchronizers is not None and len(cfg.synchronizers) == 2 + assert cfg.data_store is store + assert isinstance(cfg.data_store, AsyncFeatureStore) + assert cfg.data_store_mode is DataStoreMode.READ_ONLY + assert cfg.fdv1_fallback_synchronizer is not None + + +def test_async_data_system_config_shares_data_store_mode_enum(): + # The async config does not define its own mode enum. + assert AsyncDataSystemConfig(data_store_mode=DataStoreMode.READ_WRITE).data_store_mode \ + is DataSystemConfig(initializers=None, synchronizers=None, data_store_mode=DataStoreMode.READ_WRITE).data_store_mode diff --git a/ldclient/testing/impl/datasystem/test_async_store_view.py b/ldclient/testing/impl/datasystem/test_async_store_view.py new file mode 100644 index 00000000..c348f003 --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_async_store_view.py @@ -0,0 +1,71 @@ +# pylint: disable=missing-docstring + +from typing import Any, Dict + +import pytest + +from ldclient.impl.datasystem.async_fdv1 import _AsyncReadOnlyFeatureStoreView +from ldclient.versioned_data_kind import FEATURES, VersionedDataKind + + +def _flag_dict(key: str, version: int) -> dict: + return { + "key": key, + "version": version, + "on": True, + "variations": [True, False], + "fallthrough": {"variation": 0}, + } + + +class FakeAsyncStore: + """Async store shape: get/all are coroutines with no callback.""" + + def __init__(self, data: Dict[str, Any]): + self._data = data + + async def get(self, kind: VersionedDataKind, key: str) -> Any: + return self._data.get(key) + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + return dict(self._data) + + +@pytest.mark.asyncio +async def test_async_feature_store_view_get_decodes_dict(): + raw = _flag_dict("flag-a", 1) + view = _AsyncReadOnlyFeatureStoreView(FakeAsyncStore({"flag-a": raw})) + + result = await view.get(FEATURES, "flag-a") + + assert result == FEATURES.decode(raw) + assert not isinstance(result, dict) + + +@pytest.mark.asyncio +async def test_async_feature_store_view_get_passes_through_model(): + decoded = FEATURES.decode(_flag_dict("flag-a", 1)) + view = _AsyncReadOnlyFeatureStoreView(FakeAsyncStore({"flag-a": decoded})) + + result = await view.get(FEATURES, "flag-a") + + assert result is decoded + + +@pytest.mark.asyncio +async def test_async_feature_store_view_get_missing_returns_none(): + view = _AsyncReadOnlyFeatureStoreView(FakeAsyncStore({})) + assert await view.get(FEATURES, "missing") is None + + +@pytest.mark.asyncio +async def test_async_feature_store_view_all_decodes_each_value(): + raw_a = _flag_dict("flag-a", 1) + decoded_b = FEATURES.decode(_flag_dict("flag-b", 1)) + view = _AsyncReadOnlyFeatureStoreView(FakeAsyncStore({"flag-a": raw_a, "flag-b": decoded_b})) + + result = await view.all(FEATURES) + + assert set(result.keys()) == {"flag-a", "flag-b"} + assert result["flag-a"] == FEATURES.decode(raw_a) # dict decoded + assert result["flag-b"] is decoded_b # model passed through diff --git a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py new file mode 100644 index 00000000..eaf6c9da --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py @@ -0,0 +1,224 @@ +# pylint: disable=missing-docstring + +from typing import Any, Dict, List, Mapping, Optional + +import pytest + +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, + Change, + ChangeSet, + ChangeType, + IntentCode, + ObjectKind, + Selector +) +from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind + + +class FakeAsyncFeatureStore(AsyncFeatureStore): + """An in-memory async feature store that records the operations it receives.""" + + def __init__(self): + self._data: Dict[VersionedDataKind, Dict[str, dict]] = {FEATURES: {}, SEGMENTS: {}} + self._inited = False + + self.init_called_count = 0 + self.upsert_calls: List[tuple] = [] + self.closed = False + + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + self.init_called_count += 1 + 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]: + 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: + return await self.upsert(kind, {"key": key, "version": version, "deleted": True}) + + @property + def initialized(self) -> bool: + return self._inited + + 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 _flag(key: str, version: int, on: bool) -> dict: + return { + "key": key, + "version": version, + "on": on, + "variations": [True, False], + "fallthrough": {"variation": 0}, + } + + +def _full_changeset(key: str, version: int, on: bool) -> ChangeSet: + return ChangeSet( + intent_code=IntentCode.TRANSFER_FULL, + changes=[Change(action=ChangeType.PUT, kind=ObjectKind.FLAG, key=key, version=version, object=_flag(key, version, on))], + selector=Selector.no_selector(), + ) + + +def _delta_changeset(key: str, version: int, on: bool) -> ChangeSet: + return ChangeSet( + intent_code=IntentCode.TRANSFER_CHANGES, + changes=[Change(action=ChangeType.PUT, kind=ObjectKind.FLAG, key=key, version=version, object=_flag(key, version, on))], + selector=Selector.no_selector(), + ) + + +@pytest.mark.asyncio +async def test_apply_async_full_transfer_persists_via_init(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + await store.apply_async(_full_changeset("flag-a", 1, True), True) + + # After a full transfer the memory store is authoritative and serves reads + assert store.get_active_store() is store._memory_store + flag = store._memory_store.get(FEATURES, "flag-a") + assert flag is not None + assert flag["key"] == "flag-a" + + # The async store received the init + assert async_store.init_called_count == 1 + assert "flag-a" in async_store.snapshot()[FEATURES] + + +@pytest.mark.asyncio +async def test_apply_async_delta_persists_via_upsert(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + await store.apply_async(_full_changeset("flag-a", 1, True), True) + async_store.init_called_count = 0 + async_store.upsert_calls = [] + + await store.apply_async(_delta_changeset("flag-a", 2, False), True) + + assert any(call[1] == "flag-a" and call[2] == 2 for call in async_store.upsert_calls) + assert async_store.snapshot()[FEATURES]["flag-a"]["on"] is False + + +@pytest.mark.asyncio +async def test_apply_async_read_only_does_not_persist(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + # writable=False -> READ_ONLY: never write to the store + store.with_async_persistence(async_store, False, None) + + await store.apply_async(_full_changeset("flag-a", 1, True), True) + await store.apply_async(_delta_changeset("flag-a", 2, False), True) + + assert async_store.init_called_count == 0 + assert async_store.upsert_calls == [] + # In-memory store still updated + assert store._memory_store.get(FEATURES, "flag-a") is not None + + +@pytest.mark.asyncio +async def test_apply_async_fires_change_set_listeners(): + async_store = FakeAsyncFeatureStore() + received: List[ChangeSet] = [] + change_set_listeners = Listeners() + change_set_listeners.add(lambda cs: received.append(cs)) + + store = AsyncStore(Listeners(), change_set_listeners) + store.with_async_persistence(async_store, True, None) + + cs = _full_changeset("flag-a", 1, True) + await store.apply_async(cs, True) + + assert received == [cs] + + +@pytest.mark.asyncio +async def test_commit_async_writes_memory_to_store(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + # Populate memory without persisting yet (read-only apply through memory) + await store.apply_async(_full_changeset("flag-a", 1, True), True) + async_store.init_called_count = 0 + + err = await store.commit_async() + assert err is None + assert async_store.init_called_count == 1 + assert "flag-a" in async_store.snapshot()[FEATURES] + + +@pytest.mark.asyncio +async def test_commit_async_returns_error_on_failure(): + class FailingStore(FakeAsyncFeatureStore): + async def init(self, all_data): + raise RuntimeError("boom") + + async_store = FailingStore() + store = AsyncStore(Listeners(), Listeners()) + # Read-only so the deferred persist is skipped and memory is populated first. + store.with_async_persistence(async_store, False, None) + await store.apply_async(_full_changeset("flag-a", 1, True), True) + + # Now make it writable and commit, which triggers the failing init. + store._persistent_store_writable = True + err = await store.commit_async() + assert isinstance(err, RuntimeError) + assert str(err) == "boom" + + +@pytest.mark.asyncio +async def test_close_async_closes_async_store(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + err = await store.close_async() + assert err is None + assert async_store.closed is True + + +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 ( + StubFeatureStore + ) + + sync_store = StubFeatureStore() + store = Store(Listeners(), Listeners()) + store.with_persistence(sync_store, True, None) + + store.apply(_full_changeset("flag-a", 1, True), True) + assert sync_store.init_called_count >= 1 + assert "flag-a" in sync_store.get_data_snapshot()[FEATURES] + + store.apply(_delta_changeset("flag-a", 2, False), True) + assert any(call[1] == "flag-a" for call in sync_store.upsert_calls) diff --git a/ldclient/testing/impl/datasystem/test_store_view.py b/ldclient/testing/impl/datasystem/test_store_view.py new file mode 100644 index 00000000..6dfec9be --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_store_view.py @@ -0,0 +1,154 @@ +# pylint: disable=missing-docstring + +from typing import Any, Callable, Dict + +from ldclient.impl.datasystem.fdv1 import _ReadOnlyFeatureStoreView +from ldclient.impl.datasystem.fdv2 import _ReadOnlyStoreView +from ldclient.impl.datasystem.store import Store +from ldclient.impl.listeners import Listeners +from ldclient.interfaces import ( + Change, + ChangeSet, + ChangeType, + IntentCode, + ObjectKind, + Selector +) +from ldclient.versioned_data_kind import FEATURES, VersionedDataKind + + +def _flag_dict(key: str, version: int) -> dict: + return { + "key": key, + "version": version, + "on": True, + "variations": [True, False], + "fallthrough": {"variation": 0}, + } + + +def _full_changeset(key: str, version: int) -> ChangeSet: + return ChangeSet( + intent_code=IntentCode.TRANSFER_FULL, + changes=[Change(action=ChangeType.PUT, kind=ObjectKind.FLAG, key=key, version=version, object=_flag_dict(key, version))], + selector=Selector.no_selector(), + ) + + +class FakeSyncFeatureStore: + """Sync store: get/all take an optional callback and return the stored items.""" + + def __init__(self, data: Dict[str, Any]): + self._data = data + + def get(self, kind: VersionedDataKind, key: str, callback: Callable[[Any], Any] = lambda x: x) -> Any: + return callback(self._data.get(key)) + + def all(self, kind: VersionedDataKind, callback: Callable[[Any], Any] = lambda x: x) -> Any: + return callback(dict(self._data)) + + @property + def initialized(self) -> bool: + return True + + +def test_feature_store_view_get_decodes_dict_and_passes_callback(): + raw = _flag_dict("flag-a", 1) + view = _ReadOnlyFeatureStoreView(FakeSyncFeatureStore({"flag-a": raw})) # type: ignore[arg-type] + + result = view.get(FEATURES, "flag-a") + assert result == FEATURES.decode(raw) + assert not isinstance(result, dict) + + key = view.get(FEATURES, "flag-a", lambda flag: flag.key if flag else None) + assert key == "flag-a" + + +def test_feature_store_view_get_passes_through_model(): + decoded = FEATURES.decode(_flag_dict("flag-a", 1)) + view = _ReadOnlyFeatureStoreView(FakeSyncFeatureStore({"flag-a": decoded})) # type: ignore[arg-type] + assert view.get(FEATURES, "flag-a") is decoded + + +def test_feature_store_view_get_missing_returns_none(): + view = _ReadOnlyFeatureStoreView(FakeSyncFeatureStore({})) # type: ignore[arg-type] + assert view.get(FEATURES, "missing") is None + + +def test_feature_store_view_all_decodes_and_passes_callback(): + raw = _flag_dict("flag-a", 1) + view = _ReadOnlyFeatureStoreView(FakeSyncFeatureStore({"flag-a": raw})) # type: ignore[arg-type] + + result = view.all(FEATURES) + assert result["flag-a"] == FEATURES.decode(raw) + + count = view.all(FEATURES, lambda items: len(items)) + assert count == 1 + + +def test_feature_store_view_initialized_delegates(): + view = _ReadOnlyFeatureStoreView(FakeSyncFeatureStore({})) # type: ignore[arg-type] + assert view.initialized is True + + +def test_store_view_get_decodes_and_passes_callback(): + store = Store(Listeners(), Listeners()) + view = _ReadOnlyStoreView(store) + + store.apply(_full_changeset("flag-a", 1), False) + + result = view.get(FEATURES, "flag-a") + assert result == FEATURES.decode(_flag_dict("flag-a", 1)) + assert not isinstance(result, dict) + + key = view.get(FEATURES, "flag-a", lambda flag: flag.key if flag else None) + assert key == "flag-a" + + +def test_store_view_all_decodes_and_passes_callback(): + store = Store(Listeners(), Listeners()) + view = _ReadOnlyStoreView(store) + store.apply(_full_changeset("flag-a", 1), False) + + result = view.all(FEATURES) + assert set(result.keys()) == {"flag-a"} + assert result["flag-a"] == FEATURES.decode(_flag_dict("flag-a", 1)) + + count = view.all(FEATURES, lambda items: len(items)) + assert count == 1 + + +def test_store_view_get_missing_returns_none(): + store = Store(Listeners(), Listeners()) + view = _ReadOnlyStoreView(store) + store.apply(_full_changeset("flag-a", 1), False) + assert view.get(FEATURES, "missing") is None + + +def test_store_view_initialized_delegates(): + store = Store(Listeners(), Listeners()) + view = _ReadOnlyStoreView(store) + assert view.initialized is False + store.apply(_full_changeset("flag-a", 1), False) + assert view.initialized is True + + +def test_held_store_view_follows_active_store_swap(): + # Persistent (sync) store is active before memory has data. + persistent = FakeSyncFeatureStore({"old-flag": _flag_dict("old-flag", 1)}) + store = Store(Listeners(), Listeners()) + store.with_persistence(persistent, False, None) # type: ignore[arg-type] + + view = _ReadOnlyStoreView(store) # held across the swap + + # Before init: reads hit the persistent store, decoded. + before = view.get(FEATURES, "old-flag") + assert before == FEATURES.decode(_flag_dict("old-flag", 1)) + + # A full transfer swaps the active store to the in-memory store. + store.apply(_full_changeset("new-flag", 1), False) + + # Same held view now follows the swap to the in-memory store. + after = view.get(FEATURES, "new-flag") + assert after == FEATURES.decode(_flag_dict("new-flag", 1)) + assert view.get(FEATURES, "old-flag") is None diff --git a/ldclient/testing/test_ldclient_evaluation.py b/ldclient/testing/test_ldclient_evaluation.py index f403e953..df7e3ae7 100644 --- a/ldclient/testing/test_ldclient_evaluation.py +++ b/ldclient/testing/test_ldclient_evaluation.py @@ -188,6 +188,49 @@ def test_all_flags_returns_none_if_feature_store_throws_error(caplog): assert errlog == ['Unable to read flags for all_flag_state: NotImplementedError()'] +class RawDictFeatureStore(FeatureStore): + """A custom store that returns raw dicts instead of decoded model objects, + like a legacy integration. The read path must decode these.""" + + def __init__(self, flags): + self._flags = flags + + def get(self, kind, key, callback=lambda x: x): + return callback(self._flags.get(key) if kind == FEATURES else None) + + def all(self, kind, callback=lambda x: x): + return callback(dict(self._flags) if kind == FEATURES else {}) + + def upsert(self, kind, item): + pass + + def delete(self, kind, key, version): + pass + + def init(self, data): + pass + + @property + def initialized(self): + return True + + +def test_all_flags_state_decodes_dicts_from_custom_store(): + # A custom store that hands back raw dicts must still yield decoded values + # through all_flags_state (the read path decodes). + store = RawDictFeatureStore({'key1': flag1, 'key2': flag2}) + client = make_client(store) + state = client.all_flags_state(user) + assert state.valid + assert state.to_values_map() == {'key1': 'value1', 'key2': 'value2'} + + +def test_variation_decodes_dicts_from_custom_store(): + store = RawDictFeatureStore({'key1': flag1}) + client = make_client(store) + assert client.variation('key1', user, default='default') == 'value1' + + def test_all_flags_state_returns_state(): store = InMemoryFeatureStore() store.init({FEATURES: {'key1': flag1, 'key2': flag2}}) From 5c73793c20b6cc706833f7b9b33f73a3ca24da1e Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 19 Aug 2026 13:52:14 -0500 Subject: [PATCH 2/6] refactor: Move the FDv1 feature-store wrapper into its data system --- ldclient/client.py | 118 +------------ ldclient/impl/datastore/async_status.py | 144 --------------- ldclient/impl/datastore/status.py | 2 +- ldclient/impl/datasystem/fdv1.py | 120 ++++++++++++- .../impl/datastore/test_async_status.py | 164 ------------------ .../test_feature_store_client_wrapper.py | 2 +- 6 files changed, 119 insertions(+), 431 deletions(-) delete mode 100644 ldclient/impl/datastore/async_status.py delete mode 100644 ldclient/testing/impl/datastore/test_async_status.py diff --git a/ldclient/client.py b/ldclient/client.py index 48eee401..a26e1ada 100644 --- a/ldclient/client.py +++ b/ldclient/client.py @@ -4,13 +4,12 @@ import threading import traceback -from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple +from typing import Any, Callable, List, Optional, Tuple from uuid import uuid4 from ldclient.config import Config from ldclient.context import Context from ldclient.evaluation import EvaluationDetail, FeatureFlagsState -from ldclient.feature_store import _FeatureStoreDataSetSorter from ldclient.hook import ( EvaluationSeriesContext, Hook, @@ -36,135 +35,22 @@ from ldclient.impl.events.types import EventFactory from ldclient.impl.flag_tracker import FlagTrackerImpl from ldclient.impl.model.feature_flag import FeatureFlag -from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock from ldclient.impl.stubs import NullEventProcessor, NullUpdateProcessor from ldclient.impl.util import check_uwsgi, log from ldclient.interfaces import ( BigSegmentStoreStatusProvider, DataSourceStatusProvider, - DataStoreStatus, DataStoreStatusProvider, - DataStoreUpdateSink, - FeatureStore, FlagTracker ) from ldclient.migrations import OpTracker, Stage from ldclient.plugin import EnvironmentMetadata -from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind +from ldclient.versioned_data_kind import FEATURES, SEGMENTS from .impl import AnyNum -class _FeatureStoreClientWrapper(FeatureStore): - """Provides additional behavior that the client requires before or after feature store operations. - Currently this just means sorting the data set for init() and dealing with data store status listeners. - """ - - def __init__(self, store: FeatureStore, store_update_sink: DataStoreUpdateSink): - self.store = store - self.__store_update_sink = store_update_sink - self.__monitoring_enabled = self.is_monitoring_enabled() - - # Covers the following variables - self.__lock = ReadWriteLock() - self.__last_available = True - self.__poller: Optional[RepeatingTask] = None - - def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]): - return self.__wrapper(lambda: self.store.init(_FeatureStoreDataSetSorter.sort_all_collections(all_data))) - - def get(self, kind, key, callback): - return self.__wrapper(lambda: self.store.get(kind, key, callback)) - - def all(self, kind, callback): - return self.__wrapper(lambda: self.store.all(kind, callback)) - - def delete(self, kind, key, version): - return self.__wrapper(lambda: self.store.delete(kind, key, version)) - - def upsert(self, kind, item): - return self.__wrapper(lambda: self.store.upsert(kind, item)) - - @property - def initialized(self) -> bool: - return self.store.initialized - - def __wrapper(self, fn: Callable): - try: - return fn() - except BaseException: - if self.__monitoring_enabled: - self.__update_availability(False) - raise - - def __update_availability(self, available: bool): - with self.__lock.write(): - if available == self.__last_available: - return - self.__last_available = available - - status = DataStoreStatus(available, False) - - if available: - log.warn("Persistent store is available again") - - self.__store_update_sink.update_status(status) - - if available: - with self.__lock.write(): - if self.__poller is not None: - self.__poller.stop() - self.__poller = None - - return - - log.warn("Detected persistent store unavailability; updates will be cached until it recovers") - task = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability) - - with self.__lock.write(): - self.__poller = task - self.__poller.start() - - def __check_availability(self): - try: - if self.store.is_available(): - self.__update_availability(True) - except BaseException as e: - log.error("Unexpected error from data store status function: %s", e) - - def is_monitoring_enabled(self) -> bool: - """ - This methods determines whether the wrapped store can support enabling monitoring. - - The wrapped store must provide a monitoring_enabled method, which must - be true. But this alone is not sufficient. - - Because this class wraps all interactions with a provided store, it can - technically "monitor" any store. However, monitoring also requires that - we notify listeners when the store is available again. - - We determine this by checking the store's `available?` method, so this - is also a requirement for monitoring support. - - These extra checks won't be necessary once `available` becomes a part - of the core interface requirements and this class no longer wraps every - feature store. - """ - - if not hasattr(self.store, 'is_monitoring_enabled'): - return False - - if not hasattr(self.store, 'is_available'): - return False - - monitoring_enabled = getattr(self.store, 'is_monitoring_enabled') - if not callable(monitoring_enabled): - return False - - return monitoring_enabled() - - class LDClient: """The LaunchDarkly SDK client object. diff --git a/ldclient/impl/datastore/async_status.py b/ldclient/impl/datastore/async_status.py deleted file mode 100644 index 2b15b49c..00000000 --- a/ldclient/impl/datastore/async_status.py +++ /dev/null @@ -1,144 +0,0 @@ -""" -Async persistent-store availability tracking. - -This module provides :class:`AsyncFeatureStoreClientWrapper`, the async analog of -``ldclient.impl.datasystem.fdv2_common.FeatureStoreClientWrapper``. It wraps an -async feature store, sorts collections on ``init``, and watches the store for -outages so that recovery can be reported to a status sink. - -The wrapper reports status through a generic callable sink, so it carries no -dependency on the FDv2 data system. -""" - -import inspect -from typing import Any, Callable, Dict, Mapping, Optional - -from ldclient.feature_store import _FeatureStoreDataSetSorter -from ldclient.impl.aio.concurrency import AsyncRepeatingTask -from ldclient.impl.util import log -from ldclient.interfaces import AsyncFeatureStore, 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 - - 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 supports availability checks. - - Availability polling requires the store to provide an ``is_available`` - method so the wrapper can detect recovery. - """ - return callable(getattr(self._store, "is_available", None)) - - 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.""" - poller_to_stop = None - if not self._closed: - self._closed = True - poller_to_stop = self._poller - self._poller = None - - if poller_to_stop is not None: - poller_to_stop.stop() - await poller_to_stop.wait_stopped() - - close = getattr(self._store, "close", None) - if callable(close): - result = close() - if inspect.isawaitable(result): - await result - - -__all__ = ["AsyncFeatureStoreClientWrapper"] diff --git a/ldclient/impl/datastore/status.py b/ldclient/impl/datastore/status.py index 1e4f145b..989e3215 100644 --- a/ldclient/impl/datastore/status.py +++ b/ldclient/impl/datastore/status.py @@ -12,7 +12,7 @@ ) if TYPE_CHECKING: - from ldclient.client import _FeatureStoreClientWrapper + from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper class DataStoreUpdateSinkImpl(DataStoreUpdateSink): diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index a3771849..e5f3052b 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -1,7 +1,8 @@ from threading import Event -from typing import Any, Callable, Optional +from typing import Any, Callable, Dict, Mapping, Optional from ldclient.config import Config +from ldclient.feature_store import _FeatureStoreDataSetSorter from ldclient.impl.datasource.feature_requester import FeatureRequesterImpl from ldclient.impl.datasource.polling import PollingUpdateProcessor from ldclient.impl.datasource.status import ( @@ -20,17 +21,129 @@ ) from ldclient.impl.datasystem.store import _decode from ldclient.impl.listeners import Listeners +from ldclient.impl.repeating_task import RepeatingTask +from ldclient.impl.rwlock import ReadWriteLock from ldclient.impl.stubs import NullUpdateProcessor +from ldclient.impl.util import log from ldclient.interfaces import ( DataSourceStatusProvider, + DataStoreStatus, DataStoreStatusProvider, + DataStoreUpdateSink, FeatureStore, ReadOnlyStore, UpdateProcessor ) from ldclient.versioned_data_kind import VersionedDataKind -# Delayed import inside __init__ to avoid circular dependency with ldclient.client + +class _FeatureStoreClientWrapper(FeatureStore): + """Provides additional behavior that the client requires before or after feature store operations. + Currently this just means sorting the data set for init() and dealing with data store status listeners. + """ + + def __init__(self, store: FeatureStore, store_update_sink: DataStoreUpdateSink): + self.store = store + self.__store_update_sink = store_update_sink + self.__monitoring_enabled = self.is_monitoring_enabled() + + # Covers the following variables + self.__lock = ReadWriteLock() + self.__last_available = True + self.__poller: Optional[RepeatingTask] = None + + def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]): + return self.__wrapper(lambda: self.store.init(_FeatureStoreDataSetSorter.sort_all_collections(all_data))) + + def get(self, kind, key, callback): + return self.__wrapper(lambda: self.store.get(kind, key, callback)) + + def all(self, kind, callback): + return self.__wrapper(lambda: self.store.all(kind, callback)) + + def delete(self, kind, key, version): + return self.__wrapper(lambda: self.store.delete(kind, key, version)) + + def upsert(self, kind, item): + return self.__wrapper(lambda: self.store.upsert(kind, item)) + + @property + def initialized(self) -> bool: + return self.store.initialized + + def __wrapper(self, fn: Callable): + try: + return fn() + except BaseException: + if self.__monitoring_enabled: + self.__update_availability(False) + raise + + def __update_availability(self, available: bool): + with self.__lock.write(): + if available == self.__last_available: + return + self.__last_available = available + + status = DataStoreStatus(available, False) + + if available: + log.warn("Persistent store is available again") + + self.__store_update_sink.update_status(status) + + if available: + with self.__lock.write(): + if self.__poller is not None: + self.__poller.stop() + self.__poller = None + + return + + log.warn("Detected persistent store unavailability; updates will be cached until it recovers") + task = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability) + + with self.__lock.write(): + self.__poller = task + self.__poller.start() + + def __check_availability(self): + try: + if self.store.is_available(): + self.__update_availability(True) + except BaseException as e: + log.error("Unexpected error from data store status function: %s", e) + + def is_monitoring_enabled(self) -> bool: + """ + This methods determines whether the wrapped store can support enabling monitoring. + + The wrapped store must provide a monitoring_enabled method, which must + be true. But this alone is not sufficient. + + Because this class wraps all interactions with a provided store, it can + technically "monitor" any store. However, monitoring also requires that + we notify listeners when the store is available again. + + We determine this by checking the store's `available?` method, so this + is also a requirement for monitoring support. + + These extra checks won't be necessary once `available` becomes a part + of the core interface requirements and this class no longer wraps every + feature store. + """ + + if not hasattr(self.store, 'is_monitoring_enabled'): + return False + + if not hasattr(self.store, 'is_available'): + return False + + monitoring_enabled = getattr(self.store, 'is_monitoring_enabled') + if not callable(monitoring_enabled): + return False + + return monitoring_enabled() class _ReadOnlyFeatureStoreView(ReadOnlyStore): @@ -71,9 +184,6 @@ def __init__(self, config: Config): self._data_store_update_sink = DataStoreUpdateSinkImpl( self._data_store_listeners ) - # Import here to avoid circular import - from ldclient.client import _FeatureStoreClientWrapper - self._store_wrapper: FeatureStore = _FeatureStoreClientWrapper( self._config.feature_store, self._data_store_update_sink ) diff --git a/ldclient/testing/impl/datastore/test_async_status.py b/ldclient/testing/impl/datastore/test_async_status.py deleted file mode 100644 index d353d281..00000000 --- a/ldclient/testing/impl/datastore/test_async_status.py +++ /dev/null @@ -1,164 +0,0 @@ -# pylint: disable=missing-docstring - -import asyncio -from typing import Any, Dict, List, Mapping, Optional - -import pytest - -from ldclient.impl.datastore.async_status import AsyncFeatureStoreClientWrapper -from ldclient.interfaces import AsyncFeatureStore, DataStoreStatus -from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind - - -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_available(self) -> bool: - return self._available - - 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 - - -@pytest.mark.asyncio -async def test_is_monitoring_enabled_true_when_store_has_is_available(): - 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_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/test_feature_store_client_wrapper.py b/ldclient/testing/test_feature_store_client_wrapper.py index 9f5e936c..85839b7c 100644 --- a/ldclient/testing/test_feature_store_client_wrapper.py +++ b/ldclient/testing/test_feature_store_client_wrapper.py @@ -2,7 +2,7 @@ from typing import Callable, List from unittest.mock import Mock -from ldclient.client import _FeatureStoreClientWrapper +from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper from ldclient.impl.datastore.status import DataStoreUpdateSinkImpl from ldclient.impl.listeners import Listeners From fbec0dac9ce223e76b08c949e670d01bd359bf9d Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 19 Aug 2026 14:29:25 -0500 Subject: [PATCH 3/6] fix: Log async persist failures instead of raising them Also rename AsyncStore.apply_async/commit_async/close_async to apply/commit/close (the async suffix was a carryover; AsyncStore is a sibling of Store and inherits no sync methods to disambiguate from), and fix isort ordering in async_config.py and the wrapper test. --- ldclient/async_config.py | 3 +- ldclient/impl/datasystem/async_store.py | 25 +++++++------ .../datasystem/test_fdv2_async_persistence.py | 36 +++++++++---------- .../test_feature_store_client_wrapper.py | 2 +- 4 files changed, 34 insertions(+), 32 deletions(-) diff --git a/ldclient/async_config.py b/ldclient/async_config.py index 64cc1072..189b2a45 100644 --- a/ldclient/async_config.py +++ b/ldclient/async_config.py @@ -8,11 +8,10 @@ compatibility guarantees. """ +from dataclasses import dataclass from typing import Callable, List, Optional, Set from ldclient.async_feature_store import AsyncInMemoryFeatureStore -from dataclasses import dataclass - from ldclient.config import ( DEFAULT_BASE_URI, DEFAULT_EVENTS_URI, diff --git a/ldclient/impl/datasystem/async_store.py b/ldclient/impl/datasystem/async_store.py index 4a0f1fea..63d44383 100644 --- a/ldclient/impl/datasystem/async_store.py +++ b/ldclient/impl/datasystem/async_store.py @@ -105,7 +105,7 @@ def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optio self._persist = persist return collections if self._should_persist() else None - async def apply_async(self, change_set: ChangeSet, persist: bool) -> None: + async def apply(self, change_set: ChangeSet, persist: bool) -> None: """ Apply a changeset to the store using the async persist path. @@ -147,15 +147,18 @@ async def apply_async(self, change_set: ChangeSet, persist: bool) -> None: return async with self._async_persist_lock: - if is_full: - await store.init(pending) - else: - for kind in pending: - kind_data = pending[kind] - for key in kind_data: - await store.upsert(kind, kind_data[key]) - - async def commit_async(self) -> Optional[Exception]: + try: + if is_full: + await store.init(pending) + else: + for kind in pending: + kind_data = pending[kind] + for key in kind_data: + await store.upsert(kind, kind_data[key]) + except Exception as e: + log.error("Store: couldn't persist changeset: %s", str(e)) + + async def commit(self) -> Optional[Exception]: """ Persist the data in the memory store to the async persistent store, if configured. @@ -192,7 +195,7 @@ def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]: return e return None - async def close_async(self) -> Optional[Exception]: + async def close(self) -> Optional[Exception]: """ Close the store and the async persistent store, if configured. diff --git a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py index eaf6c9da..7879b263 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py @@ -94,12 +94,12 @@ def _delta_changeset(key: str, version: int, on: bool) -> ChangeSet: @pytest.mark.asyncio -async def test_apply_async_full_transfer_persists_via_init(): +async def test_apply_full_transfer_persists_via_init(): async_store = FakeAsyncFeatureStore() store = AsyncStore(Listeners(), Listeners()) store.with_async_persistence(async_store, True, None) - await store.apply_async(_full_changeset("flag-a", 1, True), True) + await store.apply(_full_changeset("flag-a", 1, True), True) # After a full transfer the memory store is authoritative and serves reads assert store.get_active_store() is store._memory_store @@ -113,30 +113,30 @@ async def test_apply_async_full_transfer_persists_via_init(): @pytest.mark.asyncio -async def test_apply_async_delta_persists_via_upsert(): +async def test_apply_delta_persists_via_upsert(): async_store = FakeAsyncFeatureStore() store = AsyncStore(Listeners(), Listeners()) store.with_async_persistence(async_store, True, None) - await store.apply_async(_full_changeset("flag-a", 1, True), True) + await store.apply(_full_changeset("flag-a", 1, True), True) async_store.init_called_count = 0 async_store.upsert_calls = [] - await store.apply_async(_delta_changeset("flag-a", 2, False), True) + await store.apply(_delta_changeset("flag-a", 2, False), True) assert any(call[1] == "flag-a" and call[2] == 2 for call in async_store.upsert_calls) assert async_store.snapshot()[FEATURES]["flag-a"]["on"] is False @pytest.mark.asyncio -async def test_apply_async_read_only_does_not_persist(): +async def test_apply_read_only_does_not_persist(): async_store = FakeAsyncFeatureStore() store = AsyncStore(Listeners(), Listeners()) # writable=False -> READ_ONLY: never write to the store store.with_async_persistence(async_store, False, None) - await store.apply_async(_full_changeset("flag-a", 1, True), True) - await store.apply_async(_delta_changeset("flag-a", 2, False), True) + await store.apply(_full_changeset("flag-a", 1, True), True) + await store.apply(_delta_changeset("flag-a", 2, False), True) assert async_store.init_called_count == 0 assert async_store.upsert_calls == [] @@ -145,7 +145,7 @@ async def test_apply_async_read_only_does_not_persist(): @pytest.mark.asyncio -async def test_apply_async_fires_change_set_listeners(): +async def test_apply_fires_change_set_listeners(): async_store = FakeAsyncFeatureStore() received: List[ChangeSet] = [] change_set_listeners = Listeners() @@ -155,29 +155,29 @@ async def test_apply_async_fires_change_set_listeners(): store.with_async_persistence(async_store, True, None) cs = _full_changeset("flag-a", 1, True) - await store.apply_async(cs, True) + await store.apply(cs, True) assert received == [cs] @pytest.mark.asyncio -async def test_commit_async_writes_memory_to_store(): +async def test_commit_writes_memory_to_store(): async_store = FakeAsyncFeatureStore() store = AsyncStore(Listeners(), Listeners()) store.with_async_persistence(async_store, True, None) # Populate memory without persisting yet (read-only apply through memory) - await store.apply_async(_full_changeset("flag-a", 1, True), True) + await store.apply(_full_changeset("flag-a", 1, True), True) async_store.init_called_count = 0 - err = await store.commit_async() + err = await store.commit() assert err is None assert async_store.init_called_count == 1 assert "flag-a" in async_store.snapshot()[FEATURES] @pytest.mark.asyncio -async def test_commit_async_returns_error_on_failure(): +async def test_commit_returns_error_on_failure(): class FailingStore(FakeAsyncFeatureStore): async def init(self, all_data): raise RuntimeError("boom") @@ -186,22 +186,22 @@ async def init(self, all_data): store = AsyncStore(Listeners(), Listeners()) # Read-only so the deferred persist is skipped and memory is populated first. store.with_async_persistence(async_store, False, None) - await store.apply_async(_full_changeset("flag-a", 1, True), True) + await store.apply(_full_changeset("flag-a", 1, True), True) # Now make it writable and commit, which triggers the failing init. store._persistent_store_writable = True - err = await store.commit_async() + err = await store.commit() assert isinstance(err, RuntimeError) assert str(err) == "boom" @pytest.mark.asyncio -async def test_close_async_closes_async_store(): +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_async() + err = await store.close() assert err is None assert async_store.closed is True diff --git a/ldclient/testing/test_feature_store_client_wrapper.py b/ldclient/testing/test_feature_store_client_wrapper.py index 85839b7c..8ffe7c38 100644 --- a/ldclient/testing/test_feature_store_client_wrapper.py +++ b/ldclient/testing/test_feature_store_client_wrapper.py @@ -2,8 +2,8 @@ from typing import Callable, List from unittest.mock import Mock -from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper from ldclient.impl.datastore.status import DataStoreUpdateSinkImpl +from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper from ldclient.impl.listeners import Listeners From a28145f1f5c71386add7ef3b73b87d3806bdacd2 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 19 Aug 2026 14:34:31 -0500 Subject: [PATCH 4/6] fix: Restore type imports required by the FDv2 store view MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The rebase over #504 (unused-import cleanup) silently dropped Any, Callable, and VersionedDataKind from fdv2.py — they were unused on main before the read-only store view existed, so the auto-merge removed them even though the view's signatures depend on them. --- ldclient/impl/datasystem/fdv2.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index b5d0fed7..d4515ede 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -1,7 +1,7 @@ import time from queue import Queue from threading import Event, Thread -from typing import List, Optional +from typing import Any, Callable, List, Optional from ldclient.config import Config, DataSourceBuilder, DataSystemConfig from ldclient.impl.datasystem import ( @@ -33,6 +33,7 @@ ReadOnlyStore, Synchronizer ) +from ldclient.versioned_data_kind import VersionedDataKind class _ReadOnlyStoreView(ReadOnlyStore): From 11bd6a18ddeac2a7637c841086fbf72f71a9c5bd Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 20 Aug 2026 19:13:14 -0500 Subject: [PATCH 5/6] refactor: Simplify the FDv2 store persist path and clean up docstrings Remove the _stage_persist_full/_stage_persist_delta hook: _set_basis and _apply_delta now return the collections to persist, and each apply does its own persist (sync inline under the lock, async awaited under the persist lock). Make AsyncStore.commit take its snapshot and write under the persist lock so the flush is atomic. Tidy the _StoreBase and read-only view docstrings. --- ldclient/impl/datasystem/async_fdv1.py | 8 +- ldclient/impl/datasystem/async_store.py | 46 ++++-------- ldclient/impl/datasystem/fdv1.py | 8 +- ldclient/impl/datasystem/fdv2.py | 11 ++- ldclient/impl/datasystem/store.py | 99 ++++++++++--------------- 5 files changed, 67 insertions(+), 105 deletions(-) diff --git a/ldclient/impl/datasystem/async_fdv1.py b/ldclient/impl/datasystem/async_fdv1.py index 1b18355c..958facc5 100644 --- a/ldclient/impl/datasystem/async_fdv1.py +++ b/ldclient/impl/datasystem/async_fdv1.py @@ -36,11 +36,11 @@ class _AsyncReadOnlyFeatureStoreView(AsyncReadOnlyStore): - """Exposes only async ``get``/``all`` over a stable async store, decoding dict items. + """Read-only view of an async feature store. - The wrapped store is always async and never swaps, so it is read directly. - Items stored as dicts are decoded into model objects; items already decoded - are returned unchanged. + Serves every read from the wrapped store. Items that a custom feature store + keeps as raw dicts are decoded into model objects; items that are already + models pass through unchanged. """ def __init__(self, store: AsyncReadOnlyStore): diff --git a/ldclient/impl/datasystem/async_store.py b/ldclient/impl/datasystem/async_store.py index 63d44383..9c8c0142 100644 --- a/ldclient/impl/datasystem/async_store.py +++ b/ldclient/impl/datasystem/async_store.py @@ -43,9 +43,6 @@ def __init__( self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None self._persistent_store_writable = False - # True if the data in the memory store may be persisted to the persistent store - self._persist = False - # Serializes async store writes; held only across the awaited I/O, never with self._lock. self._async_persist_lock = AsyncLock() @@ -97,23 +94,10 @@ def _on_memory_store_active(self) -> None: except Exception as e: log.warning("Failed to disable persistent store cache: %s", e) - def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]: - self._persist = persist - return collections if self._should_persist() else None - - def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]: - self._persist = persist - return collections if self._should_persist() else None - async def apply(self, change_set: ChangeSet, persist: bool) -> None: """ - Apply a changeset to the store using the async persist path. - - The in-memory update and change-set notification run under the - synchronous lock with no awaits inside it. The persistent-store write is - awaited afterwards, outside the lock, serialized by the async persist lock. - The in-memory store is authoritative, so change listeners fire before the - awaited store write completes. + Apply a changeset to the in-memory store and, if configured, the async + persistent store. Args: change_set: The changeset to apply @@ -160,10 +144,8 @@ async def apply(self, change_set: ChangeSet, persist: bool) -> None: async def commit(self) -> Optional[Exception]: """ - Persist the data in the memory store to the async persistent store, if configured. - - The memory read happens under the synchronous lock; the store write is - awaited afterwards, serialized by the async persist lock. + Persist the contents of the memory store to the async persistent store, + if configured. Returns: Exception if the commit failed, None otherwise @@ -174,21 +156,21 @@ def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]: return __mapping - 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 - store = self._persistent_store if store is None: 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)) + + if all_data is None: + return None + try: await store.init(all_data) except Exception as e: diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index e5f3052b..38655415 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -147,11 +147,11 @@ def is_monitoring_enabled(self) -> bool: class _ReadOnlyFeatureStoreView(ReadOnlyStore): - """Exposes only ``get``/``all`` over a feature store, decoding dict items. + """Read-only view of a feature store. - The wrapped store is stable and never swaps, so it is read directly. Items - stored as dicts are decoded into model objects; items already decoded are - returned unchanged, then the caller's ``callback`` is applied. + Serves every read from the wrapped store. Items that a custom feature store + keeps as raw dicts are decoded into model objects; items that are already + models pass through unchanged. """ def __init__(self, store: FeatureStore): diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index d4515ede..8939312c 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -37,13 +37,12 @@ class _ReadOnlyStoreView(ReadOnlyStore): - """Exposes only ``get``/``all`` over a store, decoding dict items. + """Read-only view of the data system store. - Resolves the active store on each read rather than at construction, so a held - instance follows the active-store swap: reads hit the persistent store before - the in-memory store has data, and the in-memory store afterwards. Items stored - as dicts are decoded into model objects; items already decoded are returned - unchanged, then the caller's ``callback`` is applied. + 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. + 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: Store): diff --git a/ldclient/impl/datasystem/store.py b/ldclient/impl/datasystem/store.py index 01903153..6192178f 100644 --- a/ldclient/impl/datasystem/store.py +++ b/ldclient/impl/datasystem/store.py @@ -160,11 +160,8 @@ class _StoreBase: synchronizer), reads serve from memory and the persistent store is no longer read from. - This base holds the in-memory store, dependency tracking, listeners, the - active-store swap, and the changeset-to-memory apply. It never references a - persistent store: subclasses own their concretely-typed store and supply the - persist step through the ``_stage_persist_full``/``_stage_persist_delta`` - hooks and the ``_on_memory_store_active`` hook. + This base owns the in-memory store, dependency tracking, listeners, and the + active-store swap. It holds no persistent store of its own. """ def __init__( @@ -198,6 +195,10 @@ def __init__( # Thread synchronization self._lock = threading.RLock() + # True if the data in the memory store may be written to the persistent + # store. Set on each apply from its persist flag. + self._persist = False + def selector(self) -> Selector: """Returns the current selector.""" with self._lock: @@ -211,25 +212,13 @@ def _on_memory_store_active(self) -> None: """ pass - def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]: - """ - Persist a full data set. Subclasses supply the persist step. - - A subclass either writes the data synchronously and returns None, or - returns the collections for the caller to persist afterwards. The - subclass owns the decision of whether to persist at all. - """ - raise NotImplementedError - - def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]: + def _should_persist(self) -> bool: """ - Persist a delta update. Subclasses supply the persist step. - - A subclass either writes the data synchronously and returns None, or - returns the collections for the caller to persist afterwards. The - subclass owns the decision of whether to persist at all. + Returns whether the current data should be written to the persistent + store. The base engine has no persistent store, so it never persists; + subclasses override this based on their store. """ - raise NotImplementedError + return False def _set_basis( self, collections: Collections, selector: Selector, persist: bool @@ -243,8 +232,10 @@ def _set_basis( persist: Whether to persist the data to the persistent store Returns: - The collections the subclass deferred for later persistence, or None. + The collections to persist, or None if there is nothing to persist. """ + self._persist = persist + # Take snapshot for change detection if we have flag listeners old_data: Optional[Collections] = None if self._flag_change_listeners.has_listeners(): @@ -266,12 +257,9 @@ def _set_basis( self._active_store = self._memory_store # In-memory store is now authoritative. The subclass reacts here (e.g. by - # disabling the persistent-store cache) before the persist step below. + # disabling the persistent-store cache) before the caller persists. self._on_memory_store_active() - # Persist through the subclass hook - pending = self._stage_persist_full(collections, persist) - # Send change events if we had listeners if old_data is not None: affected_items = self._compute_changed_items_for_full_data_set( @@ -279,7 +267,7 @@ def _set_basis( ) self._send_change_events(affected_items) - return pending + return collections if self._should_persist() else None def _apply_delta( self, collections: Collections, selector: Selector, persist: bool @@ -293,8 +281,10 @@ def _apply_delta( persist: Whether to persist the changes to the persistent store Returns: - The collections the subclass deferred for later persistence, or None. + The collections to persist, or None if there is nothing to persist. """ + self._persist = persist + ok = self._memory_store.apply_delta(collections) if ok is False: return None @@ -317,13 +307,11 @@ def _apply_delta( # Update state self._selector = selector if selector is not None else Selector.no_selector() - pending = self._stage_persist_delta(collections, persist) - # Send change events if affected_items: self._send_change_events(affected_items) - return pending + return collections if self._should_persist() else None def _changes_to_store_data(self, changes: List[Change]) -> Collections: """ @@ -424,9 +412,6 @@ def __init__( self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None self._persistent_store_writable = False - # True if the data in the memory store may be persisted to the persistent store - self._persist = False - def with_persistence( self, persistent_store: FeatureStore, @@ -456,7 +441,8 @@ def with_persistence( def apply(self, change_set: ChangeSet, persist: bool) -> None: """ - Apply a changeset to the store. + Apply a changeset to the in-memory store and, if configured, the + persistent store. Args: change_set: The changeset to apply @@ -464,12 +450,16 @@ def apply(self, change_set: ChangeSet, persist: bool) -> None: """ collections = self._changes_to_store_data(change_set.changes) + pending: Optional[Collections] = None + is_full = False + with self._lock: try: if change_set.intent_code == IntentCode.TRANSFER_FULL: - self._set_basis(collections, change_set.selector, persist) + pending = self._set_basis(collections, change_set.selector, persist) + is_full = True elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: - self._apply_delta(collections, change_set.selector, persist) + pending = self._apply_delta(collections, change_set.selector, persist) elif change_set.intent_code == IntentCode.TRANSFER_NONE: # No-op, no changes to apply return @@ -477,6 +467,18 @@ def apply(self, change_set: ChangeSet, persist: bool) -> None: # Notify changeset listeners self._change_set_listeners.notify(change_set) + # Persist synchronously, inline under the lock + if pending is not None: + store = self._persistent_store + assert store is not None + if is_full: + store.init(pending) + else: + for kind in pending: + kind_data = pending[kind] + for key in kind_data: + store.upsert(kind, kind_data[key]) + except Exception as e: # Log error but don't re-raise - matches Go behavior log.error("Store: couldn't apply changeset: %s", str(e)) @@ -502,27 +504,6 @@ def _on_memory_store_active(self) -> None: except Exception as e: log.warning("Failed to disable persistent store cache: %s", e) - def _stage_persist_full(self, collections: Collections, persist: bool) -> Optional[Collections]: - self._persist = persist - if not self._should_persist(): - return None - store = self._persistent_store - assert store is not None - store.init(collections) - return None - - def _stage_persist_delta(self, collections: Collections, persist: bool) -> Optional[Collections]: - self._persist = persist - if not self._should_persist(): - return None - store = self._persistent_store - assert store is not None - for kind in collections: - kind_data = collections[kind] - for key in kind_data: - store.upsert(kind, kind_data[key]) - return None - def commit(self) -> Optional[Exception]: """ Commit persists the data in the memory store to the persistent store, if configured. From ae96f4b301d884d53dfb9294b76e908bb5e70571 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 21 Aug 2026 09:11:16 -0500 Subject: [PATCH 6/6] fix: Only persist after a successful in-memory apply Move the persist flag and the _should_persist decision out of _StoreBase into the Store and AsyncStore subclasses; the base is now a pure in-memory engine (_set_basis/_apply_delta take no persist flag and return whether the memory apply succeeded). Each subclass sets its persist flag only after a successful apply, so a failed decode or apply can no longer flip it and let a later commit treat unapplied or empty memory as writable and overwrite the persistent store. --- ldclient/impl/datasystem/async_store.py | 22 +++++--- ldclient/impl/datasystem/store.py | 74 +++++++++++-------------- 2 files changed, 47 insertions(+), 49 deletions(-) diff --git a/ldclient/impl/datasystem/async_store.py b/ldclient/impl/datasystem/async_store.py index 9c8c0142..ce2e7742 100644 --- a/ldclient/impl/datasystem/async_store.py +++ b/ldclient/impl/datasystem/async_store.py @@ -43,6 +43,10 @@ def __init__( self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None self._persistent_store_writable = False + # True if the data in the memory store may be written to the persistent + # store. Set on each successful apply from its persist flag. + self._persist = False + # Serializes async store writes; held only across the awaited I/O, never with self._lock. self._async_persist_lock = AsyncLock() @@ -105,25 +109,29 @@ async def apply(self, change_set: ChangeSet, persist: bool) -> None: """ collections = self._changes_to_store_data(change_set.changes) - pending: Optional[Collections] = None + applied = False is_full = False with self._lock: try: if change_set.intent_code == IntentCode.TRANSFER_FULL: - pending = self._set_basis(collections, change_set.selector, persist) + applied = self._set_basis(collections, change_set.selector) is_full = True elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: - pending = self._apply_delta(collections, change_set.selector, persist) + applied = self._apply_delta(collections, change_set.selector) elif change_set.intent_code == IntentCode.TRANSFER_NONE: return self._change_set_listeners.notify(change_set) + + if applied: + # Memory now holds this data, so it may be persisted. + self._persist = persist except Exception as e: log.error("Store: couldn't apply changeset: %s", str(e)) return - if pending is None: + if not applied or not self._should_persist(): return store = self._persistent_store @@ -133,10 +141,10 @@ async def apply(self, change_set: ChangeSet, persist: bool) -> None: async with self._async_persist_lock: try: if is_full: - await store.init(pending) + await store.init(collections) else: - for kind in pending: - kind_data = pending[kind] + for kind in collections: + kind_data = collections[kind] for key in kind_data: await store.upsert(kind, kind_data[key]) except Exception as e: diff --git a/ldclient/impl/datasystem/store.py b/ldclient/impl/datasystem/store.py index 6192178f..5cb78f79 100644 --- a/ldclient/impl/datasystem/store.py +++ b/ldclient/impl/datasystem/store.py @@ -195,10 +195,6 @@ def __init__( # Thread synchronization self._lock = threading.RLock() - # True if the data in the memory store may be written to the persistent - # store. Set on each apply from its persist flag. - self._persist = False - def selector(self) -> Selector: """Returns the current selector.""" with self._lock: @@ -212,30 +208,19 @@ def _on_memory_store_active(self) -> None: """ pass - def _should_persist(self) -> bool: - """ - Returns whether the current data should be written to the persistent - store. The base engine has no persistent store, so it never persists; - subclasses override this based on their store. - """ - return False - def _set_basis( - self, collections: Collections, selector: Selector, persist: bool - ) -> Optional[Collections]: + self, collections: Collections, selector: Selector + ) -> bool: """ Set the basis of the store. Any existing data is discarded. Args: collections: The new basis data selector: The selector identifying the data - persist: Whether to persist the data to the persistent store Returns: - The collections to persist, or None if there is nothing to persist. + True if the in-memory store was updated, False if it was not. """ - self._persist = persist - # Take snapshot for change detection if we have flag listeners old_data: Optional[Collections] = None if self._flag_change_listeners.has_listeners(): @@ -245,7 +230,7 @@ def _set_basis( ok = self._memory_store.set_basis(collections) if ok is False: - return None + return False # Update dependency tracker self._reset_dependency_tracker(collections) @@ -267,27 +252,24 @@ def _set_basis( ) self._send_change_events(affected_items) - return collections if self._should_persist() else None + return True def _apply_delta( - self, collections: Collections, selector: Selector, persist: bool - ) -> Optional[Collections]: + self, collections: Collections, selector: Selector + ) -> bool: """ Apply a delta update to the store. Args: collections: The delta changes selector: The selector identifying the data - persist: Whether to persist the changes to the persistent store Returns: - The collections to persist, or None if there is nothing to persist. + True if the in-memory store was updated, False if it was not. """ - self._persist = persist - ok = self._memory_store.apply_delta(collections) if ok is False: - return None + return False has_listeners = self._flag_change_listeners.has_listeners() affected_items: Set[KindAndKey] = set() @@ -311,7 +293,7 @@ def _apply_delta( if affected_items: self._send_change_events(affected_items) - return collections if self._should_persist() else None + return True def _changes_to_store_data(self, changes: List[Change]) -> Collections: """ @@ -412,6 +394,10 @@ def __init__( self._persistent_store_status_provider: Optional[DataStoreStatusProvider] = None self._persistent_store_writable = False + # True if the data in the memory store may be written to the persistent + # store. Set on each successful apply from its persist flag. + self._persist = False + def with_persistence( self, persistent_store: FeatureStore, @@ -450,16 +436,16 @@ def apply(self, change_set: ChangeSet, persist: bool) -> None: """ collections = self._changes_to_store_data(change_set.changes) - pending: Optional[Collections] = None + applied = False is_full = False with self._lock: try: if change_set.intent_code == IntentCode.TRANSFER_FULL: - pending = self._set_basis(collections, change_set.selector, persist) + applied = self._set_basis(collections, change_set.selector) is_full = True elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: - pending = self._apply_delta(collections, change_set.selector, persist) + applied = self._apply_delta(collections, change_set.selector) elif change_set.intent_code == IntentCode.TRANSFER_NONE: # No-op, no changes to apply return @@ -467,17 +453,21 @@ def apply(self, change_set: ChangeSet, persist: bool) -> None: # Notify changeset listeners self._change_set_listeners.notify(change_set) - # Persist synchronously, inline under the lock - if pending is not None: - store = self._persistent_store - assert store is not None - if is_full: - store.init(pending) - else: - for kind in pending: - kind_data = pending[kind] - for key in kind_data: - store.upsert(kind, kind_data[key]) + if applied: + # Memory now holds this data, so it may be persisted. + self._persist = persist + + # Persist synchronously, inline under the lock + if self._should_persist(): + store = self._persistent_store + assert store is not None + if is_full: + store.init(collections) + else: + for kind in collections: + kind_data = collections[kind] + for key in kind_data: + store.upsert(kind, kind_data[key]) except Exception as e: # Log error but don't re-raise - matches Go behavior