feat!: Require feature stores to report whether a write was applied (#554)

fix: Prevent flag change listeners from firing for writes the store
rejects

Author: keelerm84

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:

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:

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)

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

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

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

ldclient/testing/feature_store_test_base.py

@@ -96,41 +96,41 @@ 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
 
     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

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

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:

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