feat!: Replace FeatureStoreCore and CachingStoreWrapper with PersistentDataStore

lib/ldclient-rb/impl/data_store/feature_store_client_wrapper.rb

@@ -67,7 +67,7 @@ def initialized?
 
         #
         # Passthrough for the FDv2 store coordinator. If the wrapped store supports
-        # disabling its cache (e.g. {LaunchDarkly::Integrations::Util::CachingStoreWrapper}),
+        # disabling its cache (e.g. {LaunchDarkly::Integrations::Util::PersistentDataStoreWrapper}),
         # forward the call; otherwise no-op.
         #
         # @return [void]

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

@@ -5,9 +5,11 @@ module Impl
     module Integrations
       module Consul
         #
-        # Internal implementation of the Consul feature store, intended to be used with CachingStoreWrapper.
+        # The Consul implementation of {LaunchDarkly::Interfaces::PersistentDataStore}.
         #
         class ConsulFeatureStoreCore
+          include LaunchDarkly::Interfaces::PersistentDataStore
+
           begin
             require "diplomat"
             require "ostruct"
@@ -28,7 +30,7 @@ def initialize(opts)
             @logger.info("ConsulFeatureStore: using Consul host at #{Diplomat.configuration.url}")
           end
 
-          def init_internal(all_data)
+          def init(all_data)
             # Start by reading the existing keys; we will later delete any of these that weren't in all_data.
             unused_old_keys = Set.new
             keys = Diplomat::Kv.get(@prefix, { keys: true, recurse: true }, :return)
@@ -39,10 +41,9 @@ def init_internal(all_data)
 
             # Insert or update every provided item
             all_data.each do |kind, items|
-              items.values.each do |item|
-                value = Model.serialize(kind, item)
-                key = item_key(kind, item[:key])
-                ops.push({ 'KV' => { 'Verb' => 'set', 'Key' => key, 'Value' => value } })
+              items.each do |item_name, item|
+                key = item_key(kind, item_name)
+                ops.push({ 'KV' => { 'Verb' => 'set', 'Key' => key, 'Value' => item.serialized_item } })
                 unused_old_keys.delete(key)
                 num_items = num_items + 1
               end
@@ -53,20 +54,20 @@ def init_internal(all_data)
               ops.push({ 'KV' => { 'Verb' => 'delete', 'Key' => key } })
             end
 
-            # Now set the special key that we check in initialized_internal?
+            # Now set the special key that we check in initialized?
             ops.push({ 'KV' => { 'Verb' => 'set', 'Key' => inited_key, 'Value' => '' } })
 
             ConsulUtil.batch_operations(ops)
 
             @logger.info { "Initialized database with #{num_items} items" }
           end
 
-          def get_internal(kind, key)
+          def get(kind, key)
             value = Diplomat::Kv.get(item_key(kind, key), {}, :return)  # :return means "don't throw an error if not found"
-            (value.nil? || value == "") ? nil : Model.deserialize(kind, value)
+            (value.nil? || value == "") ? nil : to_descriptor(value)
           end
 
-          def get_all_internal(kind)
+          def get_all(kind)
             items_out = {}
             prefix = kind_key(kind)
             Diplomat::Kv.get_all(prefix, {}, :return).each do |result|
@@ -76,39 +77,38 @@ def get_all_internal(kind)
               next unless db_key.start_with?(prefix)
               # Use the key that the item is stored under, not the key inside the item. A deleted
               # item (a "tombstone") is not guaranteed to carry a key of its own.
-              items_out[db_key[prefix.length..].to_sym] = Model.deserialize(kind, value)
+              items_out[db_key[prefix.length..].to_sym] = to_descriptor(value)
             end
             items_out
           end
 
-          def upsert_internal(kind, new_item)
-            key = item_key(kind, new_item[:key])
-            json = Model.serialize(kind, new_item)
+          def upsert(kind, item_name, item)
+            key = item_key(kind, item_name)
 
             # We will potentially keep retrying indefinitely until someone's write succeeds
             while true
               old_value = Diplomat::Kv.get(key, { decode_values: true }, :return)
               if old_value.nil? || old_value == ""
                 mod_index = 0
               else
-                old_item = Model.deserialize(kind, old_value[0]["Value"])
+                old_item = kind.deserialize(old_value[0]["Value"])
                 # Check whether the item is stale. If so, don't do the update.
-                return false if old_item[:version] >= new_item[:version]
+                return false if old_item[:version] >= 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)
+              success = Diplomat::Kv.put(key, item.serialized_item, cas: mod_index)
               return true if success
 
               # If we failed, retry the whole shebang
               @logger.debug { "Concurrent modification detected, retrying" }
             end
           end
 
-          def initialized_internal?
+          def initialized?
             # Unfortunately we need to use exceptions here, instead of the :return parameter, because with
             # :return there's no way to distinguish between a missing value and an empty string.
             begin
@@ -120,8 +120,8 @@ def initialized_internal?
           end
 
           def available?
-            # Most implementations use the initialized_internal? method as a
-            # proxy for this check. However, since `initialized_internal?`
+            # Most implementations use the initialized? method as a
+            # proxy for this check. However, since `initialized?`
             # catches a KeyNotFound exception, and that exception can be raised
             # when the server goes away, we have to modify our behavior
             # slightly.
@@ -142,7 +142,12 @@ def item_key(kind, key)
           end
 
           def kind_key(kind)
-            @prefix + kind[:namespace] + '/'
+            @prefix + kind.namespace + '/'
+          end
+
+          # Consul does not keep the version apart from the serialized item.
+          def to_descriptor(serialized)
+            LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor.new(version: 0, deleted: false, serialized_item: serialized)
           end
 
           def inited_key

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

@@ -48,9 +48,11 @@ def stop
         end
 
         #
-        # Internal implementation of the DynamoDB feature store, intended to be used with CachingStoreWrapper.
+        # The DynamoDB implementation of {LaunchDarkly::Interfaces::PersistentDataStore}.
         #
         class DynamoDBFeatureStoreCore < DynamoDBStoreImplBase
+          include LaunchDarkly::Interfaces::PersistentDataStore
+
           VERSION_ATTRIBUTE = "version"
           ITEM_JSON_ATTRIBUTE = "item"
 
@@ -70,7 +72,7 @@ def available?
             false
           end
 
-          def init_internal(all_data)
+          def init(all_data)
             # Start by reading the existing keys; we will later delete any of these that weren't in all_data.
             unused_old_keys = read_existing_keys(all_data.keys)
 
@@ -79,9 +81,9 @@ def init_internal(all_data)
 
             # Insert or update every provided item
             all_data.each do |kind, items|
-              items.values.each do |item|
-                requests.push({ put_request: { item: marshal_item(kind, item) } })
-                unused_old_keys.delete([ namespace_for_kind(kind), item[:key] ])
+              items.each do |key, item|
+                requests.push({ put_request: { item: marshal_item(kind, key, item) } })
+                unused_old_keys.delete([ namespace_for_kind(kind), key.to_s ])
                 num_items = num_items + 1
               end
             end
@@ -92,7 +94,7 @@ def init_internal(all_data)
               requests.push({ delete_request: { key: del_item } })
             end
 
-            # Now set the special key that we check in initialized_internal?
+            # Now set the special key that we check in initialized?
             inited_item = make_keys_hash(inited_key, inited_key)
             requests.push({ put_request: { item: inited_item } })
 
@@ -101,18 +103,18 @@ def init_internal(all_data)
             @logger.info { "Initialized table #{@table_name} with #{num_items} items" }
           end
 
-          def get_internal(kind, key)
-            resp = get_item_by_keys(namespace_for_kind(kind), key)
-            unmarshal_item(kind, resp.item)
+          def get(kind, key)
+            resp = get_item_by_keys(namespace_for_kind(kind), key.to_s)
+            unmarshal_item(resp.item)
           end
 
-          def get_all_internal(kind)
+          def get_all(kind)
             items_out = {}
             req = make_query_for_kind(kind)
             while true
               resp = @client.query(req)
               resp.items.each do |item|
-                item_out = unmarshal_item(kind, item)
+                item_out = unmarshal_item(item)
                 next if item_out.nil?
                 # Use the sort key that the item is stored under, not the key inside the item. A
                 # deleted item (a "tombstone") is not guaranteed to carry a key of its own.
@@ -124,8 +126,8 @@ def get_all_internal(kind)
             items_out
           end
 
-          def upsert_internal(kind, new_item)
-            encoded_item = marshal_item(kind, new_item)
+          def upsert(kind, key, item)
+            encoded_item = marshal_item(kind, key, item)
             begin
               @client.put_item({
                 table_name: @table_name,
@@ -137,7 +139,7 @@ def upsert_internal(kind, new_item)
                   "#version" => VERSION_ATTRIBUTE,
                 },
                 expression_attribute_values: {
-                  ":version" => new_item[:version],
+                  ":version" => item.version,
                 },
               })
               true
@@ -147,7 +149,7 @@ def upsert_internal(kind, new_item)
             end
           end
 
-          def initialized_internal?
+          def initialized?
             resp = get_item_by_keys(inited_key, inited_key)
             !resp.item.nil? && resp.item.length > 0
           end
@@ -159,7 +161,7 @@ def prefixed_namespace(base_str)
           end
 
           def namespace_for_kind(kind)
-            prefixed_namespace(kind[:namespace])
+            prefixed_namespace(kind.namespace)
           end
 
           def inited_key
@@ -217,18 +219,22 @@ def read_existing_keys(kinds)
             keys
           end
 
-          def marshal_item(kind, item)
-            make_keys_hash(namespace_for_kind(kind), item[:key]).merge({
-              VERSION_ATTRIBUTE => item[:version],
-              ITEM_JSON_ATTRIBUTE => Model.serialize(kind, item),
+          def marshal_item(kind, key, item)
+            make_keys_hash(namespace_for_kind(kind), key.to_s).merge({
+              VERSION_ATTRIBUTE => item.version,
+              ITEM_JSON_ATTRIBUTE => item.serialized_item,
             })
           end
 
-          def unmarshal_item(kind, item)
+          def unmarshal_item(item)
             return nil if item.nil? || item.length == 0
             json_attr = item[ITEM_JSON_ATTRIBUTE]
             raise RuntimeError.new("DynamoDB map did not contain expected item string") if json_attr.nil?
-            Model.deserialize(kind, json_attr)
+            LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor.new(
+              version: item[VERSION_ATTRIBUTE].to_i,
+              deleted: false,
+              serialized_item: json_attr
+            )
           end
         end