diff --git a/lib/ldclient-rb/impl/data_source.rb b/lib/ldclient-rb/impl/data_source.rb index 8459d81f..1be1a46b 100644 --- a/lib/ldclient-rb/impl/data_source.rb +++ b/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? diff --git a/lib/ldclient-rb/impl/integrations/consul_impl.rb b/lib/ldclient-rb/impl/integrations/consul_impl.rb index d3e8a1a2..e7ee6a8d 100644 --- a/lib/ldclient-rb/impl/integrations/consul_impl.rb +++ b/lib/ldclient-rb/impl/integrations/consul_impl.rb @@ -92,11 +92,8 @@ 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 @@ -104,7 +101,7 @@ def upsert_internal(kind, new_item) # 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" } diff --git a/lib/ldclient-rb/impl/integrations/dynamodb_impl.rb b/lib/ldclient-rb/impl/integrations/dynamodb_impl.rb index 2e0d9898..8ca35d15 100644 --- a/lib/ldclient-rb/impl/integrations/dynamodb_impl.rb +++ b/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 diff --git a/lib/ldclient-rb/impl/integrations/redis_impl.rb b/lib/ldclient-rb/impl/integrations/redis_impl.rb index 83fa7dd9..836ca6ab 100644 --- a/lib/ldclient-rb/impl/integrations/redis_impl.rb +++ b/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? diff --git a/lib/ldclient-rb/in_memory_store.rb b/lib/ldclient-rb/in_memory_store.rb index 0c6cb2d7..c0cd583e 100644 --- a/lib/ldclient-rb/in_memory_store.rb +++ b/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 diff --git a/lib/ldclient-rb/integrations/util/store_wrapper.rb b/lib/ldclient-rb/integrations/util/store_wrapper.rb index bbb600aa..ea885047 100644 --- a/lib/ldclient-rb/integrations/util/store_wrapper.rb +++ b/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 diff --git a/lib/ldclient-rb/interfaces/feature_store.rb b/lib/ldclient-rb/interfaces/feature_store.rb index 65f6e7b3..707af583 100644 --- a/lib/ldclient-rb/interfaces/feature_store.rb +++ b/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 diff --git a/spec/feature_store_spec_base.rb b/spec/feature_store_spec_base.rb index ea24bc84..4eb7b0a5 100644 --- a/spec/feature_store_spec_base.rb +++ b/spec/feature_store_spec_base.rb @@ -137,7 +137,7 @@ 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 @@ -145,7 +145,7 @@ def new_version_plus(f, delta_version, attrs = {}) 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 @@ -153,7 +153,7 @@ def new_version_plus(f, delta_version, attrs = {}) 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 @@ -161,28 +161,28 @@ def new_version_plus(f, delta_version, attrs = {}) 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 diff --git a/spec/impl/data_source_spec.rb b/spec/impl/data_source_spec.rb index 05ff23ec..5e381300 100644 --- a/spec/impl/data_source_spec.rb +++ b/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') diff --git a/spec/integrations/util/store_wrapper_spec.rb b/spec/integrations/util/store_wrapper_spec.rb index b1e0c487..004003af 100644 --- a/spec/integrations/util/store_wrapper_spec.rb +++ b/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?