From 51618e8c02a8d7c061db3eb0ec7a876d40388782 Mon Sep 17 00:00:00 2001
From: Matthew Keeler <mkeeler@launchdarkly.com>
Date: Tue, 3 Nov 2026 10:11:04 -0500
Subject: [PATCH 1/3] feat!: Pass serialized items to persistent store cores

---
 ldclient/async_feature_store_helpers.py       |  27 ++--
 ldclient/feature_store_helpers.py             |  91 +++++++----
 .../consul/consul_feature_store.py            |  44 +++---
 .../dynamodb/async_dynamodb_feature_store.py  |  55 ++++---
 .../dynamodb/dynamodb_feature_store.py        |  40 ++---
 .../redis/async_redis_feature_store.py        |  44 +++---
 .../integrations/redis/redis_feature_store.py |  40 +++--
 ldclient/interfaces.py                        | 145 ++++++++++--------
 .../impl/datasystem/test_cache_lifecycle.py   |  23 ++-
 .../datasystem/test_fdv2_async_persistence.py |  28 ++--
 ldclient/testing/serialized_items.py          |  26 ++++
 .../test_async_feature_store_helpers.py       |  78 ++++++++--
 .../testing/test_feature_store_helpers.py     |  94 ++++++++++--
 ldclient/versioned_data_kind.py               |  22 +++
 14 files changed, 519 insertions(+), 238 deletions(-)
 create mode 100644 ldclient/testing/serialized_items.py

diff --git a/ldclient/async_feature_store_helpers.py b/ldclient/async_feature_store_helpers.py
index 75455967..2a91e52b 100644
--- a/ldclient/async_feature_store_helpers.py
+++ b/ldclient/async_feature_store_helpers.py
@@ -7,7 +7,9 @@
 from ldclient.feature_store import CacheConfig
 from ldclient.feature_store_helpers import (
     _CachingStoreWrapperBase,
-    _ensure_encoded
+    _ensure_encoded,
+    _serialize,
+    _serialize_all
 )
 from ldclient.interfaces import (
     AsyncFeatureStore,
@@ -55,7 +57,7 @@ async def is_available(self) -> bool:
 
     async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None:
         """ """
-        await self._core.init_internal(all_data)
+        await self._core.init_internal(_serialize_all(all_data))
         self._cache_init(all_data)
         self._inited = True
 
@@ -64,16 +66,16 @@ async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]:
         hit, value = self._cache_get_item(kind, key)
         if hit:
             return value
-        encoded_item = await self._core.get_internal(kind, key)
-        return self._cache_put_item(kind, key, encoded_item)
+        descriptor = await self._core.get_internal(kind, key)
+        return self._cache_put_item(kind, key, descriptor)
 
     async def all(self, kind: VersionedDataKind) -> Dict[str, Any]:
         """ """
         hit, value = self._cache_get_all(kind)
         if hit:
             return value
-        encoded_items = await self._core.get_all_internal(kind)
-        return self._cache_put_all(kind, encoded_items)
+        descriptors = await self._core.get_all_internal(kind)
+        return self._cache_put_all(kind, descriptors)
 
     async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool:
         """ """
@@ -83,11 +85,14 @@ async def delete(self, kind: VersionedDataKind, key: str, version: int) -> bool:
     async def upsert(self, kind: VersionedDataKind, item: dict) -> bool:
         """ """
         encoded_item = _ensure_encoded(kind, item)
-        new_state = await self._core.upsert_internal(kind, encoded_item)
-        self._cache_put_upsert(kind, new_state)
-        # The core returns the item we passed in if the write was applied, or the existing item if it
-        # was rejected by the version check. Identity therefore tells us whether the store changed.
-        return new_state is encoded_item
+        key = encoded_item['key']
+        updated = await self._core.upsert_internal(kind, key, _serialize(encoded_item))
+        if updated:
+            self._cache_put_upsert(kind, key, encoded_item)
+        elif self._caching():
+            # The store has a newer item. Read it, so that the cache does not keep an older one.
+            self._cache_put_item(kind, key, await self._core.get_internal(kind, key))
+        return updated
 
     @property
     def initialized(self) -> bool:
diff --git a/ldclient/feature_store_helpers.py b/ldclient/feature_store_helpers.py
index 4d18e0cd..3b616fe5 100644
--- a/ldclient/feature_store_helpers.py
+++ b/ldclient/feature_store_helpers.py
@@ -2,7 +2,8 @@
 This submodule contains support code for writing feature store implementations.
 """
 
-from typing import Any, Dict, Mapping
+import json
+from typing import Any, Dict, Mapping, Optional
 
 from expiringdict import ExpiringDict
 
@@ -11,7 +12,8 @@
 from ldclient.interfaces import (
     DiagnosticDescription,
     FeatureStore,
-    FeatureStoreCore
+    FeatureStoreCore,
+    SerializedItemDescriptor
 )
 from ldclient.versioned_data_kind import VersionedDataKind
 
@@ -24,6 +26,37 @@ def _is_deleted(item):
     return item is not None and item.get('deleted') is True
 
 
+def _serialize(encoded_item: dict) -> SerializedItemDescriptor:
+    """Serializes an encoded item, which can be a deleted item, for a persistent store core."""
+    return SerializedItemDescriptor(
+        version=encoded_item.get('version', 0),
+        deleted=encoded_item.get('deleted') is True,
+        serialized_item=json.dumps(encoded_item),
+    )
+
+
+def _serialize_all(all_data) -> Dict[VersionedDataKind, Dict[str, SerializedItemDescriptor]]:
+    """Serializes a full data set. The order of the kinds and of the items does not change."""
+    return {kind: {key: _serialize(kind.encode(item)) for key, item in items.items()} for kind, items in all_data.items()}
+
+
+def _deserialize(kind, key: str, descriptor: Optional[SerializedItemDescriptor]):
+    """Deserializes an item that a persistent store core returned.
+
+    Returns None if the core found no item. A deleted item becomes a decoded placeholder with the
+    deleted flag. If the core gives a version that differs from the version in the serialized item,
+    the version from the core is used.
+    """
+    if descriptor is None:
+        return None
+    if descriptor.deleted:
+        return kind.decode({'key': key, 'version': descriptor.version, 'deleted': True})
+    data = json.loads(descriptor.serialized_item)
+    if descriptor.version != 0 and data.get('version') != descriptor.version:
+        data['version'] = descriptor.version
+    return kind.decode(data)
+
+
 class _NoopCache:
     """A cache replacement whose operations are all no-ops.
 
@@ -107,12 +140,12 @@ def _cache_get_item(self, kind, key):
         item = cached_item[0]
         return (True, None if _is_deleted(item) else item)
 
-    def _cache_put_item(self, kind, key, encoded_item):
-        """Decodes an item fetched from the core, caches it, and returns the value to return.
+    def _cache_put_item(self, kind, key, descriptor):
+        """Deserializes an item fetched from the core, caches it, and returns the value to return.
 
         The returned value is None if the item is missing or deleted.
         """
-        item = None if encoded_item is None else kind.decode(encoded_item)
+        item = _deserialize(kind, key, descriptor)
         self._cache[self._item_cache_key(kind, key)] = [item]
         return None if _is_deleted(item) else item
 
@@ -127,12 +160,12 @@ def _cache_get_all(self, kind):
             return (False, None)
         return (True, cached_items)
 
-    def _cache_put_all(self, kind, encoded_items):
-        """Decodes all items fetched from the core, drops deleted ones, caches the result, and returns it."""
+    def _cache_put_all(self, kind, descriptors):
+        """Deserializes all items fetched from the core, drops deleted ones, caches the result, and returns it."""
         all_items = {}
-        if encoded_items is not None:
-            for key, item in encoded_items.items():
-                all_items[key] = kind.decode(item)
+        if descriptors is not None:
+            for key, descriptor in descriptors.items():
+                all_items[key] = _deserialize(kind, key, descriptor)
         items = self._items_if_not_deleted(all_items)
         self._cache[self._all_cache_key(kind)] = items
         return items
@@ -152,16 +185,13 @@ def _cache_init(self, all_encoded_data):
                     decoded_items[key] = decoded_item
             cache[self._all_cache_key(kind)] = decoded_items
 
-    def _cache_put_upsert(self, kind, new_state):
-        """Updates the cache after an upsert.
-
-        Caches the item the core returned and drops the now-stale all-items entry. Returns the
-        decoded item.
-        """
-        new_decoded_item = kind.decode(new_state)
-        self._cache[self._item_cache_key(kind, new_decoded_item.get('key'))] = [new_decoded_item]
+    def _cache_put_upsert(self, kind, key, encoded_item):
+        """Updates the cache after the core wrote an item, and drops the now-stale all-items entry."""
+        self._cache[self._item_cache_key(kind, key)] = [kind.decode(encoded_item)]
         self._cache.pop(self._all_cache_key(kind), None)
-        return new_decoded_item
+
+    def _caching(self) -> bool:
+        return self._cache is not _NOOP_CACHE
 
     @staticmethod
     def _item_cache_key(kind, key):
@@ -209,7 +239,7 @@ def is_available(self) -> bool:
 
     def init(self, all_encoded_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]):
         """ """
-        self._core.init_internal(all_encoded_data)  # currently FeatureStoreCore expects to receive dicts
+        self._core.init_internal(_serialize_all(all_encoded_data))
         self._cache_init(all_encoded_data)
         self._inited = True
 
@@ -218,16 +248,16 @@ def get(self, kind, key, callback=lambda x: x):
         hit, value = self._cache_get_item(kind, key)
         if hit:
             return callback(value)
-        encoded_item = self._core.get_internal(kind, key)  # currently FeatureStoreCore returns dicts
-        return callback(self._cache_put_item(kind, key, encoded_item))
+        descriptor = self._core.get_internal(kind, key)
+        return callback(self._cache_put_item(kind, key, descriptor))
 
     def all(self, kind, callback=lambda x: x):
         """ """
         hit, value = self._cache_get_all(kind)
         if hit:
             return callback(value)
-        encoded_items = self._core.get_all_internal(kind)
-        return callback(self._cache_put_all(kind, encoded_items))
+        descriptors = self._core.get_all_internal(kind)
+        return callback(self._cache_put_all(kind, descriptors))
 
     def delete(self, kind, key, version) -> bool:
         """ """
@@ -237,11 +267,14 @@ def delete(self, kind, key, version) -> bool:
     def upsert(self, kind, encoded_item) -> bool:
         """ """
         encoded_item = _ensure_encoded(kind, encoded_item)
-        new_state = self._core.upsert_internal(kind, encoded_item)
-        self._cache_put_upsert(kind, new_state)
-        # The core returns the item that we passed in if it applied the write. It returns the
-        # existing item if the version check rejected the write. Identity thus shows the result.
-        return new_state is encoded_item
+        key = encoded_item['key']
+        updated = self._core.upsert_internal(kind, key, _serialize(encoded_item))
+        if updated:
+            self._cache_put_upsert(kind, key, encoded_item)
+        elif self._caching():
+            # The store has a newer item. Read it, so that the cache does not keep an older one.
+            self._cache_put_item(kind, key, self._core.get_internal(kind, key))
+        return updated
 
     @property
     def initialized(self) -> bool:
diff --git a/ldclient/impl/integrations/consul/consul_feature_store.py b/ldclient/impl/integrations/consul/consul_feature_store.py
index 142b089e..5d1b11ea 100644
--- a/ldclient/impl/integrations/consul/consul_feature_store.py
+++ b/ldclient/impl/integrations/consul/consul_feature_store.py
@@ -1,7 +1,9 @@
-import json
-
 from ldclient import log
-from ldclient.interfaces import DiagnosticDescription, FeatureStoreCore
+from ldclient.interfaces import (
+    DiagnosticDescription,
+    FeatureStoreCore,
+    SerializedItemDescriptor
+)
 
 have_consul = False
 try:
@@ -69,9 +71,8 @@ def init_internal(self, all_data):
         # operations (the "txn" method), so we'll write them one at a time.
         for kind, items in all_data.items():
             for key, item in items.items():
-                encoded_item = json.dumps(item)
-                db_key = self._item_key(kind, item['key'])
-                self._client.kv.put(db_key, encoded_item)
+                db_key = self._item_key(kind, key)
+                self._client.kv.put(db_key, item.serialized_item)
                 unused_old_keys.discard(db_key)
                 num_items = num_items + 1
 
@@ -86,7 +87,7 @@ def init_internal(self, all_data):
 
     def get_internal(self, kind, key):
         index, resp = self._client.kv.get(self._item_key(kind, key))
-        return None if resp is None else json.loads(resp['Value'].decode('utf-8'))
+        return None if resp is None else _descriptor(resp['Value'])
 
     def get_all_internal(self, kind):
         items_out = {}
@@ -99,33 +100,30 @@ def get_all_internal(self, kind):
             db_key = result['Key']
             if not db_key.startswith(item_key_prefix):
                 continue
-            item = json.loads(result['Value'].decode('utf-8'))
-            items_out[db_key[len(item_key_prefix):]] = item
+            items_out[db_key[len(item_key_prefix):]] = _descriptor(result['Value'])
         return items_out
 
-    def upsert_internal(self, kind, new_item):
-        key = self._item_key(kind, new_item['key'])
-        encoded_item = json.dumps(new_item)
+    def upsert_internal(self, kind, key, new_item):
+        db_key = self._item_key(kind, key)
 
         # We will potentially keep retrying indefinitely until someone's write succeeds
         while True:
-            index, old_value = self._client.kv.get(key)
+            index, old_value = self._client.kv.get(db_key)
             if old_value is None:
                 mod_index = 0
             else:
-                old_item = json.loads(old_value['Value'].decode('utf-8'))
-                # Check whether the item is stale. If so, don't do the update (and return the existing item to
-                # CachingStoreWrapper so it can be cached)
-                if old_item['version'] >= new_item['version']:
-                    return old_item
+                # Do not write an item that is not newer than the existing item.
+                old_version = kind.deserialize(old_value['Value'].decode('utf-8'))['version']
+                if old_version >= new_item.version:
+                    return False
                 mod_index = old_value['ModifyIndex']
 
             # 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 = self._client.kv.put(key, encoded_item, cas=mod_index)
+            success = self._client.kv.put(db_key, new_item.serialized_item, cas=mod_index)
             if success:
-                return new_item
+                return True
 
             log.debug('Concurrent modification detected, retrying')
 
@@ -144,3 +142,9 @@ def _item_key(self, kind, key):
 
     def _inited_key(self):
         return self._prefix + '$inited'
+
+
+def _descriptor(value: bytes) -> SerializedItemDescriptor:
+    # The store keeps only the serialized item. The SDK reads the version and the deleted state
+    # from it.
+    return SerializedItemDescriptor(version=0, deleted=False, serialized_item=value.decode('utf-8'))
diff --git a/ldclient/impl/integrations/dynamodb/async_dynamodb_feature_store.py b/ldclient/impl/integrations/dynamodb/async_dynamodb_feature_store.py
index 816231ba..ca56e438 100644
--- a/ldclient/impl/integrations/dynamodb/async_dynamodb_feature_store.py
+++ b/ldclient/impl/integrations/dynamodb/async_dynamodb_feature_store.py
@@ -1,10 +1,13 @@
 import asyncio
-import json
 from contextlib import AsyncExitStack
-from typing import Any, Mapping, Optional, cast
+from typing import Any, Mapping, Optional
 
 from ldclient.impl.util import log
-from ldclient.interfaces import AsyncFeatureStoreCore, DiagnosticDescription
+from ldclient.interfaces import (
+    AsyncFeatureStoreCore,
+    DiagnosticDescription,
+    SerializedItemDescriptor
+)
 from ldclient.versioned_data_kind import VersionedDataKind
 
 have_aioboto3 = False
@@ -85,7 +88,7 @@ async def is_available(self) -> bool:
         except BaseException:
             return False
 
-    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None:
+    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, SerializedItemDescriptor]]) -> None:
         client = await self._get_client()
         # Start by reading the existing keys; we will later delete any of these that were not in all_data.
         unused_old_keys = await self._read_existing_keys(client, all_data.keys())
@@ -96,7 +99,7 @@ async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str,
         # Insert or update every provided item
         for kind, items in all_data.items():
             for key, item in items.items():
-                encoded_item = self._marshal_item(kind, item)
+                encoded_item = self._marshal_item(kind, key, item)
                 requests.append({'PutRequest': {'Item': encoded_item}})
                 combined_key = (self._namespace_for_kind(kind), key)
                 unused_old_keys.discard(combined_key)
@@ -113,39 +116,40 @@ async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str,
         await _AsyncDynamoDBHelpers.batch_write_requests(client, self._table_name, requests)
         log.info('Initialized table %s with %d items', self._table_name, num_items)
 
-    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[dict]:
+    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[SerializedItemDescriptor]:
         client = await self._get_client()
         resp = await self._get_item_by_keys(client, self._namespace_for_kind(kind), key)
         return self._unmarshal_item(resp.get('Item'))
 
-    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, dict]:
+    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, SerializedItemDescriptor]:
         client = await self._get_client()
         items_out = {}
         paginator = client.get_paginator('query')
         async for resp in paginator.paginate(**self._make_query_for_kind(kind)):
             for item in resp['Items']:
-                # Every stored item carries the JSON attribute, so _unmarshal_item never returns None here.
-                item_out = cast(dict, self._unmarshal_item(item))
-                items_out[item_out['key']] = item_out
+                item_out = self._unmarshal_item(item)
+                if item_out is not None:
+                    # Use the sort key that each item is stored under. A deleted item (a
+                    # "tombstone") is not guaranteed to have a key of its own.
+                    items_out[item[self.SORT_KEY]['S']] = item_out
         return items_out
 
-    async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict:
+    async def upsert_internal(self, kind: VersionedDataKind, key: str, item: SerializedItemDescriptor) -> bool:
         client = await self._get_client()
-        encoded_item = self._marshal_item(kind, item)
+        encoded_item = self._marshal_item(kind, key, item)
         try:
             req = {
                 'TableName': self._table_name,
                 'Item': encoded_item,
                 'ConditionExpression': 'attribute_not_exists(#namespace) or attribute_not_exists(#key) or :version > #version',
                 'ExpressionAttributeNames': {'#namespace': self.PARTITION_KEY, '#key': self.SORT_KEY, '#version': self.VERSION_ATTRIBUTE},
-                'ExpressionAttributeValues': {':version': {'N': str(item['version'])}},
+                'ExpressionAttributeValues': {':version': {'N': str(item.version)}},
             }
             await client.put_item(**req)
         except client.exceptions.ConditionalCheckFailedException:
-            # The item was not updated because there's a newer item in the database. We must now
-            # read the item that's in the database and return it, so the wrapper can cache it.
-            return cast(dict, await self.get_internal(kind, item['key']))
-        return item
+            # The table has an item with the same or a newer version.
+            return False
+        return True
 
     async def initialized_internal(self) -> bool:
         client = await self._get_client()
@@ -197,18 +201,21 @@ async def _read_existing_keys(self, client: Any, kinds) -> set:
                     keys.add((namespace, key))
         return keys
 
-    def _marshal_item(self, kind: VersionedDataKind, item: dict) -> dict:
-        json_str = json.dumps(item)
-        ret = self._make_keys(self._namespace_for_kind(kind), item['key'])
-        ret[self.VERSION_ATTRIBUTE] = {'N': str(item['version'])}
-        ret[self.ITEM_JSON_ATTRIBUTE] = {'S': json_str}
+    def _marshal_item(self, kind: VersionedDataKind, key: str, item: SerializedItemDescriptor) -> dict:
+        ret = self._make_keys(self._namespace_for_kind(kind), key)
+        ret[self.VERSION_ATTRIBUTE] = {'N': str(item.version)}
+        ret[self.ITEM_JSON_ATTRIBUTE] = {'S': item.serialized_item}
         return ret
 
-    def _unmarshal_item(self, item: Optional[dict]) -> Optional[dict]:
+    def _unmarshal_item(self, item: Optional[dict]) -> Optional[SerializedItemDescriptor]:
         if item is None:
             return None
         json_attr = item.get(self.ITEM_JSON_ATTRIBUTE)
-        return None if json_attr is None else json.loads(json_attr['S'])
+        if json_attr is None:
+            return None
+        version_attr = item.get(self.VERSION_ATTRIBUTE)
+        version = 0 if version_attr is None else int(version_attr['N'])
+        return SerializedItemDescriptor(version=version, deleted=False, serialized_item=json_attr['S'])
 
 
 class _AsyncDynamoDBHelpers:
diff --git a/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py b/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py
index 2ff784be..4af3747a 100644
--- a/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py
+++ b/ldclient/impl/integrations/dynamodb/dynamodb_feature_store.py
@@ -1,7 +1,6 @@
-import json
 
 from ldclient import log
-from ldclient.interfaces import FeatureStoreCore
+from ldclient.interfaces import FeatureStoreCore, SerializedItemDescriptor
 
 have_dynamodb = False
 try:
@@ -75,7 +74,7 @@ def init_internal(self, all_data):
         # Insert or update every provided item
         for kind, items in all_data.items():
             for key, item in items.items():
-                encoded_item = self._marshal_item(kind, item)
+                encoded_item = self._marshal_item(kind, key, item)
                 requests.append({'PutRequest': {'Item': encoded_item}})
                 combined_key = (self._namespace_for_kind(kind), key)
                 unused_old_keys.discard(combined_key)
@@ -102,27 +101,27 @@ def get_all_internal(self, kind):
         for resp in paginator.paginate(**self._make_query_for_kind(kind)):
             for item in resp['Items']:
                 item_out = self._unmarshal_item(item)
-                # Use the sort key that each item is stored under, not the key inside the item.
-                # A deleted item (a "tombstone") is not guaranteed to have a key of its own.
-                items_out[item[self.SORT_KEY]['S']] = item_out
+                if item_out is not None:
+                    # Use the sort key that each item is stored under. A deleted item (a
+                    # "tombstone") is not guaranteed to have a key of its own.
+                    items_out[item[self.SORT_KEY]['S']] = item_out
         return items_out
 
-    def upsert_internal(self, kind, item):
-        encoded_item = self._marshal_item(kind, item)
+    def upsert_internal(self, kind, key, item):
+        encoded_item = self._marshal_item(kind, key, item)
         try:
             req = {
                 'TableName': self._table_name,
                 'Item': encoded_item,
                 'ConditionExpression': 'attribute_not_exists(#namespace) or attribute_not_exists(#key) or :version > #version',
                 'ExpressionAttributeNames': {'#namespace': self.PARTITION_KEY, '#key': self.SORT_KEY, '#version': self.VERSION_ATTRIBUTE},
-                'ExpressionAttributeValues': {':version': {'N': str(item['version'])}},
+                'ExpressionAttributeValues': {':version': {'N': str(item.version)}},
             }
             self._client.put_item(**req)
         except self._client.exceptions.ConditionalCheckFailedException:
-            # The item was not updated because there's a newer item in the database. We must now
-            # read the item that's in the database and return it, so CachingStoreWrapper can cache it.
-            return self.get_internal(kind, item['key'])
-        return item
+            # The table has an item with the same or a newer version.
+            return False
+        return True
 
     def initialized_internal(self):
         resp = self._get_item_by_keys(self._inited_key(), self._inited_key())
@@ -167,18 +166,21 @@ def _read_existing_keys(self, kinds):
                     keys.add((namespace, key))
         return keys
 
-    def _marshal_item(self, kind, item):
-        json_str = json.dumps(item)
-        ret = self._make_keys(self._namespace_for_kind(kind), item['key'])
-        ret[self.VERSION_ATTRIBUTE] = {'N': str(item['version'])}
-        ret[self.ITEM_JSON_ATTRIBUTE] = {'S': json_str}
+    def _marshal_item(self, kind, key, item):
+        ret = self._make_keys(self._namespace_for_kind(kind), key)
+        ret[self.VERSION_ATTRIBUTE] = {'N': str(item.version)}
+        ret[self.ITEM_JSON_ATTRIBUTE] = {'S': item.serialized_item}
         return ret
 
     def _unmarshal_item(self, item):
         if item is None:
             return None
         json_attr = item.get(self.ITEM_JSON_ATTRIBUTE)
-        return None if json_attr is None else json.loads(json_attr['S'])
+        if json_attr is None:
+            return None
+        version_attr = item.get(self.VERSION_ATTRIBUTE)
+        version = 0 if version_attr is None else int(version_attr['N'])
+        return SerializedItemDescriptor(version=version, deleted=False, serialized_item=json_attr['S'])
 
 
 class _DynamoDBHelpers:
diff --git a/ldclient/impl/integrations/redis/async_redis_feature_store.py b/ldclient/impl/integrations/redis/async_redis_feature_store.py
index d3fed4ad..4d840a89 100644
--- a/ldclient/impl/integrations/redis/async_redis_feature_store.py
+++ b/ldclient/impl/integrations/redis/async_redis_feature_store.py
@@ -1,9 +1,12 @@
-import json
 from typing import Any, Callable, Dict, Mapping, Optional
 
 from ldclient.feature_store_helpers import CachingStoreWrapper
 from ldclient.impl.util import log, redact_password
-from ldclient.interfaces import AsyncFeatureStoreCore, DiagnosticDescription
+from ldclient.interfaces import (
+    AsyncFeatureStoreCore,
+    DiagnosticDescription,
+    SerializedItemDescriptor
+)
 from ldclient.versioned_data_kind import VersionedDataKind
 
 have_async_redis = False
@@ -45,39 +48,37 @@ async def is_available(self) -> bool:
     def _items_key(self, kind: VersionedDataKind) -> str:
         return "{0}:{1}".format(self._prefix, kind.namespace)
 
-    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None:
+    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, SerializedItemDescriptor]]) -> None:
         all_count = 0
         async with self._client.pipeline() as pipe:
             for kind, items in all_data.items():
                 base_key = self._items_key(kind)
                 pipe.delete(base_key)
                 for key, item in items.items():
-                    pipe.hset(base_key, key, json.dumps(item))
+                    pipe.hset(base_key, key, item.serialized_item)
                 all_count = all_count + len(items)
             pipe.set(self._init_key, self._init_key)
             await pipe.execute()
         log.info("Initialized AsyncRedisFeatureStore with %d items", all_count)
 
-    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, dict]:
+    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, SerializedItemDescriptor]:
         all_items = await self._client.hgetall(self._items_key(kind))
         if not all_items:
             return {}
         results = {}
         for key, item_json in all_items.items():
-            results[key.decode('utf-8')] = json.loads(item_json.decode('utf-8'))
+            results[key.decode('utf-8')] = _descriptor(item_json)
         return results
 
-    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[dict]:
+    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[SerializedItemDescriptor]:
         item_json = await self._client.hget(self._items_key(kind), key)
         if not item_json:
             log.debug("AsyncRedisFeatureStore: key %s not found in '%s'. Returning None.", key, kind.namespace)
             return None
-        return json.loads(item_json.decode('utf-8'))
+        return _descriptor(item_json)
 
-    async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict:
+    async def upsert_internal(self, kind: VersionedDataKind, key: str, item: SerializedItemDescriptor) -> bool:
         base_key = self._items_key(kind)
-        key = item['key']
-        item_json = json.dumps(item)
 
         for _ in range(_MAX_UPSERT_RETRIES):
             async with self._client.pipeline() as pipe:
@@ -86,23 +87,24 @@ async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict:
                     old = await self.get_internal(kind, key)
                     if self.test_update_hook is not None:
                         self.test_update_hook(base_key, key)
-                    if old and old['version'] >= item['version']:
+                    old_version = None if old is None else kind.deserialize(old.serialized_item)['version']
+                    if old_version is not None and old_version >= item.version:
                         log.debug(
                             'AsyncRedisFeatureStore: Attempted to %s key: %s version %d with a version that is the same or older: %d in "%s"',
-                            'delete' if item.get('deleted') else 'update',
+                            'delete' if item.deleted else 'update',
                             key,
-                            old['version'],
-                            item['version'],
+                            old_version,
+                            item.version,
                             kind.namespace,
                         )
                         await pipe.unwatch()
-                        return old
+                        return False
                     pipe.multi()
-                    pipe.hset(base_key, key, item_json)
+                    pipe.hset(base_key, key, item.serialized_item)
                     # A concurrent change to the watched key makes execute() raise WatchError,
                     # rather than returning a null result as on some other platforms.
                     await pipe.execute()
-                    return item
+                    return True
                 except WatchError:
                     log.debug("AsyncRedisFeatureStore: concurrent modification detected, retrying")
                     continue
@@ -121,3 +123,9 @@ async def close(self) -> None:
 
     def describe_configuration(self, config) -> str:
         return 'Redis'
+
+
+def _descriptor(item_json: bytes) -> SerializedItemDescriptor:
+    # The store keeps only the serialized item. The SDK reads the version and the deleted state
+    # from it.
+    return SerializedItemDescriptor(version=0, deleted=False, serialized_item=item_json.decode('utf-8'))
diff --git a/ldclient/impl/integrations/redis/redis_feature_store.py b/ldclient/impl/integrations/redis/redis_feature_store.py
index 77fde737..5cf8e2dc 100644
--- a/ldclient/impl/integrations/redis/redis_feature_store.py
+++ b/ldclient/impl/integrations/redis/redis_feature_store.py
@@ -1,11 +1,13 @@
-import json
 from typing import Any, Dict
 
 from ldclient import log
 from ldclient.feature_store_helpers import CachingStoreWrapper
 from ldclient.impl.util import redact_password
-from ldclient.interfaces import DiagnosticDescription, FeatureStoreCore
-from ldclient.versioned_data_kind import FEATURES
+from ldclient.interfaces import (
+    DiagnosticDescription,
+    FeatureStoreCore,
+    SerializedItemDescriptor
+)
 
 have_redis = False
 try:
@@ -52,8 +54,7 @@ def init_internal(self, all_data):
             base_key = self._items_key(kind)
             pipe.delete(base_key)
             for key, item in items.items():
-                item_json = json.dumps(item)
-                pipe.hset(base_key, key, item_json)
+                pipe.hset(base_key, key, item.serialized_item)
             all_count = all_count + len(items)
 
         pipe.set(self._init_key, self._init_key)
@@ -70,7 +71,7 @@ def get_all_internal(self, kind):
         results = {}
         for key, item_json in all_items.items():
             key = key.decode('utf-8')  # necessary in Python 3
-            results[key] = json.loads(item_json.decode('utf-8'))
+            results[key] = _descriptor(item_json)
         return results
 
     def get_internal(self, kind, key):
@@ -81,13 +82,11 @@ def get_internal(self, kind, key):
             log.debug("RedisFeatureStore: key %s not found in '%s'. Returning None.", key, kind.namespace)
             return None
 
-        return json.loads(item_json.decode('utf-8'))
+        return _descriptor(item_json)
 
-    def upsert_internal(self, kind, item):
+    def upsert_internal(self, kind, key, item):
         r = redis.Redis(connection_pool=self._pool)
         base_key = self._items_key(kind)
-        key = item['key']
-        item_json = json.dumps(item)
 
         for _ in range(_MAX_UPSERT_RETRIES):
             pipeline = r.pipeline()
@@ -95,20 +94,21 @@ def upsert_internal(self, kind, item):
             old = self.get_internal(kind, key)
             if self.test_update_hook is not None:
                 self.test_update_hook(base_key, key)
-            if old and old['version'] >= item['version']:
+            old_version = None if old is None else kind.deserialize(old.serialized_item)['version']
+            if old_version is not None and old_version >= item.version:
                 log.debug(
                     'RedisFeatureStore: Attempted to %s key: %s version %d with a version that is the same or older: %d in "%s"',
-                    'delete' if item.get('deleted') else 'update',
+                    'delete' if item.deleted else 'update',
                     key,
-                    old['version'],
-                    item['version'],
+                    old_version,
+                    item.version,
                     kind.namespace,
                 )
                 pipeline.unwatch()
-                return old
+                return False
             else:
                 pipeline.multi()
-                pipeline.hset(base_key, key, item_json)
+                pipeline.hset(base_key, key, item.serialized_item)
                 try:
                     pipeline.execute()
                     # Unlike Redis implementations for other platforms, in redis-py a failed WATCH
@@ -116,7 +116,7 @@ def upsert_internal(self, kind, item):
                 except redis.exceptions.WatchError:
                     log.debug("RedisFeatureStore: concurrent modification detected, retrying")
                     continue
-            return item
+            return True
 
         raise RuntimeError("failed to update key %s in '%s' after %d attempts" % (key, kind.namespace, _MAX_UPSERT_RETRIES))
 
@@ -130,3 +130,9 @@ def describe_configuration(self, config):
     def _before_update_transaction(self, base_key, key):
         # exposed for testing
         pass
+
+
+def _descriptor(item_json: bytes) -> SerializedItemDescriptor:
+    # The store keeps only the serialized item. The SDK reads the version and the deleted state
+    # from it.
+    return SerializedItemDescriptor(version=0, deleted=False, serialized_item=item_json.decode('utf-8'))
diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py
index 758925dd..76725565 100644
--- a/ldclient/interfaces.py
+++ b/ldclient/interfaces.py
@@ -197,6 +197,30 @@ def close(self) -> None:
         """
 
 
+@dataclass(frozen=True)
+class SerializedItemDescriptor:
+    """
+    A versioned item in its serialized form, as a persistent store core reads or writes it.
+
+    The SDK serializes each item before it gives the item to a :class:`FeatureStoreCore` or an
+    :class:`AsyncFeatureStoreCore`, and deserializes each item that a core returns. A core thus
+    stores and returns opaque strings, and does not need to know the data model.
+
+    A core can store the version and the deleted state separately from the serialized string, or it
+    can store only the serialized string. In the second case, the core returns a version of 0 and
+    ``deleted`` as False. The SDK then gets both values from the serialized string.
+    """
+
+    version: int
+    """The version of the item. A core that does not store the version separately returns 0."""
+
+    deleted: bool
+    """True if the item is a placeholder for a deleted item (a "tombstone")."""
+
+    serialized_item: str
+    """The serialized item. For a deleted item, this is a placeholder that has the deleted flag."""
+
+
 class FeatureStoreCore(ABC):
     """
     Interface for a simplified subset of the functionality of :class:`FeatureStore`, to be used
@@ -204,65 +228,66 @@ class FeatureStoreCore(ABC):
     developers of custom ``FeatureStore`` implementations to avoid repeating logic that would
     commonly be needed in any such implementation, such as caching. Instead, they can implement
     only ``FeatureStoreCore`` and then create a ``CachingStoreWrapper``.
+
+    All items that go into and come out of these methods are :class:`SerializedItemDescriptor`
+    instances. The wrapper serializes and deserializes the items, encodes deleted items, and caches
+    the results. A core must not cache items.
     """
 
     @abstractmethod
-    def get_internal(self, kind: VersionedDataKind, key: str) -> dict:
+    def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[SerializedItemDescriptor]:
         """
-        Returns the object to which the specified key is mapped, or None if no such item exists.
-        The method should not attempt to filter out any items based on their deleted property,
-        nor to cache any items.
+        Returns the item to which the specified key is mapped, or None if no such item exists.
+        Return a deleted item (a "tombstone") as is. Do not filter it out.
 
-        :param kind: The kind of object to get
-        :param key: The key of the object
-        :return: The object to which the specified key is mapped, or None
+        :param kind: The kind of item to get
+        :param key: The key of the item
+        :return: The serialized item, or None
         """
 
     @abstractmethod
-    def get_all_internal(self, callback) -> Mapping[str, dict]:
+    def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, SerializedItemDescriptor]:
         """
-        Returns a dictionary of all associated objects of a given kind. The method should not attempt
-        to filter out any items based on their deleted property, nor to cache any items.
+        Returns all items of the specified kind, keyed by the key that each item is stored under.
+        Include deleted items. Do not filter them out.
 
-        :param kind: The kind of objects to get
-        :return: A dictionary of keys to items
+        :param kind: The kind of items to get
+        :return: A dictionary of keys to serialized items
         """
 
     @abstractmethod
-    def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]):
+    def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, SerializedItemDescriptor]]):
         """
-        Initializes (or re-initializes) the store with the specified set of objects. Any existing entries
-        will be removed. Implementations can assume that this set of objects is up to date-- there is no
-        need to perform individual version comparisons between the existing objects and the supplied
-        data.
+        Initializes (or re-initializes) the store with the specified set of items. Remove all
+        existing items. The items are up to date, so do not compare versions.
 
-        :param all_data: A dictionary of data kinds to item collections
+        The SDK gives the kinds and the items in an order that keeps references valid. If the store
+        cannot write all items in one transaction, write them in the given order.
+
+        :param all_data: A dictionary of data kinds to dictionaries of keys to serialized items
         """
 
     @abstractmethod
-    def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict:
+    def upsert_internal(self, kind: VersionedDataKind, key: str, item: SerializedItemDescriptor) -> bool:
         """
-        Updates or inserts the object associated with the specified key. If an item with the same key
-        already exists, it should update it only if the new item's version property is greater than
-        the old one. It should return the final state of the item, i.e. if the update succeeded then
-        it returns the item that was passed in, and if the update failed due to the version check
-        then it returns the item that is currently in the data store (this ensures that
-        ``CachingStoreWrapper`` will update the cache correctly).
+        Updates or inserts the item associated with the specified key. If an item with the same key
+        already exists, update it only if the version of the new item is greater than the version
+        of the existing item. A deleted item has a version, and the same rule applies to it.
 
-        ``CachingStoreWrapper`` compares the returned object to the object that it passed in. If they
-        are the same object, it reports the write as applied. Thus, on success, return the same
-        object that was passed in, not a copy.
+        To find the version of an existing item that the core stores only as a serialized string,
+        use :func:`ldclient.versioned_data_kind.VersionedDataKind.deserialize`.
 
-        :param kind: The kind of object to update
-        :param item: The object to update or insert
-        :return: The state of the object after the update
+        :param kind: The kind of item to update
+        :param key: The key of the item
+        :param item: The serialized item to write
+        :return: True if the core wrote the item, or False if the version check rejected it
         """
 
     @abstractmethod
     def initialized_internal(self) -> bool:
         """
         Returns true if this store has been initialized. In a shared data store, it should be able to
-        detect this even if initInternal was called in a different process, i.e. the test should be
+        detect this even if init_internal was called in a different process, i.e. the test should be
         based on looking at what is in the data store. The method does not need to worry about caching
         this value; ``CachingStoreWrapper`` will only call it when necessary.
         """
@@ -445,7 +470,7 @@ class AsyncFeatureStoreCore(ABC):
     Async equivalent of :class:`FeatureStoreCore`, for use with
     :class:`ldclient.async_feature_store_helpers.AsyncCachingStoreWrapper`. It exposes a simplified
     subset of the functionality of :class:`AsyncFeatureStore`, so a database integration only has to
-    implement the database-specific logic; the wrapper adds caching and encode/decode handling.
+    implement the database-specific logic. The wrapper adds caching and serialization.
 
     .. caution::
         This feature is experimental and should NOT be considered ready for production
@@ -453,58 +478,54 @@ class AsyncFeatureStoreCore(ABC):
         compatibility guarantees. Pin to a specific minor version and review the changelog
         before upgrading.
 
-    All items passed to and returned from these methods are plain JSON-compatible dicts, not decoded
-    model objects. The wrapper handles decoding for its cache and its callers.
+    All items that go into and come out of these methods are :class:`SerializedItemDescriptor`
+    instances, with the same rules as for :class:`FeatureStoreCore`.
     """
 
     @abstractmethod
-    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[dict]:
+    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[SerializedItemDescriptor]:
         """
-        Returns the object to which the specified key is mapped, or None if no such item exists.
-        The method should not attempt to filter out any items based on their deleted property,
-        nor to cache any items.
+        Returns the item to which the specified key is mapped, or None if no such item exists.
+        Return a deleted item (a "tombstone") as is. Do not filter it out.
 
-        :param kind: The kind of object to get
-        :param key: The key of the object
-        :return: The object to which the specified key is mapped, or None
+        :param kind: The kind of item to get
+        :param key: The key of the item
+        :return: The serialized item, or None
         """
         ...
 
     @abstractmethod
-    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, dict]:
+    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, SerializedItemDescriptor]:
         """
-        Returns a dictionary of all associated objects of a given kind. The method should not attempt
-        to filter out any items based on their deleted property, nor to cache any items.
+        Returns all items of the specified kind, keyed by the key that each item is stored under.
+        Include deleted items. Do not filter them out.
 
-        :param kind: The kind of objects to get
-        :return: A dictionary of keys to items
+        :param kind: The kind of items to get
+        :return: A dictionary of keys to serialized items
         """
         ...
 
     @abstractmethod
-    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None:
+    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, SerializedItemDescriptor]]) -> None:
         """
-        Initializes (or re-initializes) the store with the specified set of objects. Any existing
-        entries will be removed. Implementations can assume that this set of objects is up to date--
-        there is no need to perform individual version comparisons between the existing objects and
-        the supplied data.
+        Initializes (or re-initializes) the store with the specified set of items. This is the
+        same as :func:`FeatureStoreCore.init_internal`.
 
-        :param all_data: A dictionary of data kinds to item collections
+        :param all_data: A dictionary of data kinds to dictionaries of keys to serialized items
         """
         ...
 
     @abstractmethod
-    async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict:
+    async def upsert_internal(self, kind: VersionedDataKind, key: str, item: SerializedItemDescriptor) -> bool:
         """
-        Updates or inserts the object associated with the specified key. If an item with the same key
-        already exists, it should update it only if the new item's version property is greater than
-        the old one. It should return the final state of the item, that is, the item that was passed
-        in if the update succeeded, or the item that is currently in the data store if the update
-        failed the version check. This lets the wrapper update its cache correctly.
+        Updates or inserts the item associated with the specified key, if its version is greater
+        than the version of the existing item. This is the same as
+        :func:`FeatureStoreCore.upsert_internal`.
 
-        :param kind: The kind of object to update
-        :param item: The object to update or insert
-        :return: The state of the object after the update
+        :param kind: The kind of item to update
+        :param key: The key of the item
+        :param item: The serialized item to write
+        :return: True if the core wrote the item, or False if the version check rejected it
         """
         ...
 
diff --git a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py
index f6e8b44a..12c120b6 100644
--- a/ldclient/testing/impl/datasystem/test_cache_lifecycle.py
+++ b/ldclient/testing/impl/datasystem/test_cache_lifecycle.py
@@ -30,6 +30,12 @@
     ObjectKind,
     Selector
 )
+from ldclient.testing.serialized_items import (
+    from_descriptor,
+    from_descriptors,
+    to_descriptor,
+    to_descriptors
+)
 from ldclient.versioned_data_kind import FEATURES
 
 
@@ -50,24 +56,25 @@ def __init__(self):
 
     def init_internal(self, all_data):
         self.init_calls += 1
-        self.data = {kind: dict(items) for kind, items in all_data.items()}
+        self.data = from_descriptors(all_data)
         self.inited = True
 
     def get_internal(self, kind, key):
         self.get_calls += 1
-        return self.data.get(kind, {}).get(key)
+        return to_descriptor(self.data.get(kind, {}).get(key))
 
     def get_all_internal(self, kind):
-        return self.data.get(kind, {})
+        return to_descriptors(self.data.get(kind, {}))
 
-    def upsert_internal(self, kind, item):
+    def upsert_internal(self, kind, key, descriptor):
         self.upsert_calls += 1
+        item = from_descriptor(descriptor)
         items = self.data.setdefault(kind, {})
-        existing = items.get(item['key'])
+        existing = items.get(key)
         if existing is None or existing.get('version', 0) < item.get('version', 0):
-            items[item['key']] = item
-            return item
-        return existing
+            items[key] = item
+            return True
+        return False
 
     def initialized_internal(self):
         return self.inited
diff --git a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py
index 4f0552af..f2d2382f 100644
--- a/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py
+++ b/ldclient/testing/impl/datasystem/test_fdv2_async_persistence.py
@@ -23,7 +23,14 @@
     DataStoreMode,
     IntentCode,
     ObjectKind,
-    Selector
+    Selector,
+    SerializedItemDescriptor
+)
+from ldclient.testing.serialized_items import (
+    from_descriptor,
+    from_descriptors,
+    to_descriptor,
+    to_descriptors
 )
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind
 
@@ -274,19 +281,20 @@ def __init__(self):
         self.inited = False
         self.query_count = 0
 
-    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None:
-        self.data = {FEATURES: dict(all_data.get(FEATURES, {})), SEGMENTS: dict(all_data.get(SEGMENTS, {}))}
+    async def init_internal(self, all_data: Mapping[VersionedDataKind, Mapping[str, SerializedItemDescriptor]]) -> None:
+        decoded = from_descriptors(all_data)
+        self.data = {FEATURES: decoded.get(FEATURES, {}), SEGMENTS: decoded.get(SEGMENTS, {})}
         self.inited = True
 
-    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[dict]:
-        return self.data.get(kind, {}).get(key)
+    async def get_internal(self, kind: VersionedDataKind, key: str) -> Optional[SerializedItemDescriptor]:
+        return to_descriptor(self.data.get(kind, {}).get(key))
 
-    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, dict]:
-        return dict(self.data.get(kind, {}))
+    async def get_all_internal(self, kind: VersionedDataKind) -> Mapping[str, SerializedItemDescriptor]:
+        return to_descriptors(self.data.get(kind, {})) or {}
 
-    async def upsert_internal(self, kind: VersionedDataKind, item: dict) -> dict:
-        self.data[kind][item["key"]] = item
-        return item
+    async def upsert_internal(self, kind: VersionedDataKind, key: str, item: SerializedItemDescriptor) -> bool:
+        self.data[kind][key] = from_descriptor(item)
+        return True
 
     async def initialized_internal(self) -> bool:
         self.query_count += 1
diff --git a/ldclient/testing/serialized_items.py b/ldclient/testing/serialized_items.py
new file mode 100644
index 00000000..2cf0ee23
--- /dev/null
+++ b/ldclient/testing/serialized_items.py
@@ -0,0 +1,26 @@
+"""Converts between plain item dicts and serialized items, for test store cores."""
+
+import json
+from typing import Mapping, Optional
+
+from ldclient.interfaces import SerializedItemDescriptor
+
+
+def to_descriptor(item: Optional[dict]) -> Optional[SerializedItemDescriptor]:
+    if item is None:
+        return None
+    return SerializedItemDescriptor(version=item.get('version', 0), deleted=item.get('deleted') is True, serialized_item=json.dumps(item))
+
+
+def to_descriptors(items: Optional[Mapping[str, dict]]) -> Optional[Mapping[str, SerializedItemDescriptor]]:
+    if items is None:
+        return None
+    return {key: to_descriptor(item) for key, item in items.items()}  # type: ignore[misc]
+
+
+def from_descriptor(descriptor: SerializedItemDescriptor) -> dict:
+    return json.loads(descriptor.serialized_item)
+
+
+def from_descriptors(all_data) -> dict:
+    return {kind: {key: from_descriptor(d) for key, d in items.items()} for kind, items in all_data.items()}
diff --git a/ldclient/testing/test_async_feature_store_helpers.py b/ldclient/testing/test_async_feature_store_helpers.py
index c6906c7c..654467d4 100644
--- a/ldclient/testing/test_async_feature_store_helpers.py
+++ b/ldclient/testing/test_async_feature_store_helpers.py
@@ -1,10 +1,17 @@
 import asyncio
+import json
 
 import pytest
 
 from ldclient.async_feature_store_helpers import AsyncCachingStoreWrapper
 from ldclient.feature_store import CacheConfig
-from ldclient.interfaces import AsyncFeatureStoreCore
+from ldclient.interfaces import AsyncFeatureStoreCore, SerializedItemDescriptor
+from ldclient.testing.serialized_items import (
+    from_descriptor,
+    from_descriptors,
+    to_descriptor,
+    to_descriptors
+)
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind
 
 # These tests exercise the caching-wrapper logic only, using an in-memory mock core, so they run
@@ -36,22 +43,20 @@ async def close(self):
 
     async def init_internal(self, all_data):
         self._maybe_throw()
-        self.data = {}
-        for kind, items in all_data.items():
-            self.data[kind] = items.copy()
+        self.data = from_descriptors(all_data)
 
     async def get_internal(self, kind, key):
         self._maybe_throw()
         items = self.data.get(kind)
-        return None if items is None else items.get(key)
+        return None if items is None else to_descriptor(items.get(key))
 
     async def get_all_internal(self, kind):
         self._maybe_throw()
-        return self.data.get(kind)
+        return to_descriptors(self.data.get(kind))
 
-    async def upsert_internal(self, kind, item):
+    async def upsert_internal(self, kind, key, descriptor):
         self._maybe_throw()
-        key = item.get('key')
+        item = from_descriptor(descriptor)
         items = self.data.get(kind)
         if items is None:
             items = {}
@@ -59,8 +64,8 @@ async def upsert_internal(self, kind, item):
         old_item = items.get(key)
         if old_item is None or old_item.get('version') < item.get('version'):
             items[key] = item
-            return item
-        return old_item
+            return True
+        return False
 
     async def initialized_internal(self):
         self._maybe_throw()
@@ -367,3 +372,56 @@ def describe_configuration(self, config):
     async def test_describe_configuration_defaults_to_custom(self):
         wrapper = make_wrapper(MockAsyncCore(), False)
         assert wrapper.describe_configuration(None) == "custom"
+
+
+class DescriptorAsyncCore(MockAsyncCore):
+    """Returns fixed serialized items and records the serialized items that it receives."""
+
+    def __init__(self, returned=None):
+        super().__init__()
+        self.returned = returned
+        self.received = []
+
+    async def get_internal(self, kind, key):
+        return self.returned
+
+    async def get_all_internal(self, kind):
+        return {'item': self.returned}
+
+    async def upsert_internal(self, kind, key, descriptor):
+        self.received.append((key, descriptor))
+        return True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cached", [False, True])
+async def test_a_version_from_the_core_replaces_the_serialized_version(cached):
+    core = DescriptorAsyncCore(SerializedItemDescriptor(version=5, deleted=False, serialized_item=json.dumps({"key": "item", "version": 3})))
+    wrapper = make_wrapper(core, cached)
+
+    assert await wrapper.get(THINGS, "item") == {"key": "item", "version": 5}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cached", [False, True])
+async def test_an_item_that_the_core_marks_deleted_is_not_returned(cached):
+    core = DescriptorAsyncCore(SerializedItemDescriptor(version=4, deleted=True, serialized_item="placeholder"))
+    wrapper = make_wrapper(core, cached)
+
+    assert await wrapper.get(THINGS, "item") is None
+    assert await wrapper.all(THINGS) == {}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cached", [False, True])
+async def test_delete_gives_the_core_a_serialized_placeholder(cached):
+    core = DescriptorAsyncCore()
+    wrapper = make_wrapper(core, cached)
+
+    assert await wrapper.delete(THINGS, "item", 7) is True
+
+    [(key, descriptor)] = core.received
+    assert key == "item"
+    assert descriptor.version == 7
+    assert descriptor.deleted is True
+    assert json.loads(descriptor.serialized_item) == {"key": "item", "version": 7, "deleted": True}
diff --git a/ldclient/testing/test_feature_store_helpers.py b/ldclient/testing/test_feature_store_helpers.py
index 610110e6..1f379a2b 100644
--- a/ldclient/testing/test_feature_store_helpers.py
+++ b/ldclient/testing/test_feature_store_helpers.py
@@ -1,10 +1,17 @@
+import json
 from time import sleep
 
 import pytest
 
 from ldclient.feature_store import CacheConfig
 from ldclient.feature_store_helpers import CachingStoreWrapper
-from ldclient.interfaces import FeatureStoreCore
+from ldclient.interfaces import FeatureStoreCore, SerializedItemDescriptor
+from ldclient.testing.serialized_items import (
+    from_descriptor,
+    from_descriptors,
+    to_descriptor,
+    to_descriptors
+)
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind
 
 THINGS = VersionedDataKind(namespace="things", request_api_path="", stream_api_path="")
@@ -32,22 +39,20 @@ def close(self):
 
     def init_internal(self, all_data):
         self._maybe_throw()
-        self.data = {}
-        for kind, items in all_data.items():
-            self.data[kind] = items.copy()
+        self.data = from_descriptors(all_data)
 
     def get_internal(self, kind, key):
         self._maybe_throw()
         items = self.data.get(kind)
-        return None if items is None else items.get(key)
+        return None if items is None else to_descriptor(items.get(key))
 
     def get_all_internal(self, kind):
         self._maybe_throw()
-        return self.data.get(kind)
+        return to_descriptors(self.data.get(kind))
 
-    def upsert_internal(self, kind, item):
+    def upsert_internal(self, kind, key, descriptor):
         self._maybe_throw()
-        key = item.get('key')
+        item = from_descriptor(descriptor)
         items = self.data.get(kind)
         if items is None:
             items = {}
@@ -55,8 +60,8 @@ def upsert_internal(self, kind, item):
         old_item = items.get(key)
         if old_item is None or old_item.get('version') < item.get('version'):
             items[key] = item
-            return item
-        return old_item
+            return True
+        return False
 
     def initialized_internal(self):
         self._maybe_throw()
@@ -388,3 +393,72 @@ def test_cached_initialized_can_cache_false_result(self):
         # From this point on it should remain true and the method should not be called
         assert wrapper.initialized is True
         assert core.inited_query_count == 2
+
+
+class DescriptorCore(MockCore):
+    """Returns fixed serialized items and records the serialized items that it receives."""
+
+    def __init__(self, returned=None):
+        super().__init__()
+        self.returned = returned
+        self.received = []
+
+    def get_internal(self, kind, key):
+        return self.returned
+
+    def get_all_internal(self, kind):
+        return {'item': self.returned}
+
+    def upsert_internal(self, kind, key, descriptor):
+        self.received.append((key, descriptor))
+        return True
+
+
+@pytest.mark.parametrize("cached", [False, True])
+def test_a_version_from_the_core_replaces_the_serialized_version(cached):
+    core = DescriptorCore(SerializedItemDescriptor(version=5, deleted=False, serialized_item=json.dumps({"key": "item", "version": 3})))
+    wrapper = make_wrapper(core, cached)
+
+    assert wrapper.get(THINGS, "item") == {"key": "item", "version": 5}
+
+
+@pytest.mark.parametrize("cached", [False, True])
+def test_a_version_of_zero_from_the_core_keeps_the_serialized_version(cached):
+    core = DescriptorCore(SerializedItemDescriptor(version=0, deleted=False, serialized_item=json.dumps({"key": "item", "version": 3})))
+    wrapper = make_wrapper(core, cached)
+
+    assert wrapper.get(THINGS, "item") == {"key": "item", "version": 3}
+
+
+@pytest.mark.parametrize("cached", [False, True])
+def test_an_item_that_the_core_marks_deleted_is_not_returned(cached):
+    core = DescriptorCore(SerializedItemDescriptor(version=4, deleted=True, serialized_item="placeholder"))
+    wrapper = make_wrapper(core, cached)
+
+    assert wrapper.get(THINGS, "item") is None
+    assert wrapper.all(THINGS) == {}
+
+
+@pytest.mark.parametrize("cached", [False, True])
+def test_upsert_gives_the_core_the_key_version_and_serialized_item(cached):
+    core = DescriptorCore()
+    wrapper = make_wrapper(core, cached)
+    item = {"key": "item", "version": 2}
+
+    assert wrapper.upsert(THINGS, item) is True
+
+    assert core.received == [("item", SerializedItemDescriptor(version=2, deleted=False, serialized_item=json.dumps(item)))]
+
+
+@pytest.mark.parametrize("cached", [False, True])
+def test_delete_gives_the_core_a_serialized_placeholder(cached):
+    core = DescriptorCore()
+    wrapper = make_wrapper(core, cached)
+
+    assert wrapper.delete(THINGS, "item", 7) is True
+
+    [(key, descriptor)] = core.received
+    assert key == "item"
+    assert descriptor.version == 7
+    assert descriptor.deleted is True
+    assert json.loads(descriptor.serialized_item) == {"key": "item", "version": 7, "deleted": True}
diff --git a/ldclient/versioned_data_kind.py b/ldclient/versioned_data_kind.py
index 3f4f91fc..d0b19ad5 100644
--- a/ldclient/versioned_data_kind.py
+++ b/ldclient/versioned_data_kind.py
@@ -9,6 +9,7 @@
 for features or segments.
 """
 
+import json
 from collections import namedtuple
 from typing import Any, Callable, Iterable, Optional
 
@@ -44,6 +45,27 @@ def decode(self, data: Any) -> Any:
     def encode(self, item: Any) -> dict:
         return item.to_json_dict() if isinstance(item, ModelEntity) else item
 
+    def serialize(self, item: Any) -> str:
+        """
+        Serializes an item of this kind to a JSON string.
+
+        :param item: The item, as a model object or as a dict
+        :return: The JSON string
+        """
+        return json.dumps(self.encode(item))
+
+    def deserialize(self, serialized: str) -> Any:
+        """
+        Deserializes an item of this kind from a JSON string.
+
+        A persistent store core can use this method to find the version of an item that it stores
+        only as a serialized string: ``kind.deserialize(serialized)['version']``.
+
+        :param serialized: The JSON string
+        :return: The item
+        """
+        return self.decode(json.loads(serialized))
+
 
 class VersionedDataKindWithOrdering(VersionedDataKind):
     def __init__(

From 71d0450de0e2cffb452d4305de2e6c2f4690c9f9 Mon Sep 17 00:00:00 2001
From: Matthew Keeler <mkeeler@launchdarkly.com>
Date: Thu, 5 Nov 2026 09:00:20 -0500
Subject: [PATCH 2/3] test: Expect the newer item after a Redis upsert loses a
 race

---
 .../testing/integrations/test_async_redis_feature_store.py   | 5 ++++-
 ldclient/testing/integrations/test_redis.py                  | 5 ++++-
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/ldclient/testing/integrations/test_async_redis_feature_store.py b/ldclient/testing/integrations/test_async_redis_feature_store.py
index 9aa292b2..13e9f02c 100644
--- a/ldclient/testing/integrations/test_async_redis_feature_store.py
+++ b/ldclient/testing/integrations/test_async_redis_feature_store.py
@@ -134,7 +134,10 @@ def hook(base_key, key):
 
         await store.upsert(FEATURES, {'key': 'flagkey', 'version': 1})
         result = await store.get(FEATURES, 'flagkey')
-        assert result['version'] == 2
+        # The hook writes version 2 during the first attempt, which makes the transaction fail. During
+        # the retry, the core reads version 2 and rejects the write, and the hook then writes version 3.
+        # The wrapper reads the item again after a rejected write, so it returns version 3.
+        assert result['version'] == 3
     finally:
         await store.close()
 
diff --git a/ldclient/testing/integrations/test_redis.py b/ldclient/testing/integrations/test_redis.py
index 88074c4d..474fc5f6 100644
--- a/ldclient/testing/integrations/test_redis.py
+++ b/ldclient/testing/integrations/test_redis.py
@@ -103,7 +103,10 @@ def hook(base_key, key):
 
         store.upsert(FEATURES, feature)
         result = store.get(FEATURES, 'flagkey', lambda x: x)
-        assert result['version'] == 2
+        # The hook writes version 2 during the first attempt, which makes the transaction fail. During
+        # the retry, the core reads version 2 and rejects the write, and the hook then writes version 3.
+        # The wrapper reads the item again after a rejected write, so it returns version 3.
+        assert result['version'] == 3
 
     def test_upsert_race_condition_against_external_client_with_lower_version(self):
         other_client = RedisTestHelper.make_client()

From 21cd2b64c5080bdec7c34b890f1f59114b37ae2e Mon Sep 17 00:00:00 2001
From: Matthew Keeler <mkeeler@launchdarkly.com>
Date: Thu, 5 Nov 2026 09:02:50 -0500
Subject: [PATCH 3/3] fix: Drop the cached item list when a store rejects an
 older write

---
 ldclient/async_feature_store_helpers.py       |  2 +-
 ldclient/feature_store_helpers.py             | 11 ++++++++++-
 .../test_async_feature_store_helpers.py       | 19 +++++++++++++++++++
 .../testing/test_feature_store_helpers.py     | 18 ++++++++++++++++++
 4 files changed, 48 insertions(+), 2 deletions(-)

diff --git a/ldclient/async_feature_store_helpers.py b/ldclient/async_feature_store_helpers.py
index 2a91e52b..6780ad40 100644
--- a/ldclient/async_feature_store_helpers.py
+++ b/ldclient/async_feature_store_helpers.py
@@ -91,7 +91,7 @@ async def upsert(self, kind: VersionedDataKind, item: dict) -> bool:
             self._cache_put_upsert(kind, key, encoded_item)
         elif self._caching():
             # The store has a newer item. Read it, so that the cache does not keep an older one.
-            self._cache_put_item(kind, key, await self._core.get_internal(kind, key))
+            self._cache_put_rejected_upsert(kind, key, await self._core.get_internal(kind, key))
         return updated
 
     @property
diff --git a/ldclient/feature_store_helpers.py b/ldclient/feature_store_helpers.py
index 3b616fe5..a8d7e3e5 100644
--- a/ldclient/feature_store_helpers.py
+++ b/ldclient/feature_store_helpers.py
@@ -190,6 +190,15 @@ def _cache_put_upsert(self, kind, key, encoded_item):
         self._cache[self._item_cache_key(kind, key)] = [kind.decode(encoded_item)]
         self._cache.pop(self._all_cache_key(kind), None)
 
+    def _cache_put_rejected_upsert(self, kind, key, descriptor):
+        """Updates the cache after the core rejected a write, because the store has a newer item.
+
+        Caches the newer item from the core. Also drops the all-items entry, because that entry can
+        also have an older item.
+        """
+        self._cache_put_item(kind, key, descriptor)
+        self._cache.pop(self._all_cache_key(kind), None)
+
     def _caching(self) -> bool:
         return self._cache is not _NOOP_CACHE
 
@@ -273,7 +282,7 @@ def upsert(self, kind, encoded_item) -> bool:
             self._cache_put_upsert(kind, key, encoded_item)
         elif self._caching():
             # The store has a newer item. Read it, so that the cache does not keep an older one.
-            self._cache_put_item(kind, key, self._core.get_internal(kind, key))
+            self._cache_put_rejected_upsert(kind, key, self._core.get_internal(kind, key))
         return updated
 
     @property
diff --git a/ldclient/testing/test_async_feature_store_helpers.py b/ldclient/testing/test_async_feature_store_helpers.py
index 654467d4..01fb0fd3 100644
--- a/ldclient/testing/test_async_feature_store_helpers.py
+++ b/ldclient/testing/test_async_feature_store_helpers.py
@@ -302,6 +302,25 @@ async def test_cached_upsert_unsuccessful(self):
         core.force_set(THINGS, itemv3)  # bypasses cache so we can verify itemv2 is in the cache
         assert await wrapper.get(THINGS, key) == itemv2
 
+    @pytest.mark.asyncio
+    async def test_cached_upsert_unsuccessful_drops_the_all_items_cache(self):
+        # Another process writes a newer item after the wrapper caches all items. The wrapper then
+        # tries to write an older item, and the core rejects it. All items must come from the store
+        # again, so that the wrapper does not keep the older item.
+        core = MockAsyncCore()
+        wrapper = make_wrapper(core, True)
+        key = "flag"
+        itemv1 = {"key": key, "version": 1}
+        itemv2 = {"key": key, "version": 2}
+        itemv3 = {"key": key, "version": 3}
+
+        await wrapper.init({THINGS: {key: itemv1}})
+        assert await wrapper.all(THINGS) == {key: itemv1}
+
+        core.force_set(THINGS, itemv3)
+        assert await wrapper.upsert(THINGS, itemv2) is False
+        assert await wrapper.all(THINGS) == {key: itemv3}
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("cached", [False, True])
     async def test_upsert_can_throw_exception(self, cached):
diff --git a/ldclient/testing/test_feature_store_helpers.py b/ldclient/testing/test_feature_store_helpers.py
index 1f379a2b..3a561eb6 100644
--- a/ldclient/testing/test_feature_store_helpers.py
+++ b/ldclient/testing/test_feature_store_helpers.py
@@ -313,6 +313,24 @@ def test_cached_upsert_unsuccessful(self):
         core.force_set(THINGS, itemv3)  # bypasses cache so we can verify that itemv2 is in the cache
         assert wrapper.get(THINGS, key) == itemv2
 
+    def test_cached_upsert_unsuccessful_drops_the_all_items_cache(self):
+        # Another process writes a newer item after the wrapper caches all items. The wrapper then
+        # tries to write an older item, and the core rejects it. All items must come from the store
+        # again, so that the wrapper does not keep the older item.
+        core = MockCore()
+        wrapper = make_wrapper(core, True)
+        key = "flag"
+        itemv1 = {"key": key, "version": 1}
+        itemv2 = {"key": key, "version": 2}
+        itemv3 = {"key": key, "version": 3}
+
+        wrapper.init({THINGS: {key: itemv1}})
+        assert wrapper.all(THINGS) == {key: itemv1}
+
+        core.force_set(THINGS, itemv3)
+        assert wrapper.upsert(THINGS, itemv2) is False
+        assert wrapper.all(THINGS) == {key: itemv3}
+
     @pytest.mark.parametrize("cached", [False, True])
     def test_upsert_can_throw_exception(self, cached):
         core = MockCore()