From 879a6729f732f364077108abd551c1c62d14f65e Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 28 Aug 2026 11:26:45 -0500 Subject: [PATCH 1/3] fix: Make the async big-segment store status query awaitable The async big-segment store reported Available: false until its background status poll happened to run, so a status query right after start raced the poller. The sync manager avoids this by polling inline on first call, but the async status getter was a plain sync method and could not await. Make it awaitable, mirroring the is_initialized / data_availability fix: add an AsyncBigSegmentStoreStatusProvider whose get_status() is a coroutine, and have AsyncBigSegmentStoreManager.get_status() poll inline on first call (as get_user_membership already does). The status is now real on the first query instead of racing the background poll. Sync side unchanged. --- contract-tests/async_client_entity.py | 2 +- ldclient/async_client.py | 10 ++-- ldclient/impl/async_big_segments.py | 22 ++++---- ldclient/impl/big_segments_common.py | 34 ++++++++++- ldclient/interfaces.py | 56 ++++++++++++++++++- .../testing/impl/test_async_big_segments.py | 40 +++++++++++-- ldclient/testing/test_sync_async_parity.py | 12 ++++ 7 files changed, 152 insertions(+), 24 deletions(-) diff --git a/contract-tests/async_client_entity.py b/contract-tests/async_client_entity.py index a979f401..4a7d9086 100644 --- a/contract-tests/async_client_entity.py +++ b/contract-tests/async_client_entity.py @@ -185,7 +185,7 @@ def _context_response(self, c: Context) -> dict: return {"error": c.error} async def get_big_segment_store_status(self) -> dict: - status = self._client.big_segment_store_status_provider.status + status = await self._client.big_segment_store_status_provider.get_status() return {"available": status.available, "stale": status.stale} async def migration_variation(self, params: dict) -> dict: diff --git a/ldclient/async_client.py b/ldclient/async_client.py index abb9855e..031faf4d 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -41,9 +41,9 @@ from ldclient.impl.stubs import AsyncNullEventProcessor from ldclient.impl.util import log from ldclient.interfaces import ( + AsyncBigSegmentStoreStatusProvider, AsyncFeatureStore, AsyncFlagTracker, - BigSegmentStoreStatusProvider, DataSourceStatusProvider, DataStoreStatusProvider ) @@ -691,13 +691,13 @@ async def __try_execute_stage(self, method: str, hook_name: str, block: Callable return {} @property - def big_segment_store_status_provider(self) -> BigSegmentStoreStatusProvider: + def big_segment_store_status_provider(self) -> AsyncBigSegmentStoreStatusProvider: """ Returns an interface for tracking the status of a Big Segment store. - The :class:`ldclient.interfaces.BigSegmentStoreStatusProvider` has methods for checking - whether the Big Segment store is (as far as the SDK knows) currently operational and - tracking changes in this status. + The :class:`ldclient.interfaces.AsyncBigSegmentStoreStatusProvider` has methods for + checking whether the Big Segment store is (as far as the SDK knows) currently + operational and tracking changes in this status. """ return self.__big_segment_store_manager.status_provider diff --git a/ldclient/impl/async_big_segments.py b/ldclient/impl/async_big_segments.py index 23cbe856..c34552f6 100644 --- a/ldclient/impl/async_big_segments.py +++ b/ldclient/impl/async_big_segments.py @@ -7,14 +7,14 @@ from ldclient.impl.aio.concurrency import AsyncRepeatingTask from ldclient.impl.big_segments_common import ( EMPTY_MEMBERSHIP, - BigSegmentStoreStatusProviderImpl, + AsyncBigSegmentStoreStatusProviderImpl, _hash_for_user_key, is_stale ) from ldclient.impl.util import log from ldclient.interfaces import ( - BigSegmentStoreStatus, - BigSegmentStoreStatusProvider + AsyncBigSegmentStoreStatusProvider, + BigSegmentStoreStatus ) @@ -28,7 +28,7 @@ def __init__(self, config: AsyncBigSegmentsConfig): self.__store = config.store self.__stale_after_millis = config.stale_after * 1000 - self.__status_provider = BigSegmentStoreStatusProviderImpl(self.get_status) + self.__status_provider = AsyncBigSegmentStoreStatusProviderImpl(self.get_status) self.__last_status = None # type: Optional[BigSegmentStoreStatus] self.__poll_task = None # type: Optional[AsyncRepeatingTask] @@ -49,7 +49,7 @@ async def stop(self): await self.__store.stop() @property - def status_provider(self) -> BigSegmentStoreStatusProvider: + def status_provider(self) -> AsyncBigSegmentStoreStatusProvider: return self.__status_provider async def get_user_membership(self, user_key: str) -> Tuple[Optional[dict], str]: @@ -74,14 +74,16 @@ async def get_user_membership(self, user_key: str) -> Tuple[Optional[dict], str] return membership, BigSegmentsStatus.STORE_ERROR return membership, BigSegmentsStatus.STALE if status.stale else BigSegmentsStatus.HEALTHY - def get_status(self) -> BigSegmentStoreStatus: + async def get_status(self) -> BigSegmentStoreStatus: """Return the most recently polled status. - When no status has been cached yet, the sync variant polls the store - inline; the async variant (whose status getter cannot await) reports - the store as unavailable until the polling task has run. + When no status has been cached yet, poll the store inline and wait for the + result, so the status is accurate even if called immediately after start(). """ - return self.__last_status or BigSegmentStoreStatus(False, False) + status = self.__last_status + if status is None: + status = await self.poll_store_and_update_status() + return status async def poll_store_and_update_status(self) -> BigSegmentStoreStatus: new_status = BigSegmentStoreStatus(False, False) # default to "unavailable" if we don't get a new status below diff --git a/ldclient/impl/big_segments_common.py b/ldclient/impl/big_segments_common.py index 08f4f62e..d548d982 100644 --- a/ldclient/impl/big_segments_common.py +++ b/ldclient/impl/big_segments_common.py @@ -9,10 +9,11 @@ import base64 import time from hashlib import sha256 -from typing import Callable, Optional +from typing import Awaitable, Callable, Optional from ldclient.impl.listeners import Listeners from ldclient.interfaces import ( + AsyncBigSegmentStoreStatusProvider, BigSegmentStoreStatus, BigSegmentStoreStatusProvider ) @@ -61,3 +62,34 @@ def _update_status(self, new_status: BigSegmentStoreStatus): elif new_status.available != last.available or new_status.stale != last.stale: self.__last_status = new_status self.__status_listeners.notify(new_status) + + +class AsyncBigSegmentStoreStatusProviderImpl(AsyncBigSegmentStoreStatusProvider): + """ + Default implementation of the AsyncBigSegmentStoreStatusProvider interface. + + Mirrors :class:`BigSegmentStoreStatusProviderImpl`, except the status getter passed in + is a coroutine, so :meth:`get_status` is a coroutine too and awaits it. + """ + + def __init__(self, status_getter: Callable[[], Awaitable[BigSegmentStoreStatus]]): + self.__status_getter = status_getter + self.__status_listeners = Listeners() + self.__last_status = None # type: Optional[BigSegmentStoreStatus] + + async def get_status(self) -> BigSegmentStoreStatus: + return await self.__status_getter() + + def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.add(listener) + + def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.remove(listener) + + def _update_status(self, new_status: BigSegmentStoreStatus): + last = self.__last_status + if last is None: + self.__last_status = new_status + elif new_status.available != last.available or new_status.stale != last.stale: + self.__last_status = new_status + self.__status_listeners.notify(new_status) diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index ff28659e..2c9c245f 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -894,9 +894,8 @@ def status(self) -> BigSegmentStoreStatus: """ Gets the current status of the store. - Before the first poll completes, the synchronous SDK performs a blocking store query and - returns its result, while the async SDK (``AsyncLDClient``) returns the last polled status - without blocking -- ``available=False`` until the first background poll completes. + Before the first poll completes, this performs a blocking store query and returns its + result, so the status is accurate even if called immediately after the client starts. :return: the status """ @@ -925,6 +924,57 @@ def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> pass +class AsyncBigSegmentStoreStatusProvider(ABC): + """ + Async interface for querying the status of a Big Segment store, for use with the async + client (:class:`ldclient.async_client.AsyncLDClient`). It mirrors + :class:`BigSegmentStoreStatusProvider`, except the current status is read with the + coroutine :meth:`get_status` instead of a synchronous ``status`` property (a property + cannot await). + + .. 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. + + An implementation of this abstract class is returned by + :meth:`ldclient.async_client.AsyncLDClient.big_segment_store_status_provider`. Application + code never needs to implement this interface. + """ + + @abstractmethod + async def get_status(self) -> BigSegmentStoreStatus: + """ + Gets the current status of the store. + + Before the first poll completes, this awaits a store query and returns its result, so + the status is accurate even if called immediately after the client starts. + + :return: the status + """ + pass + + @abstractmethod + def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + """ + Subscribes for notifications of status changes. Behaves identically to + :meth:`BigSegmentStoreStatusProvider.add_listener`. + + :param listener: the listener to add + """ + pass + + @abstractmethod + def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + """ + Unsubscribes from notifications of status changes. + + :param listener: a listener that was previously added with :func:`add_listener()`; if it was not, + this method does nothing + """ + pass + + class DataSourceState(Enum): """ Enumeration representing the states a data source can be in at any given time. diff --git a/ldclient/testing/impl/test_async_big_segments.py b/ldclient/testing/impl/test_async_big_segments.py index 1fedc48f..b2779d17 100644 --- a/ldclient/testing/impl/test_async_big_segments.py +++ b/ldclient/testing/impl/test_async_big_segments.py @@ -302,8 +302,8 @@ async def test_stop_stops_store(): @pytest.mark.asyncio -async def test_status_provider_status_is_synchronous(): - """BigSegmentStoreStatusProvider.status must be readable synchronously without an await.""" +async def test_status_provider_get_status_after_poll(): + """AsyncBigSegmentStoreStatusProvider.get_status() reflects the last polled status.""" store = MockAsyncBigSegmentStore() store.setup_metadata_always_up_to_date() @@ -311,8 +311,40 @@ async def test_status_provider_status_is_synchronous(): try: # Poll once to populate __last_status await manager.poll_store_and_update_status() - # status property is sync — calling it should not raise - status = manager.status_provider.status + status = await manager.status_provider.get_status() assert status.available is True finally: await manager.stop() + + +@pytest.mark.asyncio +async def test_get_status_polls_inline_before_background_poll_runs(): + """ + get_status() must not race the background AsyncRepeatingTask: calling it immediately + after start() -- before the poll task has had a chance to run -- should still report an + accurate, available status by polling inline, mirroring the sync manager's behavior. + """ + store = MockAsyncBigSegmentStore() + store.setup_metadata_always_up_to_date() + + config = AsyncBigSegmentsConfig(store=store) + manager = AsyncBigSegmentStoreManager(config) + manager.start() + try: + status = await manager.get_status() + assert status.available is True + assert status.stale is False + finally: + await manager.stop() + + +@pytest.mark.asyncio +async def test_get_status_with_no_store_configured(): + """With no store configured, get_status() should report unavailable rather than hang or error.""" + config = AsyncBigSegmentsConfig(store=None) + manager = AsyncBigSegmentStoreManager(config) + try: + status = await manager.get_status() + assert status.available is False + finally: + await manager.stop() diff --git a/ldclient/testing/test_sync_async_parity.py b/ldclient/testing/test_sync_async_parity.py index 845fc20d..d1e42042 100644 --- a/ldclient/testing/test_sync_async_parity.py +++ b/ldclient/testing/test_sync_async_parity.py @@ -18,6 +18,10 @@ from ldclient.feature_store import InMemoryFeatureStore from ldclient.impl.async_evaluator import AsyncEvaluator from ldclient.impl.evaluator import Evaluator +from ldclient.interfaces import ( + AsyncBigSegmentStoreStatusProvider, + BigSegmentStoreStatusProvider +) from ldclient.migrations import AsyncMigratorBuilder, MigratorBuilder @@ -48,6 +52,14 @@ def _public_surface(cls) -> set: ), pytest.param(Evaluator, AsyncEvaluator, set(), set(), id="evaluator"), pytest.param(MigratorBuilder, AsyncMigratorBuilder, set(), set(), id="migrator_builder"), + pytest.param( + BigSegmentStoreStatusProvider, AsyncBigSegmentStoreStatusProvider, + # status is a synchronous property on the sync provider; a property can't await, + # so the async provider exposes the same readiness check as the get_status coroutine. + {"status"}, + {"get_status"}, + id="big_segment_store_status_provider", + ), ] From f5ea3d32f819cb58ec18cfc012c74f6bf3f9706a Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 28 Aug 2026 11:52:16 -0500 Subject: [PATCH 2/3] chore: Move the async big-segment status provider out of _common AsyncBigSegmentStoreStatusProviderImpl is async-only (used only by async_big_segments), so it belongs there, not in big_segments_common (which is for logic shared by the sync and async managers). --- ldclient/impl/async_big_segments.py | 35 ++++++++++++++++++++-- ldclient/impl/big_segments_common.py | 43 ++++------------------------ 2 files changed, 39 insertions(+), 39 deletions(-) diff --git a/ldclient/impl/async_big_segments.py b/ldclient/impl/async_big_segments.py index c34552f6..ce4aecd0 100644 --- a/ldclient/impl/async_big_segments.py +++ b/ldclient/impl/async_big_segments.py @@ -1,4 +1,4 @@ -from typing import Optional, Tuple +from typing import Awaitable, Callable, Optional, Tuple from expiringdict import ExpiringDict @@ -7,10 +7,10 @@ from ldclient.impl.aio.concurrency import AsyncRepeatingTask from ldclient.impl.big_segments_common import ( EMPTY_MEMBERSHIP, - AsyncBigSegmentStoreStatusProviderImpl, _hash_for_user_key, is_stale ) +from ldclient.impl.listeners import Listeners from ldclient.impl.util import log from ldclient.interfaces import ( AsyncBigSegmentStoreStatusProvider, @@ -18,6 +18,37 @@ ) +class AsyncBigSegmentStoreStatusProviderImpl(AsyncBigSegmentStoreStatusProvider): + """ + Default implementation of the AsyncBigSegmentStoreStatusProvider interface. + + Mirrors :class:`BigSegmentStoreStatusProviderImpl`, except the status getter passed in + is a coroutine, so :meth:`get_status` is a coroutine too and awaits it. + """ + + def __init__(self, status_getter: Callable[[], Awaitable[BigSegmentStoreStatus]]): + self.__status_getter = status_getter + self.__status_listeners = Listeners() + self.__last_status = None # type: Optional[BigSegmentStoreStatus] + + async def get_status(self) -> BigSegmentStoreStatus: + return await self.__status_getter() + + def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.add(listener) + + def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.remove(listener) + + def _update_status(self, new_status: BigSegmentStoreStatus): + last = self.__last_status + if last is None: + self.__last_status = new_status + elif new_status.available != last.available or new_status.stale != last.stale: + self.__last_status = new_status + self.__status_listeners.notify(new_status) + + class AsyncBigSegmentStoreManager: """ Internal component that decorates the Big Segment store with caching behavior, and also polls the diff --git a/ldclient/impl/big_segments_common.py b/ldclient/impl/big_segments_common.py index d548d982..f181e43f 100644 --- a/ldclient/impl/big_segments_common.py +++ b/ldclient/impl/big_segments_common.py @@ -1,19 +1,19 @@ """ -Shared, I/O-free big-segments status provider used by both the sync +I/O-free helpers for big-segments status tracking, plus the sync +:class:`BigSegmentStoreStatusProviderImpl`. The helpers (`_hash_for_user_key`, +`is_stale`, `EMPTY_MEMBERSHIP`) are shared by both the sync :mod:`ldclient.impl.big_segments` and the async -:mod:`ldclient.impl.async_big_segments`. It holds the last known status and -notifies listeners; nothing here touches the store or network, so it is -identical across the two managers. +:mod:`ldclient.impl.async_big_segments` managers; nothing here touches the +store or network. """ import base64 import time from hashlib import sha256 -from typing import Awaitable, Callable, Optional +from typing import Callable, Optional from ldclient.impl.listeners import Listeners from ldclient.interfaces import ( - AsyncBigSegmentStoreStatusProvider, BigSegmentStoreStatus, BigSegmentStoreStatusProvider ) @@ -62,34 +62,3 @@ def _update_status(self, new_status: BigSegmentStoreStatus): elif new_status.available != last.available or new_status.stale != last.stale: self.__last_status = new_status self.__status_listeners.notify(new_status) - - -class AsyncBigSegmentStoreStatusProviderImpl(AsyncBigSegmentStoreStatusProvider): - """ - Default implementation of the AsyncBigSegmentStoreStatusProvider interface. - - Mirrors :class:`BigSegmentStoreStatusProviderImpl`, except the status getter passed in - is a coroutine, so :meth:`get_status` is a coroutine too and awaits it. - """ - - def __init__(self, status_getter: Callable[[], Awaitable[BigSegmentStoreStatus]]): - self.__status_getter = status_getter - self.__status_listeners = Listeners() - self.__last_status = None # type: Optional[BigSegmentStoreStatus] - - async def get_status(self) -> BigSegmentStoreStatus: - return await self.__status_getter() - - def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: - self.__status_listeners.add(listener) - - def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: - self.__status_listeners.remove(listener) - - def _update_status(self, new_status: BigSegmentStoreStatus): - last = self.__last_status - if last is None: - self.__last_status = new_status - elif new_status.available != last.available or new_status.stale != last.stale: - self.__last_status = new_status - self.__status_listeners.notify(new_status) From 9af34884290d67ff020620865aa75ed9db7213a0 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 28 Aug 2026 12:02:10 -0500 Subject: [PATCH 3/3] chore: Move the sync big-segment status provider back to big_segments BigSegmentStoreStatusProviderImpl originally lived in big_segments.py; the async work moved it into big_segments_common to share it. Now that async has its own provider, the sync one is single-sided again, so move it back. big_segments_common now holds only the shared I/O-free helpers. --- ldclient/impl/big_segments.py | 37 ++++++++++++++++++++-- ldclient/impl/big_segments_common.py | 47 ++-------------------------- 2 files changed, 38 insertions(+), 46 deletions(-) diff --git a/ldclient/impl/big_segments.py b/ldclient/impl/big_segments.py index 0f42c07d..cf2dec61 100644 --- a/ldclient/impl/big_segments.py +++ b/ldclient/impl/big_segments.py @@ -1,4 +1,4 @@ -from typing import Optional, Tuple +from typing import Callable, Optional, Tuple from expiringdict import ExpiringDict @@ -6,10 +6,10 @@ from ldclient.evaluation import BigSegmentsStatus from ldclient.impl.big_segments_common import ( EMPTY_MEMBERSHIP, - BigSegmentStoreStatusProviderImpl, _hash_for_user_key, is_stale ) +from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.util import log from ldclient.interfaces import ( @@ -18,6 +18,39 @@ ) +class BigSegmentStoreStatusProviderImpl(BigSegmentStoreStatusProvider): + """ + Default implementation of the BigSegmentStoreStatusProvider interface. + + The real implementation of getting the status is in the big segment store manager - we pass in a lambda that + allows us to get the current status from that class. So this class provides a facade for that, and + also adds the listener mechanism. + """ + + def __init__(self, status_getter: Callable[[], BigSegmentStoreStatus]): + self.__status_getter = status_getter + self.__status_listeners = Listeners() + self.__last_status = None # type: Optional[BigSegmentStoreStatus] + + @property + def status(self) -> BigSegmentStoreStatus: + return self.__status_getter() + + def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.add(listener) + + def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: + self.__status_listeners.remove(listener) + + def _update_status(self, new_status: BigSegmentStoreStatus): + last = self.__last_status + if last is None: + self.__last_status = new_status + elif new_status.available != last.available or new_status.stale != last.stale: + self.__last_status = new_status + self.__status_listeners.notify(new_status) + + class BigSegmentStoreManager: """ Internal component that decorates the Big Segment store with caching behavior, and also polls the diff --git a/ldclient/impl/big_segments_common.py b/ldclient/impl/big_segments_common.py index f181e43f..182c2bcc 100644 --- a/ldclient/impl/big_segments_common.py +++ b/ldclient/impl/big_segments_common.py @@ -1,8 +1,7 @@ """ -I/O-free helpers for big-segments status tracking, plus the sync -:class:`BigSegmentStoreStatusProviderImpl`. The helpers (`_hash_for_user_key`, -`is_stale`, `EMPTY_MEMBERSHIP`) are shared by both the sync -:mod:`ldclient.impl.big_segments` and the async +This module holds the shared, I/O-free helpers for big-segments status +tracking. The helpers (`_hash_for_user_key`, `is_stale`, `EMPTY_MEMBERSHIP`) +are shared by both the sync :mod:`ldclient.impl.big_segments` and the async :mod:`ldclient.impl.async_big_segments` managers; nothing here touches the store or network. """ @@ -10,13 +9,6 @@ import base64 import time from hashlib import sha256 -from typing import Callable, Optional - -from ldclient.impl.listeners import Listeners -from ldclient.interfaces import ( - BigSegmentStoreStatus, - BigSegmentStoreStatusProvider -) # use EMPTY_MEMBERSHIP as a singleton whenever a membership query returns None; it's safe to reuse it # because we will never modify the membership properties after they're queried @@ -29,36 +21,3 @@ def _hash_for_user_key(user_key: str) -> str: def is_stale(timestamp: int, stale_after_millis) -> bool: return (timestamp is None) or ((int(time.time() * 1000) - timestamp) >= stale_after_millis) - - -class BigSegmentStoreStatusProviderImpl(BigSegmentStoreStatusProvider): - """ - Default implementation of the BigSegmentStoreStatusProvider interface. - - The real implementation of getting the status is in the big segment store manager - we pass in a lambda that - allows us to get the current status from that class. So this class provides a facade for that, and - also adds the listener mechanism. - """ - - def __init__(self, status_getter: Callable[[], BigSegmentStoreStatus]): - self.__status_getter = status_getter - self.__status_listeners = Listeners() - self.__last_status = None # type: Optional[BigSegmentStoreStatus] - - @property - def status(self) -> BigSegmentStoreStatus: - return self.__status_getter() - - def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: - self.__status_listeners.add(listener) - - def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None: - self.__status_listeners.remove(listener) - - def _update_status(self, new_status: BigSegmentStoreStatus): - last = self.__last_status - if last is None: - self.__last_status = new_status - elif new_status.available != last.available or new_status.stale != last.stale: - self.__last_status = new_status - self.__status_listeners.notify(new_status)