feat!: Pass serialized items to persistent store cores

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_rejected_upsert(kind, key, await self._core.get_internal(kind, key))
+        return updated
 
     @property
     def initialized(self) -> bool:

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,22 @@ 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.
+    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)
 
-        Caches the item the core returned and drops the now-stale all-items entry. Returns the
-        decoded item.
+    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.
         """
-        new_decoded_item = kind.decode(new_state)
-        self._cache[self._item_cache_key(kind, new_decoded_item.get('key'))] = [new_decoded_item]
+        self._cache_put_item(kind, key, descriptor)
         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 +248,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 +257,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 +276,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_rejected_upsert(kind, key, self._core.get_internal(kind, key))
+        return updated
 
     @property
     def initialized(self) -> bool:

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'))