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