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 5ad02540..c5164b7a 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 @@ -92,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 @@ -489,5 +638,4 @@ def store(self) -> ReadOnlyStore: 'DataSourceStatusProviderImpl', 'DataStoreStatusProviderImpl', 'FDv2', - 'FeatureStoreClientWrapper', ] diff --git a/ldclient/impl/datasystem/fdv2_common.py b/ldclient/impl/datasystem/fdv2_common.py index 13b1ae6f..b88837d8 100644 --- a/ldclient/impl/datasystem/fdv2_common.py +++ b/ldclient/impl/datasystem/fdv2_common.py @@ -2,20 +2,18 @@ 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 ( @@ -27,7 +25,6 @@ DataStoreStatusProvider, FeatureStore ) -from ldclient.versioned_data_kind import VersionedDataKind class DataSourceStatusProviderImpl(DataSourceStatusProvider): @@ -112,154 +109,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 +271,6 @@ def target_availability(self) -> DataAvailability: 'ConditionDirective', 'DataSourceStatusProviderImpl', 'DataStoreStatusProviderImpl', - 'FeatureStoreClientWrapper', 'fallback_condition', 'recovery_condition', ] 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()) )