@@ -86,10 +86,16 @@ def get_internal(self, kind, key):
 
     def get_all_internal(self, kind):
         items_out = {}
+        # Use the 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.
+        item_key_prefix = self._kind_key(kind) + '/'
         index, results = self._client.kv.get(self._kind_key(kind), recurse=True)
         for result in results:
+            db_key = result['Key']
+            if not db_key.startswith(item_key_prefix):
+                continue
             item = json.loads(result['Value'].decode('utf-8'))
-            items_out[item['key']] = item
+            items_out[db_key[len(item_key_prefix):]] = item
         return items_out
 
     def upsert_internal(self, kind, new_item):
 
@@ -99,7 +99,9 @@ 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)
-                items_out[item_out['key']] = item_out
+                # 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
         return items_out
 
     def upsert_internal(self, kind, item):
 
@@ -104,11 +104,13 @@ def __init__(self, data: dict):
         # be absent even if they are really required in the schema. That's for backward compatibility
         # with test logic that constructed incomplete JSON, and also with the file data source which
         # previously allowed users to get away with leaving out a lot of properties in the JSON.
-        self._key = req_str(data, 'key')
         self._version = req_int(data, 'version')
         self._deleted = opt_bool(data, 'deleted')
         if self._deleted:
+            # Tombstones are not guaranteed to have a key.
+            self._key = opt_str(data, 'key') or ''
             return
+        self._key = req_str(data, 'key')
         self._variations = opt_list(data, 'variations')
         self._on = opt_bool(data, 'on')
         self._off_variation = opt_int(data, 'offVariation')
 
@@ -73,11 +73,13 @@ def __init__(self, data: dict):
         # be absent even if they are really required in the schema. That's for backward compatibility
         # with test logic that constructed incomplete JSON, and also with the file data source which
         # previously allowed users to get away with leaving out a lot of properties in the JSON.
-        self._key = req_str(data, 'key')
         self._version = req_int(data, 'version')
         self._deleted = opt_bool(data, 'deleted')
         if self._deleted:
+            # Tombstones are not guaranteed to have a key.
+            self._key = opt_str(data, 'key') or ''
             return
+        self._key = req_str(data, 'key')
         self._included = set(opt_str_list(data, 'included'))
         self._excluded = set(opt_str_list(data, 'excluded'))
         self._included_contexts = list(SegmentTarget(item) for item in opt_dict_list(data, 'includedContexts'))
 
@@ -5,6 +5,7 @@
 
 from ldclient.impl.model import *
 from ldclient.testing.builders import *
+from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
 
 def test_flag_targets_are_stored_as_sets():
@@ -41,3 +42,38 @@ def test_clause_values_preprocessed_with_time_operator(op):
     flag = make_boolean_flag_with_clauses(make_clause(None, "attr", op, 1000, "1970-01-01T00:00:02Z", True))
     assert flag.rules[0].clauses[0]._values == [1000, "1970-01-01T00:00:02Z", True]
     assert list(x.as_time for x in flag.rules[0].clauses[0]._values_preprocessed) == [1000, 2000, None]
+
+
+@pytest.mark.parametrize('kind', [FEATURES, SEGMENTS])
+def test_tombstone_without_key_can_be_decoded(kind):
+    # Other LaunchDarkly SDKs write deleted items to a persistent store with only the version,
+    # so we must be able to read them back.
+    item = kind.decode({"version": 5, "deleted": True})
+    assert item.version == 5
+    assert item.deleted is True
+    assert item.key == ''
+    # The original data must round-trip unchanged, because the store re-serializes it.
+    assert item.to_json_dict() == {"version": 5, "deleted": True}
+
+
+@pytest.mark.parametrize('kind', [FEATURES, SEGMENTS])
+def test_tombstone_with_placeholder_key_can_be_decoded(kind):
+    # The Go SDK and the Relay Proxy write deleted items with a placeholder key.
+    item = kind.decode({"key": "$deleted", "version": 5, "deleted": True})
+    assert item.version == 5
+    assert item.deleted is True
+    assert item.key == '$deleted'
+
+
+@pytest.mark.parametrize('kind', [FEATURES, SEGMENTS])
+def test_tombstone_still_requires_version(kind):
+    with pytest.raises(ValueError):
+        kind.decode({"deleted": True})
+
+
+@pytest.mark.parametrize('kind', [FEATURES, SEGMENTS])
+def test_item_that_is_not_deleted_still_requires_key(kind):
+    with pytest.raises(ValueError):
+        kind.decode({"version": 5})
+    with pytest.raises(ValueError):
+        kind.decode({"version": 5, "deleted": False})
@@ -40,6 +40,18 @@ def clear_data(self, prefix: str):
         """
         pass
 
+    @abstractmethod
+    def write_raw_item(self, prefix: str, kind, key: str, item: dict):
+        """
+        Override this method to write an item straight to the database, with no help from the
+        store. This lets a test set up data in a shape that the store itself does not write.
+        :param prefix: the prefix parameter for the store constructor - may be None or empty to use the default
+        :param kind: the kind of data, such as FEATURES
+        :param key: the key to store the item under
+        :param item: the item, to be stored as JSON
+        """
+        pass
+
     def create_feature_store(self) -> FeatureStore:
         return self.create_persistent_feature_store(self.prefix, self.caching)
 
@@ -63,6 +75,16 @@ def tester(self, request):
     def clear_data_before_each(self, tester):
         tester.clear_data(tester.prefix)
 
+    def test_all_reads_tombstone_with_no_key(self, tester):
+        # Other LaunchDarkly SDKs write a deleted item with only a version, and no key of its
+        # own. The store must read these back with the key that the item is stored under.
+        with self.inited_store(tester) as store:
+            tester.write_raw_item(tester.prefix, FEATURES, 'deleted-flag', {'version': 5, 'deleted': True})
+
+            items = store.all(FEATURES, lambda x: x)
+            assert items == {'foo': self.make_feature('foo', 10), 'bar': self.make_feature('bar', 10)}
+            assert store.get(FEATURES, 'deleted-flag', lambda x: x) is None
+
     def test_stores_with_different_prefixes_are_independent(self):
         # This verifies that init(), get(), all(), and upsert() are all correctly using the specified key prefix.
         # The delete() method isn't tested separately because it's implemented as a variant of upsert().
 
@@ -1,3 +1,5 @@
+import json
+
 import pytest
 
 from ldclient.integrations import Consul
@@ -39,6 +41,11 @@ def clear_data(self, prefix):
         for key in keys or []:
             client.kv.delete(key)
 
+    def write_raw_item(self, prefix, kind, key, item):
+        client = consul.Consul()
+        db_key = "%s/%s/%s" % (prefix or Consul.DEFAULT_PREFIX, kind.namespace, key)
+        client.kv.put(db_key, json.dumps(item))
+
 
 class TestConsulFeatureStore(PersistentFeatureStoreTestBase):
     @property
 
@@ -1,3 +1,4 @@
+import json
 import time
 
 from ldclient.impl.integrations.dynamodb.dynamodb_big_segment_store import (
@@ -106,6 +107,19 @@ def create_persistent_feature_store(self, prefix, caching) -> FeatureStore:
     def clear_data(self, prefix):
         DynamoDBTestHelper.clear_data_for_prefix(prefix)
 
+    def write_raw_item(self, prefix, kind, key, item):
+        client = DynamoDBTestHelper.make_client()
+        namespace = (prefix + ":" if prefix else "") + kind.namespace
+        client.put_item(
+            TableName=DynamoDBTestHelper.table_name,
+            Item={
+                _DynamoDBFeatureStoreCore.PARTITION_KEY: {'S': namespace},
+                _DynamoDBFeatureStoreCore.SORT_KEY: {'S': key},
+                _DynamoDBFeatureStoreCore.VERSION_ATTRIBUTE: {'N': str(item['version'])},
+                _DynamoDBFeatureStoreCore.ITEM_JSON_ATTRIBUTE: {'S': json.dumps(item)},
+            },
+        )
+
 
 class DynamoDBBigSegmentTester(BigSegmentStoreTester):
     def __init__(self):
 
@@ -54,6 +54,11 @@ def create_persistent_feature_store(self, prefix, caching) -> FeatureStore:
     def clear_data(self, prefix):
         RedisTestHelper.clear_data_for_prefix(prefix or Redis.DEFAULT_PREFIX)
 
+    def write_raw_item(self, prefix, kind, key, item):
+        r = RedisTestHelper.make_client()
+        items_key = "%s:%s" % (prefix or Redis.DEFAULT_PREFIX, kind.namespace)
+        r.hset(items_key, key, json.dumps(item))
+
 
 class RedisBigSegmentStoreTester(BigSegmentStoreTester):
     def create_big_segment_store(self, prefix) -> BigSegmentStore:
 
@@ -4,7 +4,7 @@
 
 from ldclient.async_feature_store_helpers import AsyncCachingStoreWrapper
 from ldclient.feature_store import CacheConfig
-from ldclient.versioned_data_kind import VersionedDataKind
+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
 # without a Redis instance. They mirror ldclient.testing.test_feature_store_helpers for the sync
@@ -205,6 +205,21 @@ async def test_get_all_removes_deleted_items(self, cached):
         core.force_set(THINGS, item2)
         assert await wrapper.all(THINGS) == {item1["key"]: item1}
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("kind", [FEATURES, SEGMENTS])
+    @pytest.mark.parametrize("cached", [False, True])
+    async def test_get_all_tolerates_tombstone_with_no_key(self, cached, kind):
+        # Other LaunchDarkly SDKs write deleted items to a persistent store with only the
+        # version. The store knows the key, because it is the key the item is stored under.
+        core = MockAsyncCore()
+        wrapper = make_wrapper(core, cached)
+        live_item = {"key": "item1", "version": 1}
+        tombstone = {"version": 2, "deleted": True}
+        core.data[kind] = {"item1": live_item, "item2": tombstone}
+
+        assert await wrapper.all(kind) == {"item1": kind.decode(live_item)}
+        assert await wrapper.get(kind, "item2") is None
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("cached", [False, True])
     async def test_get_all_changes_None_to_empty_dict(self, cached):