refactor: Make the async readiness gate awaitable end-to-end

Addresses review feedback on the interim warm-start refresh. The async
readiness path is now awaitable at every layer instead of a private
two-step refresh feeding a synchronous gate:

- AsyncDataSystem.data_availability and AsyncLDClient.is_initialized()
  become coroutines; the eval path awaits data_availability() directly.
- AsyncFeatureStore gains an abstract async is_initialized(); the
  getattr duck-type is dropped, so a custom store that omits it fails
  at construction instead of silently serving DEFAULTS forever.
- The warm-start store error is caught inside data_availability and
  degrades to DEFAULTS, so evaluation never propagates a store error.
- Removes the interim refresh_availability, cold-path gating, and
  start-time refresh, which are subsumed by the awaitable gate.
- Adds a per-iteration stop-event check in the synchronizer loop so a
  perpetually-ready queue cannot starve the stop signal.

The sync data system is unchanged.

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

@@ -227,11 +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)
 
-        # 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:
+        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.")
@@ -383,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.
@@ -495,13 +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
 
-        # 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:
+        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)
@@ -571,13 +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)
 
-        # 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:
+        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

@@ -100,22 +100,23 @@ def initialized(self) -> bool:
         """Returns the store's last observed initialized state."""
         return self._inited
 
-    async def refresh_initialized(self) -> None:
-        """Refreshes :attr:`initialized` from the store's own initialized state.
+    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.
         """
         if self._inited:
-            return
+            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
 
@@ -232,18 +232,6 @@ 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,10 +124,6 @@ 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.
@@ -147,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:

ldclient/impl/datasystem/async_fdv2.py

@@ -21,7 +21,11 @@
     join_handle,
     spawn_handle
 )
-from ldclient.impl.datasystem import AsyncDataSystem, DiagnosticSource
+from ldclient.impl.datasystem import (
+    AsyncDataSystem,
+    DataAvailability,
+    DiagnosticSource
+)
 from ldclient.impl.datasystem.async_store import AsyncStore
 from ldclient.impl.datasystem.fdv2_common import (
     ConditionDirective,
@@ -92,16 +96,13 @@ async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool:
     def initialized(self) -> bool:
         return self._store.initialized
 
-    async def refresh_initialized(self) -> None:
-        """Refreshes the inner store's initialized state, if it supports it.
+    async def is_initialized(self) -> bool:
+        """Queries the inner store's initialized state.
 
         Runs through the availability wrapper so a failed query marks the store
         unavailable like any other operation.
         """
-        refresh = getattr(self._store, "refresh_initialized", None)
-        if refresh is None:
-            return
-        await self._wrap(refresh)
+        return await self._wrap(lambda: self._store.is_initialized())
 
     def disable_cache(self) -> None:
         """Disables the inner store's cache if it supports it."""
@@ -547,6 +548,10 @@ async def reader():
             sync_reader = spawn_handle("AsyncFDv2-sync-reader", reader)
 
             while True:
+                # Honor a stop request every iteration so a queue that always has
+                # an item ready cannot starve the check.
+                if self._stop_event.is_set():
+                    return ConditionDirective.FALLBACK
                 update = await action_queue.get()
                 if isinstance(update, str):
                     if update == "quit":
@@ -608,16 +613,21 @@ def store(self) -> AsyncReadOnlyStore:
         """Get the underlying store for flag evaluation."""
         return self._store_view
 
-    async def refresh_availability(self) -> None:
-        """Refreshes the persistent store's initialized state so a store populated
-        by another process satisfies the availability gate before a synchronizer
-        supplies a basis. A store error is logged and swallowed so the gate
-        degrades to DEFAULTS and evaluation returns the default rather than
-        propagating the error."""
+    async def data_availability(self) -> DataAvailability:  # type: ignore[override]
+        """Reports what form of data is currently available, awaiting the store's
+        readiness so a persistent store populated by another process is recognized
+        before a synchronizer supplies a basis. A persistent-store error is treated
+        as no data: it is logged and reported as ``DEFAULTS`` rather than raised."""
+        if self._store.selector().is_defined():
+            return DataAvailability.REFRESHED
+        if not self._configured_with_data_sources:
+            return DataAvailability.CACHED
         try:
-            await self._store.refresh_persistent_initialized()
+            ready = await self._store.is_ready()
         except Exception as e:
-            log.warning("Failed to refresh persistent store initialized state: %s", e)
+            log.warning("Error checking persistent store readiness: %s", e)
+            return DataAvailability.DEFAULTS
+        return DataAvailability.CACHED if ready else DataAvailability.DEFAULTS
 
 
 __all__ = [

ldclient/impl/datasystem/async_store.py

@@ -206,21 +206,18 @@ def get_data_store_status_provider(self) -> Optional[DataStoreStatusProvider]:
         with self._lock:
             return self._persistent_store_status_provider
 
-    async def refresh_persistent_initialized(self) -> None:
-        """Refreshes the persistent store's initialized state while it is active.
+    async def is_ready(self) -> bool:
+        """Reports whether the active store holds usable data.
 
-        Once the in-memory store is active, the persistent store is no longer read
-        from, so its initialized state no longer gates reads and no query is made.
+        Once the in-memory store is active its readiness is authoritative and no
+        query is made. While the persistent store is active, its readiness is
+        queried (awaiting the store), so a store populated by another process is
+        recognized.
         """
         store = self._persistent_store
-        if store is None:
-            return
-        if self._active_store is self._memory_store:
-            return
-        refresh = getattr(store, "refresh_initialized", None)
-        if refresh is None:
-            return
-        await refresh()
+        if store is None or self._active_store is self._memory_store:
+            return self._active_store.initialized
+        return await store.is_initialized()
 
 
 __all__ = ["AsyncStore"]

ldclient/interfaces.py

@@ -408,7 +408,16 @@ async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool:
     @abstractmethod
     def initialized(self) -> bool:
         """
-        Returns whether the store has been initialized yet or not.
+        Returns the store's last observed initialized state without querying it.
+        """
+
+    @abstractmethod
+    async def is_initialized(self) -> bool:
+        """
+        Queries whether the store has been initialized, awaiting the store if a query is required.
+
+        A persistent store may have been populated by another process, so this can require I/O.
+        Implementations should latch a positive result: once the store is initialized it stays so.
         """
 
     async def close(self) -> None: