From 47d9be996e0cd42fb17414495b67d5d57094f299 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Mon, 2 Nov 2026 13:03:18 -0500 Subject: [PATCH 1/5] feat!: Require status and close methods on feature stores Custom FeatureStore and AsyncFeatureStore implementations must now implement is_monitoring_enabled, is_available, and close. Custom FeatureStoreCore and AsyncFeatureStoreCore implementations must now implement is_available and close. These methods are abstract, so a store that omits one cannot be instantiated. The SDK no longer probes for these methods. CachingStoreWrapper and AsyncCachingStoreWrapper always report status monitoring as enabled and delegate is_available and close to the core. The async in-memory store's is_available is now a coroutine, to match the async interface. --- ldclient/async_feature_store.py | 2 +- ldclient/async_feature_store_helpers.py | 18 +- ldclient/feature_store.py | 3 + ldclient/feature_store_helpers.py | 14 +- ldclient/impl/datasystem/async_fdv2.py | 28 +-- ldclient/impl/datasystem/fdv1.py | 38 ++- ldclient/impl/datasystem/fdv2.py | 39 +--- ldclient/impl/datasystem/fdv2_common.py | 4 +- ldclient/impl/datasystem/store.py | 6 +- .../consul/consul_feature_store.py | 4 + .../dynamodb/dynamodb_feature_store.py | 3 + .../integrations/redis/redis_feature_store.py | 4 + ldclient/interfaces.py | 218 ++++++++++-------- .../impl/datasystem/test_async_config.py | 9 + .../impl/datasystem/test_async_fdv2.py | 53 ++--- .../datasystem/test_async_fdv2_persistence.py | 3 + .../impl/datasystem/test_cache_lifecycle.py | 30 +-- .../datasystem/test_fdv2_async_persistence.py | 12 + .../impl/datasystem/test_fdv2_persistence.py | 3 + ldclient/testing/stub_util.py | 9 + .../test_async_feature_store_helpers.py | 72 +++--- .../test_async_in_memory_feature_store.py | 7 +- .../testing/test_feature_store_helpers.py | 51 ++-- .../testing/test_feature_store_interfaces.py | 94 ++++++++ .../testing/test_in_memory_feature_store.py | 4 +- ldclient/testing/test_ldclient_evaluation.py | 18 ++ ldclient/testing/test_sync_async_parity.py | 2 +- 27 files changed, 420 insertions(+), 328 deletions(-) create mode 100644 ldclient/testing/test_feature_store_interfaces.py diff --git a/ldclient/async_feature_store.py b/ldclient/async_feature_store.py index c29041e3..8f47807d 100644 --- a/ldclient/async_feature_store.py +++ b/ldclient/async_feature_store.py @@ -28,7 +28,7 @@ def __init__(self): def is_monitoring_enabled(self) -> bool: return False - def is_available(self) -> bool: + async def is_available(self) -> bool: return True async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: diff --git a/ldclient/async_feature_store_helpers.py b/ldclient/async_feature_store_helpers.py index 94aa2756..75455967 100644 --- a/ldclient/async_feature_store_helpers.py +++ b/ldclient/async_feature_store_helpers.py @@ -2,7 +2,6 @@ This submodule contains support code for writing async feature store implementations. """ -import inspect from typing import Any, Dict, Mapping, Optional from ldclient.feature_store import CacheConfig @@ -48,16 +47,11 @@ def __init__(self, core: AsyncFeatureStoreCore, cache_config: CacheConfig): :param cache_config: the caching parameters """ self._core = core - self._has_available_method = callable(getattr(core, 'is_available', None)) super().__init__(cache_config) async def is_available(self) -> bool: - """Tests whether the underlying store seems to be reachable. - - Returns False if the core does not provide an availability check. - """ - # We know is_available exists since we are checking _has_available_method. - return await self._core.is_available() if self._has_available_method else False # type: ignore + """Returns the result of the availability check of the core.""" + return await self._core.is_available() async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: """ """ @@ -119,13 +113,9 @@ async def is_initialized(self) -> bool: return result async def close(self) -> None: - """Releases the cache and closes the underlying core if it supports it.""" + """Releases the cache and closes the core.""" self.disable_cache() - core_close = getattr(self._core, "close", None) - if callable(core_close): - result = core_close() - if inspect.isawaitable(result): - await result + await self._core.close() def describe_configuration(self, config) -> str: describe = getattr(self._core, 'describe_configuration', None) diff --git a/ldclient/feature_store.py b/ldclient/feature_store.py index 974063f8..2cd50286 100644 --- a/ldclient/feature_store.py +++ b/ldclient/feature_store.py @@ -84,6 +84,9 @@ def is_monitoring_enabled(self) -> bool: def is_available(self) -> bool: return True + def close(self) -> None: + """Does nothing. The in-memory store has no resources to release.""" + def get(self, kind: VersionedDataKind, key: str, callback: Callable[[Any], Any] = lambda x: x) -> Any: """ """ with self._lock.read(): diff --git a/ldclient/feature_store_helpers.py b/ldclient/feature_store_helpers.py index 3b8b4159..4d18e0cd 100644 --- a/ldclient/feature_store_helpers.py +++ b/ldclient/feature_store_helpers.py @@ -58,7 +58,6 @@ class _CachingStoreWrapperBase: _cache: Any _inited: bool - _has_available_method: bool def __init__(self, cache_config: CacheConfig): """Sets up the cache from the caching parameters. @@ -72,7 +71,8 @@ def __init__(self, cache_config: CacheConfig): self._inited = False def is_monitoring_enabled(self) -> bool: - return self._has_available_method + """Returns True. The wrapper reports status monitoring as enabled for all cores.""" + return True def disable_cache(self) -> None: """Replace the in-memory cache with a no-op so further operations don't populate it. @@ -201,12 +201,11 @@ def __init__(self, core: FeatureStoreCore, cache_config: CacheConfig): :param cache_config: the caching parameters """ self._core = core - self._has_available_method = callable(getattr(core, 'is_available', None)) super().__init__(cache_config) def is_available(self) -> bool: - # We know is_available exists since we are checking _has_available_method - return self._core.is_available() if self._has_available_method else False # type: ignore + """Returns the result of the availability check of the core.""" + return self._core.is_available() def init(self, all_encoded_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]): """ """ @@ -258,10 +257,9 @@ def initialized(self) -> bool: return result def close(self) -> None: - """Release the cache and close the underlying core if it supports it.""" + """Releases the cache and closes the core.""" self.disable_cache() - if hasattr(self._core, "close"): - self._core.close() # type: ignore + self._core.close() def describe_configuration(self, config): describe = getattr(self._core, 'describe_configuration', None) diff --git a/ldclient/impl/datasystem/async_fdv2.py b/ldclient/impl/datasystem/async_fdv2.py index d6b6c3e3..dc30d03a 100644 --- a/ldclient/impl/datasystem/async_fdv2.py +++ b/ldclient/impl/datasystem/async_fdv2.py @@ -113,16 +113,12 @@ def disable_cache(self) -> None: inner_disable() def is_monitoring_enabled(self) -> bool: - """Returns whether the inner store opts in to availability monitoring. + """Returns the monitoring setting of the inner store.""" + return self._store.is_monitoring_enabled() - Delegates to the store's own opt-in. A store that does not report - availability is not polled, so it is never marked unavailable with no - path back to recovery. - """ - store_check = getattr(self._store, "is_monitoring_enabled", None) - if not callable(store_check): - return False - return store_check() + async def is_available(self) -> bool: + """Returns the result of the availability check of the inner store.""" + return await self._store.is_available() async def _wrap(self, fn: Callable): try: @@ -162,7 +158,7 @@ def _update_availability(self, available: bool) -> None: async def _check_availability(self) -> None: try: - if await self._store.is_available(): # type: ignore[attr-defined] + if await self._store.is_available(): self._update_availability(True) except BaseException as e: log.error("Unexpected error from data store status function: %s", e) @@ -185,14 +181,10 @@ async def close(self) -> None: except asyncio.TimeoutError: log.warning("Timed out waiting for the persistent store availability poller to stop") - close = getattr(self._store, "close", None) - if callable(close): - try: - result = close() - if inspect.isawaitable(result): - await result - except Exception as e: - log.warning("Error closing the persistent store: %s", e) + try: + await self._store.close() + except Exception as e: + log.warning("Error closing the persistent store: %s", e) class _AsyncReadOnlyStoreView(AsyncReadOnlyStore): diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index 477b6e0f..affc06ec 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -115,35 +115,23 @@ def __check_availability(self): 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. + """Returns the monitoring setting of the wrapped store.""" + return self.store.is_monitoring_enabled() - 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. + def is_available(self) -> bool: + """Returns the result of the availability check of the wrapped store.""" + return self.store.is_available() - 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 + def close(self) -> None: + """Stops the availability poller and closes the wrapped store.""" + with self.__lock.write(): + poller_to_stop = self.__poller + self.__poller = None - monitoring_enabled = getattr(self.store, 'is_monitoring_enabled') - if not callable(monitoring_enabled): - return False + if poller_to_stop is not None: + poller_to_stop.stop() - return monitoring_enabled() + self.store.close() class _ReadOnlyFeatureStoreView(ReadOnlyStore): diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index afb80088..685beac9 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -155,40 +155,16 @@ def __check_availability(self): 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 + """Returns the monitoring setting of the wrapped store.""" + return self.store.is_monitoring_enabled() - return monitoring_enabled() + def is_available(self) -> bool: + """Returns the result of the availability check of the wrapped store.""" + return self.store.is_available() def close(self): """ - Close the wrapper and stop the repeating task poller if it's running. - Also forwards the close call to the underlying store if it has a close method. + Closes the wrapper. Stops the availability poller if it runs, then closes the wrapped store. """ poller_to_stop = None @@ -202,8 +178,7 @@ def close(self): if poller_to_stop is not None: poller_to_stop.stop() - if hasattr(self.store, "close"): - self.store.close() + self.store.close() class FDv2(_FDv2Base, DataSystem): diff --git a/ldclient/impl/datasystem/fdv2_common.py b/ldclient/impl/datasystem/fdv2_common.py index f692919f..cc50814d 100644 --- a/ldclient/impl/datasystem/fdv2_common.py +++ b/ldclient/impl/datasystem/fdv2_common.py @@ -97,10 +97,8 @@ def status(self) -> DataStoreStatus: def is_monitoring_enabled(self) -> bool: if self.__store is None: return False - if hasattr(self.__store, "is_monitoring_enabled") is False: - return False - return self.__store.is_monitoring_enabled() # type: ignore + return self.__store.is_monitoring_enabled() def add_listener(self, listener: Callable[[DataStoreStatus], None]): self.__listeners.add(listener) diff --git a/ldclient/impl/datasystem/store.py b/ldclient/impl/datasystem/store.py index b6532605..92b6c7d5 100644 --- a/ldclient/impl/datasystem/store.py +++ b/ldclient/impl/datasystem/store.py @@ -526,11 +526,7 @@ def close(self) -> None: with self._lock: if self._persistent_store is not None: try: - # Most FeatureStore implementations don't have close methods - # but we'll try to call it if it exists - close = getattr(self._persistent_store, "close", None) - if callable(close): - close() + self._persistent_store.close() except Exception as e: log.warning("Error closing the persistent store: %s", e) diff --git a/ldclient/impl/integrations/consul/consul_feature_store.py b/ldclient/impl/integrations/consul/consul_feature_store.py index 07f8d842..142b089e 100644 --- a/ldclient/impl/integrations/consul/consul_feature_store.py +++ b/ldclient/impl/integrations/consul/consul_feature_store.py @@ -52,6 +52,10 @@ def is_available(self) -> bool: except BaseException: return False + def close(self) -> None: + """Closes the HTTP session of the Consul client.""" + self._client.http.session.close() + def init_internal(self, all_data): # Start by reading the existing keys; we will later delete any of these that weren't in all_data. index, keys = self._client.kv.get(self._prefix, recurse=True, keys=True) diff --git a/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py b/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py index bc025502..2ff784be 100644 --- a/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py +++ b/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py @@ -62,6 +62,9 @@ def is_available(self) -> bool: except BaseException: return False + def close(self) -> None: + """Does nothing. The boto3 client has no close method in all supported versions.""" + def init_internal(self, all_data): # Start by reading the existing keys; we will later delete any of these that weren't in all_data. unused_old_keys = self._read_existing_keys(all_data.keys()) diff --git a/ldclient/impl/integrations/redis/redis_feature_store.py b/ldclient/impl/integrations/redis/redis_feature_store.py index b44f39c0..77fde737 100644 --- a/ldclient/impl/integrations/redis/redis_feature_store.py +++ b/ldclient/impl/integrations/redis/redis_feature_store.py @@ -36,6 +36,10 @@ def is_available(self) -> bool: except BaseException: return False + def close(self) -> None: + """Disconnects all connections in the connection pool.""" + self._pool.disconnect() + def _items_key(self, kind): return "{0}:{1}".format(self._prefix, kind.namespace) diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index 0ecd6026..03ea0a1d 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -150,79 +150,51 @@ def initialized(self) -> bool: Returns whether the store has been initialized yet or not """ - # WARN: This isn't a required method on a FeatureStore yet. The SDK will - # currently check if the provided store responds to this method, and if - # it does, will take appropriate action based on the documented behavior - # below. This will become required in a future major version release of - # the SDK. - # - # @abstractmethod - # def is_monitoring_enabled(self) -> bool: - # """ - # Returns true if this data store implementation supports status - # monitoring. - # - # This is normally only true for persistent data stores but it could also - # be true for any custom :class:`FeatureStore` implementation. - # - # Returning true means that the store guarantees that if it ever enters - # an invalid state (that is, an operation has failed or it knows that - # operations cannot succeed at the moment), it will publish a status - # update, and will then publish another status update once it has - # returned to a valid state. - # - # Custom implementations must implement :func:`FeatureStore.is_available` - # which synchronously checks if the store is available. Without this - # method, the SDK cannot ensure status updates will occur once the store - # has gone offline. - # - # The same value will be returned from - # :func:`DataStoreStatusProvider.is_monitoring_enabled`. - # """ - - # WARN: This isn't a required method on a FeatureStore. The SDK will - # check if the provided store responds to this method, and if it does, - # will take appropriate action based on the documented behavior below. - # Usage of this method will be dropped in a future version of the SDK. - # - # @abstractmethod - # def is_available(self) -> bool: - # """ - # Tests whether the data store seems to be functioning normally. - # - # This should not be a detailed test of different kinds of operations, - # but just the smallest possible operation to determine whether (for - # instance) we can reach the database. - # - # Whenever one of the store's other methods throws an exception, the SDK - # will assume that it may have become unavailable (e.g. the database - # connection was lost). The SDK will then call is_available at intervals - # until it returns true. - # - # :return: true if the underlying data store is reachable - # """ - - # WARN: This isn't a required method on a FeatureStore. The SDK will - # check if the provided store responds to this method, and if it does, - # will call it during shutdown to release any resources (such as database - # connections or connection pools) that the store may be using. - # - # @abstractmethod - # def close(self): - # """ - # Releases any resources used by the data store implementation. - # - # This method will be called by the SDK during shutdown to ensure proper - # cleanup of resources such as database connections, connection pools, - # network sockets, or other resources that should be explicitly released. - # - # Implementations should be idempotent - calling close() multiple times - # should be safe and have no additional effect after the first call. - # - # This is particularly important for persistent data stores that maintain - # connection pools or other long-lived resources that should be properly - # cleaned up when the SDK is shut down. - # """ + @abstractmethod + def is_monitoring_enabled(self) -> bool: + """ + Returns True if this store supports status monitoring. + + This is normally True only for persistent data stores. A custom :class:`FeatureStore` + implementation can also return True. + + If this method returns True, the store must publish a status update when it becomes + unavailable. An unavailable store is a store where an operation failed, or where the store + knows that operations cannot succeed. The store must then publish another status update + when it becomes available again. The SDK uses :func:`is_available` to detect this change. + + In-memory stores must return False. + + The SDK returns the same value from :func:`DataStoreStatusProvider.is_monitoring_enabled`. + + :return: True if the store supports status monitoring + """ + + @abstractmethod + def is_available(self) -> bool: + """ + Tests whether the data store operates normally. + + Do not test different kinds of operations in detail. Do the smallest possible operation + that shows that the store works. For example, make sure that the database is reachable. + + When one of the other methods of the store raises an exception, the SDK assumes that the + store can be unavailable. If :func:`is_monitoring_enabled` returns True, the SDK then calls + this method at intervals until it returns True. + + A store that cannot become unavailable, such as an in-memory store, must return True. + + :return: True if the underlying data store is reachable + """ + + @abstractmethod + def close(self) -> None: + """ + Releases the resources that the store uses, such as database connections. + + The SDK calls this method when the client closes. A later call must have no effect. + A store that has no resources to release can do nothing. + """ class FeatureStoreCore(ABC): @@ -295,27 +267,27 @@ def initialized_internal(self) -> bool: this value; ``CachingStoreWrapper`` will only call it when necessary. """ - # WARN: This isn't a required method on a FeatureStoreCore. The SDK will - # check if the provided store responds to this method, and if it does, - # will take appropriate action based on the documented behavior below. - # Usage of this method will be dropped in a future version of the SDK. - # - # @abstractmethod - # def is_available(self) -> bool: - # """ - # Tests whether the data store seems to be functioning normally. - # - # This should not be a detailed test of different kinds of operations, - # but just the smallest possible operation to determine whether (for - # instance) we can reach the database. - # - # Whenever one of the store's other methods throws an exception, the SDK - # will assume that it may have become unavailable (e.g. the database - # connection was lost). The SDK will then call is_available at intervals - # until it returns true. - # - # :return: true if the underlying data store is reachable - # """ + @abstractmethod + def is_available(self) -> bool: + """ + Tests whether the data store operates normally. This is the same as + :func:`FeatureStore.is_available`. + + :class:`ldclient.feature_store_helpers.CachingStoreWrapper` reports status monitoring as + enabled for all cores. When an operation fails, the SDK calls this method at intervals + until it returns True. + + :return: True if the underlying data store is reachable + """ + + @abstractmethod + def close(self) -> None: + """ + Releases the resources that the core uses, such as database connections. + + :class:`ldclient.feature_store_helpers.CachingStoreWrapper` calls this method when the + wrapper closes. A later call must have no effect. + """ class AsyncReadOnlyStore(Protocol): @@ -434,14 +406,38 @@ async def is_initialized(self) -> bool: Implementations should latch a positive result: once the store is initialized it stays so. """ + @abstractmethod + def is_monitoring_enabled(self) -> bool: + """ + Returns True if this store supports status monitoring. + + This has the same meaning as :func:`FeatureStore.is_monitoring_enabled`. In-memory stores + must return False. + + :return: True if the store supports status monitoring + """ + + @abstractmethod + async def is_available(self) -> bool: + """ + Tests whether the data store operates normally. + + This has the same meaning as :func:`FeatureStore.is_available`. If + :func:`is_monitoring_enabled` returns True, the SDK calls this method at intervals after an + operation fails, until it returns True. A store that cannot become unavailable must return + True. + + :return: True if the underlying data store is reachable + """ + + @abstractmethod async def close(self) -> None: """ - Releases any resources used by the store implementation. + Releases the resources that the store uses, such as database connections. - The default implementation is a no-op. Persistent store implementations should - override this to close connections and release resources on shutdown. + The SDK calls this method when the client closes. A later call must have no effect. + A store that has no resources to release can do nothing. """ - pass class AsyncFeatureStoreCore(ABC): @@ -521,9 +517,29 @@ async def initialized_internal(self) -> bool: """ ... - # An implementation may also define ``async def is_available(self) -> bool`` and - # ``async def close(self) -> None``. The wrapper detects and uses them if present, so they are - # not declared here as required methods. + @abstractmethod + async def is_available(self) -> bool: + """ + Tests whether the data store operates normally. This is the same as + :func:`AsyncFeatureStore.is_available`. + + :class:`ldclient.async_feature_store_helpers.AsyncCachingStoreWrapper` reports status + monitoring as enabled for all cores. When an operation fails, the SDK calls this method at + intervals until it returns True. + + :return: True if the underlying data store is reachable + """ + ... + + @abstractmethod + async def close(self) -> None: + """ + Releases the resources that the core uses, such as database connections. + + :class:`ldclient.async_feature_store_helpers.AsyncCachingStoreWrapper` calls this method + when the wrapper closes. A later call must have no effect. + """ + ... # Internal use only. Common methods for components that perform a task in the background. diff --git a/ldclient/testing/impl/datasystem/test_async_config.py b/ldclient/testing/impl/datasystem/test_async_config.py index 143a7b06..c3758117 100644 --- a/ldclient/testing/impl/datasystem/test_async_config.py +++ b/ldclient/testing/impl/datasystem/test_async_config.py @@ -38,6 +38,15 @@ def initialized(self) -> bool: async def is_initialized(self) -> bool: return True + def is_monitoring_enabled(self) -> bool: + return False + + async def is_available(self) -> bool: + return True + + async def close(self) -> None: + pass + def test_async_data_system_config_defaults(): cfg = AsyncDataSystemConfig() diff --git a/ldclient/testing/impl/datasystem/test_async_fdv2.py b/ldclient/testing/impl/datasystem/test_async_fdv2.py index 1a0a2d88..ff6e05ca 100644 --- a/ldclient/testing/impl/datasystem/test_async_fdv2.py +++ b/ldclient/testing/impl/datasystem/test_async_fdv2.py @@ -654,39 +654,9 @@ async def close(self) -> None: self.closed = True -class StoreWithoutAvailability(AsyncFeatureStore): - async def init(self, all_data): - pass - - async def get(self, kind, key): - return None - - async def all(self, kind): - return {} - - async def upsert(self, kind, item): - return True - - async def delete(self, kind, key, version): - return True - - @property - def initialized(self) -> bool: - return True - - async def is_initialized(self) -> bool: - return True - - -class StoreWithAvailabilityNoOptIn(StoreWithoutAvailability): - """Reports availability but does not opt in to monitoring. - - Models a custom store that provides is_available yet omits - is_monitoring_enabled, which must not turn on availability polling. - """ - - async def is_available(self) -> bool: - return True +class StoreWithoutMonitoring(FakeAsyncStore): + def is_monitoring_enabled(self) -> bool: + return False @pytest.mark.asyncio @@ -696,16 +666,19 @@ async def test_is_monitoring_enabled_true_when_store_opts_in(): @pytest.mark.asyncio -async def test_is_monitoring_enabled_false_without_is_available(): - wrapper = _AsyncFeatureStoreClientWrapper(StoreWithoutAvailability(), lambda _s: None) - assert wrapper.is_monitoring_enabled() is False +async def test_is_monitoring_enabled_false_when_store_does_not_opt_in(): + wrapper = _AsyncFeatureStoreClientWrapper(StoreWithoutMonitoring(), lambda _s: None) + assert not wrapper.is_monitoring_enabled() @pytest.mark.asyncio -async def test_is_monitoring_enabled_false_when_store_does_not_opt_in(): - # A store with is_available but no is_monitoring_enabled must not be polled. - wrapper = _AsyncFeatureStoreClientWrapper(StoreWithAvailabilityNoOptIn(), lambda _s: None) - assert wrapper.is_monitoring_enabled() is False +async def test_is_available_delegates_to_store(): + store = FakeAsyncStore() + wrapper = _AsyncFeatureStoreClientWrapper(store, lambda _s: None) + assert await wrapper.is_available() + + store._available = False + assert not await wrapper.is_available() @pytest.mark.asyncio diff --git a/ldclient/testing/impl/datasystem/test_async_fdv2_persistence.py b/ldclient/testing/impl/datasystem/test_async_fdv2_persistence.py index 2d5cad5e..b0eae019 100644 --- a/ldclient/testing/impl/datasystem/test_async_fdv2_persistence.py +++ b/ldclient/testing/impl/datasystem/test_async_fdv2_persistence.py @@ -103,6 +103,9 @@ def initialized(self) -> bool: async def is_initialized(self) -> bool: return self._initialized + def is_monitoring_enabled(self) -> bool: + return False + async def is_available(self) -> bool: return self._available diff --git a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py index aa710e48..f6e8b44a 100644 --- a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py +++ b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py @@ -72,6 +72,9 @@ def upsert_internal(self, kind, item): def initialized_internal(self): return self.inited + def is_available(self): + return True + def close(self): self.closed = True @@ -438,30 +441,3 @@ def test_close_propagates_through_feature_store_client_wrapper(self): assert inner._cache is _NOOP_CACHE assert core.closed is True - - def test_close_when_core_has_no_close_method(self): - class CoreWithoutClose: - def __init__(self): - self.data: dict = {} - self.inited = False - - def init_internal(self, all_data): - self.data = dict(all_data) - self.inited = True - - def get_internal(self, kind, key): - return self.data.get(kind, {}).get(key) - - def get_all_internal(self, kind): - return self.data.get(kind, {}) - - def upsert_internal(self, kind, item): - return item - - def initialized_internal(self): - return self.inited - - wrapper = CachingStoreWrapper(CoreWithoutClose(), CacheConfig.default()) # type: ignore[arg-type] - - wrapper.close() # must not raise - assert wrapper._cache is _NOOP_CACHE diff --git a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py index dcfab21b..4f0552af 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py @@ -72,6 +72,12 @@ def initialized(self) -> bool: async def is_initialized(self) -> bool: return self._inited + def is_monitoring_enabled(self) -> bool: + return False + + async def is_available(self) -> bool: + return True + async def close(self) -> None: self.closed = True @@ -286,6 +292,12 @@ async def initialized_internal(self) -> bool: self.query_count += 1 return self.inited + async def is_available(self) -> bool: + return True + + async def close(self) -> None: + pass + @pytest.mark.asyncio async def test_wrapper_is_initialized_reflects_external_init_and_latches(): diff --git a/ldclient/testing/impl/datasystem/test_fdv2_persistence.py b/ldclient/testing/impl/datasystem/test_fdv2_persistence.py index ce1b87a3..682178e1 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_persistence.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_persistence.py @@ -97,6 +97,9 @@ def is_monitoring_enabled(self) -> bool: """For monitoring support""" return self._monitoring_enabled + def close(self) -> None: + pass + def set_available(self, available: bool): """Test helper to simulate availability changes""" self._available = available diff --git a/ldclient/testing/stub_util.py b/ldclient/testing/stub_util.py index 28b11758..fa1a493d 100644 --- a/ldclient/testing/stub_util.py +++ b/ldclient/testing/stub_util.py @@ -203,6 +203,15 @@ def upsert(self, kind, item): def initialized(self): return True + def is_monitoring_enabled(self): + return False + + def is_available(self): + return True + + def close(self): + pass + @property def received_data(self): return self.data diff --git a/ldclient/testing/test_async_feature_store_helpers.py b/ldclient/testing/test_async_feature_store_helpers.py index 4fb4c508..c6906c7c 100644 --- a/ldclient/testing/test_async_feature_store_helpers.py +++ b/ldclient/testing/test_async_feature_store_helpers.py @@ -4,6 +4,7 @@ from ldclient.async_feature_store_helpers import AsyncCachingStoreWrapper from ldclient.feature_store import CacheConfig +from ldclient.interfaces import AsyncFeatureStoreCore from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind # These tests exercise the caching-wrapper logic only, using an in-memory mock core, so they run @@ -18,12 +19,20 @@ def make_wrapper(core, cached): return AsyncCachingStoreWrapper(core, CacheConfig(expiration=30) if cached else CacheConfig.disabled()) -class MockAsyncCore: +class MockAsyncCore(AsyncFeatureStoreCore): def __init__(self): self.data = {} self.inited = False self.inited_query_count = 0 self.error = None + self.available = True + self.closed = False + + async def is_available(self): + return self.available + + async def close(self): + self.closed = True async def init_internal(self, all_data): self._maybe_throw() @@ -75,39 +84,37 @@ def force_remove(self, kind, key): items.pop(key, None) -class AvailableCore(MockAsyncCore): - def __init__(self, available): - super().__init__() - self._available = available - - async def is_available(self): - return self._available - - class CustomError(Exception): pass class TestAsyncCachingStoreWrapper: - @pytest.mark.parametrize("available", [False, True]) - def test_monitoring_enabled_if_available_is_defined(self, available: bool): - wrapper = make_wrapper(AvailableCore(available), False) - assert wrapper.is_monitoring_enabled() is True - - @pytest.mark.asyncio - @pytest.mark.parametrize("available", [False, True]) - async def test_is_available_reflects_core(self, available: bool): - wrapper = make_wrapper(AvailableCore(available), False) - assert await wrapper.is_available() is available - - def test_monitoring_not_enabled_if_available_is_not_defined(self): + def test_monitoring_is_enabled(self): wrapper = make_wrapper(MockAsyncCore(), False) - assert wrapper.is_monitoring_enabled() is False + assert wrapper.is_monitoring_enabled() @pytest.mark.asyncio - async def test_is_available_false_if_not_defined(self): - wrapper = make_wrapper(MockAsyncCore(), False) - assert await wrapper.is_available() is False + @pytest.mark.parametrize("available", [False, True]) + async def test_is_available_delegates_to_core(self, available: bool): + core = MockAsyncCore() + core.available = available + wrapper = make_wrapper(core, False) + assert await wrapper.is_available() == available + + @pytest.mark.parametrize("missing", ["is_available", "close"]) + def test_core_without_required_method_cannot_be_instantiated(self, missing: str): + methods = {name: getattr(MockAsyncCore, name) for name in ("is_available", "close")} + del methods[missing] + core_class = type("IncompleteCore", (AsyncFeatureStoreCore,), { + "get_internal": MockAsyncCore.get_internal, + "get_all_internal": MockAsyncCore.get_all_internal, + "init_internal": MockAsyncCore.init_internal, + "upsert_internal": MockAsyncCore.upsert_internal, + "initialized_internal": MockAsyncCore.initialized_internal, + **methods, + }) + with pytest.raises(TypeError): + core_class() @pytest.mark.asyncio @pytest.mark.parametrize("cached", [False, True]) @@ -341,16 +348,11 @@ async def test_initialized_after_init(self): assert wrapper.initialized is True @pytest.mark.asyncio - async def test_close_closes_core_if_supported(self): - closed = {"value": False} - - class ClosableCore(MockAsyncCore): - async def close(self): - closed["value"] = True - - wrapper = make_wrapper(ClosableCore(), True) + async def test_close_closes_core(self): + core = MockAsyncCore() + wrapper = make_wrapper(core, True) await wrapper.close() - assert closed["value"] is True + assert core.closed @pytest.mark.asyncio async def test_describe_configuration_delegates_to_core(self): diff --git a/ldclient/testing/test_async_in_memory_feature_store.py b/ldclient/testing/test_async_in_memory_feature_store.py index afed356b..abedeb9c 100644 --- a/ldclient/testing/test_async_in_memory_feature_store.py +++ b/ldclient/testing/test_async_in_memory_feature_store.py @@ -8,11 +8,12 @@ ) -def test_async_in_memory_status_checks(): +@pytest.mark.asyncio +async def test_async_in_memory_status_checks(): store = AsyncInMemoryFeatureStore() - assert store.is_monitoring_enabled() is False - assert store.is_available() is True + assert not store.is_monitoring_enabled() + assert await store.is_available() class AsyncInMemoryFeatureStoreTester(AsyncFeatureStoreTester): diff --git a/ldclient/testing/test_feature_store_helpers.py b/ldclient/testing/test_feature_store_helpers.py index e4105d56..610110e6 100644 --- a/ldclient/testing/test_feature_store_helpers.py +++ b/ldclient/testing/test_feature_store_helpers.py @@ -1,10 +1,10 @@ from time import sleep -from unittest.mock import Mock import pytest from ldclient.feature_store import CacheConfig from ldclient.feature_store_helpers import CachingStoreWrapper +from ldclient.interfaces import FeatureStoreCore from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind THINGS = VersionedDataKind(namespace="things", request_api_path="", stream_api_path="") @@ -15,12 +15,20 @@ def make_wrapper(core, cached): return CachingStoreWrapper(core, CacheConfig(expiration=30) if cached else CacheConfig.disabled()) -class MockCore: +class MockCore(FeatureStoreCore): def __init__(self): self.data = {} self.inited = False self.inited_query_count = 0 self.error = None + self.available = True + self.closed = False + + def is_available(self): + return self.available + + def close(self): + self.closed = True def init_internal(self, all_data): self._maybe_throw() @@ -77,20 +85,37 @@ class CustomError(Exception): class TestCachingStoreWrapper: - @pytest.mark.parametrize("available", [False, True]) - def test_monitoring_enabled_if_available_is_defined(self, available: bool): - core = Mock() - core.is_available = lambda: available + def test_monitoring_is_enabled(self): + wrapper = make_wrapper(MockCore(), False) + assert wrapper.is_monitoring_enabled() + @pytest.mark.parametrize("available", [False, True]) + def test_is_available_delegates_to_core(self, available: bool): + core = MockCore() + core.available = available wrapper = make_wrapper(core, False) + assert wrapper.is_available() == available - assert wrapper.is_monitoring_enabled() is True - assert wrapper.is_available() is available - - @pytest.mark.parametrize("available", [False, True]) - def test_monitoring_not_enabled_if_available_is_not_defined(self, available: bool): - wrapper = make_wrapper(MockCore(), False) - assert wrapper.is_monitoring_enabled() is False + def test_close_closes_core(self): + core = MockCore() + wrapper = make_wrapper(core, False) + wrapper.close() + assert core.closed + + @pytest.mark.parametrize("missing", ["is_available", "close"]) + def test_core_without_required_method_cannot_be_instantiated(self, missing: str): + methods = {name: getattr(MockCore, name) for name in ("is_available", "close")} + del methods[missing] + core_class = type("IncompleteCore", (FeatureStoreCore,), { + "get_internal": MockCore.get_internal, + "get_all_internal": MockCore.get_all_internal, + "init_internal": MockCore.init_internal, + "upsert_internal": MockCore.upsert_internal, + "initialized_internal": MockCore.initialized_internal, + **methods, + }) + with pytest.raises(TypeError): + core_class() @pytest.mark.parametrize("cached", [False, True]) def test_get_item(self, cached): diff --git a/ldclient/testing/test_feature_store_interfaces.py b/ldclient/testing/test_feature_store_interfaces.py new file mode 100644 index 00000000..3c3a1d5c --- /dev/null +++ b/ldclient/testing/test_feature_store_interfaces.py @@ -0,0 +1,94 @@ +import pytest + +from ldclient.interfaces import AsyncFeatureStore, FeatureStore + +STATUS_METHODS = ["is_monitoring_enabled", "is_available", "close"] + + +class CompleteStore(FeatureStore): + def get(self, kind, key, callback=lambda x: x): + return callback(None) + + def all(self, kind, callback=lambda x: x): + return callback({}) + + def init(self, all_data): + pass + + def delete(self, kind, key, version): + return False + + def upsert(self, kind, item): + return False + + @property + def initialized(self): + return True + + def is_monitoring_enabled(self): + return False + + def is_available(self): + return True + + def close(self): + pass + + +class CompleteAsyncStore(AsyncFeatureStore): + async def get(self, kind, key): + return None + + async def all(self, kind): + return {} + + async def init(self, all_data): + pass + + async def upsert(self, kind, item): + return False + + async def delete(self, kind, key, version): + return False + + @property + def initialized(self): + return True + + async def is_initialized(self): + return True + + def is_monitoring_enabled(self): + return False + + async def is_available(self): + return True + + async def close(self): + pass + + +def _without(base, interface, missing): + # Copy every method of the complete store except the missing one. The copy + # subclasses the interface directly, so the interface sees the gap. + members = {name: value for name, value in vars(base).items() if not name.startswith("__") and name != missing} + return type("IncompleteStore", (interface,), members) + + +def test_complete_store_can_be_instantiated(): + CompleteStore() + CompleteAsyncStore() + + +@pytest.mark.parametrize("missing", STATUS_METHODS) +def test_store_without_status_method_cannot_be_instantiated(missing): + store_class = _without(CompleteStore, FeatureStore, missing) + with pytest.raises(TypeError): + store_class() + + +@pytest.mark.parametrize("missing", STATUS_METHODS) +def test_async_store_without_status_method_cannot_be_instantiated(missing): + store_class = _without(CompleteAsyncStore, AsyncFeatureStore, missing) + with pytest.raises(TypeError): + store_class() diff --git a/ldclient/testing/test_in_memory_feature_store.py b/ldclient/testing/test_in_memory_feature_store.py index 093f9e88..2c83de2a 100644 --- a/ldclient/testing/test_in_memory_feature_store.py +++ b/ldclient/testing/test_in_memory_feature_store.py @@ -11,8 +11,8 @@ def test_in_memory_status_checks(): store = InMemoryFeatureStore() - assert store.is_monitoring_enabled() is False - assert store.is_available() is True + assert not store.is_monitoring_enabled() + assert store.is_available() class InMemoryFeatureStoreTester(FeatureStoreTester): diff --git a/ldclient/testing/test_ldclient_evaluation.py b/ldclient/testing/test_ldclient_evaluation.py index 75605ee8..4411ac0f 100644 --- a/ldclient/testing/test_ldclient_evaluation.py +++ b/ldclient/testing/test_ldclient_evaluation.py @@ -38,6 +38,15 @@ def init(self, data): def initialized(self): return True + def is_monitoring_enabled(self): + return False + + def is_available(self): + return True + + def close(self): + pass + def get_log_lines(caplog, level): loglines = caplog.records @@ -215,6 +224,15 @@ def init(self, data): def initialized(self): return True + def is_monitoring_enabled(self): + return False + + def is_available(self): + return True + + def close(self): + pass + def test_all_flags_state_decodes_dicts_from_custom_store(): # A custom store that hands back raw dicts must still yield decoded values diff --git a/ldclient/testing/test_sync_async_parity.py b/ldclient/testing/test_sync_async_parity.py index d1e42042..69f670a7 100644 --- a/ldclient/testing/test_sync_async_parity.py +++ b/ldclient/testing/test_sync_async_parity.py @@ -47,7 +47,7 @@ def _public_surface(cls) -> set: set(), # is_initialized is an awaitable readiness check; the sync store exposes # readiness through the synchronous ``initialized`` property instead. - {"close", "is_initialized"}, + {"is_initialized"}, id="feature_store", ), pytest.param(Evaluator, AsyncEvaluator, set(), set(), id="evaluator"), From a0047f91aadd473144a44c4568ffc13cd148066a Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Mon, 2 Nov 2026 15:43:19 -0500 Subject: [PATCH 2/5] fix: Close the feature store when an FDv1 client closes FDv1 stop now closes the feature store, as FDv2 does, so persistent stores release their connections when the client closes. The sync store wrapper also records that it is closed, so a store failure after close does not start a new availability poller. --- ldclient/impl/datasystem/async_fdv1.py | 1 + ldclient/impl/datasystem/fdv1.py | 8 ++- .../testing/impl/datasystem/test_fdv1_stop.py | 71 +++++++++++++++++++ .../test_feature_store_client_wrapper.py | 27 +++++++ 4 files changed, 106 insertions(+), 1 deletion(-) create mode 100644 ldclient/testing/impl/datasystem/test_fdv1_stop.py diff --git a/ldclient/impl/datasystem/async_fdv1.py b/ldclient/impl/datasystem/async_fdv1.py index 96fbc660..ed5fc86b 100644 --- a/ldclient/impl/datasystem/async_fdv1.py +++ b/ldclient/impl/datasystem/async_fdv1.py @@ -119,6 +119,7 @@ def start(self, set_on_ready: AsyncEvent): async def stop(self): if self._update_processor is not None: await self._update_processor.stop() + await self._store.close() @property def store(self) -> AsyncReadOnlyStore: diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index affc06ec..67236f35 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -51,6 +51,7 @@ def __init__(self, store: FeatureStore, store_update_sink: DataStoreUpdateSink): self.__lock = ReadWriteLock() self.__last_available = True self.__poller: Optional[RepeatingTask] = None + self.__closed = False 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))) @@ -81,7 +82,7 @@ def __wrapper(self, fn: Callable): def __update_availability(self, available: bool): with self.__lock.write(): - if available == self.__last_available: + if self.__closed or available == self.__last_available: return self.__last_available = available @@ -104,6 +105,9 @@ def __update_availability(self, available: bool): task = RepeatingTask.at_interval("ldclient.check-availability", 0.5, 0, self.__check_availability) with self.__lock.write(): + # Do not start a poller after close() has run. + if self.__closed: + return self.__poller = task self.__poller.start() @@ -125,6 +129,7 @@ def is_available(self) -> bool: def close(self) -> None: """Stops the availability poller and closes the wrapped store.""" with self.__lock.write(): + self.__closed = True poller_to_stop = self.__poller self.__poller = None @@ -216,6 +221,7 @@ def start(self, set_on_ready: Event): def stop(self): if self._update_processor is not None: self._update_processor.stop() + self._store_wrapper.close() @property def store(self) -> ReadOnlyStore: diff --git a/ldclient/testing/impl/datasystem/test_fdv1_stop.py b/ldclient/testing/impl/datasystem/test_fdv1_stop.py new file mode 100644 index 00000000..2962bfe1 --- /dev/null +++ b/ldclient/testing/impl/datasystem/test_fdv1_stop.py @@ -0,0 +1,71 @@ +# pylint: disable=missing-docstring + +from threading import Event + +import pytest + +from ldclient.async_config import AsyncConfig +from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from ldclient.config import Config +from ldclient.feature_store import InMemoryFeatureStore +from ldclient.impl.aio.concurrency import AsyncEvent +from ldclient.impl.datasystem.async_fdv1 import AsyncFDv1 +from ldclient.impl.datasystem.fdv1 import FDv1 + + +class _ClosingStore(InMemoryFeatureStore): + def __init__(self): + super().__init__() + self.close_count = 0 + + def close(self) -> None: + self.close_count += 1 + + +class _AsyncClosingStore(AsyncInMemoryFeatureStore): + def __init__(self): + super().__init__() + self.close_count = 0 + + async def close(self) -> None: + self.close_count += 1 + + +def test_stop_closes_the_feature_store(): + store = _ClosingStore() + data_system = FDv1(Config(sdk_key="key", offline=True, feature_store=store)) + data_system.start(Event()) + + data_system.stop() + + assert store.close_count == 1 + + +def test_stop_closes_the_feature_store_when_not_started(): + store = _ClosingStore() + data_system = FDv1(Config(sdk_key="key", offline=True, feature_store=store)) + + data_system.stop() + + assert store.close_count == 1 + + +@pytest.mark.asyncio +async def test_async_stop_closes_the_feature_store(): + store = _AsyncClosingStore() + data_system = AsyncFDv1(AsyncConfig(sdk_key="key", offline=True), store, lambda: None) + data_system.start(AsyncEvent()) + + await data_system.stop() + + assert store.close_count == 1 + + +@pytest.mark.asyncio +async def test_async_stop_closes_the_feature_store_when_not_started(): + store = _AsyncClosingStore() + data_system = AsyncFDv1(AsyncConfig(sdk_key="key", offline=True), store, lambda: None) + + await data_system.stop() + + assert store.close_count == 1 diff --git a/ldclient/testing/test_feature_store_client_wrapper.py b/ldclient/testing/test_feature_store_client_wrapper.py index 8ffe7c38..a9746dcb 100644 --- a/ldclient/testing/test_feature_store_client_wrapper.py +++ b/ldclient/testing/test_feature_store_client_wrapper.py @@ -103,3 +103,30 @@ def set_event(status): assert len(statuses) == 2 assert statuses[0].available is False assert statuses[1].available is True + + +def test_store_failure_after_close_does_not_report_or_poll(): + store = Mock() + store.is_monitoring_enabled = lambda: True + store.is_available = Mock(return_value=False) + store.init = raise_an_error + + listener = RecordStatusListener() + listeners = Listeners() + listeners.add(listener) + sink = DataStoreUpdateSinkImpl(listeners) + + wrapper = _FeatureStoreClientWrapper(store, sink) + wrapper.close() + try: + wrapper.init({}) + raise Exception("init should have raised an exception") + except BaseException: + pass + + # The poller checks every 0.5 seconds, so this wait covers one check. + Event().wait(0.7) + + assert len(listener.statuses) == 0 + store.is_available.assert_not_called() + store.close.assert_called_once() From 34d5902e6b3b47a304c3685fc97012c917bbdb07 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 3 Nov 2026 08:59:58 -0500 Subject: [PATCH 3/5] fix: Wait for FDv1 data source threads before closing the store --- ldclient/impl/datasource/polling.py | 11 +++-- ldclient/impl/datasource/streaming.py | 14 +++++- ldclient/impl/repeating_task.py | 16 ++++++- .../impl/datasource/test_polling_processor.py | 33 +++++++++++++ .../testing/impl/datasource/test_streaming.py | 48 +++++++++++++++++++ ldclient/testing/impl/test_repeating_task.py | 37 ++++++++++++++ 6 files changed, 154 insertions(+), 5 deletions(-) diff --git a/ldclient/impl/datasource/polling.py b/ldclient/impl/datasource/polling.py index 2eacc6ad..1ad1fae3 100644 --- a/ldclient/impl/datasource/polling.py +++ b/ldclient/impl/datasource/polling.py @@ -35,6 +35,9 @@ UpdateProcessor ) +# Seconds that stop() waits for the worker thread to finish. +_STOP_TIMEOUT = 5.0 + @runtime_checkable class _FeatureRequesterWithHeaders(Protocol): @@ -70,10 +73,12 @@ def stop(self): log.info("Stopping PollingUpdateProcessor") self._task.stop() - if self._data_source_update_sink is None: - return + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.OFF, None) - self._data_source_update_sink.update_status(DataSourceState.OFF, None) + # Wait for an in-flight poll, so it cannot write to a closed store. + if not self._task.join(_STOP_TIMEOUT): + log.warning("Polling thread did not stop within %s seconds", _STOP_TIMEOUT) def _poll(self) -> None: """Makes one poll request and records the outcome on the retry state.""" diff --git a/ldclient/impl/datasource/streaming.py b/ldclient/impl/datasource/streaming.py index 9e425e3f..b3999413 100644 --- a/ldclient/impl/datasource/streaming.py +++ b/ldclient/impl/datasource/streaming.py @@ -2,7 +2,7 @@ import time from threading import TIMEOUT_MAX from threading import Event as ThreadEvent -from threading import Thread +from threading import Thread, current_thread from typing import Callable, Optional from urllib import parse @@ -43,6 +43,10 @@ stream_read_timeout = 5 * 60 +# Seconds that stop() waits for the stream thread to finish. +_STOP_TIMEOUT = 5.0 + + class StreamingUpdateProcessor(Thread, UpdateProcessor): """Reads flag data from LaunchDarkly's streaming endpoint on its own thread. @@ -173,6 +177,14 @@ def stop(self): if self._sse: self._sse.close() + # Wait for the stream thread, so it cannot write to a closed store. A + # call from the stream thread itself, such as from a listener, does + # not wait. + if self.is_alive() and current_thread() is not self: + self.join(_STOP_TIMEOUT) + if self.is_alive(): + log.warning("Streaming thread did not stop within %s seconds", _STOP_TIMEOUT) + def _interrupt_stream(self): """Drops the stream connection so the next read reconnects. The SSE client reports the close as a Fault with no error, and the flag tells diff --git a/ldclient/impl/repeating_task.py b/ldclient/impl/repeating_task.py index c8c1fd62..b4c2a8e2 100644 --- a/ldclient/impl/repeating_task.py +++ b/ldclient/impl/repeating_task.py @@ -1,4 +1,4 @@ -from threading import TIMEOUT_MAX, Event, Thread +from threading import TIMEOUT_MAX, Event, Thread, current_thread from typing import Any, Callable from ldclient.impl.delay import DelaySource, FixedDelay @@ -64,6 +64,20 @@ def stop(self): """ self.__stop.set() + def join(self, timeout: float) -> bool: + """ + Waits up to ``timeout`` seconds for the worker thread to finish. + + Call :meth:`stop` first. A call from the worker thread itself does not + wait, so a callback can stop its own task safely. + + :return: True if the worker thread is not running when the call returns + """ + if not self.__started or current_thread() is self.__thread: + return True + self.__thread.join(timeout) + return not self.__thread.is_alive() + def _run(self): if self.__initial_delay > 0: if self.__stop.wait(min(self.__initial_delay, TIMEOUT_MAX)): diff --git a/ldclient/testing/impl/datasource/test_polling_processor.py b/ldclient/testing/impl/datasource/test_polling_processor.py index 913438a4..53a78302 100644 --- a/ldclient/testing/impl/datasource/test_polling_processor.py +++ b/ldclient/testing/impl/datasource/test_polling_processor.py @@ -399,6 +399,39 @@ def test_a_poll_finishing_after_stop_reports_nothing(): assert [status.state for status in spy.statuses] == [DataSourceState.OFF] +class BlockingFeatureRequester(MockFeatureRequester): + """Blocks each request until the test releases it.""" + + def __init__(self): + super().__init__() + self.entered = threading.Event() + self.release = threading.Event() + + def get_all_data(self): + self.entered.set() + self.release.wait(5) + return super().get_all_data() + + +def test_stop_waits_for_an_in_flight_poll(): + """stop() must not return while a poll can still write to the store.""" + requester = BlockingFeatureRequester() + requester.all_data = {FEATURES: {}, SEGMENTS: {}} + processor = PollingUpdateProcessor(Config("SDK_KEY"), requester, store, ready) + processor.start() + assert requester.entered.wait(2) + + stopper = threading.Thread(target=processor.stop) + stopper.start() + stopper.join(0.2) + assert stopper.is_alive() + + requester.release.set() + stopper.join(2) + assert not stopper.is_alive() + assert processor._task.join(0) is True + + def test_valid_status_is_reported_before_ready_is_set(): # Mirrors go-server-sdk#442: a caller that wakes on readiness must not # still be able to read INITIALIZING. diff --git a/ldclient/testing/impl/datasource/test_streaming.py b/ldclient/testing/impl/datasource/test_streaming.py index 0fa3b803..0e6eb341 100644 --- a/ldclient/testing/impl/datasource/test_streaming.py +++ b/ldclient/testing/impl/datasource/test_streaming.py @@ -534,6 +534,54 @@ def test_stop_before_start_does_not_raise(): sp.stop() +def test_stop_waits_for_the_stream_thread(): + store = InMemoryFeatureStore() + ready = Event() + + with start_server() as server: + with stream_content(make_put_event()) as stream: + config = Config(sdk_key='sdk-key', stream_uri=server.uri) + server.for_path('/all', stream) + + sp = StreamingUpdateProcessor(config, store, ready, None) + sp.start() + assert ready.wait(start_wait) + + sp.stop() + + assert not sp.is_alive() + + +def test_stop_from_a_listener_on_the_stream_thread_does_not_wait(): + store = InMemoryFeatureStore() + ready = Event() + elapsed: List[float] = [] + sp = None + + def stop_on_valid(status): + if status.state == DataSourceState.VALID: + started = time.monotonic() + sp.stop() + elapsed.append(time.monotonic() - started) + + listeners = Listeners() + listeners.add(stop_on_valid) + + with start_server() as server: + with stream_content(make_put_event()) as stream: + config = Config(sdk_key='sdk-key', stream_uri=server.uri) + config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) + server.for_path('/all', stream) + + sp = StreamingUpdateProcessor(config, store, ready, None) + sp.start() + wait_until(lambda: len(elapsed) > 0, start_wait) + sp.join(start_wait) + + assert elapsed[0] < 1 + assert not sp.is_alive() + + def test_a_stop_before_the_connection_exists_still_ends_the_run(): """stop() has nothing to close when run() has not built the client yet, so the run itself must not go on to read a connection nobody will close.""" diff --git a/ldclient/testing/impl/test_repeating_task.py b/ldclient/testing/impl/test_repeating_task.py index 979652c4..6f0e5dc3 100644 --- a/ldclient/testing/impl/test_repeating_task.py +++ b/ldclient/testing/impl/test_repeating_task.py @@ -182,3 +182,40 @@ def do_task(): assert counter == 2 finally: task.stop() + + +def test_join_waits_for_a_running_callback(): + entered = Event() + release = Event() + task = RepeatingTask.at_interval("ldclient.testing.block", 0.01, 0, lambda: (entered.set(), release.wait(5))) + task.start() + assert entered.wait(2) + + task.stop() + assert task.join(0.1) is False + + release.set() + assert task.join(2) is True + + +def test_join_before_start_does_not_wait(): + task = RepeatingTask.at_interval("ldclient.testing.never-started", 0.01, 0, lambda: None) + task.stop() + + assert task.join(5) is True + + +def test_join_from_the_callback_does_not_wait(): + results: Queue = Queue() + + def stop_self(): + task.stop() + started = time.monotonic() + results.put((task.join(5), time.monotonic() - started)) + + task = RepeatingTask.at_interval("ldclient.testing.stop-self", 0.01, 0, stop_self) + task.start() + + joined, elapsed = results.get(timeout=2) + assert joined is True + assert elapsed < 1 From 30132578b54b24122d2f3300296e63472d33156a Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 3 Nov 2026 09:38:29 -0500 Subject: [PATCH 4/5] fix: Prevent FDv1 writes to the feature store after the client closes --- ldclient/impl/datasource/polling.py | 11 +--- ldclient/impl/datasource/streaming.py | 14 +---- ldclient/impl/datasystem/fdv1.py | 35 +++++++++--- ldclient/impl/repeating_task.py | 16 +----- .../impl/datasource/test_polling_processor.py | 33 ----------- .../testing/impl/datasource/test_streaming.py | 48 ---------------- ldclient/testing/impl/test_repeating_task.py | 37 ------------ .../test_feature_store_client_wrapper.py | 57 ++++++++++++++----- 8 files changed, 75 insertions(+), 176 deletions(-) diff --git a/ldclient/impl/datasource/polling.py b/ldclient/impl/datasource/polling.py index 1ad1fae3..2eacc6ad 100644 --- a/ldclient/impl/datasource/polling.py +++ b/ldclient/impl/datasource/polling.py @@ -35,9 +35,6 @@ UpdateProcessor ) -# Seconds that stop() waits for the worker thread to finish. -_STOP_TIMEOUT = 5.0 - @runtime_checkable class _FeatureRequesterWithHeaders(Protocol): @@ -73,12 +70,10 @@ def stop(self): log.info("Stopping PollingUpdateProcessor") self._task.stop() - if self._data_source_update_sink is not None: - self._data_source_update_sink.update_status(DataSourceState.OFF, None) + if self._data_source_update_sink is None: + return - # Wait for an in-flight poll, so it cannot write to a closed store. - if not self._task.join(_STOP_TIMEOUT): - log.warning("Polling thread did not stop within %s seconds", _STOP_TIMEOUT) + self._data_source_update_sink.update_status(DataSourceState.OFF, None) def _poll(self) -> None: """Makes one poll request and records the outcome on the retry state.""" diff --git a/ldclient/impl/datasource/streaming.py b/ldclient/impl/datasource/streaming.py index b3999413..9e425e3f 100644 --- a/ldclient/impl/datasource/streaming.py +++ b/ldclient/impl/datasource/streaming.py @@ -2,7 +2,7 @@ import time from threading import TIMEOUT_MAX from threading import Event as ThreadEvent -from threading import Thread, current_thread +from threading import Thread from typing import Callable, Optional from urllib import parse @@ -43,10 +43,6 @@ stream_read_timeout = 5 * 60 -# Seconds that stop() waits for the stream thread to finish. -_STOP_TIMEOUT = 5.0 - - class StreamingUpdateProcessor(Thread, UpdateProcessor): """Reads flag data from LaunchDarkly's streaming endpoint on its own thread. @@ -177,14 +173,6 @@ def stop(self): if self._sse: self._sse.close() - # Wait for the stream thread, so it cannot write to a closed store. A - # call from the stream thread itself, such as from a listener, does - # not wait. - if self.is_alive() and current_thread() is not self: - self.join(_STOP_TIMEOUT) - if self.is_alive(): - log.warning("Streaming thread did not stop within %s seconds", _STOP_TIMEOUT) - def _interrupt_stream(self): """Drops the stream connection so the next read reconnects. The SSE client reports the close as a Fault with no error, and the flag tells diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index 67236f35..74684fff 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -1,4 +1,4 @@ -from threading import Event +from threading import Event, Lock from typing import Any, Callable, Dict, Mapping, Optional from ldclient.config import Config @@ -47,6 +47,10 @@ def __init__(self, store: FeatureStore, store_update_sink: DataStoreUpdateSink): self.__store_update_sink = store_update_sink self.__monitoring_enabled = self.is_monitoring_enabled() + # Held for each write and by close(). A write that starts after + # close() does nothing, and close() waits for a write in progress. + self.__write_lock = Lock() + # Covers the following variables self.__lock = ReadWriteLock() self.__last_available = True @@ -54,7 +58,10 @@ def __init__(self, store: FeatureStore, store_update_sink: DataStoreUpdateSink): self.__closed = False 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))) + with self.__write_lock: + if self.__closed: + 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)) @@ -63,10 +70,16 @@ def all(self, kind, callback): return self.__wrapper(lambda: self.store.all(kind, callback)) def delete(self, kind, key, version) -> bool: - return self.__wrapper(lambda: self.store.delete(kind, key, version)) + with self.__write_lock: + if self.__closed: + return False + return self.__wrapper(lambda: self.store.delete(kind, key, version)) def upsert(self, kind, item) -> bool: - return self.__wrapper(lambda: self.store.upsert(kind, item)) + with self.__write_lock: + if self.__closed: + return False + return self.__wrapper(lambda: self.store.upsert(kind, item)) @property def initialized(self) -> bool: @@ -127,11 +140,15 @@ def is_available(self) -> bool: return self.store.is_available() def close(self) -> None: - """Stops the availability poller and closes the wrapped store.""" - with self.__lock.write(): - self.__closed = True - poller_to_stop = self.__poller - self.__poller = None + """Stops the availability poller and closes the wrapped store. + + Waits for a write in progress. Later writes do nothing. + """ + with self.__write_lock: + with self.__lock.write(): + self.__closed = True + poller_to_stop = self.__poller + self.__poller = None if poller_to_stop is not None: poller_to_stop.stop() diff --git a/ldclient/impl/repeating_task.py b/ldclient/impl/repeating_task.py index b4c2a8e2..c8c1fd62 100644 --- a/ldclient/impl/repeating_task.py +++ b/ldclient/impl/repeating_task.py @@ -1,4 +1,4 @@ -from threading import TIMEOUT_MAX, Event, Thread, current_thread +from threading import TIMEOUT_MAX, Event, Thread from typing import Any, Callable from ldclient.impl.delay import DelaySource, FixedDelay @@ -64,20 +64,6 @@ def stop(self): """ self.__stop.set() - def join(self, timeout: float) -> bool: - """ - Waits up to ``timeout`` seconds for the worker thread to finish. - - Call :meth:`stop` first. A call from the worker thread itself does not - wait, so a callback can stop its own task safely. - - :return: True if the worker thread is not running when the call returns - """ - if not self.__started or current_thread() is self.__thread: - return True - self.__thread.join(timeout) - return not self.__thread.is_alive() - def _run(self): if self.__initial_delay > 0: if self.__stop.wait(min(self.__initial_delay, TIMEOUT_MAX)): diff --git a/ldclient/testing/impl/datasource/test_polling_processor.py b/ldclient/testing/impl/datasource/test_polling_processor.py index 53a78302..913438a4 100644 --- a/ldclient/testing/impl/datasource/test_polling_processor.py +++ b/ldclient/testing/impl/datasource/test_polling_processor.py @@ -399,39 +399,6 @@ def test_a_poll_finishing_after_stop_reports_nothing(): assert [status.state for status in spy.statuses] == [DataSourceState.OFF] -class BlockingFeatureRequester(MockFeatureRequester): - """Blocks each request until the test releases it.""" - - def __init__(self): - super().__init__() - self.entered = threading.Event() - self.release = threading.Event() - - def get_all_data(self): - self.entered.set() - self.release.wait(5) - return super().get_all_data() - - -def test_stop_waits_for_an_in_flight_poll(): - """stop() must not return while a poll can still write to the store.""" - requester = BlockingFeatureRequester() - requester.all_data = {FEATURES: {}, SEGMENTS: {}} - processor = PollingUpdateProcessor(Config("SDK_KEY"), requester, store, ready) - processor.start() - assert requester.entered.wait(2) - - stopper = threading.Thread(target=processor.stop) - stopper.start() - stopper.join(0.2) - assert stopper.is_alive() - - requester.release.set() - stopper.join(2) - assert not stopper.is_alive() - assert processor._task.join(0) is True - - def test_valid_status_is_reported_before_ready_is_set(): # Mirrors go-server-sdk#442: a caller that wakes on readiness must not # still be able to read INITIALIZING. diff --git a/ldclient/testing/impl/datasource/test_streaming.py b/ldclient/testing/impl/datasource/test_streaming.py index 0e6eb341..0fa3b803 100644 --- a/ldclient/testing/impl/datasource/test_streaming.py +++ b/ldclient/testing/impl/datasource/test_streaming.py @@ -534,54 +534,6 @@ def test_stop_before_start_does_not_raise(): sp.stop() -def test_stop_waits_for_the_stream_thread(): - store = InMemoryFeatureStore() - ready = Event() - - with start_server() as server: - with stream_content(make_put_event()) as stream: - config = Config(sdk_key='sdk-key', stream_uri=server.uri) - server.for_path('/all', stream) - - sp = StreamingUpdateProcessor(config, store, ready, None) - sp.start() - assert ready.wait(start_wait) - - sp.stop() - - assert not sp.is_alive() - - -def test_stop_from_a_listener_on_the_stream_thread_does_not_wait(): - store = InMemoryFeatureStore() - ready = Event() - elapsed: List[float] = [] - sp = None - - def stop_on_valid(status): - if status.state == DataSourceState.VALID: - started = time.monotonic() - sp.stop() - elapsed.append(time.monotonic() - started) - - listeners = Listeners() - listeners.add(stop_on_valid) - - with start_server() as server: - with stream_content(make_put_event()) as stream: - config = Config(sdk_key='sdk-key', stream_uri=server.uri) - config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners()) - server.for_path('/all', stream) - - sp = StreamingUpdateProcessor(config, store, ready, None) - sp.start() - wait_until(lambda: len(elapsed) > 0, start_wait) - sp.join(start_wait) - - assert elapsed[0] < 1 - assert not sp.is_alive() - - def test_a_stop_before_the_connection_exists_still_ends_the_run(): """stop() has nothing to close when run() has not built the client yet, so the run itself must not go on to read a connection nobody will close.""" diff --git a/ldclient/testing/impl/test_repeating_task.py b/ldclient/testing/impl/test_repeating_task.py index 6f0e5dc3..979652c4 100644 --- a/ldclient/testing/impl/test_repeating_task.py +++ b/ldclient/testing/impl/test_repeating_task.py @@ -182,40 +182,3 @@ def do_task(): assert counter == 2 finally: task.stop() - - -def test_join_waits_for_a_running_callback(): - entered = Event() - release = Event() - task = RepeatingTask.at_interval("ldclient.testing.block", 0.01, 0, lambda: (entered.set(), release.wait(5))) - task.start() - assert entered.wait(2) - - task.stop() - assert task.join(0.1) is False - - release.set() - assert task.join(2) is True - - -def test_join_before_start_does_not_wait(): - task = RepeatingTask.at_interval("ldclient.testing.never-started", 0.01, 0, lambda: None) - task.stop() - - assert task.join(5) is True - - -def test_join_from_the_callback_does_not_wait(): - results: Queue = Queue() - - def stop_self(): - task.stop() - started = time.monotonic() - results.put((task.join(5), time.monotonic() - started)) - - task = RepeatingTask.at_interval("ldclient.testing.stop-self", 0.01, 0, stop_self) - task.start() - - joined, elapsed = results.get(timeout=2) - assert joined is True - assert elapsed < 1 diff --git a/ldclient/testing/test_feature_store_client_wrapper.py b/ldclient/testing/test_feature_store_client_wrapper.py index a9746dcb..29f87847 100644 --- a/ldclient/testing/test_feature_store_client_wrapper.py +++ b/ldclient/testing/test_feature_store_client_wrapper.py @@ -1,10 +1,11 @@ -from threading import Event +from threading import Event, Thread from typing import Callable, List from unittest.mock import Mock from ldclient.impl.datastore.status import DataStoreUpdateSinkImpl from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper from ldclient.impl.listeners import Listeners +from ldclient.versioned_data_kind import FEATURES class CallbackListener: @@ -105,11 +106,9 @@ def set_event(status): assert statuses[1].available is True -def test_store_failure_after_close_does_not_report_or_poll(): +def test_writes_after_close_do_not_reach_the_store(): store = Mock() store.is_monitoring_enabled = lambda: True - store.is_available = Mock(return_value=False) - store.init = raise_an_error listener = RecordStatusListener() listeners = Listeners() @@ -118,15 +117,47 @@ def test_store_failure_after_close_does_not_report_or_poll(): wrapper = _FeatureStoreClientWrapper(store, sink) wrapper.close() - try: - wrapper.init({}) - raise Exception("init should have raised an exception") - except BaseException: - pass - # The poller checks every 0.5 seconds, so this wait covers one check. - Event().wait(0.7) + wrapper.init({}) + assert wrapper.upsert(FEATURES, {'key': 'flag', 'version': 1}) is False + assert wrapper.delete(FEATURES, 'flag', 2) is False - assert len(listener.statuses) == 0 - store.is_available.assert_not_called() + store.init.assert_not_called() + store.upsert.assert_not_called() + store.delete.assert_not_called() store.close.assert_called_once() + assert len(listener.statuses) == 0 + + +def test_close_waits_for_a_write_in_progress(): + entered = Event() + release = Event() + calls: List[str] = [] + + def slow_upsert(kind, item): + entered.set() + release.wait(5) + calls.append('upsert') + return True + + store = Mock() + store.is_monitoring_enabled = lambda: True + store.upsert = slow_upsert + store.close = lambda: calls.append('close') + + wrapper = _FeatureStoreClientWrapper(store, DataStoreUpdateSinkImpl(Listeners())) + + writer = Thread(target=lambda: wrapper.upsert(FEATURES, {'key': 'flag', 'version': 1})) + writer.start() + assert entered.wait(2) + + closer = Thread(target=wrapper.close) + closer.start() + closer.join(0.2) + assert closer.is_alive() + + release.set() + closer.join(2) + writer.join(2) + assert not closer.is_alive() + assert calls == ['upsert', 'close'] From 6c452a001659126583509aa3bdc2a42e1fdcb247 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Tue, 3 Nov 2026 10:03:01 -0500 Subject: [PATCH 5/5] fix: Report FDv1 store write failures after releasing the write lock --- ldclient/impl/datasystem/fdv1.py | 28 ++++++++------- .../test_feature_store_client_wrapper.py | 34 +++++++++++++++++++ 2 files changed, 50 insertions(+), 12 deletions(-) diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index 74684fff..5052d811 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -58,10 +58,7 @@ def __init__(self, store: FeatureStore, store_update_sink: DataStoreUpdateSink): self.__closed = False def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]): - with self.__write_lock: - if self.__closed: - return - self.__wrapper(lambda: self.store.init(_FeatureStoreDataSetSorter.sort_all_collections(all_data))) + self.__write(lambda: self.store.init(_FeatureStoreDataSetSorter.sort_all_collections(all_data)), None) def get(self, kind, key, callback): return self.__wrapper(lambda: self.store.get(kind, key, callback)) @@ -70,16 +67,10 @@ def all(self, kind, callback): return self.__wrapper(lambda: self.store.all(kind, callback)) def delete(self, kind, key, version) -> bool: - with self.__write_lock: - if self.__closed: - return False - return self.__wrapper(lambda: self.store.delete(kind, key, version)) + return self.__write(lambda: self.store.delete(kind, key, version), False) def upsert(self, kind, item) -> bool: - with self.__write_lock: - if self.__closed: - return False - return self.__wrapper(lambda: self.store.upsert(kind, item)) + return self.__write(lambda: self.store.upsert(kind, item), False) @property def initialized(self) -> bool: @@ -93,6 +84,19 @@ def __wrapper(self, fn: Callable): self.__update_availability(False) raise + def __write(self, fn: Callable, result_if_closed: Any): + # The failure is reported after the lock is released. A status listener can then + # write or close the client without a deadlock. + try: + with self.__write_lock: + if self.__closed: + return result_if_closed + return fn() + except BaseException: + if self.__monitoring_enabled: + self.__update_availability(False) + raise + def __update_availability(self, available: bool): with self.__lock.write(): if self.__closed or available == self.__last_available: diff --git a/ldclient/testing/test_feature_store_client_wrapper.py b/ldclient/testing/test_feature_store_client_wrapper.py index 29f87847..ed744378 100644 --- a/ldclient/testing/test_feature_store_client_wrapper.py +++ b/ldclient/testing/test_feature_store_client_wrapper.py @@ -161,3 +161,37 @@ def slow_upsert(kind, item): writer.join(2) assert not closer.is_alive() assert calls == ['upsert', 'close'] + + +def test_a_listener_can_close_the_wrapper_after_a_failed_write(): + store = Mock() + store.is_monitoring_enabled = lambda: True + store.is_available = Mock(return_value=False) + store.init = raise_an_error + + wrapper = None + closed = Event() + + def close_on_failure(status): + if not status.available: + wrapper.close() + closed.set() + + listeners = Listeners() + listeners.add(CallbackListener(close_on_failure)) + wrapper = _FeatureStoreClientWrapper(store, DataStoreUpdateSinkImpl(listeners)) + + writer = Thread(target=lambda: _ignore_errors(lambda: wrapper.init({})), daemon=True) + writer.start() + writer.join(2) + + assert not writer.is_alive() + assert closed.is_set() + store.close.assert_called_once() + + +def _ignore_errors(fn): + try: + fn() + except Exception: + pass