diff --git a/ldclient/async_config.py b/ldclient/async_config.py index 2af0a1fb..189b2a45 100644 --- a/ldclient/async_config.py +++ b/ldclient/async_config.py @@ -8,6 +8,7 @@ compatibility guarantees. """ +from dataclasses import dataclass from typing import Callable, List, Optional, Set from ldclient.async_feature_store import AsyncInMemoryFeatureStore @@ -17,8 +18,8 @@ DEFAULT_STREAM_URI, GET_LATEST_FEATURES_PATH, STREAM_FLAGS_PATH, + DataSourceBuilder, DataSourceBuilderConfig, - DataSystemConfig, HTTPConfig, PrivateAttributesConfig ) @@ -34,7 +35,10 @@ AsyncDataSourceUpdateSink, AsyncEventProcessor, AsyncFeatureStore, - AsyncUpdateProcessor + AsyncInitializer, + AsyncSynchronizer, + AsyncUpdateProcessor, + DataStoreMode ) from ldclient.plugin import AsyncPlugin @@ -91,6 +95,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 +175,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 +503,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..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,143 +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() - - -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 +136,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 +422,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/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/async_fdv1.py b/ldclient/impl/datasystem/async_fdv1.py index 8fc84f92..958facc5 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): + """Read-only view of an async feature store. + + 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): + 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..ce2e7742 --- /dev/null +++ b/ldclient/impl/datasystem/async_store.py @@ -0,0 +1,210 @@ +""" +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 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() + + 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) + + async def apply(self, change_set: ChangeSet, persist: bool) -> None: + """ + Apply a changeset to the in-memory store and, if configured, the async + persistent 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) + + applied = False + is_full = False + + with self._lock: + try: + if change_set.intent_code == IntentCode.TRANSFER_FULL: + applied = self._set_basis(collections, change_set.selector) + is_full = True + elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: + 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 not applied or not self._should_persist(): + return + + store = self._persistent_store + if store is None: + return + + async with self._async_persist_lock: + try: + if is_full: + await store.init(collections) + else: + for kind in collections: + kind_data = collections[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 contents of the memory store to the async persistent store, + if configured. + + 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 + + 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: + return e + return None + + async def close(self) -> Optional[Exception]: + """ + Close the store and the async persistent store, if configured. + + Returns: + Exception if closing failed, None otherwise + """ + 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..38655415 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -1,7 +1,8 @@ from threading import Event -from typing import 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 ( @@ -18,17 +19,155 @@ DataSystem, DiagnosticAccumulator ) +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): + """Read-only view of a feature store. + + 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): + 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): @@ -45,12 +184,10 @@ 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 ) + self._store_view = _ReadOnlyFeatureStoreView(self._store_wrapper) self._data_store_status_provider_impl = DataStoreStatusProviderImpl( self._store_wrapper, self._data_store_update_sink ) @@ -94,7 +231,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..8939312c 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 ( @@ -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 @@ -33,6 +33,32 @@ ReadOnlyStore, Synchronizer ) +from ldclient.versioned_data_kind import VersionedDataKind + + +class _ReadOnlyStoreView(ReadOnlyStore): + """Read-only view of the data system store. + + Serves every read from the store's active store, so a held instance follows + the swap from the persistent store to the in-memory store once it has data. + 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): + 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): @@ -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..5cb78f79 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,17 @@ 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. + Shared, store-agnostic engine for the FDv2 store. - 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. + 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 owns the in-memory store, dependency tracking, listeners, and the + active-store swap. It holds no persistent store of its own. """ def __init__( @@ -163,16 +170,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 +186,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 +195,31 @@ 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. - - Args: - change_set: The changeset to apply - persist: Whether the changes should be persisted to the persistent 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. """ - 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)) + pass def _set_basis( - self, collections: Collections, selector: Selector, persist: bool - ) -> None: + self, collections: Collections, selector: Selector + ) -> bool: """ Set the basis of the store. Any existing data is discarded. Args: - change_set: The changeset containing the new basis data - persist: Whether to persist the data to the persistent store + collections: The new basis data + selector: The selector identifying the data + + Returns: + True if the in-memory store was updated, False if it was not. """ # Take snapshot for change detection if we have flag listeners old_data: Optional[Collections] = None @@ -286,33 +230,20 @@ def _set_basis( ok = self._memory_store.set_basis(collections) if ok is False: - return + return False # 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) - - # Persist to persistent store if configured and writable - if self._should_persist(): - self._persistent_store.init(collections) # type: ignore + # In-memory store is now authoritative. The subclass reacts here (e.g. by + # disabling the persistent-store cache) before the caller persists. + self._on_memory_store_active() # Send change events if we had listeners if old_data is not None: @@ -321,19 +252,24 @@ def _set_basis( ) self._send_change_events(affected_items) + return True + def _apply_delta( - self, collections: Collections, selector: Selector, persist: bool - ) -> None: + self, collections: Collections, selector: Selector + ) -> bool: """ Apply a delta update to the store. Args: - change_set: The changeset containing the delta changes - persist: Whether to persist the changes to the persistent store + collections: The delta changes + selector: The selector identifying the data + + Returns: + True if the in-memory store was updated, False if it was not. """ ok = self._memory_store.apply_delta(collections) if ok is False: - return + return False has_listeners = self._flag_change_listeners.has_listeners() affected_items: Set[KindAndKey] = set() @@ -351,27 +287,13 @@ 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 - # 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 True def _changes_to_store_data(self, changes: List[Change]) -> Collections: """ @@ -441,6 +363,137 @@ 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 written to the persistent + # store. Set on each successful apply from its persist flag. + 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 in-memory store and, if configured, the + persistent 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) + + applied = False + is_full = False + + with self._lock: + try: + if change_set.intent_code == IntentCode.TRANSFER_FULL: + applied = self._set_basis(collections, change_set.selector) + is_full = True + elif change_set.intent_code == IntentCode.TRANSFER_CHANGES: + applied = self._apply_delta(collections, change_set.selector) + 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) + + 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 + 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 commit(self) -> Optional[Exception]: """ Commit persists the data in the memory store to the persistent store, if configured. @@ -456,24 +509,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/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..7879b263 --- /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_full_transfer_persists_via_init(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + 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 + 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_delta_persists_via_upsert(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + await store.apply(_full_changeset("flag-a", 1, True), True) + async_store.init_called_count = 0 + async_store.upsert_calls = [] + + 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_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(_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 == [] + # In-memory store still updated + assert store._memory_store.get(FEATURES, "flag-a") is not None + + +@pytest.mark.asyncio +async def test_apply_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(cs, True) + + assert received == [cs] + + +@pytest.mark.asyncio +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(_full_changeset("flag-a", 1, True), True) + async_store.init_called_count = 0 + + 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_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(_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() + assert isinstance(err, RuntimeError) + assert str(err) == "boom" + + +@pytest.mark.asyncio +async def test_close_closes_async_store(): + async_store = FakeAsyncFeatureStore() + store = AsyncStore(Listeners(), Listeners()) + store.with_async_persistence(async_store, True, None) + + err = await store.close() + assert err is None + 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_feature_store_client_wrapper.py b/ldclient/testing/test_feature_store_client_wrapper.py index 9f5e936c..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.client import _FeatureStoreClientWrapper from ldclient.impl.datastore.status import DataStoreUpdateSinkImpl +from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper from ldclient.impl.listeners import Listeners 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}})