From a86f9d996615d6dab2b41424c822e648dafe2669 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 13 Aug 2026 09:05:04 -0500 Subject: [PATCH 1/3] feat: Add async persistent feature store foundation and Redis adapter --- ldclient/async_feature_store_helpers.py | 118 ++++++ ldclient/feature_store_helpers.py | 232 +++++++----- .../redis/async_redis_feature_store.py | 118 ++++++ ldclient/integrations/__init__.py | 46 +++ ldclient/interfaces.py | 82 ++++ .../test_async_redis_feature_store.py | 180 +++++++++ .../test_async_feature_store_helpers.py | 352 ++++++++++++++++++ 7 files changed, 1044 insertions(+), 84 deletions(-) create mode 100644 ldclient/async_feature_store_helpers.py create mode 100644 ldclient/impl/integrations/redis/async_redis_feature_store.py create mode 100644 ldclient/testing/integrations/test_async_redis_feature_store.py create mode 100644 ldclient/testing/test_async_feature_store_helpers.py diff --git a/ldclient/async_feature_store_helpers.py b/ldclient/async_feature_store_helpers.py new file mode 100644 index 00000000..5a2bc876 --- /dev/null +++ b/ldclient/async_feature_store_helpers.py @@ -0,0 +1,118 @@ +""" +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 +from ldclient.feature_store_helpers import ( + _CachingStoreWrapperBase, + _ensure_encoded +) +from ldclient.interfaces import ( + AsyncFeatureStore, + AsyncFeatureStoreCore, + DiagnosticDescription +) +from ldclient.versioned_data_kind import VersionedDataKind + + +class AsyncCachingStoreWrapper(_CachingStoreWrapperBase, DiagnosticDescription, AsyncFeatureStore): + """A partial implementation of :class:`ldclient.interfaces.AsyncFeatureStore`. + + This class delegates the database-specific work to an implementation of + :class:`ldclient.interfaces.AsyncFeatureStoreCore`, while adding optional caching behavior and + other logic that would otherwise be repeated in every async feature store implementation. This + makes it easier to create new async database integrations by implementing only the + database-specific logic. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. Pin to a specific minor version and review the changelog + before upgrading. + + The cache is a plain in-memory dict, which is safe for concurrent access within a single asyncio + event loop because its reads and writes never suspend between one another. + """ + + _core: AsyncFeatureStoreCore + + def __init__(self, core: AsyncFeatureStoreCore, cache_config: CacheConfig): + """Constructs an instance by wrapping a core implementation object. + + :param core: the implementation object + :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 + + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + """ """ + await self._core.init_internal(all_data) + self._cache_init(all_data) + self._inited = True + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + """ """ + hit, value = self._cache_get_item(kind, key) + if hit: + return value + encoded_item = await self._core.get_internal(kind, key) + return self._cache_put_item(kind, key, encoded_item) + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + """ """ + hit, value = self._cache_get_all(kind) + if hit: + return value + encoded_items = await self._core.get_all_internal(kind) + return self._cache_put_all(kind, encoded_items) + + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: + """ """ + deleted_item = {"key": key, "version": version, "deleted": True} + return await self.upsert(kind, deleted_item) + + async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: + """ """ + encoded_item = _ensure_encoded(kind, item) + new_state = await self._core.upsert_internal(kind, encoded_item) + self._cache_put_upsert(kind, new_state) + # The core returns the item we passed in if the write was applied, or the existing item if it + # was rejected by the version check. Identity therefore tells us whether the store changed. + return new_state is encoded_item + + @property + def initialized(self) -> bool: + """Returns whether ``init`` has completed in this process. + + This property does not query the store: it is synchronous, but a persistent-store query is + a coroutine, so it reflects only whether this process has initialized the store. + """ + return self._inited + + async def close(self) -> None: + """Releases the cache and closes the underlying core if it supports it.""" + self.disable_cache() + core_close = getattr(self._core, "close", None) + if callable(core_close): + result = core_close() + if inspect.isawaitable(result): + await result + + def describe_configuration(self, config) -> str: + describe = getattr(self._core, 'describe_configuration', None) + if callable(describe): + return describe(config) + return "custom" diff --git a/ldclient/feature_store_helpers.py b/ldclient/feature_store_helpers.py index ff7a0e1b..19cec91c 100644 --- a/ldclient/feature_store_helpers.py +++ b/ldclient/feature_store_helpers.py @@ -30,7 +30,7 @@ class _NoopCache: Used both when caching is disabled at config time and when the FDv2 in-memory store has taken over and the persistent-store cache is no longer useful. Implements only the subset of the dict-like surface - that CachingStoreWrapper exercises. + that the caching wrappers exercise. """ __slots__ = () @@ -51,26 +51,20 @@ def clear(self): _NOOP_CACHE = _NoopCache() -class CachingStoreWrapper(DiagnosticDescription, FeatureStore): - """A partial implementation of :class:`ldclient.interfaces.FeatureStore`. - - This class delegates the basic functionality to an implementation of - :class:`ldclient.interfaces.FeatureStoreCore` - while adding optional caching behavior and other logic - that would otherwise be repeated in every feature store implementation. This makes it easier to create - new database integrations by implementing only the database-specific logic. +class _CachingStoreWrapperBase: + """Provides common cache methods for a feature store wrapper. Subclass it to reuse the cache + setup, the sans-I/O cache bookkeeping over ``self._cache``, and the shared non-I/O helpers. """ - __INITED_CACHE_KEY__ = "$inited" + _cache: Any + _inited: bool + _has_available_method: bool - def __init__(self, core: FeatureStoreCore, cache_config: CacheConfig): - """Constructs an instance by wrapping a core implementation object. + def __init__(self, cache_config: CacheConfig): + """Sets up the cache from the caching parameters. - :param core: the implementation object :param cache_config: the caching parameters """ - self._core = core - self.__has_available_method = callable(getattr(core, 'is_available', None)) - if cache_config.enabled: self._cache = ExpiringDict(max_len=cache_config.capacity, max_age_seconds=cache_config.expiration) else: @@ -78,58 +72,163 @@ def __init__(self, core: FeatureStoreCore, cache_config: CacheConfig): self._inited = False def is_monitoring_enabled(self) -> bool: - return self.__has_available_method + return self._has_available_method - 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 + def disable_cache(self) -> None: + """Replace the in-memory cache with a no-op so further operations don't populate it. - def init(self, all_encoded_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]): - """ """ - self._core.init_internal(all_encoded_data) # currently FeatureStoreCore expects to receive dicts + Called by the FDv2 store coordinator once the in-memory store has become the + source of truth and the persistent-store cache is no longer useful. Safe to + call multiple times. Internal -- not part of the public API. + """ + cache = self._cache + if cache is _NOOP_CACHE: + return + self._cache = _NOOP_CACHE # readers from this point forward see the no-op + try: + cache.clear() # release the entries the old dict was holding + except Exception as e: + log.warning("Error clearing persistent store cache: %s", e) + log.debug("Persistent store cache replaced with no-op; in-memory store is now active") + + # The methods below hold the cache logic that both wrappers share. They do no I/O; each one is + # the pre-work or post-work that surrounds a single core call in a wrapper method. + + def _cache_get_item(self, kind, key): + """Looks up a single item in the cache. + + Returns a ``(hit, value)`` pair. ``hit`` is True if the item was in the cache. ``value`` is + the item to return, which is None if the cached entry is missing or deleted. + """ + cached_item = self._cache.get(self._item_cache_key(kind, key)) + # note, cached items are wrapped in an array so we can cache None values + if cached_item is None: + return (False, None) + item = cached_item[0] + return (True, None if _is_deleted(item) else item) + + def _cache_put_item(self, kind, key, encoded_item): + """Decodes an item fetched from the core, caches it, and returns the value to return. + + The returned value is None if the item is missing or deleted. + """ + item = None if encoded_item is None else kind.decode(encoded_item) + self._cache[self._item_cache_key(kind, key)] = [item] + return None if _is_deleted(item) else item + + def _cache_get_all(self, kind): + """Looks up the full set of items of a kind in the cache. + + Returns a ``(hit, value)`` pair. ``hit`` is True if the set was in the cache, in which case + ``value`` is the cached dict of items. + """ + cached_items = self._cache.get(self._all_cache_key(kind)) + if cached_items is None: + return (False, None) + return (True, cached_items) + + def _cache_put_all(self, kind, encoded_items): + """Decodes all items fetched from the core, drops deleted ones, caches the result, and returns it.""" + all_items = {} + if encoded_items is not None: + for key, item in encoded_items.items(): + all_items[key] = kind.decode(item) + items = self._items_if_not_deleted(all_items) + self._cache[self._all_cache_key(kind)] = items + return items + + def _cache_init(self, all_encoded_data): + """Populates the cache from a full data set. Does nothing when caching is off (a no-op cache).""" cache = self._cache if cache is _NOOP_CACHE: - # Skip the per-item decode loop when there's nothing to cache. - self._inited = True return cache.clear() for kind, items in all_encoded_data.items(): - decoded_items = {} # we don't want to cache dicts, we want to cache FeatureFlags/Segments + decoded_items = {} # we cache FeatureFlags/Segments, not raw dicts for key, item in items.items(): decoded_item = kind.decode(item) cache[self._item_cache_key(kind, key)] = [decoded_item] # note array wrapper if not _is_deleted(decoded_item): decoded_items[key] = decoded_item cache[self._all_cache_key(kind)] = decoded_items + + def _cache_put_upsert(self, kind, new_state): + """Updates the cache after an upsert. + + Caches the item the core returned and drops the now-stale all-items entry. Returns the + decoded item. + """ + new_decoded_item = kind.decode(new_state) + self._cache[self._item_cache_key(kind, new_decoded_item.get('key'))] = [new_decoded_item] + self._cache.pop(self._all_cache_key(kind), None) + return new_decoded_item + + @staticmethod + def _item_cache_key(kind, key): + return "{0}:{1}".format(kind.namespace, key) + + @staticmethod + def _all_cache_key(kind): + return kind.namespace + + @staticmethod + def _items_if_not_deleted(items): + results = {} + if items is not None: + for key, item in items.items(): + if not item.get('deleted', False): + results[key] = item + return results + + +class CachingStoreWrapper(_CachingStoreWrapperBase, DiagnosticDescription, FeatureStore): + """A partial implementation of :class:`ldclient.interfaces.FeatureStore`. + + This class delegates the basic functionality to an implementation of + :class:`ldclient.interfaces.FeatureStoreCore` - while adding optional caching behavior and other logic + that would otherwise be repeated in every feature store implementation. This makes it easier to create + new database integrations by implementing only the database-specific logic. + """ + + __INITED_CACHE_KEY__ = "$inited" + + _core: FeatureStoreCore + + def __init__(self, core: FeatureStoreCore, cache_config: CacheConfig): + """Constructs an instance by wrapping a core implementation object. + + :param core: the implementation object + :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 + + def init(self, all_encoded_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]): + """ """ + self._core.init_internal(all_encoded_data) # currently FeatureStoreCore expects to receive dicts + self._cache_init(all_encoded_data) self._inited = True def get(self, kind, key, callback=lambda x: x): """ """ - cache_key = self._item_cache_key(kind, key) - cached_item = self._cache.get(cache_key) - # note, cached items are wrapped in an array so we can cache None values - if cached_item is not None: - item = cached_item[0] - return callback(None if _is_deleted(item) else item) + hit, value = self._cache_get_item(kind, key) + if hit: + return callback(value) encoded_item = self._core.get_internal(kind, key) # currently FeatureStoreCore returns dicts - item = None if encoded_item is None else kind.decode(encoded_item) - self._cache[cache_key] = [item] - return callback(None if _is_deleted(item) else item) + return callback(self._cache_put_item(kind, key, encoded_item)) def all(self, kind, callback=lambda x: x): """ """ - cache_key = self._all_cache_key(kind) - cached_items = self._cache.get(cache_key) - if cached_items is not None: - return callback(cached_items) + hit, value = self._cache_get_all(kind) + if hit: + return callback(value) encoded_items = self._core.get_all_internal(kind) - all_items = {} - if encoded_items is not None: - for key, item in encoded_items.items(): - all_items[key] = kind.decode(item) - items = self._items_if_not_deleted(all_items) - self._cache[cache_key] = items - return callback(items) + return callback(self._cache_put_all(kind, encoded_items)) def delete(self, kind, key, version): """ """ @@ -140,9 +239,7 @@ def upsert(self, kind, encoded_item): """ """ encoded_item = _ensure_encoded(kind, encoded_item) new_state = self._core.upsert_internal(kind, encoded_item) - new_decoded_item = kind.decode(new_state) - self._cache[self._item_cache_key(kind, new_decoded_item.get('key'))] = [new_decoded_item] - self._cache.pop(self._all_cache_key(kind), None) + self._cache_put_upsert(kind, new_state) @property def initialized(self) -> bool: @@ -157,23 +254,6 @@ def initialized(self) -> bool: self._inited = True return result - def disable_cache(self) -> None: - """Replace the in-memory cache with a no-op so further operations don't populate it. - - Called by the FDv2 store coordinator once the in-memory store has become the - source of truth and the persistent-store cache is no longer useful. Safe to - call multiple times. Internal -- not part of the public API. - """ - cache = self._cache - if cache is _NOOP_CACHE: - return - self._cache = _NOOP_CACHE # readers from this point forward see the no-op - try: - cache.clear() # release the entries the old dict was holding - except Exception as e: - log.warning("Error clearing persistent store cache: %s", e) - log.debug("Persistent store cache replaced with no-op; in-memory store is now active") - def close(self) -> None: """Release the cache and close the underlying core if it supports it.""" self.disable_cache() @@ -181,23 +261,7 @@ def close(self) -> None: self._core.close() # type: ignore def describe_configuration(self, config): - if callable(getattr(self._core, 'describe_configuration', None)): - return self._core.describe_configuration(config) + describe = getattr(self._core, 'describe_configuration', None) + if callable(describe): + return describe(config) return "custom" - - @staticmethod - def _item_cache_key(kind, key): - return "{0}:{1}".format(kind.namespace, key) - - @staticmethod - def _all_cache_key(kind): - return kind.namespace - - @staticmethod - def _items_if_not_deleted(items): - results = {} - if items is not None: - for key, item in items.items(): - if not item.get('deleted', False): - results[key] = item - return results diff --git a/ldclient/impl/integrations/redis/async_redis_feature_store.py b/ldclient/impl/integrations/redis/async_redis_feature_store.py new file mode 100644 index 00000000..8391cfca --- /dev/null +++ b/ldclient/impl/integrations/redis/async_redis_feature_store.py @@ -0,0 +1,118 @@ +import json +from typing import Any, Callable, Dict, Mapping, Optional + +from ldclient.feature_store_helpers import CachingStoreWrapper +from ldclient.impl.util import log, redact_password +from ldclient.interfaces import AsyncFeatureStoreCore, DiagnosticDescription +from ldclient.versioned_data_kind import VersionedDataKind + +have_async_redis = False +try: + import redis.asyncio as redis_client + from redis.exceptions import WatchError + + have_async_redis = True +except ImportError: + pass + + +class _AsyncRedisFeatureStoreCore(DiagnosticDescription, AsyncFeatureStoreCore): + """Async Redis implementation of :class:`ldclient.interfaces.AsyncFeatureStoreCore`. + + It stores data in the same Redis key layout as the synchronous Redis feature store, so an async + and a synchronous SDK can share one Redis instance. + """ + + def __init__(self, url: str, prefix: Optional[str], redis_opts: Dict[str, Any]): + if not have_async_redis: + raise NotImplementedError("Cannot use async Redis feature store because redis package is not installed") + self._prefix = prefix or 'launchdarkly' + self._init_key = "{0}:{1}".format(self._prefix, CachingStoreWrapper.__INITED_CACHE_KEY__) + self._client = redis_client.from_url(url, **redis_opts) + self.test_update_hook: Optional[Callable[[str, str], None]] = None # exposed for testing + log.info("Started AsyncRedisFeatureStore connected to URL: " + redact_password(url) + " using prefix: " + self._prefix) + + async def is_available(self) -> bool: + try: + await self.initialized_internal() + return True + except BaseException: + return False + + def _items_key(self, kind: VersionedDataKind) -> str: + return "{0}:{1}".format(self._prefix, kind.namespace) + + async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + all_count = 0 + async with self._client.pipeline() as pipe: + for kind, items in all_data.items(): + base_key = self._items_key(kind) + pipe.delete(base_key) + for key, item in items.items(): + pipe.hset(base_key, key, json.dumps(item)) + all_count = all_count + len(items) + pipe.set(self._init_key, self._init_key) + await pipe.execute() + log.info("Initialized AsyncRedisFeatureStore with %d items", all_count) + + async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, dict]: + all_items = await self._client.hgetall(self._items_key(kind)) + if not all_items: + return {} + results = {} + for key, item_json in all_items.items(): + results[key.decode('utf-8')] = json.loads(item_json.decode('utf-8')) + return results + + async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[dict]: + item_json = await self._client.hget(self._items_key(kind), key) + if not item_json: + log.debug("AsyncRedisFeatureStore: key %s not found in '%s'. Returning None.", key, kind.namespace) + return None + return json.loads(item_json.decode('utf-8')) + + async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict: + base_key = self._items_key(kind) + key = item['key'] + item_json = json.dumps(item) + + while True: + async with self._client.pipeline() as pipe: + try: + await pipe.watch(base_key) + old = await self.get_internal(kind, key) + if self.test_update_hook is not None: + self.test_update_hook(base_key, key) + if old and old['version'] >= item['version']: + log.debug( + 'AsyncRedisFeatureStore: Attempted to %s key: %s version %d with a version that is the same or older: %d in "%s"', + 'delete' if item.get('deleted') else 'update', + key, + old['version'], + item['version'], + kind.namespace, + ) + await pipe.unwatch() + return old + pipe.multi() + pipe.hset(base_key, key, item_json) + # A concurrent change to the watched key makes execute() raise WatchError, + # rather than returning a null result as on some other platforms. + await pipe.execute() + return item + except WatchError: + log.debug("AsyncRedisFeatureStore: concurrent modification detected, retrying") + continue + + async def initialized_internal(self) -> bool: + return bool(await self._client.exists(self._init_key)) + + async def close(self) -> None: + # Prefer aclose() (redis-py 5.0.1+); older supported versions (>= 4.2) only have close(). + if hasattr(self._client, "aclose"): + await self._client.aclose() + else: + await self._client.close() + + def describe_configuration(self, config) -> str: + return 'Redis' diff --git a/ldclient/integrations/__init__.py b/ldclient/integrations/__init__.py index a16347ca..3f729589 100644 --- a/ldclient/integrations/__init__.py +++ b/ldclient/integrations/__init__.py @@ -237,6 +237,52 @@ def new_big_segment_store(url: str = 'redis://localhost:6379/0', prefix: str = ' return _RedisBigSegmentStore(url, prefix, redis_opts) + @staticmethod + def async_feature_store(url: str = 'redis://localhost:6379/0', prefix: Optional[str] = None, caching: CacheConfig = CacheConfig.default(), redis_opts: Dict[str, Any] = {}): + """ + Creates an async Redis-backed implementation of :class:`~ldclient.interfaces.AsyncFeatureStore`. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. Pin to a specific minor version and review the changelog + before upgrading. + + For more details about how and why you can use a persistent feature store, see the + `SDK reference guide `_. + + To use this method, you must first install the ``redis`` package (version >=5.0.1). Then, put + the object returned by this method into the ``feature_store`` property of your client + configuration when constructing an ``AsyncLDClient``. + :: + + from ldclient.config import Config + from ldclient.integrations import Redis + store = Redis.async_feature_store() + config = Config(feature_store=store) + + The data layout matches :func:`new_feature_store`, so an async and a synchronous SDK can share + one Redis instance. + + :param url: the URL of the Redis host; defaults to ``DEFAULT_URL`` + :param prefix: a namespace prefix to be prepended to all Redis keys; defaults to + ``DEFAULT_PREFIX`` + :param caching: specifies whether local caching should be enabled and if so, + sets the cache properties; defaults to :func:`ldclient.feature_store.CacheConfig.default()`. + See :class:`ldclient.feature_store.CacheConfig`. + :param redis_opts: extra options forwarded to ``redis.asyncio.from_url`` + """ + from ldclient.async_feature_store_helpers import ( + AsyncCachingStoreWrapper + ) + from ldclient.impl.integrations.redis.async_redis_feature_store import ( + _AsyncRedisFeatureStoreCore + ) + core = _AsyncRedisFeatureStoreCore(url, prefix, redis_opts) + wrapper = AsyncCachingStoreWrapper(core, caching) + wrapper._core = core # exposed for testing + return wrapper + @staticmethod def async_big_segment_store(url: str = 'redis://localhost:6379/0', prefix: Optional[str] = None, redis_opts: Dict[str, Any] = {}): """ diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index 7c2e17bb..3772c491 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -421,6 +421,88 @@ async def close(self) -> None: pass +class AsyncFeatureStoreCore(ABC): + """ + Async equivalent of :class:`FeatureStoreCore`, for use with + :class:`ldclient.async_feature_store_helpers.AsyncCachingStoreWrapper`. It exposes a simplified + subset of the functionality of :class:`AsyncFeatureStore`, so a database integration only has to + implement the database-specific logic; the wrapper adds caching and encode/decode handling. + + .. caution:: + This feature is experimental and should NOT be considered ready for production + use. It may change or be removed without notice and is not subject to backwards + compatibility guarantees. Pin to a specific minor version and review the changelog + before upgrading. + + All items passed to and returned from these methods are plain JSON-compatible dicts, not decoded + model objects. The wrapper handles decoding for its cache and its callers. + """ + + @abstractmethod + async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[dict]: + """ + Returns the object to which the specified key is mapped, or None if no such item exists. + The method should not attempt to filter out any items based on their deleted property, + nor to cache any items. + + :param kind: The kind of object to get + :param key: The key of the object + :return: The object to which the specified key is mapped, or None + """ + ... + + @abstractmethod + async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, dict]: + """ + Returns a dictionary of all associated objects of a given kind. The method should not attempt + to filter out any items based on their deleted property, nor to cache any items. + + :param kind: The kind of objects to get + :return: A dictionary of keys to items + """ + ... + + @abstractmethod + async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + """ + Initializes (or re-initializes) the store with the specified set of objects. Any existing + entries will be removed. Implementations can assume that this set of objects is up to date-- + there is no need to perform individual version comparisons between the existing objects and + the supplied data. + + :param all_data: A dictionary of data kinds to item collections + """ + ... + + @abstractmethod + async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict: + """ + Updates or inserts the object associated with the specified key. If an item with the same key + already exists, it should update it only if the new item's version property is greater than + the old one. It should return the final state of the item, that is, the item that was passed + in if the update succeeded, or the item that is currently in the data store if the update + failed the version check. This lets the wrapper update its cache correctly. + + :param kind: The kind of object to update + :param item: The object to update or insert + :return: The state of the object after the update + """ + ... + + @abstractmethod + async def initialized_internal(self) -> bool: + """ + Returns true if this store has been initialized. In a shared data store, it should be able to + detect this even if init_internal was called in a different process, so the test should be + based on what is in the data store. + """ + ... + + # 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. + + # Internal use only. Common methods for components that perform a task in the background. class BackgroundOperation: diff --git a/ldclient/testing/integrations/test_async_redis_feature_store.py b/ldclient/testing/integrations/test_async_redis_feature_store.py new file mode 100644 index 00000000..9aa292b2 --- /dev/null +++ b/ldclient/testing/integrations/test_async_redis_feature_store.py @@ -0,0 +1,180 @@ +""" +Integration tests for the async Redis feature store (_AsyncRedisFeatureStoreCore wrapped by +AsyncCachingStoreWrapper). + +These tests require a real Redis instance running on localhost:6379. They are skipped when the +redis package is not installed or when the LD_SKIP_DATABASE_TESTS environment variable is set to +'1'. The caching-wrapper logic itself is covered without Redis in +ldclient.testing.test_async_feature_store_helpers. +""" + +import json + +import pytest + +from ldclient.async_feature_store_helpers import AsyncCachingStoreWrapper +from ldclient.feature_store import CacheConfig +from ldclient.integrations import Redis +from ldclient.interfaces import AsyncFeatureStore +from ldclient.testing.async_feature_store_test_base import ( + AsyncFeatureStoreTestBase, + AsyncFeatureStoreTester +) +from ldclient.testing.test_util import skip_database_tests +from ldclient.versioned_data_kind import FEATURES + +have_async_redis = False +try: + import redis.asyncio as aioredis + + have_async_redis = True +except ImportError: + pass + +try: + import redis as _sync_redis + + have_sync_redis = True +except ImportError: + have_sync_redis = False + +pytestmark = pytest.mark.skipif( + not have_async_redis, + reason="skipping async Redis tests because redis package is not installed" +) + +DEFAULT_PREFIX = 'launchdarkly' + + +def sync_redis_client(): + """Return a synchronous Redis client for test setup and teardown.""" + import redis + return redis.StrictRedis(host="localhost", port=6379, db=0) + + +def clear_data(prefix): + r = sync_redis_client() + for key in r.keys("%s:*" % (prefix or DEFAULT_PREFIX)): + r.delete(key) + + +class AsyncRedisFeatureStoreTester(AsyncFeatureStoreTester): + def __init__(self, prefix=None, caching=None): + self.prefix = prefix + self.caching = caching if caching is not None else CacheConfig.disabled() + + async def create_feature_store(self) -> AsyncFeatureStore: + return Redis.async_feature_store(prefix=self.prefix, caching=self.caching) + + +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +class TestAsyncRedisFeatureStore(AsyncFeatureStoreTestBase): + @pytest.fixture(params=[(False, False), (True, False), (False, True), (True, True)]) + def tester(self, request): + specify_prefix, use_caching = request.param + prefix = "testprefix" if specify_prefix else None + caching = CacheConfig.default() if use_caching else CacheConfig.disabled() + return AsyncRedisFeatureStoreTester(prefix, caching) + + @pytest.fixture(autouse=True) + def clear_data_before_each(self, tester): + clear_data(tester.prefix) + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_stores_with_different_prefixes_are_independent(): + clear_data("a") + clear_data("b") + + flag_a1 = {'key': 'flagA1', 'version': 1} + flag_a2 = {'key': 'flagA2', 'version': 1} + flag_b1 = {'key': 'flagB1', 'version': 1} + flag_b2 = {'key': 'flagB2', 'version': 1} + + store_a = Redis.async_feature_store(prefix="a") + store_b = Redis.async_feature_store(prefix="b") + try: + await store_a.init({FEATURES: {'flagA1': flag_a1}}) + await store_a.upsert(FEATURES, flag_a2) + + await store_b.init({FEATURES: {'flagB1': flag_b1}}) + await store_b.upsert(FEATURES, flag_b2) + + assert await store_a.get(FEATURES, 'flagA1') == FEATURES.decode(flag_a1) + assert await store_a.get(FEATURES, 'flagB1') is None + assert await store_a.all(FEATURES) == {'flagA1': FEATURES.decode(flag_a1), 'flagA2': FEATURES.decode(flag_a2)} + + assert await store_b.get(FEATURES, 'flagB1') == FEATURES.decode(flag_b1) + assert await store_b.get(FEATURES, 'flagA1') is None + assert await store_b.all(FEATURES) == {'flagB1': FEATURES.decode(flag_b1), 'flagB2': FEATURES.decode(flag_b2)} + finally: + await store_a.close() + await store_b.close() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_upsert_race_condition_against_external_client_with_higher_version(): + other_client = sync_redis_client() + store = Redis.async_feature_store() + try: + await store.init({FEATURES: {}}) + + other_version = {'key': 'flagkey', 'version': 2} + + def hook(base_key, key): + if other_version['version'] <= 4: + other_client.hset(base_key, key, json.dumps(other_version)) + other_version['version'] = other_version['version'] + 1 + + store._core.test_update_hook = hook + + await store.upsert(FEATURES, {'key': 'flagkey', 'version': 1}) + result = await store.get(FEATURES, 'flagkey') + assert result['version'] == 2 + finally: + await store.close() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +@pytest.mark.skipif(not have_sync_redis, reason="skipping: sync redis not available for test setup") +async def test_upsert_race_condition_against_external_client_with_lower_version(): + other_client = sync_redis_client() + store = Redis.async_feature_store() + try: + await store.init({FEATURES: {}}) + + other_version = {'key': 'flagkey', 'version': 2} + + def hook(base_key, key): + if other_version['version'] <= 4: + other_client.hset(base_key, key, json.dumps(other_version)) + other_version['version'] = other_version['version'] + 1 + + store._core.test_update_hook = hook + + await store.upsert(FEATURES, {'key': 'flagkey', 'version': 5}) + result = await store.get(FEATURES, 'flagkey') + assert result['version'] == 5 + finally: + await store.close() + + +@pytest.mark.asyncio +@pytest.mark.skipif(skip_database_tests, reason="skipping database tests") +async def test_available_and_monitoring(): + store = Redis.async_feature_store() + try: + assert store.is_monitoring_enabled() is True + assert await store.is_available() is True + finally: + await store.close() + + +def test_async_feature_store_is_caching_wrapper(): + store = Redis.async_feature_store() + assert isinstance(store, AsyncCachingStoreWrapper) diff --git a/ldclient/testing/test_async_feature_store_helpers.py b/ldclient/testing/test_async_feature_store_helpers.py new file mode 100644 index 00000000..63529207 --- /dev/null +++ b/ldclient/testing/test_async_feature_store_helpers.py @@ -0,0 +1,352 @@ +import asyncio + +import pytest + +from ldclient.async_feature_store_helpers import AsyncCachingStoreWrapper +from ldclient.feature_store import CacheConfig +from ldclient.versioned_data_kind import VersionedDataKind + +# These tests exercise the caching-wrapper logic only, using an in-memory mock core, so they run +# without a Redis instance. They mirror ldclient.testing.test_feature_store_helpers for the sync +# CachingStoreWrapper. + +THINGS = VersionedDataKind(namespace="things", request_api_path="", stream_api_path="") +WRONG_THINGS = VersionedDataKind(namespace="wrong", request_api_path="", stream_api_path="") + + +def make_wrapper(core, cached): + return AsyncCachingStoreWrapper(core, CacheConfig(expiration=30) if cached else CacheConfig.disabled()) + + +class MockAsyncCore: + def __init__(self): + self.data = {} + self.inited = False + self.inited_query_count = 0 + self.error = None + + async def init_internal(self, all_data): + self._maybe_throw() + self.data = {} + for kind, items in all_data.items(): + self.data[kind] = items.copy() + + async def get_internal(self, kind, key): + self._maybe_throw() + items = self.data.get(kind) + return None if items is None else items.get(key) + + async def get_all_internal(self, kind): + self._maybe_throw() + return self.data.get(kind) + + async def upsert_internal(self, kind, item): + self._maybe_throw() + key = item.get('key') + items = self.data.get(kind) + if items is None: + items = {} + self.data[kind] = items + old_item = items.get(key) + if old_item is None or old_item.get('version') < item.get('version'): + items[key] = item + return item + return old_item + + async def initialized_internal(self): + self._maybe_throw() + self.inited_query_count = self.inited_query_count + 1 + return self.inited + + def _maybe_throw(self): + if self.error is not None: + raise self.error + + def force_set(self, kind, item): + items = self.data.get(kind) + if items is None: + items = {} + self.data[kind] = items + items[item.get('key')] = item + + def force_remove(self, kind, key): + items = self.data.get(kind) + if items is not None: + 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): + wrapper = make_wrapper(MockAsyncCore(), False) + assert wrapper.is_monitoring_enabled() is False + + @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.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_item(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + key = "flag" + itemv1 = {"key": key, "version": 1} + itemv2 = {"key": key, "version": 2} + + core.force_set(THINGS, itemv1) + assert await wrapper.get(THINGS, key) == itemv1 + + core.force_set(THINGS, itemv2) + # if cached, we will not see the new underlying value yet + assert await wrapper.get(THINGS, key) == (itemv1 if cached else itemv2) + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_deleted_item(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + key = "flag" + itemv1 = {"key": key, "version": 1, "deleted": True} + itemv2 = {"key": key, "version": 2} + + core.force_set(THINGS, itemv1) + assert await wrapper.get(THINGS, key) is None # filtered out because deleted is true + + core.force_set(THINGS, itemv2) + assert await wrapper.get(THINGS, key) == (None if cached else itemv2) + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_missing_item(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + key = "flag" + item = {"key": key, "version": 1} + + assert await wrapper.get(THINGS, key) is None + + core.force_set(THINGS, item) + # the cache can retain a None result + assert await wrapper.get(THINGS, key) == (None if cached else item) + + def test_cached_get_uses_values_from_init(self): + async def run(): + core = MockAsyncCore() + wrapper = make_wrapper(core, True) + item1 = {"key": "flag1", "version": 1} + item2 = {"key": "flag2", "version": 1} + + await wrapper.init({THINGS: {item1["key"]: item1, item2["key"]: item2}}) + core.force_remove(THINGS, item1["key"]) + assert await wrapper.get(THINGS, item1["key"]) == item1 + asyncio.run(run()) + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_can_throw_exception(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + core.error = CustomError() + with pytest.raises(CustomError): + await wrapper.get(THINGS, "key") + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_all(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + item1 = {"key": "flag1", "version": 1} + item2 = {"key": "flag2", "version": 1} + + core.force_set(THINGS, item1) + core.force_set(THINGS, item2) + assert await wrapper.all(THINGS) == {item1["key"]: item1, item2["key"]: item2} + + core.force_remove(THINGS, item2["key"]) + if cached: + assert await wrapper.all(THINGS) == {item1["key"]: item1, item2["key"]: item2} + else: + assert await wrapper.all(THINGS) == {item1["key"]: item1} + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_all_removes_deleted_items(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + item1 = {"key": "flag1", "version": 1} + item2 = {"key": "flag2", "version": 1, "deleted": True} + + core.force_set(THINGS, item1) + core.force_set(THINGS, item2) + assert await wrapper.all(THINGS) == {item1["key"]: item1} + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_all_changes_None_to_empty_dict(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + assert await wrapper.all(WRONG_THINGS) == {} + + def test_cached_get_all_uses_values_from_init(self): + async def run(): + core = MockAsyncCore() + wrapper = make_wrapper(core, True) + item1 = {"key": "flag1", "version": 1} + item2 = {"key": "flag2", "version": 1} + both = {item1["key"]: item1, item2["key"]: item2} + + await wrapper.init({THINGS: both}) + core.force_remove(THINGS, item1["key"]) + assert await wrapper.all(THINGS) == both + asyncio.run(run()) + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_get_all_can_throw_exception(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + core.error = CustomError() + with pytest.raises(CustomError): + await wrapper.all(THINGS) + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_upsert_successful(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + key = "flag" + itemv1 = {"key": key, "version": 1} + itemv2 = {"key": key, "version": 2} + + assert await wrapper.upsert(THINGS, itemv1) is True + assert core.data[THINGS][key] == itemv1 + + assert await wrapper.upsert(THINGS, itemv2) is True + assert core.data[THINGS][key] == itemv2 + + # if we have a cache, verify that the new item is now cached by writing a different value + # to the underlying data - get should still return the cached item + if cached: + itemv3 = {"key": key, "version": 3} + core.force_set(THINGS, itemv3) + + assert await wrapper.get(THINGS, key) == itemv2 + + @pytest.mark.asyncio + async def test_cached_upsert_unsuccessful(self): + core = MockAsyncCore() + wrapper = make_wrapper(core, True) + key = "flag" + itemv1 = {"key": key, "version": 1} + itemv2 = {"key": key, "version": 2} + + assert await wrapper.upsert(THINGS, itemv2) is True + assert core.data[THINGS][key] == itemv2 + + assert await wrapper.upsert(THINGS, itemv1) is False + assert core.data[THINGS][key] == itemv2 # value in store remains the same + + itemv3 = {"key": key, "version": 3} + core.force_set(THINGS, itemv3) # bypasses cache so we can verify itemv2 is in the cache + assert await wrapper.get(THINGS, key) == itemv2 + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_upsert_can_throw_exception(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + core.error = CustomError() + with pytest.raises(CustomError): + await wrapper.upsert(THINGS, {"key": "x", "version": 1}) + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_delete(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + key = "flag" + itemv1 = {"key": key, "version": 1} + itemv2 = {"key": key, "version": 2, "deleted": True} + itemv3 = {"key": key, "version": 3} + + core.force_set(THINGS, itemv1) + assert await wrapper.get(THINGS, key) == itemv1 + + assert await wrapper.delete(THINGS, key, 2) is True + assert core.data[THINGS][key] == itemv2 + + core.force_set(THINGS, itemv3) # make a change that bypasses the cache + assert await wrapper.get(THINGS, key) == (None if cached else itemv3) + + @pytest.mark.asyncio + @pytest.mark.parametrize("cached", [False, True]) + async def test_delete_can_throw_exception(self, cached): + core = MockAsyncCore() + wrapper = make_wrapper(core, cached) + core.error = CustomError() + with pytest.raises(CustomError): + await wrapper.delete(THINGS, "x", 1) + + @pytest.mark.asyncio + async def test_not_initialized_before_init(self): + core = MockAsyncCore() + wrapper = make_wrapper(core, False) + assert wrapper.initialized is False + + @pytest.mark.asyncio + async def test_initialized_after_init(self): + core = MockAsyncCore() + wrapper = make_wrapper(core, False) + await wrapper.init({}) + 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) + await wrapper.close() + assert closed["value"] is True + + @pytest.mark.asyncio + async def test_describe_configuration_delegates_to_core(self): + class DescribedCore(MockAsyncCore): + def describe_configuration(self, config): + return "MyStore" + + wrapper = make_wrapper(DescribedCore(), False) + assert wrapper.describe_configuration(None) == "MyStore" + + @pytest.mark.asyncio + async def test_describe_configuration_defaults_to_custom(self): + wrapper = make_wrapper(MockAsyncCore(), False) + assert wrapper.describe_configuration(None) == "custom" From 1f4d0f747ba3a6a95973f943f108eed782872437 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 13 Aug 2026 10:40:25 -0500 Subject: [PATCH 2/3] feat: Cap the async Redis feature store upsert retry loop at 10 attempts --- .../impl/integrations/redis/async_redis_feature_store.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/ldclient/impl/integrations/redis/async_redis_feature_store.py b/ldclient/impl/integrations/redis/async_redis_feature_store.py index 8391cfca..d3fed4ad 100644 --- a/ldclient/impl/integrations/redis/async_redis_feature_store.py +++ b/ldclient/impl/integrations/redis/async_redis_feature_store.py @@ -15,6 +15,9 @@ except ImportError: pass +# Cap the WATCH-retry loop so a hot-contended key can't starve upsert_internal forever; matches the LaunchDarkly Go Redis stores. +_MAX_UPSERT_RETRIES = 10 + class _AsyncRedisFeatureStoreCore(DiagnosticDescription, AsyncFeatureStoreCore): """Async Redis implementation of :class:`ldclient.interfaces.AsyncFeatureStoreCore`. @@ -76,7 +79,7 @@ async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict: key = item['key'] item_json = json.dumps(item) - while True: + for _ in range(_MAX_UPSERT_RETRIES): async with self._client.pipeline() as pipe: try: await pipe.watch(base_key) @@ -104,6 +107,8 @@ async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict: log.debug("AsyncRedisFeatureStore: concurrent modification detected, retrying") continue + raise RuntimeError("failed to update key %s in '%s' after %d attempts" % (key, kind.namespace, _MAX_UPSERT_RETRIES)) + async def initialized_internal(self) -> bool: return bool(await self._client.exists(self._init_key)) From aefa42819890ef19258294ae46540c37d541b79a Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 13 Aug 2026 10:50:19 -0500 Subject: [PATCH 3/3] fix: Cap the sync Redis feature store upsert retry loop at 10 attempts --- ldclient/impl/integrations/redis/redis_feature_store.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/ldclient/impl/integrations/redis/redis_feature_store.py b/ldclient/impl/integrations/redis/redis_feature_store.py index 95a95d14..b44f39c0 100644 --- a/ldclient/impl/integrations/redis/redis_feature_store.py +++ b/ldclient/impl/integrations/redis/redis_feature_store.py @@ -15,6 +15,9 @@ except ImportError: pass +# Cap the WATCH-retry loop so a hot-contended key can't starve upsert_internal forever; matches the LaunchDarkly Go Redis stores. +_MAX_UPSERT_RETRIES = 10 + class _RedisFeatureStoreCore(DiagnosticDescription, FeatureStoreCore): def __init__(self, url, prefix, redis_opts: Dict[str, Any]): @@ -82,7 +85,7 @@ def upsert_internal(self, kind, item): key = item['key'] item_json = json.dumps(item) - while True: + for _ in range(_MAX_UPSERT_RETRIES): pipeline = r.pipeline() pipeline.watch(base_key) old = self.get_internal(kind, key) @@ -111,6 +114,8 @@ def upsert_internal(self, kind, item): continue return item + raise RuntimeError("failed to update key %s in '%s' after %d attempts" % (key, kind.namespace, _MAX_UPSERT_RETRIES)) + def initialized_internal(self): r = redis.Redis(connection_pool=self._pool) return r.exists(self._init_key)