From a8750a444597e2fef81d182ba231714b2da34af1 Mon Sep 17 00:00:00 2001
From: Matthew Keeler <mkeeler@launchdarkly.com>
Date: Tue, 3 Nov 2026 11:37:50 -0500
Subject: [PATCH] feat!: Replace FeatureStoreCore and CachingStoreWrapper with
 PersistentDataStore

A database integration now implements LaunchDarkly::Interfaces::PersistentDataStore and wraps it in LaunchDarkly::Integrations::Util::PersistentDataStoreWrapper. This matches the persistent data store design of the Go, Java, .NET and js-core SDKs, and lets an integration in its own gem use only public SDK types.

The store receives and returns SerializedItemDescriptor values. The wrapper serializes and deserializes items with the public data kinds, so a store no longer calls LaunchDarkly::Impl::Model. The store methods are init, get, get_all, upsert, initialized?, available? and stop. upsert takes the key and the serialized item, does the version check, and returns true only if it wrote the item. The wrapper treats any other result as false.

The wrapper behaves as CachingStoreWrapper did, with the cache rules of the Go SDK's wrapper: it caches misses, latches initialized?, re-reads an item into the cache after a write that the store did not apply, keeps its cache entry if that read fails, and clears the cache if init fails.

PersistentDataStore only documents the methods. A stub method in an included module hides a method that the store inherits from a superclass, so the module defines none, and a missing method raises NoMethodError.

The Redis, DynamoDB and Consul stores implement the new interface. Their on-disk format does not change. The Redis store no longer has the RedisFeatureStore facade class; Redis.new_feature_store returns the wrapper, as the DynamoDB and Consul factories do.

FeatureStoreCore and CachingStoreWrapper are removed. A custom store core must move to the new interface.
---
 .../feature_store_client_wrapper.rb           |   2 +-
 .../impl/integrations/consul_impl.rb          |  47 +--
 .../impl/integrations/dynamodb_impl.rb        |  50 ++--
 .../impl/integrations/redis_impl.rb           | 148 ++--------
 lib/ldclient-rb/integrations.rb               |   2 +-
 lib/ldclient-rb/integrations/consul.rb        |   4 +-
 lib/ldclient-rb/integrations/dynamodb.rb      |   4 +-
 lib/ldclient-rb/integrations/redis.rb         |   4 +-
 .../util/persistent_data_store_wrapper.rb     | 235 +++++++++++++++
 .../integrations/util/store_wrapper.rb        | 277 ------------------
 lib/ldclient-rb/interfaces.rb                 |   1 +
 .../interfaces/persistent_data_store.rb       | 111 +++++++
 spec/feature_store_spec_base.rb               |   4 +-
 ... => persistent_data_store_wrapper_spec.rb} | 201 +++++++++++--
 14 files changed, 609 insertions(+), 481 deletions(-)
 create mode 100644 lib/ldclient-rb/integrations/util/persistent_data_store_wrapper.rb
 delete mode 100644 lib/ldclient-rb/integrations/util/store_wrapper.rb
 create mode 100644 lib/ldclient-rb/interfaces/persistent_data_store.rb
 rename spec/integrations/util/{store_wrapper_spec.rb => persistent_data_store_wrapper_spec.rb} (62%)

diff --git a/lib/ldclient-rb/impl/data_store/feature_store_client_wrapper.rb b/lib/ldclient-rb/impl/data_store/feature_store_client_wrapper.rb
index 0a0e43af..88547b43 100644
--- a/lib/ldclient-rb/impl/data_store/feature_store_client_wrapper.rb
+++ b/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]
diff --git a/lib/ldclient-rb/impl/integrations/consul_impl.rb b/lib/ldclient-rb/impl/integrations/consul_impl.rb
index e7ee6a8d..527b99ce 100644
--- a/lib/ldclient-rb/impl/integrations/consul_impl.rb
+++ b/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,7 +54,7 @@ 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)
@@ -61,12 +62,12 @@ def init_internal(all_data)
             @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,14 +77,13 @@ 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
@@ -91,16 +91,16 @@ def upsert_internal(kind, new_item)
               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
@@ -108,7 +108,7 @@ def upsert_internal(kind, new_item)
             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
diff --git a/lib/ldclient-rb/impl/integrations/dynamodb_impl.rb b/lib/ldclient-rb/impl/integrations/dynamodb_impl.rb
index 8ca35d15..0728d507 100644
--- a/lib/ldclient-rb/impl/integrations/dynamodb_impl.rb
+++ b/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
 
diff --git a/lib/ldclient-rb/impl/integrations/redis_impl.rb b/lib/ldclient-rb/impl/integrations/redis_impl.rb
index 836ca6ab..4901c86f 100644
--- a/lib/ldclient-rb/impl/integrations/redis_impl.rb
+++ b/lib/ldclient-rb/impl/integrations/redis_impl.rb
@@ -6,99 +6,6 @@ module LaunchDarkly
   module Impl
     module Integrations
       module Redis
-        #
-        # An implementation of the LaunchDarkly client's feature store that uses a Redis
-        # instance.  This object holds feature flags and related data received from the
-        # streaming API.  Feature data can also be further cached in memory to reduce overhead
-        # of calls to Redis.
-        #
-        # To use this class, you must first have the `redis` and `connection-pool` gems
-        # installed.  Then, create an instance and store it in the `feature_store` property
-        # of your client configuration.
-        #
-        class RedisFeatureStore
-          include LaunchDarkly::Interfaces::FeatureStore
-
-          # Note that this class is now just a facade around CachingStoreWrapper, which is in turn delegating
-          # to RedisFeatureStoreCore where the actual database logic is. This class was retained for historical
-          # reasons, so that existing code can still call RedisFeatureStore.new. In the future, we will migrate
-          # away from exposing these concrete classes and use factory methods instead.
-
-          #
-          # Constructor for a RedisFeatureStore instance.
-          #
-          # @param opts [Hash] the configuration options
-          # @option opts [String] :redis_url  URL of the Redis instance (shortcut for omitting redis_opts)
-          # @option opts [Hash] :redis_opts  options to pass to the Redis constructor (if you want to specify more than just redis_url)
-          # @option opts [String] :prefix  namespace prefix to add to all hash keys used by LaunchDarkly
-          # @option opts [Logger] :logger  a `Logger` instance; defaults to `Config.default_logger`
-          # @option opts [Integer] :max_connections  size of the Redis connection pool
-          # @option opts [Integer] :expiration  expiration time for the in-memory cache, in seconds; 0 for no local caching
-          # @option opts [Integer] :capacity  maximum number of feature flags (or related objects) to cache locally
-          # @option opts [Object] :pool  custom connection pool, if desired
-          # @option opts [Boolean] :pool_shutdown_on_close whether calling `close` should shutdown the custom connection pool.
-          #
-          def initialize(opts = {})
-            core = RedisFeatureStoreCore.new(opts)
-            @wrapper = LaunchDarkly::Integrations::Util::CachingStoreWrapper.new(core, opts)
-          end
-
-          def monitoring_enabled?
-            true
-          end
-
-          def available?
-            @wrapper.available?
-          end
-
-          #
-          # Default value for the `redis_url` constructor parameter; points to an instance of Redis
-          # running at `localhost` with its default port.
-          #
-          def self.default_redis_url
-            LaunchDarkly::Integrations::Redis::default_redis_url
-          end
-
-          #
-          # Default value for the `prefix` constructor parameter.
-          #
-          def self.default_prefix
-            LaunchDarkly::Integrations::Redis::default_prefix
-          end
-
-          def get(kind, key)
-            @wrapper.get(kind, key)
-          end
-
-          def all(kind)
-            @wrapper.all(kind)
-          end
-
-          def delete(kind, key, version)
-            @wrapper.delete(kind, key, version)
-          end
-
-          def init(all_data)
-            @wrapper.init(all_data)
-          end
-
-          def upsert(kind, item)
-            @wrapper.upsert(kind, item)
-          end
-
-          def initialized?
-            @wrapper.initialized?
-          end
-
-          def stop
-            @wrapper.stop
-          end
-
-          def disable_cache
-            @wrapper.disable_cache
-          end
-        end
-
         class RedisStoreImplBase
           begin
             require "redis"
@@ -158,9 +65,11 @@ def stop
         end
 
         #
-        # Internal implementation of the Redis feature store, intended to be used with CachingStoreWrapper.
+        # The Redis implementation of {LaunchDarkly::Interfaces::PersistentDataStore}.
         #
         class RedisFeatureStoreCore < RedisStoreImplBase
+          include LaunchDarkly::Interfaces::PersistentDataStore
+
           def initialize(opts)
             super(opts)
 
@@ -168,8 +77,8 @@ def initialize(opts)
           end
 
           def available?
-            # We don't care what the status is, only that we can connect
-            initialized_internal?
+            # The result of the query does not matter. A successful query shows that Redis operates.
+            initialized?
             true
           rescue
             false
@@ -179,7 +88,7 @@ def description
             "RedisFeatureStore"
           end
 
-          def init_internal(all_data)
+          def init(all_data)
             count = 0
             with_connection do |redis|
               redis.multi do |multi|
@@ -187,7 +96,7 @@ def init_internal(all_data)
                   multi.del(items_key(kind))
                   count = count + items.count
                   items.each do |key, item|
-                    multi.hset(items_key(kind), key, Model.serialize(kind,item))
+                    multi.hset(items_key(kind), key, item.serialized_item)
                   end
                 end
                 multi.set(inited_key, inited_key)
@@ -196,37 +105,36 @@ def init_internal(all_data)
             @logger.info { "RedisFeatureStore: initialized with #{count} items" }
           end
 
-          def get_internal(kind, key)
+          def get(kind, key)
             with_connection do |redis|
-              get_redis(redis, kind, key)
+              to_descriptor(redis.hget(items_key(kind), key))
             end
           end
 
-          def get_all_internal(kind)
-            fs = {}
+          def get_all(kind)
+            items = {}
             with_connection do |redis|
-              hashfs = redis.hgetall(items_key(kind))
-              hashfs.each do |k, json_item|
-                fs[k.to_sym] = Model.deserialize(kind, json_item)
+              redis.hgetall(items_key(kind)).each do |key, serialized|
+                items[key.to_sym] = to_descriptor(serialized)
               end
             end
-            fs
+            items
           end
 
-          def upsert_internal(kind, new_item)
+          def upsert(kind, key, item)
             base_key = items_key(kind)
-            key = new_item[:key]
             try_again = true
             updated = true
             while try_again
               try_again = false
               with_connection do |redis|
                 redis.watch(base_key) do
-                  old_item = get_redis(redis, kind, key)
+                  old_serialized = redis.hget(base_key, key)
                   before_update_transaction(base_key, key)
-                  if old_item.nil? || old_item[:version] < new_item[:version]
+                  old_version = old_serialized.nil? ? nil : kind.deserialize(old_serialized)[:version]
+                  if old_version.nil? || old_version < item.version
                     result = redis.multi do |multi|
-                      multi.hset(base_key, key, Model.serialize(kind, new_item))
+                      multi.hset(base_key, key, item.serialized_item)
                     end
                     if result.nil?
                       @logger.debug { "RedisFeatureStore: concurrent modification detected, retrying" }
@@ -234,9 +142,9 @@ def upsert_internal(kind, new_item)
                     end
                   else
                     updated = false
-                    action = new_item[:deleted] ? "delete" : "update"
+                    action = 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]}" }
+                    @logger.warn { "RedisFeatureStore: attempted to #{action} #{key} version: #{old_version} in '#{kind.namespace}' with a version that is the same or older: #{item.version}" }
                   end
                   redis.unwatch
                 end
@@ -245,7 +153,7 @@ def upsert_internal(kind, new_item)
             updated
           end
 
-          def initialized_internal?
+          def initialized?
             with_connection { |redis| redis.exists?(inited_key) }
           end
 
@@ -256,19 +164,17 @@ def before_update_transaction(base_key, key)
           end
 
           def items_key(kind)
-            @prefix + ":" + kind[:namespace]
-          end
-
-          def cache_key(kind, key)
-            kind[:namespace] + ":" + key.to_s
+            @prefix + ":" + kind.namespace
           end
 
           def inited_key
             @prefix + ":$inited"
           end
 
-          def get_redis(redis, kind, key)
-            Model.deserialize(kind, redis.hget(items_key(kind), key))
+          # Redis does not keep the version apart from the serialized item.
+          def to_descriptor(serialized)
+            return nil if serialized.nil?
+            LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor.new(version: 0, deleted: false, serialized_item: serialized)
           end
         end
 
diff --git a/lib/ldclient-rb/integrations.rb b/lib/ldclient-rb/integrations.rb
index 2a2ac216..aa5122ef 100644
--- a/lib/ldclient-rb/integrations.rb
+++ b/lib/ldclient-rb/integrations.rb
@@ -3,4 +3,4 @@
 require "ldclient-rb/integrations/file_data"
 require "ldclient-rb/integrations/redis"
 require "ldclient-rb/integrations/test_data"
-require "ldclient-rb/integrations/util/store_wrapper"
+require "ldclient-rb/integrations/util/persistent_data_store_wrapper"
diff --git a/lib/ldclient-rb/integrations/consul.rb b/lib/ldclient-rb/integrations/consul.rb
index e1ba4abd..c9e350b9 100644
--- a/lib/ldclient-rb/integrations/consul.rb
+++ b/lib/ldclient-rb/integrations/consul.rb
@@ -1,5 +1,5 @@
 require "ldclient-rb/impl/integrations/consul_impl"
-require "ldclient-rb/integrations/util/store_wrapper"
+require "ldclient-rb/integrations/util/persistent_data_store_wrapper"
 
 module LaunchDarkly
   module Integrations
@@ -42,7 +42,7 @@ def self.default_prefix
       #
       def self.new_feature_store(opts = {})
         core = LaunchDarkly::Impl::Integrations::Consul::ConsulFeatureStoreCore.new(opts)
-        LaunchDarkly::Integrations::Util::CachingStoreWrapper.new(core, opts)
+        LaunchDarkly::Integrations::Util::PersistentDataStoreWrapper.new(core, opts)
       end
     end
   end
diff --git a/lib/ldclient-rb/integrations/dynamodb.rb b/lib/ldclient-rb/integrations/dynamodb.rb
index 7f4b4425..a1b73562 100644
--- a/lib/ldclient-rb/integrations/dynamodb.rb
+++ b/lib/ldclient-rb/integrations/dynamodb.rb
@@ -1,5 +1,5 @@
 require "ldclient-rb/impl/integrations/dynamodb_impl"
-require "ldclient-rb/integrations/util/store_wrapper"
+require "ldclient-rb/integrations/util/persistent_data_store_wrapper"
 
 module LaunchDarkly
   module Integrations
@@ -52,7 +52,7 @@ module DynamoDB
       #
       def self.new_feature_store(table_name, opts = {})
         core = LaunchDarkly::Impl::Integrations::DynamoDB::DynamoDBFeatureStoreCore.new(table_name, opts)
-        LaunchDarkly::Integrations::Util::CachingStoreWrapper.new(core, opts)
+        LaunchDarkly::Integrations::Util::PersistentDataStoreWrapper.new(core, opts)
       end
 
       #
diff --git a/lib/ldclient-rb/integrations/redis.rb b/lib/ldclient-rb/integrations/redis.rb
index e78b22ba..d08c15ae 100644
--- a/lib/ldclient-rb/integrations/redis.rb
+++ b/lib/ldclient-rb/integrations/redis.rb
@@ -1,4 +1,5 @@
 require "ldclient-rb/impl/integrations/redis_impl"
+require "ldclient-rb/integrations/util/persistent_data_store_wrapper"
 
 module LaunchDarkly
   module Integrations
@@ -63,7 +64,8 @@ def self.default_prefix
       # @return [LaunchDarkly::Interfaces::FeatureStore]  a feature store object
       #
       def self.new_feature_store(opts = {})
-        LaunchDarkly::Impl::Integrations::Redis::RedisFeatureStore.new(opts)
+        core = LaunchDarkly::Impl::Integrations::Redis::RedisFeatureStoreCore.new(opts)
+        LaunchDarkly::Integrations::Util::PersistentDataStoreWrapper.new(core, opts)
       end
 
       #
diff --git a/lib/ldclient-rb/integrations/util/persistent_data_store_wrapper.rb b/lib/ldclient-rb/integrations/util/persistent_data_store_wrapper.rb
new file mode 100644
index 00000000..fa47f332
--- /dev/null
+++ b/lib/ldclient-rb/integrations/util/persistent_data_store_wrapper.rb
@@ -0,0 +1,235 @@
+require "concurrent/atomics"
+
+require "ldclient-rb/impl/expiring_cache"
+require "ldclient-rb/interfaces"
+
+module LaunchDarkly
+  module Integrations
+    #
+    # Support code that may be helpful in creating integrations.
+    #
+    # @since 5.5.0
+    #
+    module Util
+      #
+      # Gives the SDK's {LaunchDarkly::Interfaces::FeatureStore} interface to a
+      # {LaunchDarkly::Interfaces::PersistentDataStore}.
+      #
+      # The wrapper serializes the items that the SDK writes and deserializes the items that the
+      # store returns. It removes deleted items from query results. It can also keep the items in
+      # an in-memory cache, so that the SDK does not query the store for each flag evaluation.
+      #
+      # A database integration implements {LaunchDarkly::Interfaces::PersistentDataStore} and
+      # gives the SDK a wrapper around its store.
+      #
+      # Cache behavior:
+      #
+      # - The wrapper caches each item that it reads, and each list of all items of a kind. It also
+      #   caches an item that the store does not have, so that a query for a missing flag does not
+      #   go to the store each time.
+      # - After a successful write, the wrapper caches the item that the store has.
+      # - If a write fails, the wrapper does not change the cache.
+      # - After the store reports that it is initialized, the wrapper does not ask again. It caches
+      #   a "not initialized" result until the cache entry expires.
+      #
+      class PersistentDataStoreWrapper
+        include LaunchDarkly::Interfaces::FeatureStore
+
+        #
+        # Creates a wrapper around a persistent data store.
+        #
+        # @param core [LaunchDarkly::Interfaces::PersistentDataStore] the store
+        # @param opts [Hash] options; the wrapper ignores options that it does not use
+        # @option opts [Float] :expiration (15) the number of seconds to keep an item in the cache;
+        #   0 or less means no cache
+        # @option opts [Integer] :capacity (1000) the maximum number of entries in the cache
+        # @option opts [Logger] :logger logs data validation problems in items from the store
+        #
+        def initialize(core, opts = {})
+          @core = core
+          @logger = opts[:logger]
+
+          expiration_seconds = opts[:expiration] || 15
+          if expiration_seconds > 0
+            capacity = opts[:capacity] || 1000
+            @cache = Impl::ExpiringCache.new(capacity, expiration_seconds)
+          else
+            @cache = nil
+          end
+
+          @inited = Concurrent::AtomicBoolean.new(false)
+        end
+
+        def monitoring_enabled?
+          true
+        end
+
+        def available?
+          @core.available?
+        end
+
+        def init(all_data)
+          serialized = all_data.to_h do |kind, items|
+            [kind, items.to_h { |key, item| [key, serialize(kind, item)] }]
+          end
+
+          begin
+            @core.init(serialized)
+          rescue
+            @cache&.clear
+            raise
+          end
+
+          cache = @cache
+          unless cache.nil?
+            cache.clear
+            all_data.each do |kind, items|
+              cache[all_cache_key(kind)] = items_if_not_deleted(items)
+              items.each do |key, item|
+                cache[item_cache_key(kind, key)] = [item]
+              end
+            end
+          end
+
+          @inited.make_true
+        end
+
+        def get(kind, key)
+          cache = @cache
+          cache_key = item_cache_key(kind, key)
+          unless cache.nil?
+            # An item entry is an array, so that the cache can keep a nil result.
+            cached = cache[cache_key]
+            return item_if_not_deleted(cached[0]) unless cached.nil?
+          end
+
+          item = deserialize(kind, key, @core.get(kind, key))
+          cache[cache_key] = [item] unless cache.nil?
+          item_if_not_deleted(item)
+        end
+
+        def all(kind)
+          cache = @cache
+          unless cache.nil?
+            items = cache[all_cache_key(kind)]
+            return items unless items.nil?
+          end
+
+          items = @core.get_all(kind).to_h do |key, descriptor|
+            [key.to_sym, deserialize(kind, key, descriptor)]
+          end
+          items = items_if_not_deleted(items)
+          cache[all_cache_key(kind)] = items unless cache.nil?
+          items
+        end
+
+        def upsert(kind, item)
+          key = item[:key]
+          updated = @core.upsert(kind, key, serialize(kind, item)) == true
+
+          cache = @cache
+          unless cache.nil?
+            cache.delete(all_cache_key(kind))
+            if updated
+              cache[item_cache_key(kind, key)] = [item]
+            else
+              refresh_cached_item(cache, kind, key)
+            end
+          end
+
+          updated
+        end
+
+        def delete(kind, key, version)
+          upsert(kind, { key: key, version: version, deleted: true })
+        end
+
+        def initialized?
+          return true if @inited.value
+
+          cache = @cache
+          return false if !cache.nil? && cache[inited_cache_key] == false
+
+          result = @core.initialized? == true
+          if result
+            @inited.make_true
+            cache&.delete(inited_cache_key)
+          elsif !cache.nil?
+            cache[inited_cache_key] = false
+          end
+          result
+        end
+
+        #
+        # Disables the in-memory cache. Later operations go to the store. It is safe to call this
+        # method more than one time.
+        #
+        # The FDv2 data system calls this method after its in-memory store has the data. The
+        # cache then has no use. This method is not part of the public API.
+        #
+        # @return [void]
+        #
+        def disable_cache
+          cache = @cache
+          return if cache.nil?
+          @cache = nil
+          cache.clear
+        end
+
+        def stop
+          @core.stop
+        end
+
+        private
+
+        def serialize(kind, item)
+          LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor.new(
+            version: item[:version],
+            deleted: item[:deleted] == true,
+            serialized_item: kind.serialize(item)
+          )
+        end
+
+        def deserialize(kind, key, descriptor)
+          return nil if descriptor.nil?
+          if descriptor.serialized_item.nil?
+            return { key: key.to_s, version: descriptor.version, deleted: true }
+          end
+
+          kind.deserialize(descriptor.serialized_item, @logger)
+        end
+
+        # After a write that the store did not apply, the store has a newer item than the cache.
+        # If the read fails, the cache keeps its entry.
+        def refresh_cached_item(cache, kind, key)
+          cache[item_cache_key(kind, key)] = [deserialize(kind, key, @core.get(kind, key))]
+        rescue
+          nil
+        end
+
+        # One cache holds three types of entry. An item entry has a key like "features:my-flag".
+        def item_cache_key(kind, key)
+          kind.namespace + ":" + key.to_s
+        end
+
+        # A list of all items of a kind uses the kind as its key.
+        def all_cache_key(kind)
+          kind
+        end
+
+        # A "not initialized" result uses this key.
+        def inited_cache_key
+          "$inited"
+        end
+
+        def item_if_not_deleted(item)
+          (item.nil? || item[:deleted]) ? nil : item
+        end
+
+        def items_if_not_deleted(items)
+          items.reject { |_key, item| item[:deleted] }
+        end
+      end
+    end
+  end
+end
diff --git a/lib/ldclient-rb/integrations/util/store_wrapper.rb b/lib/ldclient-rb/integrations/util/store_wrapper.rb
deleted file mode 100644
index ea885047..00000000
--- a/lib/ldclient-rb/integrations/util/store_wrapper.rb
+++ /dev/null
@@ -1,277 +0,0 @@
-require "concurrent/atomics"
-
-require "ldclient-rb/impl/expiring_cache"
-require "ldclient-rb/interfaces"
-
-module LaunchDarkly
-  module Integrations
-    #
-    # Support code that may be helpful in creating integrations.
-    #
-    # @since 5.5.0
-    #
-    module Util
-      #
-      # CachingStoreWrapper is a partial implementation of the {LaunchDarkly::Interfaces::FeatureStore}
-      # pattern that delegates part of its behavior to another object, while providing optional caching
-      # behavior and other logic that would otherwise be repeated in every feature store implementation.
-      # This makes it easier to create new database integrations by implementing only the database-specific
-      # logic.
-      #
-      # The mixin {FeatureStoreCore} describes the methods that need to be supported by the inner
-      # implementation object.
-      #
-      class CachingStoreWrapper
-        include LaunchDarkly::Interfaces::FeatureStore
-
-        #
-        # Creates a new store wrapper instance.
-        #
-        # @param core [Object]  an object that implements the {FeatureStoreCore} methods
-        # @param opts [Hash]  a hash that may include cache-related options; all others will be ignored
-        # @option opts [Float] :expiration (15)  cache TTL; zero means no caching
-        # @option opts [Integer] :capacity (1000)  maximum number of items in the cache
-        #
-        def initialize(core, opts)
-          @core = core
-
-          expiration_seconds = opts[:expiration] || 15
-          if expiration_seconds > 0
-            capacity = opts[:capacity] || 1000
-            @cache = Impl::ExpiringCache.new(capacity, expiration_seconds)
-          else
-            @cache = nil
-          end
-
-          @inited = Concurrent::AtomicBoolean.new(false)
-        end
-
-        def monitoring_enabled?
-          true
-        end
-
-        def available?
-          @core.available?
-        end
-
-        def init(all_data)
-          @core.init_internal(all_data)
-          @inited.make_true
-
-          cache = @cache
-          unless cache.nil?
-            cache.clear
-            all_data.each do |kind, items|
-              cache[kind] = items_if_not_deleted(items)
-              items.each do |key, item|
-                cache[item_cache_key(kind, key)] = [item]
-              end
-            end
-          end
-        end
-
-        def get(kind, key)
-          cache = @cache
-          cache_key = item_cache_key(kind, key)
-          unless cache.nil?
-            cached = cache[cache_key] # note, item entries in the cache are wrapped in an array so we can cache nil values
-            return item_if_not_deleted(cached[0]) unless cached.nil?
-          end
-
-          item = @core.get_internal(kind, key)
-
-          cache[cache_key] = [item] unless cache.nil?
-
-          item_if_not_deleted(item)
-        end
-
-        def all(kind)
-          cache = @cache
-          unless cache.nil?
-            items = cache[all_cache_key(kind)]
-            return items unless items.nil?
-          end
-
-          items = items_if_not_deleted(@core.get_all_internal(kind))
-          cache[all_cache_key(kind)] = items unless cache.nil?
-          items
-        end
-
-        def upsert(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)
-          upsert(kind, { key: key, version: version, deleted: true })
-        end
-
-        def initialized?
-          return true if @inited.value
-
-          cache = @cache
-          if cache.nil?
-            result = @core.initialized_internal?
-          else
-            result = cache[inited_cache_key]
-            if result.nil?
-              result = @core.initialized_internal?
-              cache[inited_cache_key] = result
-            end
-          end
-
-          @inited.make_true if result
-          result
-        end
-
-        #
-        # Disable the in-memory cache. Releases the cache reference so subsequent operations
-        # bypass it and go directly to the underlying core. Safe to call multiple times.
-        #
-        # Called by the FDv2 store coordinator once the in-memory store has become the
-        # source of truth and the persistent-store cache is no longer useful. Internal --
-        # not part of the public API.
-        #
-        # @return [void]
-        #
-        def disable_cache
-          cache = @cache
-          return if cache.nil?
-          @cache = nil
-          cache.clear
-        end
-
-        def stop
-          @core.stop
-        end
-
-        private
-
-        # We use just one cache for 3 kinds of objects. Individual entities use a key like 'features:my-flag'.
-        def item_cache_key(kind, key)
-          kind[:namespace] + ":" + key.to_s
-        end
-
-        # The result of a call to get_all_internal is cached using the "kind" object as a key.
-        def all_cache_key(kind)
-          kind
-        end
-
-        # The result of initialized_internal? is cached using this key.
-        def inited_cache_key
-          "$inited"
-        end
-
-        def item_if_not_deleted(item)
-          (item.nil? || item[:deleted]) ? nil : item
-        end
-
-        def items_if_not_deleted(items)
-          items.select { |key, item| !item[:deleted] }
-        end
-      end
-
-      #
-      # This module describes the methods that you must implement on your own object in order to
-      # use {CachingStoreWrapper}.
-      #
-      module FeatureStoreCore
-        #
-        # Initializes the store. This is the same as {LaunchDarkly::Interfaces::FeatureStore#init},
-        # but the wrapper will take care of updating the cache if caching is enabled.
-        #
-        # If possible, the store should update the entire data set atomically. If that is not possible,
-        # it should iterate through the outer hash and then the inner hash using the existing iteration
-        # order of those hashes (the SDK will ensure that the items were inserted into the hashes in
-        # the correct order), storing each item, and then delete any leftover items at the very end.
-        #
-        # @param all_data [Hash]  a hash where each key is one of the data kind objects, and each
-        #   value is in turn a hash of string keys to entities
-        # @return [void]
-        #
-        def init_internal(all_data)
-        end
-
-        #
-        # Retrieves a single entity. This is the same as {LaunchDarkly::Interfaces::FeatureStore#get}
-        # except that 1. the wrapper will take care of filtering out deleted entities by checking the
-        # `:deleted` property, so you can just return exactly what was in the data store, and 2. the
-        # wrapper will take care of checking and updating the cache if caching is enabled.
-        #
-        # @param kind [Object]  the kind of entity to get
-        # @param key [String]  the unique key of the entity to get
-        # @return [Hash]  the entity; nil if the key was not found
-        #
-        def get_internal(kind, key)
-        end
-
-        #
-        # Retrieves all entities of the specified kind. This is the same as {LaunchDarkly::Interfaces::FeatureStore#all}
-        # except that 1. the wrapper will take care of filtering out deleted entities by checking the
-        # `:deleted` property, so you can just return exactly what was in the data store, and 2. the
-        # wrapper will take care of checking and updating the cache if caching is enabled.
-        #
-        # @param kind [Object]  the kind of entity to get
-        # @return [Hash]  a hash where each key is the entity's `:key` property and each value
-        #   is the entity
-        #
-        def get_all_internal(kind)
-        end
-
-        #
-        # Attempts to add or update an entity. This is the same as {LaunchDarkly::Interfaces::FeatureStore#upsert}
-        # 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 [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
-
-        #
-        # Checks whether this store has been initialized. This is the same as
-        # {LaunchDarkly::Interfaces::FeatureStore#initialized?} except that there is less of a concern
-        # for efficiency, because the wrapper will use caching and memoization in order to call the method
-        # as little as possible.
-        #
-        # @return [Boolean]  true if the store is in an initialized state
-        #
-        def initialized_internal?
-        end
-
-        #
-        # Tests whether the data store operates normally. This is the same as
-        # {LaunchDarkly::Interfaces::FeatureStore#available?}.
-        #
-        # The wrapper reports status monitoring as enabled for all cores. When an operation
-        # fails, the SDK calls this method at intervals until it returns true.
-        #
-        # @return [Boolean] true if the underlying data store is reachable
-        #
-        def available?
-        end
-
-        #
-        # Performs any necessary cleanup to shut down the store when the client is being shut down.
-        #
-        # @return [void]
-        #
-        def stop
-        end
-      end
-    end
-  end
-end
diff --git a/lib/ldclient-rb/interfaces.rb b/lib/ldclient-rb/interfaces.rb
index 95d1e9b9..7abdd8dc 100644
--- a/lib/ldclient-rb/interfaces.rb
+++ b/lib/ldclient-rb/interfaces.rb
@@ -1,5 +1,6 @@
 require "observer"
 require "ldclient-rb/interfaces/feature_store"
+require "ldclient-rb/interfaces/persistent_data_store"
 require "ldclient-rb/interfaces/flag_tracker"
 require "ldclient-rb/interfaces/data_store"
 require "ldclient-rb/interfaces/data_source"
diff --git a/lib/ldclient-rb/interfaces/persistent_data_store.rb b/lib/ldclient-rb/interfaces/persistent_data_store.rb
new file mode 100644
index 00000000..69da18fd
--- /dev/null
+++ b/lib/ldclient-rb/interfaces/persistent_data_store.rb
@@ -0,0 +1,111 @@
+module LaunchDarkly
+  module Interfaces
+    #
+    # The interface for a database integration that keeps feature flag data, such as Redis.
+    #
+    # A persistent data store keeps items in serialized form. It does not parse, cache or filter
+    # them. The SDK wraps the store in {LaunchDarkly::Integrations::Util::PersistentDataStoreWrapper},
+    # which serializes and deserializes the items, keeps an optional in-memory cache, and removes
+    # deleted items from query results.
+    #
+    # Each item is a {LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor}. Each data
+    # kind is a {LaunchDarkly::Interfaces::DataStore::DataKind}. The store uses the kind's
+    # {LaunchDarkly::Interfaces::DataStore::DataKind#namespace} to keep the items of each kind apart.
+    #
+    # The store must keep deleted items. The SDK uses a deleted item to stop an older version of
+    # the item from replacing the deletion.
+    #
+    # This module only documents the methods. It does not define them, so a store that includes
+    # it keeps the methods it inherits from a superclass. A store must supply all of the methods.
+    # If an operation fails, the method must raise an error. The SDK then marks the store as
+    # unavailable and calls {#available?} at intervals until it returns true.
+    #
+    module PersistentDataStore
+      #
+      # @!method init(all_data)
+      #   Replaces all of the data in the store.
+      #
+      #   If possible, the store replaces the data in one atomic operation. If that is not possible,
+      #   it writes the kinds and the items in the order of the hashes, and then deletes the items
+      #   that are not in the new data. The SDK puts the data in an order that keeps the store
+      #   consistent during a write that is not atomic.
+      #
+      #   The store also records that it is initialized. See {#initialized?}.
+      #
+      #   @param all_data [Hash{LaunchDarkly::Interfaces::DataStore::DataKind => Hash{Symbol => LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor}}]
+      #     the items of each kind, by key
+      #   @return [void]
+      #
+
+      #
+      # @!method get(kind, key)
+      #   Gets one item.
+      #
+      #   A store that keeps the version apart from the serialized item returns that version. A
+      #   store that does not returns a version of 0, and the SDK reads the version from the
+      #   serialized item.
+      #
+      #   @param kind [LaunchDarkly::Interfaces::DataStore::DataKind]
+      #   @param key [String, Symbol]
+      #   @return [LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor, nil] the item, or
+      #     nil if the store has no item with this key. A deleted item is an item.
+      #
+
+      #
+      # @!method get_all(kind)
+      #   Gets all of the items of one kind, including deleted items.
+      #
+      #   @param kind [LaunchDarkly::Interfaces::DataStore::DataKind]
+      #   @return [Hash{Symbol => LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor}]
+      #     the items, by key
+      #
+
+      #
+      # @!method upsert(kind, key, item)
+      #   Adds or replaces one item, if its version is higher than the version in the store.
+      #
+      #   The store does the version check and the write in one atomic operation. If the store has
+      #   no item with this key, it writes the item. If the store has an item with the same or a
+      #   higher version, it does not write the item. A deleted item has a version, and the same
+      #   rule applies to it.
+      #
+      #   To get the version of a stored item that it keeps only in serialized form, the store uses
+      #   {LaunchDarkly::Interfaces::DataStore::DataKind#deserialize}.
+      #
+      #   @param kind [LaunchDarkly::Interfaces::DataStore::DataKind]
+      #   @param key [String, Symbol]
+      #   @param item [LaunchDarkly::Interfaces::DataStore::SerializedItemDescriptor]
+      #   @return [Boolean] true if the store wrote the item; false if the store already had an
+      #     item with the same or a higher version
+      #
+
+      #
+      # @!method initialized?
+      #   Tells whether the store has data from an {#init}.
+      #
+      #   The result must be true if any process has called {#init} on a store with the same
+      #   configuration, so that an SDK can use data that another SDK or the Relay Proxy wrote.
+      #   The SDK caches a true result.
+      #
+      #   @return [Boolean]
+      #
+
+      #
+      # @!method available?
+      #   Tells whether the store can operate.
+      #
+      #   The SDK calls this method at intervals after an operation fails, until the result is true.
+      #   The method can do a simple query. It must not raise an error.
+      #
+      #   @return [Boolean] true if the store can operate
+      #
+
+      #
+      # @!method stop
+      #   Releases the resources of the store. The SDK calls this method when the client closes.
+      #
+      #   @return [void]
+      #
+    end
+  end
+end
diff --git a/spec/feature_store_spec_base.rb b/spec/feature_store_spec_base.rb
index 6f7a012e..e6d1230d 100644
--- a/spec/feature_store_spec_base.rb
+++ b/spec/feature_store_spec_base.rb
@@ -38,7 +38,7 @@
 
 # Rather than testing with feature flag or segment data, we'll use this fake data kind
 # to make it clear that feature stores need to be able to handle arbitrary data.
-$things_kind = { namespace: "things" }
+$things_kind = LaunchDarkly::Interfaces::DataStore::DataKind.new(namespace: "things", priority: 2)
 
 $key1 = "$thing1"
 $thing1 = {
@@ -206,7 +206,7 @@ def new_version_plus(f, delta_version, attrs = {})
 
   # We'll loop through permutations of the following parameters. Note: in the future, the caching logic will
   # be separated out and implemented at a higher level of the SDK, so we won't have to test it for individual
-  # persistent store implementations. Currently caching *is* implemented in a shared class (CachingStoreWrapper),
+  # persistent store implementations. Currently caching *is* implemented in a shared class (PersistentDataStoreWrapper),
   # but the individual store implementations are wrapping themselves in that class, so they can't be tested
   # separately from it.
 
diff --git a/spec/integrations/util/store_wrapper_spec.rb b/spec/integrations/util/persistent_data_store_wrapper_spec.rb
similarity index 62%
rename from spec/integrations/util/store_wrapper_spec.rb
rename to spec/integrations/util/persistent_data_store_wrapper_spec.rb
index 004003af..229fd36f 100644
--- a/spec/integrations/util/store_wrapper_spec.rb
+++ b/spec/integrations/util/persistent_data_store_wrapper_spec.rb
@@ -2,10 +2,10 @@
 
 module LaunchDarkly
   module Integrations
-    describe Util::CachingStoreWrapper do
-      subject { Util::CachingStoreWrapper }
+    describe Util::PersistentDataStoreWrapper do
+      subject { Util::PersistentDataStoreWrapper }
 
-      THINGS = { namespace: "things" }
+      THINGS = Interfaces::DataStore::DataKind.new(namespace: "things", priority: 0)
 
       it "reports monitoring enabled and gets availability from the core" do
         [true, false].each do |expected|
@@ -27,6 +27,19 @@ module Integrations
       shared_examples "tests" do |cached|
         opts = cached ? { expiration: 30 } : { expiration: 0 }
 
+        it "gives the core serialized items" do
+          core = MockCore.new
+          wrapper = subject.new(core, opts)
+
+          wrapper.upsert(THINGS, { key: "flag", version: 3 })
+          wrapper.delete(THINGS, "other", 4)
+
+          expect(core.data[THINGS]["flag"]).to eq Interfaces::DataStore::SerializedItemDescriptor.new(
+            version: 3, deleted: false, serialized_item: '{"key":"flag","version":3}')
+          expect(core.data[THINGS]["other"]).to eq Interfaces::DataStore::SerializedItemDescriptor.new(
+            version: 4, deleted: true, serialized_item: '{"key":"other","version":4,"deleted":true}')
+        end
+
         it "gets item" do
           core = MockCore.new
           wrapper = subject.new(core, opts)
@@ -55,6 +68,16 @@ module Integrations
           expect(wrapper.get(THINGS, key)).to eq(cached ? nil : itemv2)  # if cached, we will not see the new underlying value yet
         end
 
+        it "treats an item with no serialized data as deleted" do
+          core = MockCore.new
+          wrapper = subject.new(core, opts)
+          core.force_set_descriptor(THINGS, "flag",
+            Interfaces::DataStore::SerializedItemDescriptor.new(version: 5, deleted: true, serialized_item: nil))
+
+          expect(wrapper.get(THINGS, "flag")).to be_nil
+          expect(wrapper.all(THINGS)).to eq({})
+        end
+
         it "gets missing item" do
           core = MockCore.new
           wrapper = subject.new(core, opts)
@@ -75,12 +98,10 @@ module Integrations
 
           core.force_set(THINGS, item1)
           core.force_set(THINGS, item2)
-          expect(wrapper.all(THINGS)).to eq({ item1[:key] => item1, item2[:key] => item2 })
+          expect(wrapper.all(THINGS)).to eq({ flag1: item1, flag2: item2 })
 
           core.force_remove(THINGS, item2[:key])
-          expect(wrapper.all(THINGS)).to eq(cached ?
-            { item1[:key] => item1, item2[:key] => item2 } :
-            { item1[:key] => item1 })
+          expect(wrapper.all(THINGS)).to eq(cached ? { flag1: item1, flag2: item2 } : { flag1: item1 })
         end
 
         it "gets all items filtering out deleted items" do
@@ -91,7 +112,7 @@ module Integrations
 
           core.force_set(THINGS, item1)
           core.force_set(THINGS, item2)
-          expect(wrapper.all(THINGS)).to eq({ item1[:key] => item1 })
+          expect(wrapper.all(THINGS)).to eq({ flag1: item1 })
         end
 
         it "upserts item successfully" do
@@ -102,10 +123,10 @@ module Integrations
           itemv2 = { key: key, version: 2 }
 
           expect(wrapper.upsert(THINGS, itemv1)).to be true
-          expect(core.data[THINGS][key]).to eq itemv1
+          expect(core.stored(THINGS, key)).to eq itemv1
 
           expect(wrapper.upsert(THINGS, itemv2)).to be true
-          expect(core.data[THINGS][key]).to eq itemv2
+          expect(core.stored(THINGS, key)).to eq itemv2
 
           # if we have a cache, verify that the new item is now cached by writing a different value
           # to the underlying data - Get should still return the cached item
@@ -129,7 +150,7 @@ module Integrations
           expect(wrapper.get(THINGS, key)).to eq itemv1
 
           expect(wrapper.delete(THINGS, key, 2)).to be true
-          expect(core.data[THINGS][key]).to eq itemv2
+          expect(core.stored(THINGS, key)).to eq itemv2
 
           core.force_set(THINGS, itemv3)  # make a change that bypasses the cache
 
@@ -144,7 +165,7 @@ module Integrations
 
           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
+          expect(core.stored(THINGS, "flag")).to eq itemv2
         end
 
         it "returns false when delete has a version that is not newer" do
@@ -154,7 +175,33 @@ module Integrations
           core.force_set(THINGS, itemv2)
 
           expect(wrapper.delete(THINGS, "flag", 2)).to be false
-          expect(core.data[THINGS]["flag"]).to eq itemv2
+          expect(core.stored(THINGS, "flag")).to eq itemv2
+        end
+
+        it "returns false when the core does not return true from upsert" do
+          core = MockCore.new
+          core.upsert_result = nil
+          wrapper = subject.new(core, opts)
+
+          expect(wrapper.upsert(THINGS, { key: "flag", version: 1 })).to be false
+        end
+
+        it "raises an error from init and does not report initialized" do
+          core = MockCore.new
+          core.error = RuntimeError.new("sorry")
+          wrapper = subject.new(core, opts)
+
+          expect { wrapper.init({ THINGS => {} }) }.to raise_error("sorry")
+          core.error = nil
+          expect(wrapper.initialized?).to be false
+        end
+
+        it "raises an error from upsert" do
+          core = MockCore.new
+          core.error = RuntimeError.new("sorry")
+          wrapper = subject.new(core, opts)
+
+          expect { wrapper.upsert(THINGS, { key: "flag", version: 1 }) }.to raise_error("sorry")
         end
       end
 
@@ -169,7 +216,7 @@ module Integrations
           item1 = { key: "flag1", version: 1 }
           item2 = { key: "flag2", version: 1 }
 
-          wrapper.init({ THINGS => { item1[:key] => item1, item2[:key] => item2 } })
+          wrapper.init({ THINGS => { flag1: item1, flag2: item2 } })
           core.force_remove(THINGS, item1[:key])
 
           expect(wrapper.get(THINGS, item1[:key])).to eq item1
@@ -181,10 +228,37 @@ module Integrations
           item1 = { key: "flag1", version: 1 }
           item2 = { key: "flag2", version: 1 }
 
-          wrapper.init({ THINGS => { item1[:key] => item1, item2[:key] => item2 } })
+          wrapper.init({ THINGS => { flag1: item1, flag2: item2 } })
           core.force_remove(THINGS, item1[:key])
 
-          expect(wrapper.all(THINGS)).to eq({ item1[:key] => item1, item2[:key] => item2 })
+          expect(wrapper.all(THINGS)).to eq({ flag1: item1, flag2: item2 })
+        end
+
+        it "init clears the cache if the core fails" do
+          core = MockCore.new
+          wrapper = subject.new(core, cached_opts)
+          item = { key: "flag", version: 1 }
+          wrapper.init({ THINGS => { flag: item } })
+
+          core.error = RuntimeError.new("sorry")
+          expect { wrapper.init({ THINGS => {} }) }.to raise_error("sorry")
+          core.error = nil
+
+          core.force_remove(THINGS, "flag")
+          expect(wrapper.get(THINGS, "flag")).to be_nil
+        end
+
+        it "upsert does not change the cache if the core fails" do
+          core = MockCore.new
+          wrapper = subject.new(core, cached_opts)
+          itemv1 = { key: "flag", version: 1 }
+          wrapper.upsert(THINGS, itemv1)
+
+          core.error = RuntimeError.new("sorry")
+          expect { wrapper.upsert(THINGS, { key: "flag", version: 2 }) }.to raise_error("sorry")
+          core.error = nil
+
+          expect(wrapper.get(THINGS, "flag")).to eq itemv1
         end
 
         it "upsert doesn't update cache if unsuccessful" do
@@ -199,10 +273,10 @@ module Integrations
           itemv2 = { key: key, version: 2 }
 
           wrapper.upsert(THINGS, itemv2)
-          expect(core.data[THINGS][key]).to eq itemv2
+          expect(core.stored(THINGS, key)).to eq itemv2
 
           wrapper.upsert(THINGS, itemv1)
-          expect(core.data[THINGS][key]).to eq itemv2  # value in store remains the same
+          expect(core.stored(THINGS, key)).to eq itemv2  # value in store remains the same
 
           itemv3 = { key: key, version: 3 }
           core.force_set(THINGS, itemv3)  # bypasses cache so we can verify that itemv2 is in the cache
@@ -224,6 +298,32 @@ module Integrations
           expect(wrapper.get(THINGS, key)).to eq itemv3
         end
 
+        it "upsert keeps the cached item if the store read fails after an unsuccessful write" do
+          core = MockCore.new
+          wrapper = subject.new(core, cached_opts)
+          itemv2 = { key: "flag", version: 2 }
+          wrapper.upsert(THINGS, itemv2)
+
+          core.get_error = RuntimeError.new("sorry")
+          expect(wrapper.upsert(THINGS, { key: "flag", version: 1 })).to be false
+          core.get_error = nil
+
+          core.force_set(THINGS, { key: "flag", version: 3 })
+          expect(wrapper.get(THINGS, "flag")).to eq itemv2
+        end
+
+        it "upsert drops the cached list of all items" do
+          core = MockCore.new
+          wrapper = subject.new(core, cached_opts)
+          item1 = { key: "flag1", version: 1 }
+          item2 = { key: "flag2", version: 1 }
+          wrapper.init({ THINGS => { flag1: item1 } })
+          expect(wrapper.all(THINGS)).to eq({ flag1: item1 })
+
+          wrapper.upsert(THINGS, item2)
+          expect(wrapper.all(THINGS)).to eq({ flag1: item1, flag2: item2 })
+        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
@@ -343,49 +443,88 @@ module Integrations
         end
       end
 
+      describe Interfaces::PersistentDataStore do
+        it "does not hide a method that a store inherits from a superclass" do
+          base = Class.new do
+            def stop
+              :stopped
+            end
+          end
+          store = Class.new(base) { include Interfaces::PersistentDataStore }.new
+
+          expect(store.stop).to eq :stopped
+        end
+
+        it "does not supply a method that a store does not implement" do
+          store = Class.new { include Interfaces::PersistentDataStore }.new
+
+          expect { store.stop }.to raise_error(NoMethodError)
+        end
+      end
+
+      #
+      # A store core that keeps serialized items in memory. Like Redis, it does not keep the version
+      # apart from the serialized item.
+      #
       class MockCore
+        include Interfaces::PersistentDataStore
+
         def initialize
           @data = {}
           @inited = false
           @inited_query_count = 0
+          @upsert_result = :default
         end
 
         attr_reader :data
         attr_reader :inited_query_count
-        attr_accessor :inited
+        attr_accessor :inited, :error, :get_error, :upsert_result
 
         def force_set(kind, item)
+          force_set_descriptor(kind, item[:key], Interfaces::DataStore::SerializedItemDescriptor.new(
+            version: 0, deleted: false, serialized_item: kind.serialize(item)))
+        end
+
+        def force_set_descriptor(kind, key, descriptor)
           @data[kind] = {} unless @data.has_key?(kind)
-          @data[kind][item[:key]] = item
+          @data[kind][key] = descriptor
         end
 
         def force_remove(kind, key)
           @data[kind].delete(key) if @data.has_key?(kind)
         end
 
-        def init_internal(all_data)
-          @data = all_data
+        def stored(kind, key)
+          kind.deserialize(@data[kind][key].serialized_item)
+        end
+
+        def init(all_data)
+          raise @error unless @error.nil?
+          @data = all_data.to_h { |kind, items| [kind, items.transform_keys(&:to_s)] }
           @inited = true
         end
 
-        def get_internal(kind, key)
+        def get(kind, key)
+          raise @get_error unless @get_error.nil?
           items = @data[kind]
-          items.nil? ? nil : items[key]
+          items.nil? ? nil : items[key.to_s]
         end
 
-        def get_all_internal(kind)
-          @data[kind]
+        def get_all(kind)
+          @data[kind] || {}
         end
 
-        def upsert_internal(kind, item)
+        def upsert(kind, key, item)
+          raise @error unless @error.nil?
+          return @upsert_result unless @upsert_result == :default
           @data[kind] = {} unless @data.has_key?(kind)
-          old_item = @data[kind][item[:key]]
-          return false if !old_item.nil? && old_item[:version] >= item[:version]
-          @data[kind][item[:key]] = item
+          old = @data[kind][key.to_s]
+          return false if !old.nil? && kind.deserialize(old.serialized_item)[:version] >= item.version
+          @data[kind][key.to_s] = item
           true
         end
 
-        def initialized_internal?
+        def initialized?
           @inited_query_count = @inited_query_count + 1
           @inited
         end