feat!: Require feature stores to report whether an upsert or delete applied

lib/ldclient-rb/impl/data_source.rb

@@ -107,10 +107,9 @@ def init(all_data)
         end
 
         def upsert(kind, item)
-          monitor_store_update { @data_store.upsert(kind, item) }
+          updated = monitor_store_update { @data_store.upsert(kind, item) }
+          return unless updated
 
-          # TODO(sc-197908): We only want to do this if the store successfully
-          # updates the record.
           @dependency_tracker.update_dependencies_from(kind, item[:key], item)
           if @flag_change_broadcaster.has_listeners?
             affected_items = Set.new
@@ -120,7 +119,8 @@ def upsert(kind, item)
         end
 
         def delete(kind, key, version)
-          monitor_store_update { @data_store.delete(kind, key, version) }
+          deleted = monitor_store_update { @data_store.delete(kind, key, version) }
+          return unless deleted
 
           @dependency_tracker.update_dependencies_from(kind, key, nil)
           if @flag_change_broadcaster.has_listeners?

lib/ldclient-rb/impl/integrations/consul_impl.rb

@@ -92,19 +92,16 @@ def upsert_internal(kind, new_item)
                 mod_index = 0
               else
                 old_item = Model.deserialize(kind, old_value[0]["Value"])
-                # Check whether the item is stale. If so, don't do the update (and return the existing item to
-                # FeatureStoreWrapper so it can be cached)
-                if old_item[:version] >= new_item[:version]
-                  return old_item
-                end
+                # Check whether the item is stale. If so, don't do the update.
+                return false if old_item[:version] >= new_item[:version]
                 mod_index = old_value[0]["ModifyIndex"]
               end
 
               # Otherwise, try to write. We will do a compare-and-set operation, so the write will only succeed if
               # the key's ModifyIndex is still equal to the previous value. If the previous ModifyIndex was zero,
               # it means the key did not previously exist and the write will only succeed if it still doesn't exist.
               success = Diplomat::Kv.put(key, json, cas: mod_index)
-              return new_item if success
+              return true if success
 
               # If we failed, retry the whole shebang
               @logger.debug { "Concurrent modification detected, retrying" }

lib/ldclient-rb/impl/integrations/dynamodb_impl.rb

@@ -140,11 +140,10 @@ def upsert_internal(kind, new_item)
                   ":version" => new_item[:version],
                 },
               })
-              new_item
+              true
             rescue Aws::DynamoDB::Errors::ConditionalCheckFailedException
               # The item was not updated because there's a newer item in the database.
-              # We must now read the item that's in the database and return it, so CachingStoreWrapper can cache it.
-              get_internal(kind, new_item[:key])
+              false
             end
           end
 

lib/ldclient-rb/impl/integrations/redis_impl.rb

@@ -217,7 +217,7 @@ def upsert_internal(kind, new_item)
             base_key = items_key(kind)
             key = new_item[:key]
             try_again = true
-            final_item = new_item
+            updated = true
             while try_again
               try_again = false
               with_connection do |redis|
@@ -233,7 +233,7 @@ def upsert_internal(kind, new_item)
                       try_again = true
                     end
                   else
-                    final_item = old_item
+                    updated = false
                     action = new_item[:deleted] ? "delete" : "update"
                     # rubocop:disable Layout/LineLength
                     @logger.warn { "RedisFeatureStore: attempted to #{action} #{key} version: #{old_item[:version]} in '#{kind[:namespace]}' with a version that is the same or older: #{new_item[:version]}" }
@@ -242,7 +242,7 @@ def upsert_internal(kind, new_item)
                 end
               end
             end
-            final_item
+            updated
           end
 
           def initialized_internal?

lib/ldclient-rb/in_memory_store.rb

@@ -47,10 +47,10 @@ def delete(kind, key, version)
           @items[kind] = coll
         end
         old = coll[key.to_sym]
+        return false unless old.nil? || old[:version] < version
 
-        if old.nil? || old[:version] < version
-          coll[key.to_sym] = { deleted: true, version: version }
-        end
+        coll[key.to_sym] = { deleted: true, version: version }
+        true
       end
     end
 
@@ -69,10 +69,10 @@ def upsert(kind, item)
           @items[kind] = coll
         end
         old = coll[item[:key].to_sym]
+        return false unless old.nil? || old[:version] < item[:version]
 
-        if old.nil? || old[:version] < item[:version]
-          coll[item[:key].to_sym] = item
-        end
+        coll[item[:key].to_sym] = item
+        true
       end
     end
 

lib/ldclient-rb/integrations/util/store_wrapper.rb

@@ -98,13 +98,16 @@ def all(kind)
         end
 
         def upsert(kind, item)
-          new_state = @core.upsert_internal(kind, item)
+          updated = @core.upsert_internal(kind, item)
 
           cache = @cache
           unless cache.nil?
+            new_state = updated ? item : @core.get_internal(kind, item[:key])
             cache[item_cache_key(kind, item[:key])] = [new_state]
             cache.delete(all_cache_key(kind))
           end
+
+          updated
         end
 
         def delete(kind, key, version)
@@ -225,17 +228,15 @@ def get_all_internal(kind)
 
         #
         # Attempts to add or update an entity. This is the same as {LaunchDarkly::Interfaces::FeatureStore#upsert}
-        # except that 1. the wrapper will take care of updating the cache if caching is enabled, and 2.
-        # the method is expected to return the final state of the entity (i.e. either the `item`
-        # parameter if the update succeeded, or the previously existing entity in the store if the
-        # update failed; this is used for the caching logic).
+        # except that the wrapper updates the cache if caching is enabled.
         #
         # Note that FeatureStoreCore does not have a `delete` method. This is because {CachingStoreWrapper}
         # implements `delete` by simply calling `upsert` with an item whose `:deleted` property is true.
         #
         # @param kind [Object]  the kind of entity to add or update
         # @param item [Hash]  the entity to add or update
-        # @return [Hash]  the entity as it now exists in the store after the update
+        # @return [Boolean]  true if the store applied the update; false if the store already had an
+        #   item with the same or a higher version
         #
         def upsert_internal(kind, item)
         end

lib/ldclient-rb/interfaces/feature_store.rb

@@ -68,9 +68,12 @@ def all(kind)
       # should only succeed if the new item's `:version` is greater than the old one;
       # otherwise, the method should do nothing.
       #
+      # The SDK uses the return value to decide whether to send flag change events.
+      #
       # @param kind [Object]  the kind of entity to add or update
       # @param item [Hash]  the entity to add or update
-      # @return [void]
+      # @return [Boolean]  true if the store applied the update; false if the store already
+      #   had an item with the same or a higher version
       #
       def upsert(kind, item)
       end
@@ -80,10 +83,13 @@ def upsert(kind, item)
       # `version` parameter is greater than the existing entity's `:version`; otherwise, the
       # method should do nothing.
       #
+      # The SDK uses the return value to decide whether to send flag change events.
+      #
       # @param kind [Object]  the kind of entity to delete
       # @param key [String]  the unique key of the entity
       # @param version [Integer]  the entity must have a lower version than this to be deleted
-      # @return [void]
+      # @return [Boolean]  true if the store applied the deletion; false if the store already
+      #   had an item with the same or a higher version
       #
       def delete(kind, key, version)
       end

spec/feature_store_spec_base.rb

@@ -137,52 +137,52 @@ def new_version_plus(f, delta_version, attrs = {})
 
   it "can add new item" do
     with_inited_store([]) do |store|
-      store.upsert($things_kind, $thing1)
+      expect(store.upsert($things_kind, $thing1)).to be true
       expect(store.get($things_kind, $key1)).to eq $thing1
     end
   end
 
   it "can update item with newer version" do
     with_inited_store([ $thing1 ]) do |store|
       $thing1_mod = new_version_plus($thing1, 1, { name: $thing1[:name] + ' updated' })
-      store.upsert($things_kind, $thing1_mod)
+      expect(store.upsert($things_kind, $thing1_mod)).to be true
       expect(store.get($things_kind, $key1)).to eq $thing1_mod
     end
   end
 
   it "cannot update item with same version" do
     with_inited_store([ $thing1 ]) do |store|
       $thing1_mod = $thing1.clone.merge({ name: $thing1[:name] + ' updated' })
-      store.upsert($things_kind, $thing1_mod)
+      expect(store.upsert($things_kind, $thing1_mod)).to be false
       expect(store.get($things_kind, $key1)).to eq $thing1
     end
   end
 
   it "cannot update feature with older version" do
     with_inited_store([ $thing1 ]) do |store|
       $thing1_mod = new_version_plus($thing1, -1, { name: $thing1[:name] + ' updated' })
-      store.upsert($things_kind, $thing1_mod)
+      expect(store.upsert($things_kind, $thing1_mod)).to be false
       expect(store.get($things_kind, $key1)).to eq $thing1
     end
   end
 
   it "can delete item with newer version" do
     with_inited_store([ $thing1 ]) do |store|
-      store.delete($things_kind, $key1, $thing1[:version] + 1)
+      expect(store.delete($things_kind, $key1, $thing1[:version] + 1)).to be true
       expect(store.get($things_kind, $key1)).to be_nil
     end
   end
 
   it "cannot delete item with same version" do
     with_inited_store([ $thing1 ]) do |store|
-      store.delete($things_kind, $key1, $thing1[:version])
+      expect(store.delete($things_kind, $key1, $thing1[:version])).to be false
       expect(store.get($things_kind, $key1)).to eq $thing1
     end
   end
 
   it "cannot delete item with older version" do
     with_inited_store([ $thing1 ]) do |store|
-      store.delete($things_kind, $key1, $thing1[:version] - 1)
+      expect(store.delete($things_kind, $key1, $thing1[:version] - 1)).to be false
       expect(store.get($things_kind, $key1)).to eq $thing1
     end
   end

spec/impl/data_source_spec.rb

@@ -161,9 +161,8 @@ module Impl
           flag_change_broadcaster.add_listener(listener)
 
           sink.upsert(LaunchDarkly::Impl::DataStore::FEATURES, LaunchDarkly::Impl::Model::FeatureFlag.new({ key: 'flag1', version: 2 }))
-          # TODO(sc-197908): Once the store starts returning a success status on upsert, the flag change notification
-          # can start ignoring duplicate requests like this.
-          # sink.upsert(LaunchDarkly::Impl::DataStore::FEATURES, LaunchDarkly::Impl::Model::FeatureFlag.new({ key: 'flag1', version: 2 }))
+          # A duplicate update does not change the store, so it sends no event.
+          sink.upsert(LaunchDarkly::Impl::DataStore::FEATURES, LaunchDarkly::Impl::Model::FeatureFlag.new({ key: 'flag1', version: 2 }))
 
           expect(listener.statuses.count).to eq(1)
           expect(listener.statuses[0].key).to eq('flag1')
@@ -176,9 +175,8 @@ module Impl
           flag_change_broadcaster.add_listener(listener)
 
           sink.delete(LaunchDarkly::Impl::DataStore::FEATURES, "flag1", 2)
-          # TODO(sc-197908): Once the store starts returning a success status on delete, the flag change notification
-          # can start ignoring duplicate requests like this.
-          # sink.delete(LaunchDarkly::Impl::DataStore::FEATURES, :flag1, 2)
+          # A duplicate update does not change the store, so it sends no event.
+          sink.delete(LaunchDarkly::Impl::DataStore::FEATURES, "flag1", 2)
 
           expect(listener.statuses.count).to eq(1)
           expect(listener.statuses[0].key).to eq("flag1")
@@ -191,9 +189,8 @@ module Impl
           flag_change_broadcaster.add_listener(listener)
 
           sink.upsert(LaunchDarkly::Impl::DataStore::SEGMENTS, LaunchDarkly::Impl::Model::Segment.new({ key: 'segment2', version: 2 }))
-          # TODO(sc-197908): Once the store starts returning a success status on upsert, the flag change notification
-          # can start ignoring duplicate requests like this.
-          # sink.upsert(LaunchDarkly::Impl::Model::Segment.new({ key: 'segment2', version: 2 }))
+          # A duplicate update does not change the store, so it sends no event.
+          sink.upsert(LaunchDarkly::Impl::DataStore::SEGMENTS, LaunchDarkly::Impl::Model::Segment.new({ key: 'segment2', version: 2 }))
 
           expect(listener.statuses.count).to eq(1)
           expect(listener.statuses[0].key).to eq('flag3')
@@ -324,9 +321,8 @@ module Impl
           flag_change_broadcaster.add_listener(listener)
 
           sink.upsert(LaunchDarkly::Impl::DataStore::SEGMENTS, LaunchDarkly::Impl::Model::Segment.new({ key: 'segment1', version: 2 }))
-          # TODO(sc-197908): Once the store starts returning a success status on upsert, the flag change notification
-          # can start ignoring duplicate requests like this.
-          # sink.upsert(LaunchDarkly::Impl::DataStore::SEGMENTS, LaunchDarkly::Impl::Model::Segment.new({ key: 'segment1', version: 2 }))
+          # A duplicate update does not change the store, so it sends no event.
+          sink.upsert(LaunchDarkly::Impl::DataStore::SEGMENTS, LaunchDarkly::Impl::Model::Segment.new({ key: 'segment1', version: 2 }))
 
           expect(listener.statuses.count).to eq(3)
           expect(listener.statuses[0].key).to eq('flag6')
@@ -341,9 +337,8 @@ module Impl
           flag_change_broadcaster.add_listener(listener)
 
           sink.delete(LaunchDarkly::Impl::DataStore::SEGMENTS, 'segment2', 2)
-          # TODO(sc-197908): Once the store starts returning a success status on upsert, the flag change notification
-          # can start ignoring duplicate requests like this.
-          # sink.delete(LaunchDarkly::Impl::DataStore::SEGMENTS, 'segment2', 2)
+          # A duplicate update does not change the store, so it sends no event.
+          sink.delete(LaunchDarkly::Impl::DataStore::SEGMENTS, 'segment2', 2)
 
           expect(listener.statuses.count).to eq(3)
           expect(listener.statuses[0].key).to eq('flag6')

spec/integrations/util/store_wrapper_spec.rb

@@ -101,10 +101,10 @@ module Integrations
           itemv1 = { key: key, version: 1 }
           itemv2 = { key: key, version: 2 }
 
-          wrapper.upsert(THINGS, itemv1)
+          expect(wrapper.upsert(THINGS, itemv1)).to be true
           expect(core.data[THINGS][key]).to eq itemv1
 
-          wrapper.upsert(THINGS, itemv2)
+          expect(wrapper.upsert(THINGS, itemv2)).to be true
           expect(core.data[THINGS][key]).to eq itemv2
 
           # if we have a cache, verify that the new item is now cached by writing a different value
@@ -128,13 +128,34 @@ module Integrations
           core.force_set(THINGS, itemv1)
           expect(wrapper.get(THINGS, key)).to eq itemv1
 
-          wrapper.delete(THINGS, key, 2)
+          expect(wrapper.delete(THINGS, key, 2)).to be true
           expect(core.data[THINGS][key]).to eq itemv2
 
           core.force_set(THINGS, itemv3)  # make a change that bypasses the cache
 
           expect(wrapper.get(THINGS, key)).to eq(cached ? nil : itemv3)
         end
+
+        it "returns false when upsert has a version that is not newer" do
+          core = MockCore.new
+          wrapper = subject.new(core, opts)
+          itemv2 = { key: "flag", version: 2 }
+          core.force_set(THINGS, itemv2)
+
+          expect(wrapper.upsert(THINGS, { key: "flag", version: 2 })).to be false
+          expect(wrapper.upsert(THINGS, { key: "flag", version: 1 })).to be false
+          expect(core.data[THINGS]["flag"]).to eq itemv2
+        end
+
+        it "returns false when delete has a version that is not newer" do
+          core = MockCore.new
+          wrapper = subject.new(core, opts)
+          itemv2 = { key: "flag", version: 2 }
+          core.force_set(THINGS, itemv2)
+
+          expect(wrapper.delete(THINGS, "flag", 2)).to be false
+          expect(core.data[THINGS]["flag"]).to eq itemv2
+        end
       end
 
       context "cached" do
@@ -188,6 +209,21 @@ module Integrations
           expect(wrapper.get(THINGS, key)).to eq itemv2
         end
 
+        it "upsert caches the item from the store if unsuccessful" do
+          core = MockCore.new
+          wrapper = subject.new(core, cached_opts)
+          key = "flag"
+          itemv1 = { key: key, version: 1 }
+          itemv3 = { key: key, version: 3 }
+
+          wrapper.upsert(THINGS, itemv1)
+          core.force_set(THINGS, itemv3)  # another process writes a newer version
+
+          expect(wrapper.upsert(THINGS, { key: key, version: 2 })).to be false
+          core.force_remove(THINGS, key)  # bypasses cache so we can verify that itemv3 is in the cache
+          expect(wrapper.get(THINGS, key)).to eq itemv3
+        end
+
         it "initialized? can cache false result" do
           core = MockCore.new
           wrapper = subject.new(core, { expiration: 0.2 })  # use a shorter cache TTL for this test
@@ -344,9 +380,9 @@ def get_all_internal(kind)
         def upsert_internal(kind, item)
           @data[kind] = {} unless @data.has_key?(kind)
           old_item = @data[kind][item[:key]]
-          return old_item if !old_item.nil? && old_item[:version] >= item[:version]
+          return false if !old_item.nil? && old_item[:version] >= item[:version]
           @data[kind][item[:key]] = item
-          item
+          true
         end
 
         def initialized_internal?