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_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/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..5052d811 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,13 +47,18 @@ 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 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))) + 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)) @@ -62,10 +67,10 @@ 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)) + return self.__write(lambda: self.store.delete(kind, key, version), False) def upsert(self, kind, item) -> bool: - return self.__wrapper(lambda: self.store.upsert(kind, item)) + return self.__write(lambda: self.store.upsert(kind, item), False) @property def initialized(self) -> bool: @@ -79,9 +84,22 @@ 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 available == self.__last_available: + if self.__closed or available == self.__last_available: return self.__last_available = available @@ -104,6 +122,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() @@ -115,35 +136,28 @@ 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. + def close(self) -> None: + """Stops the availability poller and closes the wrapped store. - 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. + 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 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 + if poller_to_stop is not None: + poller_to_stop.stop() - return monitoring_enabled() + self.store.close() class _ReadOnlyFeatureStoreView(ReadOnlyStore): @@ -228,6 +242,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/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_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/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_client_wrapper.py b/ldclient/testing/test_feature_store_client_wrapper.py index 8ffe7c38..ed744378 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: @@ -103,3 +104,94 @@ def set_event(status): assert len(statuses) == 2 assert statuses[0].available is False assert statuses[1].available is True + + +def test_writes_after_close_do_not_reach_the_store(): + store = Mock() + store.is_monitoring_enabled = lambda: True + + listener = RecordStatusListener() + listeners = Listeners() + listeners.add(listener) + sink = DataStoreUpdateSinkImpl(listeners) + + wrapper = _FeatureStoreClientWrapper(store, sink) + wrapper.close() + + wrapper.init({}) + assert wrapper.upsert(FEATURES, {'key': 'flag', 'version': 1}) is False + assert wrapper.delete(FEATURES, 'flag', 2) is False + + 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'] + + +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 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"),