From 48742ef748879411a730777c61c430573e39f87f Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Mon, 2 Nov 2026 11:39:06 -0500 Subject: [PATCH] feat!: Require feature stores to report whether a write was applied fix: Flag change listeners no longer fire for writes the store rejects FeatureStore.upsert and FeatureStore.delete must now return a bool. Return True if the store applied the write, or False if it rejected the write because the stored version is equal or newer. This matches the async store contract. CachingStoreWrapper derives the result from FeatureStoreCore.upsert_internal, so custom cores need no change if they return the passed-in item on success. The data source update sink now updates dependencies and sends flag change events only when the store applied the write. --- ldclient/feature_store.py | 11 ++++--- ldclient/feature_store_helpers.py | 9 ++++-- ldclient/impl/datasource/status.py | 24 ++++++++------- ldclient/impl/datasystem/fdv1.py | 4 +-- ldclient/impl/datasystem/fdv2.py | 4 +-- ldclient/interfaces.py | 20 +++++++++++-- ldclient/testing/feature_store_test_base.py | 14 ++++----- .../impl/datasystem/test_cache_lifecycle.py | 4 +-- .../impl/datasystem/test_fdv2_persistence.py | 8 +++-- ldclient/testing/impl/test_data_sink.py | 29 ++++++++++++------- ldclient/testing/stub_util.py | 4 +-- .../testing/test_feature_store_helpers.py | 10 +++---- ldclient/testing/test_ldclient_evaluation.py | 10 +++---- 13 files changed, 94 insertions(+), 57 deletions(-) diff --git a/ldclient/feature_store.py b/ldclient/feature_store.py index f327e4cd..974063f8 100644 --- a/ldclient/feature_store.py +++ b/ldclient/feature_store.py @@ -119,16 +119,17 @@ def init(self, all_data): log.debug("Initialized '%s' store with %d items", k.namespace, len(all_data[k])) # noinspection PyShadowingNames - def delete(self, kind, key: str, version: int): + def delete(self, kind, key: str, version: int) -> bool: """ """ with self._lock.write(): itemsOfKind = self._items[kind] i = itemsOfKind.get(key) if i is None or i['version'] < version: - i = {'deleted': True, 'version': version} - itemsOfKind[key] = i + itemsOfKind[key] = {'deleted': True, 'version': version} + return True + return False - def upsert(self, kind, item): + def upsert(self, kind, item) -> bool: """ """ decoded_item = kind.decode(item) key = item['key'] @@ -138,6 +139,8 @@ def upsert(self, kind, item): if i is None or i['version'] < item['version']: itemsOfKind[key] = decoded_item log.debug("Updated %s in '%s' to version %d", key, kind.namespace, item['version']) + return True + return False @property def initialized(self) -> bool: diff --git a/ldclient/feature_store_helpers.py b/ldclient/feature_store_helpers.py index 19cec91c..3b8b4159 100644 --- a/ldclient/feature_store_helpers.py +++ b/ldclient/feature_store_helpers.py @@ -230,16 +230,19 @@ def all(self, kind, callback=lambda x: x): encoded_items = self._core.get_all_internal(kind) return callback(self._cache_put_all(kind, encoded_items)) - def delete(self, kind, key, version): + def delete(self, kind, key, version) -> bool: """ """ deleted_item = {"key": key, "version": version, "deleted": True} - self.upsert(kind, deleted_item) + return self.upsert(kind, deleted_item) - def upsert(self, kind, encoded_item): + def upsert(self, kind, encoded_item) -> bool: """ """ encoded_item = _ensure_encoded(kind, encoded_item) new_state = self._core.upsert_internal(kind, encoded_item) self._cache_put_upsert(kind, new_state) + # The core returns the item that we passed in if it applied the write. It returns the + # existing item if the version check rejected the write. Identity thus shows the result. + return new_state is encoded_item @property def initialized(self) -> bool: diff --git a/ldclient/impl/datasource/status.py b/ldclient/impl/datasource/status.py index cec2ed24..fbe761bd 100644 --- a/ldclient/impl/datasource/status.py +++ b/ldclient/impl/datasource/status.py @@ -1,5 +1,5 @@ import time -from typing import Callable, Mapping, Optional, Protocol, Set +from typing import Any, Callable, Mapping, Optional, Protocol, Set from ldclient.impl.dependency_tracker import DependencyTracker, KindAndKey from ldclient.impl.listeners import Listeners @@ -63,16 +63,20 @@ def init_store(): self.__send_change_events(self.__compute_changed_items_for_full_data_set(old_data, all_data)) def upsert(self, kind: VersionedDataKind, item: dict): - self.__monitor_store_update(lambda: self.__store.upsert(kind, item)) + updated = self.__monitor_store_update(lambda: self.__store.upsert(kind, item)) - # TODO(SDK-62): We only want to do this if the store successfully wrote the record, - # which requires upsert to return a bool result (breaking change). - key = item.get('key', '') - self.__update_dependency_for_single_item(kind, key, item) + # Update dependencies and send change events only if the store applied the write. + # A rejected write does not change the stored data. + if updated: + key = item.get('key', '') + self.__update_dependency_for_single_item(kind, key, item) def delete(self, kind: VersionedDataKind, key: str, version: int): - self.__monitor_store_update(lambda: self.__store.delete(kind, key, version)) - self.__update_dependency_for_single_item(kind, key, None) + deleted = self.__monitor_store_update(lambda: self.__store.delete(kind, key, version)) + + # Update dependencies and send change events only if the store applied the delete. + if deleted: + self.__update_dependency_for_single_item(kind, key, None) def update_status(self, new_state: DataSourceState, new_error: Optional[DataSourceErrorInfo]): status_to_broadcast = None @@ -98,9 +102,9 @@ def update_status(self, new_state: DataSourceState, new_error: Optional[DataSour if status_to_broadcast is not None: self.__status_listeners.notify(status_to_broadcast) - def __monitor_store_update(self, fn: Callable[[], None]): + def __monitor_store_update(self, fn: Callable[[], Any]) -> Any: try: - fn() + return fn() except Exception as e: error_info = DataSourceErrorInfo(DataSourceErrorKind.STORE_ERROR, 0, time.time(), str(e)) self.update_status(DataSourceState.INTERRUPTED, error_info) diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index 09ca6939..477b6e0f 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -61,10 +61,10 @@ def get(self, kind, key, callback): def all(self, kind, callback): return self.__wrapper(lambda: self.store.all(kind, callback)) - def delete(self, kind, key, version): + def delete(self, kind, key, version) -> bool: return self.__wrapper(lambda: self.store.delete(kind, key, version)) - def upsert(self, kind, item): + def upsert(self, kind, item) -> bool: return self.__wrapper(lambda: self.store.upsert(kind, item)) @property diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 1fbcbd33..afb80088 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -83,10 +83,10 @@ def get(self, kind, key, callback): def all(self, kind, callback): return self.__wrapper(lambda: self.store.all(kind, callback)) - def delete(self, kind, key, version): + def delete(self, kind, key, version) -> bool: return self.__wrapper(lambda: self.store.delete(kind, key, version)) - def upsert(self, kind, item): + def upsert(self, kind, item) -> bool: return self.__wrapper(lambda: self.store.upsert(kind, item)) @property diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index de403354..0ecd6026 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -111,26 +111,36 @@ def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]): """ @abstractmethod - def delete(self, kind: VersionedDataKind, key: str, version: int): + def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: """ Deletes the object associated with the specified key, if it exists and its version is less than the specified version. The object should be replaced in the data store by a - placeholder with the specified version and a "deleted" property of TErue. + placeholder with the specified version and a "deleted" property of True. + + Implementations must return whether they applied the delete. The SDK uses this result to + decide whether to send flag change events. :param kind: The kind of object to delete :param key: The key of the object to be deleted :param version: The version for the delete operation + :return: True if the store was updated (a newer delete placeholder was written); False if the + delete was rejected because the stored version was equal or newer """ @abstractmethod - def upsert(self, kind: VersionedDataKind, item: dict): + def upsert(self, kind: VersionedDataKind, item: dict) -> bool: """ 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. + Implementations must return whether they applied the update. The SDK uses this result to + decide whether to send flag change events. + :param kind: The kind of object to update :param item: The object to update or insert + :return: True if the store was updated (the item was new or a newer version); False if the + update was rejected because the stored version was equal or newer """ @property @@ -267,6 +277,10 @@ def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict: then it returns the item that is currently in the data store (this ensures that ``CachingStoreWrapper`` will update the cache correctly). + ``CachingStoreWrapper`` compares the returned object to the object that it passed in. If they + are the same object, it reports the write as applied. Thus, on success, return the same + object that was passed in, not a copy. + :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 diff --git a/ldclient/testing/feature_store_test_base.py b/ldclient/testing/feature_store_test_base.py index c37b1aa3..dc2e1fd8 100644 --- a/ldclient/testing/feature_store_test_base.py +++ b/ldclient/testing/feature_store_test_base.py @@ -96,35 +96,35 @@ def test_get_all_versions(self, tester): def test_upsert_with_newer_version(self, tester): with self.inited_store(tester) as store: new_ver = self.make_feature('foo', 11) - store.upsert(FEATURES, new_ver) + assert store.upsert(FEATURES, new_ver) is True assert store.get(FEATURES, 'foo', lambda x: x) == new_ver def test_upsert_with_older_version(self, tester): with self.inited_store(tester) as store: new_ver = self.make_feature('foo', 9) expected = self.make_feature('foo', 10) - store.upsert(FEATURES, new_ver) + assert store.upsert(FEATURES, new_ver) is False assert store.get(FEATURES, 'foo', lambda x: x) == expected def test_upsert_with_new_feature(self, tester): with self.inited_store(tester) as store: new_ver = self.make_feature('biz', 1) - store.upsert(FEATURES, new_ver) + assert store.upsert(FEATURES, new_ver) is True assert store.get(FEATURES, 'biz', lambda x: x) == new_ver def test_delete_with_newer_version(self, tester): with self.inited_store(tester) as store: - store.delete(FEATURES, 'foo', 11) + assert store.delete(FEATURES, 'foo', 11) is True assert store.get(FEATURES, 'foo', lambda x: x) is None def test_delete_unknown_feature(self, tester): with self.inited_store(tester) as store: - store.delete(FEATURES, 'biz', 11) + assert store.delete(FEATURES, 'biz', 11) is True assert store.get(FEATURES, 'biz', lambda x: x) is None def test_delete_with_older_version(self, tester): with self.inited_store(tester) as store: - store.delete(FEATURES, 'foo', 9) + assert store.delete(FEATURES, 'foo', 9) is False expected = self.make_feature('foo', 10) assert store.get(FEATURES, 'foo', lambda x: x) == expected @@ -132,5 +132,5 @@ def test_upsert_older_version_after_delete(self, tester): with self.inited_store(tester) as store: store.delete(FEATURES, 'foo', 11) old_ver = self.make_feature('foo', 9) - store.upsert(FEATURES, old_ver) + assert store.upsert(FEATURES, old_ver) is False assert store.get(FEATURES, 'foo', lambda x: x) is None diff --git a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py index 52e9cdc7..aa710e48 100644 --- a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py +++ b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py @@ -295,10 +295,10 @@ def all(self, kind, callback=lambda x: x): return callback({}) def delete(self, kind, key, version): - pass + return False def upsert(self, kind, item): - pass + return False @property def initialized(self): diff --git a/ldclient/testing/impl/datasystem/test_fdv2_persistence.py b/ldclient/testing/impl/datasystem/test_fdv2_persistence.py index d9f68bd5..ce1b87a3 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_persistence.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_persistence.py @@ -68,18 +68,22 @@ def all( items = self._data.get(kind, {}) return {key: callback(value) for key, value in items.items()} - def delete(self, kind: VersionedDataKind, key: str, version: int): + def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool: self.delete_calls.append((kind, key, version)) existing = self._data.get(kind, {}).get(key) if existing and existing.get("version", 0) < version: self._data[kind][key] = {"key": key, "version": version, "deleted": True} + return True + return False - def upsert(self, kind: VersionedDataKind, item: dict): + def upsert(self, kind: VersionedDataKind, item: dict) -> bool: self.upsert_calls.append((kind, item.get("key"), item.get("version"))) key = item["key"] existing = self._data.get(kind, {}).get(key) if not existing or existing.get("version", 0) < item.get("version", 0): self._data[kind][key] = item + return True + return False @property def initialized(self) -> bool: diff --git a/ldclient/testing/impl/test_data_sink.py b/ldclient/testing/impl/test_data_sink.py index 1ec5d49e..9c901860 100644 --- a/ldclient/testing/impl/test_data_sink.py +++ b/ldclient/testing/impl/test_data_sink.py @@ -194,6 +194,19 @@ def test_upsert_triggers_flag_listener(basic_data): assert spy.statuses[0].key == 'flag1' +def test_rejected_upsert_does_not_trigger_flag_listener(basic_data): + flag_change_listener = Listeners() + sink = DataSourceUpdateSinkImpl(InMemoryFeatureStore(), Listeners(), flag_change_listener) + sink.init(basic_data) + + spy = SpyListener() + flag_change_listener.add(spy) + # The store already has version 1 of this flag, so it rejects this write. + sink.upsert(FEATURES, FlagBuilder('flag1').version(1).on(False).build()) + + assert len(spy.statuses) == 0 + + def test_delete_triggers_flag_listener(basic_data): flag_change_listener = Listeners() sink = DataSourceUpdateSinkImpl(InMemoryFeatureStore(), Listeners(), flag_change_listener) @@ -202,10 +215,8 @@ def test_delete_triggers_flag_listener(basic_data): spy = SpyListener() flag_change_listener.add(spy) sink.delete(FEATURES, 'flag1', 2) - - # TODO(sc-212471): Once the store starts returning a success status on delete, the flag change - # notification can start ignoring duplicate requests like this. - # sink.delete(FEATURES, 'flag1', 2) + # The store rejects this duplicate delete, so it sends no change event. + sink.delete(FEATURES, 'flag1', 2) assert len(spy.statuses) == 1 assert spy.statuses[0].key == 'flag1' @@ -304,9 +315,8 @@ def test_triggers_dependent_segment_is_modified(prereq_data): flag_change_listener.add(spy) sink.upsert(SEGMENTS, SegmentBuilder('segment1').version(2).build()) - # TODO(sc-212471): Once the store starts returning a success status on upsert, the flag change - # notification can start ignoring duplicate requests like this. - # sink.upsert(SEGMENTS, SegmentBuilder('segment1').version(2).build()) + # The store rejects this duplicate upsert, so it sends no change events. + sink.upsert(SEGMENTS, SegmentBuilder('segment1').version(2).build()) assert len(spy.statuses) == 3 @@ -325,9 +335,8 @@ def test_triggers_if_dependent_segment_removed(prereq_data): flag_change_listener.add(spy) sink.delete(SEGMENTS, 'segment2', 2) - # TODO(sc-212471): Once the store starts returning a success status on delete, the flag change - # notification can start ignoring duplicate requests like this. - # sink.delete(SEGMENTS, 'segment2', 2) + # The store rejects this duplicate delete, so it sends no change events. + sink.delete(SEGMENTS, 'segment2', 2) assert len(spy.statuses) == 3 diff --git a/ldclient/testing/stub_util.py b/ldclient/testing/stub_util.py index 7f5f6aab..28b11758 100644 --- a/ldclient/testing/stub_util.py +++ b/ldclient/testing/stub_util.py @@ -194,10 +194,10 @@ def all(self, kind, callback=lambda x: x): pass def delete(self, kind, key, version): - pass + return False def upsert(self, kind, item): - pass + return False @property def initialized(self): diff --git a/ldclient/testing/test_feature_store_helpers.py b/ldclient/testing/test_feature_store_helpers.py index 97e1652a..e4105d56 100644 --- a/ldclient/testing/test_feature_store_helpers.py +++ b/ldclient/testing/test_feature_store_helpers.py @@ -248,10 +248,10 @@ def test_upsert_successful(self, cached): itemv1 = {"key": key, "version": 1} itemv2 = {"key": key, "version": 2} - wrapper.upsert(THINGS, itemv1) + assert wrapper.upsert(THINGS, itemv1) is True assert core.data[THINGS][key] == itemv1 - wrapper.upsert(THINGS, itemv2) + assert 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 @@ -273,10 +273,10 @@ def test_cached_upsert_unsuccessful(self): itemv1 = {"key": key, "version": 1} itemv2 = {"key": key, "version": 2} - wrapper.upsert(THINGS, itemv2) + assert wrapper.upsert(THINGS, itemv2) is True assert core.data[THINGS][key] == itemv2 - wrapper.upsert(THINGS, itemv1) + assert wrapper.upsert(THINGS, itemv1) is False assert core.data[THINGS][key] == itemv2 # value in store remains the same itemv3 = {"key": key, "version": 3} @@ -303,7 +303,7 @@ def test_delete(self, cached): core.force_set(THINGS, itemv1) assert wrapper.get(THINGS, key) == itemv1 - wrapper.delete(THINGS, key, 2) + assert 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 diff --git a/ldclient/testing/test_ldclient_evaluation.py b/ldclient/testing/test_ldclient_evaluation.py index 8e6e0a3a..75605ee8 100644 --- a/ldclient/testing/test_ldclient_evaluation.py +++ b/ldclient/testing/test_ldclient_evaluation.py @@ -26,10 +26,10 @@ def all(self, kind, callback=lambda x: x): raise NotImplementedError() def upsert(self, kind, item): - pass + return False - def delete(self, key, version): - pass + def delete(self, kind, key, version): + return False def init(self, data): pass @@ -203,10 +203,10 @@ def all(self, kind, callback=lambda x: x): return callback(dict(self._flags) if kind == FEATURES else {}) def upsert(self, kind, item): - pass + return False def delete(self, kind, key, version): - pass + return False def init(self, data): pass