feat: Add read-only store views and async persistence foundation for the data system

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

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())

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.
     """

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.
     """

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):

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):
         """