keelerm84 · Nov 2, 2026 · Nov 2, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026
@@ -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]:

@@ -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)

@@ -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():

@@ -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)

@@ -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:

@@ -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):

@@ -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:

@@ -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):

@@ -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)

@@ -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)
 

@@ -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)