From eee6c193281a560cc2aac092b76a2816fae47bb8 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 28 Aug 2026 10:33:55 -0500 Subject: [PATCH 1/3] refactor: Move the sync FeatureStoreClientWrapper into the FDv2 data system MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FeatureStoreClientWrapper lived in fdv2_common.py, which is meant for logic shared between the sync FDv2 and the async AsyncFDv2 (via _FDv2Base). This wrapper is sync-only — only fdv2.py uses it; the async side has its own wrapper in async_fdv2.py. Move it into fdv2.py, next to its sole user, and drop the now-unused imports and __all__ entry from fdv2_common.py. This mirrors the earlier FDv1 wrapper move (5948b78). Pure relocation, no behavior change. --- ldclient/impl/datasystem/fdv2.py | 153 ++++++++++++++++++++++- ldclient/impl/datasystem/fdv2_common.py | 159 +----------------------- 2 files changed, 154 insertions(+), 158 deletions(-) diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 5ad02540..0baae866 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -1,15 +1,15 @@ import time from queue import Queue from threading import Event, Thread -from typing import Any, Callable, List, Optional +from typing import Any, Callable, Dict, List, Mapping, Optional from ldclient.config import Config, DataSourceBuilder, DataSystemConfig +from ldclient.feature_store import _FeatureStoreDataSetSorter from ldclient.impl.datasystem import DataSystem, DiagnosticSource from ldclient.impl.datasystem.fdv2_common import ( ConditionDirective, DataSourceStatusProviderImpl, DataStoreStatusProviderImpl, - FeatureStoreClientWrapper, _FDv2Base, fallback_condition, recovery_condition @@ -25,6 +25,7 @@ DataSourceState, DataStoreMode, DataStoreStatus, + FeatureStore, ReadOnlyStore, Synchronizer ) @@ -56,6 +57,154 @@ def initialized(self) -> bool: return self._store.is_initialized() +class FeatureStoreClientWrapper(FeatureStore): + """Provides additional behavior that the client requires before or after feature store operations. + Currently this just means sorting the data set for init() and dealing with data store status listeners. + """ + + def __init__(self, store: FeatureStore, store_update_sink: DataStoreStatusProviderImpl): + self.store = store + self.__store_update_sink = store_update_sink + self.__monitoring_enabled = self.is_monitoring_enabled() + + # 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))) + + def get(self, kind, key, callback): + return self.__wrapper(lambda: self.store.get(kind, key, callback)) + + def all(self, kind, callback): + return self.__wrapper(lambda: self.store.all(kind, callback)) + + def delete(self, kind, key, version): + return self.__wrapper(lambda: self.store.delete(kind, key, version)) + + def upsert(self, kind, item): + return self.__wrapper(lambda: self.store.upsert(kind, item)) + + @property + def initialized(self) -> bool: + return self.store.initialized + + def disable_cache(self) -> None: + def _do_disable(): + try: + inner = self.store + if hasattr(inner, "disable_cache"): + inner.disable_cache() # type: ignore[attr-defined] + except Exception as e: + log.warning("disable_cache failed on inner store: %s", e) + + self.__wrapper(_do_disable) + + def __wrapper(self, fn: Callable): + try: + return fn() + except BaseException: + if self.__monitoring_enabled: + self.__update_availability(False) + raise + + def __update_availability(self, available: bool): + state_changed = False + poller_to_stop = None + task_to_start = None + + with self.__lock.write(): + if self.__closed: + return + if available == self.__last_available: + return + + state_changed = True + self.__last_available = available + + if available: + poller_to_stop = self.__poller + self.__poller = None + elif self.__poller is None: + task_to_start = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability) + self.__poller = task_to_start + + if available: + log.warning("Persistent store is available again") + else: + log.warning("Detected persistent store unavailability; updates will be cached until it recovers") + + status = DataStoreStatus(available, True) + self.__store_update_sink.update_status(status) + + if poller_to_stop is not None: + poller_to_stop.stop() + + if task_to_start is not None: + task_to_start.start() + + def __check_availability(self): + try: + if self.store.is_available(): + self.__update_availability(True) + except BaseException as e: + 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 + + return monitoring_enabled() + + 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. + """ + poller_to_stop = None + + with self.__lock.write(): + if self.__closed: + return + self.__closed = True + poller_to_stop = self.__poller + self.__poller = None + + if poller_to_stop is not None: + poller_to_stop.stop() + + if hasattr(self.store, "close"): + self.store.close() + + class FDv2(_FDv2Base, DataSystem): """ FDv2 is an implementation of the DataSystem interface that uses the Flag Delivery V2 protocol diff --git a/ldclient/impl/datasystem/fdv2_common.py b/ldclient/impl/datasystem/fdv2_common.py index 13b1ae6f..06a9d477 100644 --- a/ldclient/impl/datasystem/fdv2_common.py +++ b/ldclient/impl/datasystem/fdv2_common.py @@ -2,22 +2,19 @@ Support classes shared by the sync and async FDv2 data system coordinators. These are synchronous (thread-based) components used identically by both -``FDv2`` and ``AsyncFDv2``: status providers, the persistent-store wrapper, -and the condition directive enum. +``FDv2`` and ``AsyncFDv2``: status providers, and the condition directive +enum. """ import time from copy import copy from enum import Enum -from typing import Any, Callable, Dict, Mapping, Optional +from typing import Callable, Optional -from ldclient.feature_store import _FeatureStoreDataSetSorter from ldclient.impl.datasystem import DataAvailability, DiagnosticAccumulator from ldclient.impl.datasystem.store import _StoreBase from ldclient.impl.listeners import Listeners -from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock -from ldclient.impl.util import log from ldclient.interfaces import ( DataSourceErrorInfo, DataSourceState, @@ -27,7 +24,6 @@ DataStoreStatusProvider, FeatureStore ) -from ldclient.versioned_data_kind import VersionedDataKind class DataSourceStatusProviderImpl(DataSourceStatusProvider): @@ -112,154 +108,6 @@ def remove_listener(self, listener: Callable[[DataStoreStatus], None]): self.__listeners.remove(listener) -class FeatureStoreClientWrapper(FeatureStore): - """Provides additional behavior that the client requires before or after feature store operations. - Currently this just means sorting the data set for init() and dealing with data store status listeners. - """ - - def __init__(self, store: FeatureStore, store_update_sink: DataStoreStatusProviderImpl): - self.store = store - self.__store_update_sink = store_update_sink - self.__monitoring_enabled = self.is_monitoring_enabled() - - # 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))) - - def get(self, kind, key, callback): - return self.__wrapper(lambda: self.store.get(kind, key, callback)) - - def all(self, kind, callback): - return self.__wrapper(lambda: self.store.all(kind, callback)) - - def delete(self, kind, key, version): - return self.__wrapper(lambda: self.store.delete(kind, key, version)) - - def upsert(self, kind, item): - return self.__wrapper(lambda: self.store.upsert(kind, item)) - - @property - def initialized(self) -> bool: - return self.store.initialized - - def disable_cache(self) -> None: - def _do_disable(): - try: - inner = self.store - if hasattr(inner, "disable_cache"): - inner.disable_cache() # type: ignore[attr-defined] - except Exception as e: - log.warning("disable_cache failed on inner store: %s", e) - - self.__wrapper(_do_disable) - - def __wrapper(self, fn: Callable): - try: - return fn() - except BaseException: - if self.__monitoring_enabled: - self.__update_availability(False) - raise - - def __update_availability(self, available: bool): - state_changed = False - poller_to_stop = None - task_to_start = None - - with self.__lock.write(): - if self.__closed: - return - if available == self.__last_available: - return - - state_changed = True - self.__last_available = available - - if available: - poller_to_stop = self.__poller - self.__poller = None - elif self.__poller is None: - task_to_start = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability) - self.__poller = task_to_start - - if available: - log.warning("Persistent store is available again") - else: - log.warning("Detected persistent store unavailability; updates will be cached until it recovers") - - status = DataStoreStatus(available, True) - self.__store_update_sink.update_status(status) - - if poller_to_stop is not None: - poller_to_stop.stop() - - if task_to_start is not None: - task_to_start.start() - - def __check_availability(self): - try: - if self.store.is_available(): - self.__update_availability(True) - except BaseException as e: - 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 - - return monitoring_enabled() - - 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. - """ - poller_to_stop = None - - with self.__lock.write(): - if self.__closed: - return - self.__closed = True - poller_to_stop = self.__poller - self.__poller = None - - if poller_to_stop is not None: - poller_to_stop.stop() - - if hasattr(self.store, "close"): - self.store.close() - - class ConditionDirective(str, Enum): """ ConditionDirective represents the possible directives that can be returned from a condition check. @@ -422,7 +270,6 @@ def target_availability(self) -> DataAvailability: 'ConditionDirective', 'DataSourceStatusProviderImpl', 'DataStoreStatusProviderImpl', - 'FeatureStoreClientWrapper', 'fallback_condition', 'recovery_condition', ] From d086049f3a88aaaaca47e708ab4fba2590006d96 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 28 Aug 2026 11:36:23 -0500 Subject: [PATCH 2/3] chore: Privatize the FDv2 feature-store client wrappers Rename FeatureStoreClientWrapper -> _FeatureStoreClientWrapper (sync) and AsyncFeatureStoreClientWrapper -> _AsyncFeatureStoreClientWrapper (async), and drop both from their modules' __all__. They are internal helpers in ldclient.impl.datasystem, matching the FDv1 wrapper's private convention. --- ldclient/impl/datasystem/async_fdv2.py | 5 ++--- ldclient/impl/datasystem/fdv2.py | 5 ++--- .../testing/impl/datasystem/test_async_fdv2.py | 16 ++++++++-------- .../impl/datasystem/test_cache_lifecycle.py | 10 +++++----- 4 files changed, 17 insertions(+), 19 deletions(-) diff --git a/ldclient/impl/datasystem/async_fdv2.py b/ldclient/impl/datasystem/async_fdv2.py index 8749721e..02d9c826 100644 --- a/ldclient/impl/datasystem/async_fdv2.py +++ b/ldclient/impl/datasystem/async_fdv2.py @@ -52,7 +52,7 @@ from ldclient.versioned_data_kind import VersionedDataKind -class AsyncFeatureStoreClientWrapper(AsyncFeatureStore): +class _AsyncFeatureStoreClientWrapper(AsyncFeatureStore): """Adds availability tracking around an async feature store. Every store operation runs through a wrapper that watches for failures. When @@ -261,7 +261,7 @@ def __init__( writable = data_system_config.data_store_mode == DataStoreMode.READ_WRITE # The async wrapper reports status through a plain callable sink, so # pass the provider's update method rather than the provider itself. - wrapper = AsyncFeatureStoreClientWrapper(data_system_config.data_store, self._data_store_status_provider.update_status) + wrapper = _AsyncFeatureStoreClientWrapper(data_system_config.data_store, self._data_store_status_provider.update_status) self._store.with_async_persistence(wrapper, writable, self._data_store_status_provider) self._store_view = _AsyncReadOnlyStoreView(self._store) @@ -646,7 +646,6 @@ async def data_availability(self) -> DataAvailability: # type: ignore[override] __all__ = [ 'AsyncFDv2', - 'AsyncFeatureStoreClientWrapper', 'ConditionDirective', 'DataSourceStatusProviderImpl', 'DataStoreStatusProviderImpl', diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 0baae866..c5164b7a 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -57,7 +57,7 @@ def initialized(self) -> bool: return self._store.is_initialized() -class FeatureStoreClientWrapper(FeatureStore): +class _FeatureStoreClientWrapper(FeatureStore): """Provides additional behavior that the client requires before or after feature store operations. Currently this just means sorting the data set for init() and dealing with data store status listeners. """ @@ -241,7 +241,7 @@ def __init__( if data_system_config.data_store is not None: self._data_store_status_provider = DataStoreStatusProviderImpl(data_system_config.data_store, self._data_store_listeners) writable = data_system_config.data_store_mode == DataStoreMode.READ_WRITE - wrapper = FeatureStoreClientWrapper(data_system_config.data_store, self._data_store_status_provider) + wrapper = _FeatureStoreClientWrapper(data_system_config.data_store, self._data_store_status_provider) self._store.with_persistence(wrapper, writable, self._data_store_status_provider) # Threading @@ -638,5 +638,4 @@ def store(self) -> ReadOnlyStore: 'DataSourceStatusProviderImpl', 'DataStoreStatusProviderImpl', 'FDv2', - 'FeatureStoreClientWrapper', ] diff --git a/ldclient/testing/impl/datasystem/test_async_fdv2.py b/ldclient/testing/impl/datasystem/test_async_fdv2.py index 8421368d..2411ac9e 100644 --- a/ldclient/testing/impl/datasystem/test_async_fdv2.py +++ b/ldclient/testing/impl/datasystem/test_async_fdv2.py @@ -14,7 +14,7 @@ from ldclient.impl.datasystem import DataAvailability from ldclient.impl.datasystem.async_fdv2 import ( AsyncFDv2, - AsyncFeatureStoreClientWrapper + _AsyncFeatureStoreClientWrapper ) from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, _Success from ldclient.integrations.test_datav2 import TestDataV2 @@ -690,27 +690,27 @@ async def is_available(self) -> bool: @pytest.mark.asyncio async def test_is_monitoring_enabled_true_when_store_opts_in(): - wrapper = AsyncFeatureStoreClientWrapper(FakeAsyncStore(), lambda _s: None) + wrapper = _AsyncFeatureStoreClientWrapper(FakeAsyncStore(), lambda _s: None) assert wrapper.is_monitoring_enabled() is True @pytest.mark.asyncio async def test_is_monitoring_enabled_false_without_is_available(): - wrapper = AsyncFeatureStoreClientWrapper(StoreWithoutAvailability(), lambda _s: None) + wrapper = _AsyncFeatureStoreClientWrapper(StoreWithoutAvailability(), lambda _s: None) assert wrapper.is_monitoring_enabled() is False @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) + wrapper = _AsyncFeatureStoreClientWrapper(StoreWithAvailabilityNoOptIn(), lambda _s: None) assert wrapper.is_monitoring_enabled() is False @pytest.mark.asyncio async def test_init_sorts_and_delegates(): store = FakeAsyncStore() - wrapper = AsyncFeatureStoreClientWrapper(store, lambda _s: None) + wrapper = _AsyncFeatureStoreClientWrapper(store, lambda _s: None) await wrapper.init({FEATURES: {}, SEGMENTS: {}}) assert len(store.init_calls) == 1 assert wrapper.initialized is True @@ -720,7 +720,7 @@ async def test_init_sorts_and_delegates(): async def test_failure_marks_unavailable_polls_and_recovers(): store = FakeAsyncStore() statuses: List[DataStoreStatus] = [] - wrapper = AsyncFeatureStoreClientWrapper(store, lambda s: statuses.append(s)) + wrapper = _AsyncFeatureStoreClientWrapper(store, lambda s: statuses.append(s)) # Make the next operation fail. store.fail = True @@ -752,7 +752,7 @@ async def test_failure_marks_unavailable_polls_and_recovers(): async def test_close_stops_poller_and_closes_inner(): store = FakeAsyncStore() statuses: List[DataStoreStatus] = [] - wrapper = AsyncFeatureStoreClientWrapper(store, lambda s: statuses.append(s)) + wrapper = _AsyncFeatureStoreClientWrapper(store, lambda s: statuses.append(s)) # Trigger an outage so a poller is running. store.fail = True @@ -771,7 +771,7 @@ async def test_close_stops_poller_and_closes_inner(): @pytest.mark.asyncio async def test_successful_ops_pass_through(): store = FakeAsyncStore() - wrapper = AsyncFeatureStoreClientWrapper(store, lambda _s: None) + wrapper = _AsyncFeatureStoreClientWrapper(store, lambda _s: None) await wrapper.upsert(FEATURES, {"key": "flag-a", "version": 1}) got = await wrapper.get(FEATURES, "flag-a") diff --git a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py index ed02c168..52e9cdc7 100644 --- a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py +++ b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py @@ -17,7 +17,7 @@ ) from ldclient.impl.datasystem.fdv2 import ( DataStoreStatusProviderImpl, - FeatureStoreClientWrapper + _FeatureStoreClientWrapper ) from ldclient.impl.datasystem.store import Store from ldclient.impl.listeners import Listeners @@ -119,12 +119,12 @@ def _delta_changeset(flag_key: str, version: int = 2) -> ChangeSet: def _build_store_with_persistent(persistent) -> Store: - """Build a Store wired the same way fdv2.py does: outer FeatureStoreClientWrapper + """Build a Store wired the same way fdv2.py does: outer _FeatureStoreClientWrapper over the user's persistent store. """ listeners = Listeners() status_provider = DataStoreStatusProviderImpl(persistent, listeners) - outer = FeatureStoreClientWrapper(persistent, status_provider) + outer = _FeatureStoreClientWrapper(persistent, status_provider) store = Store(Listeners(), Listeners()) store.with_persistence(outer, True, status_provider) return store @@ -261,7 +261,7 @@ def clear(self): class TestStoreDisablesPersistentCache: def test_set_basis_disables_cache_through_feature_store_client_wrapper(self): - """End-to-end: Store._set_basis -> FeatureStoreClientWrapper.disable_cache + """End-to-end: Store._set_basis -> _FeatureStoreClientWrapper.disable_cache -> CachingStoreWrapper.disable_cache.""" core = RecordingCore() inner = CachingStoreWrapper(core, CacheConfig.default()) @@ -430,7 +430,7 @@ def test_close_on_caching_store_wrapper_disables_and_closes_core(self): def test_close_propagates_through_feature_store_client_wrapper(self): core = RecordingCore() inner = CachingStoreWrapper(core, CacheConfig.default()) - outer = FeatureStoreClientWrapper( + outer = _FeatureStoreClientWrapper( inner, DataStoreStatusProviderImpl(inner, Listeners()) ) From 81f59336d9e8970ce44af6346419cae1e0fc989b Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 28 Aug 2026 11:46:14 -0500 Subject: [PATCH 3/3] fix: Keep the log import in fdv2_common used by the FDv2 availability gate The relocation removed the log import (only the moved wrapper used it on the pre-#506 main), but #506's warm-start guard in _FDv2Base.data_availability uses log.error. After rebasing onto current main the guard is present, so the import must stay. --- ldclient/impl/datasystem/fdv2_common.py | 1 + 1 file changed, 1 insertion(+) diff --git a/ldclient/impl/datasystem/fdv2_common.py b/ldclient/impl/datasystem/fdv2_common.py index 06a9d477..b88837d8 100644 --- a/ldclient/impl/datasystem/fdv2_common.py +++ b/ldclient/impl/datasystem/fdv2_common.py @@ -15,6 +15,7 @@ from ldclient.impl.datasystem.store import _StoreBase from ldclient.impl.listeners import Listeners from ldclient.impl.rwlock import ReadWriteLock +from ldclient.impl.util import log from ldclient.interfaces import ( DataSourceErrorInfo, DataSourceState,