feat: Add the async FDv2 data system

Author: jsonbailey

ldclient/async_client.py

@@ -197,6 +197,12 @@ async def __start_up(self, start_wait: float):
         # Start the big-segment status poll now that a loop is running.
         self.__big_segment_store_manager.start()
 
+        # FDv2 builds its data sources from builders; wire the shared session into
+        # them before starting (FDv1 pulls the session itself via its provider).
+        datasystem_config = self._config.datasystem_config
+        if datasystem_config is not None and not self._config.offline:
+            self._wire_data_source_sessions(datasystem_config)
+
         if self._config.offline:
             log.info("Started LaunchDarkly Client in offline mode")
 
@@ -221,6 +227,10 @@ async def __start_up(self, start_wait: float):
                 log.info("Waiting up to " + str(start_wait) + " seconds for LaunchDarkly client to initialize...")
                 await update_processor_ready.wait(start_wait)
 
+        # Warm the persistent store's initialized state so a store populated by
+        # another process is recognized before the readiness check.
+        await self._data_system.refresh_availability()
+
         if self.is_initialized() is True:
             log.info("Started LaunchDarkly Client: OK")
         else:
@@ -243,7 +253,9 @@ def _make_data_system(self) -> AsyncDataSystem:
 
             return AsyncFDv1(self._config, self._select_feature_store(), self._get_session)
 
-        raise NotImplementedError("FDv2 is not yet supported in the async client")
+        from ldclient.impl.datasystem.async_fdv2 import AsyncFDv2
+
+        return AsyncFDv2(self._config, datasystem_config)
 
     def _select_feature_store(self) -> AsyncFeatureStore:
         """Choose the async feature store for the v1 data system based on the
@@ -253,6 +265,34 @@ def _select_feature_store(self) -> AsyncFeatureStore:
             return AsyncInMemoryFeatureStore()
         return feature_store
 
+    def _wire_data_source_sessions(self, data_system_config) -> None:
+        """Provide the client's aiohttp session to any async data source
+        builders so the sources they build share the client's connection pool."""
+        from ldclient.impl.datasourcev2.async_polling import (
+            AsyncFallbackToFDv1PollingDataSourceBuilder,
+            AsyncPollingDataSourceBuilder
+        )
+        from ldclient.impl.datasourcev2.async_streaming import (
+            AsyncStreamingDataSourceBuilder
+        )
+
+        builders = list(data_system_config.initializers or []) + list(
+            data_system_config.synchronizers or []
+        )
+        if data_system_config.fdv1_fallback_synchronizer is not None:
+            builders.append(data_system_config.fdv1_fallback_synchronizer)
+
+        for builder in builders:
+            if isinstance(
+                builder,
+                (
+                    AsyncFallbackToFDv1PollingDataSourceBuilder,
+                    AsyncPollingDataSourceBuilder,
+                    AsyncStreamingDataSourceBuilder,
+                ),
+            ):
+                builder.session(self._get_session())
+
     async def __register_plugins(self, environment_metadata: EnvironmentMetadata):
         for plugin in self._config.plugins:
             try:
@@ -455,6 +495,11 @@ async def _evaluate_internal(self, key: str, context: Context, default: Any, eve
         if self._config.offline:
             return EvaluationDetail(default, None, error_reason('CLIENT_NOT_READY')), None
 
+        # Refresh the store's initialized state only while still uninitialized;
+        # once initialized or a basis arrives, the gate reads a cached value.
+        if self._data_system.data_availability == DataAvailability.DEFAULTS:
+            await self._data_system.refresh_availability()
+
         if self._data_system.data_availability != DataAvailability.REFRESHED:
             if self._data_system.data_availability == DataAvailability.CACHED:
                 log.warning("Feature Flag evaluation attempted before client has initialized - using last known values from feature store for feature key: " + key)
@@ -526,6 +571,11 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState
             log.warning("all_flags_state() called, but client is in offline mode. Returning empty state")
             return FeatureFlagsState(False)
 
+        # Refresh the store's initialized state only while still uninitialized;
+        # once initialized or a basis arrives, the gate reads a cached value.
+        if self._data_system.data_availability == DataAvailability.DEFAULTS:
+            await self._data_system.refresh_availability()
+
         if self._data_system.data_availability != DataAvailability.REFRESHED:
             if self._data_system.data_availability == DataAvailability.CACHED:
                 log.warning("all_flags_state() called before client has finished initializing! Using last known values from feature store")

ldclient/async_feature_store_helpers.py

@@ -37,6 +37,8 @@ class AsyncCachingStoreWrapper(_CachingStoreWrapperBase, DiagnosticDescription,
     event loop because its reads and writes never suspend between one another.
     """
 
+    __INITED_CACHE_KEY__ = "$inited"
+
     _core: AsyncFeatureStoreCore
 
     def __init__(self, core: AsyncFeatureStoreCore, cache_config: CacheConfig):
@@ -95,12 +97,25 @@ async def upsert(self, kind: VersionedDataKind, item: dict) -> bool:
 
     @property
     def initialized(self) -> bool:
-        """Returns whether ``init`` has completed in this process.
+        """Returns the store's last observed initialized state."""
+        return self._inited
 
-        This property does not query the store: it is synchronous, but a persistent-store query is
-        a coroutine, so it reflects only whether this process has initialized the store.
+    async def refresh_initialized(self) -> None:
+        """Refreshes :attr:`initialized` from the store's own initialized state.
+
+        Honors the cache: with caching off the store is queried on every call;
+        with a TTL it is queried once per interval; with an infinite TTL it is
+        queried once. Once the store reports initialized the state latches and
+        later calls return without I/O.
         """
-        return self._inited
+        if self._inited:
+            return
+        result = self._cache.get(AsyncCachingStoreWrapper.__INITED_CACHE_KEY__)
+        if result is None:
+            result = bool(await self._core.initialized_internal())
+            self._cache[AsyncCachingStoreWrapper.__INITED_CACHE_KEY__] = result
+        if result:
+            self._inited = True
 
     async def close(self) -> None:
         """Releases the cache and closes the underlying core if it supports it."""

ldclient/impl/datasystem/__init__.py

@@ -232,6 +232,18 @@ def store(self) -> AsyncReadOnlyStore:
         """
         raise NotImplementedError
 
+    @abstractmethod
+    async def refresh_availability(self) -> None:
+        """
+        Refreshes any awaited state that :attr:`data_availability` depends on.
+
+        The client awaits this before reading :attr:`data_availability` so that a
+        persistent store initialized by another process is reflected while a data
+        source is configured but has not yet supplied a basis. Data systems with
+        no such state treat this as a no-op.
+        """
+        raise NotImplementedError
+
 
 class DiagnosticAccumulator(Protocol):
     def record_stream_init(self, timestamp, duration, failed):

ldclient/impl/datasystem/async_fdv1.py

@@ -124,6 +124,10 @@ async def stop(self):
     def store(self) -> AsyncReadOnlyStore:
         return self._store_view
 
+    async def refresh_availability(self) -> None:
+        """No-op; ``data_availability`` reads the store's initialized state directly."""
+        return None
+
     def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator):
         """
         Sets the diagnostic accumulator for streaming initialization metrics.