feat: Add async FDv2 data system (#486)

## Overview

Part of the async Python SDK work (epic SDK-60). Adds the async FDv2
**data system** (coordinator) and wires it into the async client.
Targets `main` (its predecessor, #485 async FDv2 data sources, has
merged).

This is experimental and should not be considered production-ready.

## What this PR adds

- `impl/datasystem/async_fdv2.py` — `AsyncFDv2`, the async data system
that coordinates the async initializers and synchronizers, mirrors the
sync `FDv2` fallback/recovery behavior, and exposes the async data
source status and flag tracking. Includes
`AsyncFeatureStoreClientWrapper` for persistent-store availability
polling.
- `async_client.py` wiring: `_make_data_system` builds `AsyncFDv2`;
`_wire_data_source_sessions` shares the client's aiohttp session into
the async data source builders so they reuse the connection pool.

## Async readiness gate (awaitable end-to-end)

Readiness and availability are now awaitable so a persistent store
populated by another process (warm start / daemon) is recognized:

- `AsyncDataSystem.data_availability` and
`AsyncLDClient.is_initialized()` are coroutines; the eval path awaits
`data_availability()` directly.
- `AsyncFeatureStore` gains an abstract async `is_initialized()` (with
`$inited` caching + monotonic latch in `AsyncCachingStoreWrapper`,
surfaced via `AsyncStore.is_ready()`). The `getattr` duck-type is
dropped, so a custom store that omits it fails at construction rather
than silently serving defaults.
- The gate catches a persistent-store error and degrades to `DEFAULTS`
rather than raising, so `variation()` never throws on a store outage.
(Sync counterpart: #506.)

## Wrapper hardening

- `is_monitoring_enabled()` delegates to the store's own opt-in
(matching sync), so a store that cannot report availability is not
polled and left stuck unavailable.
- `close()` is idempotent, bounds the availability-poller stop with a
timeout, and logs errors from the inner store close.
- `AsyncStore.close()` logs and swallows close errors instead of
returning them.

## Shared refactor

- `impl/datasystem/fdv2_common.py` gains module-level
`fallback_condition` / `recovery_condition`, shared by the sync and
async data systems.

## Testing

- `LD_SKIP_DATABASE_TESTS=1 uv run pytest
ldclient/testing/impl/datasystem/` and
`ldclient/testing/test_async_client.py` — green.
- `make lint` (mypy, isort, pycodestyle) — clean.

Tracked internally: SDK-2870

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Overview**
> Adds **async Flag Delivery v2** support by introducing `AsyncFDv2`,
wiring it into `AsyncLDClient` when `datasystem_config` is set
(replacing the previous `NotImplementedError`), and sharing the client's
aiohttp session with async polling/streaming data source builders via
`_wire_data_source_sessions`.
> 
> **Readiness and availability are now async** so persistent stores
populated by another process can be detected:
`AsyncLDClient.is_initialized()`, `AsyncDataSystem.data_availability()`,
and a new `AsyncFeatureStore.is_initialized()` (with caching/latching in
`AsyncCachingStoreWrapper` and `AsyncStore.is_ready()`). Contract-test
harnesses await these checks accordingly.
> 
> Sync **FDv2** is refactored to share wiring through `_FDv2Base` and
module-level `fallback_condition` / `recovery_condition` in
`fdv2_common.py`. `AsyncFDv2` mirrors initializer/synchronizer
coordination, FDv1 fallback, persistent-store outage recovery, and adds
`AsyncFeatureStoreClientWrapper` for availability polling.
> 
> `AsyncStore.close()` logs and swallows close errors instead of
returning them. Large new test suites cover `AsyncFDv2`, async
persistence, and readiness gating.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
111bb37. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: jsonbailey

contract-tests/async_client_entity.py

@@ -104,8 +104,8 @@ async def start(self):
         await self._client.start(start_wait / 1000.0)
         self._listeners = AsyncListenerRegistry(self._client.flag_tracker)
 
-    def is_initializing(self) -> bool:
-        return self._client.is_initialized() if self._client else False
+    async def is_initializing(self) -> bool:
+        return await self._client.is_initialized() if self._client else False
 
     async def evaluate(self, params: dict) -> dict:
         response = {}

contract-tests/async_service.py

@@ -102,7 +102,7 @@ async def handle_create_client(request: aiohttp.web.Request) -> aiohttp.web.Resp
         await client.close()
         return aiohttp.web.Response(text=str(e), status=500)
 
-    if not client.is_initializing() and not options['configuration'].get('initCanFail', False):
+    if not await client.is_initializing() and not options['configuration'].get('initCanFail', False):
         await client.close()
         return aiohttp.web.Response(text='Failed to initialize', status=500)
 

ldclient/async_client.py

@@ -197,6 +197,12 @@ async def __start_up(self, start_wait: float):
         # Start the big-segment status poll now that a loop is running.
         self.__big_segment_store_manager.start()
 
+        # FDv2 builds its data sources from builders; wire the shared session into
+        # them before starting (FDv1 pulls the session itself via its provider).
+        datasystem_config = self._config.datasystem_config
+        if datasystem_config is not None and not self._config.offline:
+            self._wire_data_source_sessions(datasystem_config)
+
         if self._config.offline:
             log.info("Started LaunchDarkly Client in offline mode")
 
@@ -221,7 +227,7 @@ async def __start_up(self, start_wait: float):
                 log.info("Waiting up to " + str(start_wait) + " seconds for LaunchDarkly client to initialize...")
                 await update_processor_ready.wait(start_wait)
 
-        if self.is_initialized() is True:
+        if await self.is_initialized() is True:
             log.info("Started LaunchDarkly Client: OK")
         else:
             log.warning("Initialization timeout exceeded for LaunchDarkly Client or an error occurred. " "Feature Flags may not yet be available.")
@@ -243,7 +249,9 @@ def _make_data_system(self) -> AsyncDataSystem:
 
             return AsyncFDv1(self._config, self._select_feature_store(), self._get_session)
 
-        raise NotImplementedError("FDv2 is not yet supported in the async client")
+        from ldclient.impl.datasystem.async_fdv2 import AsyncFDv2
+
+        return AsyncFDv2(self._config, datasystem_config)
 
     def _select_feature_store(self) -> AsyncFeatureStore:
         """Choose the async feature store for the v1 data system based on the
@@ -253,6 +261,34 @@ def _select_feature_store(self) -> AsyncFeatureStore:
             return AsyncInMemoryFeatureStore()
         return feature_store
 
+    def _wire_data_source_sessions(self, data_system_config) -> None:
+        """Provide the client's aiohttp session to any async data source
+        builders so the sources they build share the client's connection pool."""
+        from ldclient.impl.datasourcev2.async_polling import (
+            AsyncFallbackToFDv1PollingDataSourceBuilder,
+            AsyncPollingDataSourceBuilder
+        )
+        from ldclient.impl.datasourcev2.async_streaming import (
+            AsyncStreamingDataSourceBuilder
+        )
+
+        builders = list(data_system_config.initializers or []) + list(
+            data_system_config.synchronizers or []
+        )
+        if data_system_config.fdv1_fallback_synchronizer is not None:
+            builders.append(data_system_config.fdv1_fallback_synchronizer)
+
+        for builder in builders:
+            if isinstance(
+                builder,
+                (
+                    AsyncFallbackToFDv1PollingDataSourceBuilder,
+                    AsyncPollingDataSourceBuilder,
+                    AsyncStreamingDataSourceBuilder,
+                ),
+            ):
+                builder.session(self._get_session())
+
     async def __register_plugins(self, environment_metadata: EnvironmentMetadata):
         for plugin in self._config.plugins:
             try:
@@ -343,18 +379,20 @@ def is_offline(self) -> bool:
         """Returns true if the client is in offline mode."""
         return self._config.offline
 
-    def is_initialized(self) -> bool:
+    async def is_initialized(self) -> bool:
         """Returns true if the client has successfully connected to LaunchDarkly.
 
         If this returns false, it means that the client has not yet successfully connected to LaunchDarkly.
         It might still be in the process of starting up, or it might be attempting to reconnect after an
         unsuccessful attempt, or it might have received an unrecoverable error (such as an invalid SDK key)
         and given up.
+
+        This is a coroutine because determining readiness may query a persistent store.
         """
         if self.is_offline() or self._config.use_ldd:
             return True
 
-        return self._data_system.data_availability.at_least(DataAvailability.CACHED)
+        return (await self._data_system.data_availability()).at_least(DataAvailability.CACHED)
 
     async def flush(self):
         """Flushes all pending analytics events.
@@ -455,8 +493,9 @@ async def _evaluate_internal(self, key: str, context: Context, default: Any, eve
         if self._config.offline:
             return EvaluationDetail(default, None, error_reason('CLIENT_NOT_READY')), None
 
-        if self._data_system.data_availability != DataAvailability.REFRESHED:
-            if self._data_system.data_availability == DataAvailability.CACHED:
+        availability = await self._data_system.data_availability()
+        if availability != DataAvailability.REFRESHED:
+            if availability == DataAvailability.CACHED:
                 log.warning("Feature Flag evaluation attempted before client has initialized - using last known values from feature store for feature key: " + key)
             else:
                 log.warning("Feature Flag evaluation attempted before client has initialized! Feature store unavailable - returning default: " + str(default) + " for feature key: " + key)
@@ -526,8 +565,9 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState
             log.warning("all_flags_state() called, but client is in offline mode. Returning empty state")
             return FeatureFlagsState(False)
 
-        if self._data_system.data_availability != DataAvailability.REFRESHED:
-            if self._data_system.data_availability == DataAvailability.CACHED:
+        availability = await self._data_system.data_availability()
+        if availability != DataAvailability.REFRESHED:
+            if availability == DataAvailability.CACHED:
                 log.warning("all_flags_state() called before client has finished initializing! Using last known values from feature store")
             else:
                 log.warning("all_flags_state() called before client has finished initializing! Feature store unavailable - returning empty state")

ldclient/async_feature_store.py

@@ -88,6 +88,10 @@ def initialized(self) -> bool:
         """ """
         return self._initialized
 
+    async def is_initialized(self) -> bool:
+        """ """
+        return self._initialized
+
     async def close(self) -> None:
         """ """
         pass

ldclient/async_feature_store_helpers.py

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

ldclient/impl/datasystem/__init__.py

@@ -208,11 +208,11 @@ def flag_change_listeners(self) -> Listeners:
         """
         raise NotImplementedError
 
-    @property
     @abstractmethod
-    def data_availability(self) -> DataAvailability:
+    async def data_availability(self) -> DataAvailability:
         """
-        Indicates what form of data is currently available.
+        Indicates what form of data is currently available, awaiting the store's
+        readiness so a persistent store populated by another process is recognized.
         """
         raise NotImplementedError
 

ldclient/impl/datasystem/async_fdv1.py

@@ -143,18 +143,22 @@ def data_store_status_provider(self) -> DataStoreStatusProvider:
     def flag_change_listeners(self) -> Listeners:
         return self._flag_change_listeners
 
-    @property
-    def data_availability(self) -> DataAvailability:
+    async def data_availability(self) -> DataAvailability:
         if self._config.offline:
             return DataAvailability.DEFAULTS
 
         if self._update_processor is not None and self._update_processor.initialized():
             return DataAvailability.REFRESHED
 
-        if self._store.initialized:
-            return DataAvailability.CACHED
+        # Awaits the store so a persistent store populated by another process is
+        # recognized. A persistent-store error is logged and reported as no data.
+        try:
+            ready = await self._store.is_initialized()
+        except Exception as e:
+            log.warning("Error checking persistent store readiness: %s", e)
+            return DataAvailability.DEFAULTS
 
-        return DataAvailability.DEFAULTS
+        return DataAvailability.CACHED if ready else DataAvailability.DEFAULTS
 
     @property
     def target_availability(self) -> DataAvailability: