Cache verified external payload bytes

Author: durable-workflow-ops

README.md

@@ -205,7 +205,7 @@ failures surfaced as workflow failure commands.
 - **Polyglot**: Works alongside PHP workers on the same task queue
 - **HTTP/JSON protocol**: No gRPC, no protobuf dependencies
 - **Codec envelopes**: Avro payloads by default, with JSON decode compatibility for existing history
-- **External payload references**: opt-in reference envelopes and a local filesystem driver for large-payload offload experiments
+- **External payload references**: opt-in reference envelopes, a local filesystem driver, and a bounded verified-byte cache for large-payload offload experiments
 - **Payload-size warnings**: Structured warnings before oversized workflow, activity, schedule, signal, update, query, or search-attribute payloads reach the server
 - **Workflow definition guard**: Worker registration refuses same-id hot reloads when a workflow class definition changed
 - **Deterministic workflow helpers**: `ctx.now()`, `ctx.random()`, `ctx.uuid4()`, and `ctx.uuid7()` replay from workflow state

src/durable_workflow/__init__.py

@@ -56,6 +56,7 @@
 )
 from .external_storage import (
     EXTERNAL_PAYLOAD_REFERENCE_SCHEMA,
+    ExternalPayloadCache,
     ExternalPayloadIntegrityError,
     ExternalPayloadReference,
     ExternalStorageDriver,
@@ -171,6 +172,7 @@
     "workflow",
     "DurableWorkflowError",
     "EXTERNAL_PAYLOAD_REFERENCE_SCHEMA",
+    "ExternalPayloadCache",
     "ExternalPayloadIntegrityError",
     "ExternalPayloadReference",
     "ExternalStorageDriver",

src/durable_workflow/external_storage.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import hashlib
+from collections import OrderedDict
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Protocol
@@ -76,6 +77,63 @@ def from_dict(cls, data: object) -> ExternalPayloadReference:
         return cls(uri=uri, sha256=sha256, size_bytes=size_bytes, codec=codec, schema=schema)
 
 
+class ExternalPayloadCache:
+    """Bounded cache for verified external payload bytes during replay.
+
+    Cache entries are keyed by the complete reference identity. Bytes are
+    inserted only by :func:`fetch_external_payload` after size and sha256
+    verification has succeeded, so cache hits preserve the same integrity
+    contract as a fresh driver fetch.
+    """
+
+    def __init__(self, *, max_entries: int = 128, max_bytes: int = 16 * 1024 * 1024) -> None:
+        if max_entries < 1:
+            raise ValueError("external payload cache max_entries must be at least 1")
+        if max_bytes < 1:
+            raise ValueError("external payload cache max_bytes must be at least 1")
+        self.max_entries = max_entries
+        self.max_bytes = max_bytes
+        self.current_bytes = 0
+        self._entries: OrderedDict[tuple[str, str, int, str], bytes] = OrderedDict()
+
+    def get(self, reference: ExternalPayloadReference) -> bytes | None:
+        key = self._key(reference)
+        data = self._entries.get(key)
+        if data is None:
+            return None
+        self._entries.move_to_end(key)
+        return data
+
+    def put(self, reference: ExternalPayloadReference, data: bytes) -> None:
+        if len(data) > self.max_bytes:
+            return
+
+        key = self._key(reference)
+        existing = self._entries.pop(key, None)
+        if existing is not None:
+            self.current_bytes -= len(existing)
+
+        self._entries[key] = data
+        self.current_bytes += len(data)
+        self._evict()
+
+    def clear(self) -> None:
+        self._entries.clear()
+        self.current_bytes = 0
+
+    def __len__(self) -> int:
+        return len(self._entries)
+
+    @staticmethod
+    def _key(reference: ExternalPayloadReference) -> tuple[str, str, int, str]:
+        return (reference.uri, reference.sha256, reference.size_bytes, reference.codec)
+
+    def _evict(self) -> None:
+        while len(self._entries) > self.max_entries or self.current_bytes > self.max_bytes:
+            _, data = self._entries.popitem(last=False)
+            self.current_bytes -= len(data)
+
+
 class LocalFilesystemExternalStorage:
     """Dependency-free external storage driver for development and tests."""
 
@@ -136,15 +194,24 @@ def store_external_payload(
 def fetch_external_payload(
     driver: ExternalStorageDriver,
     reference: ExternalPayloadReference,
+    *,
+    cache: ExternalPayloadCache | None = None,
 ) -> bytes:
     """Fetch payload bytes and verify size/hash before replay or decode."""
+    if cache is not None:
+        cached = cache.get(reference)
+        if cached is not None:
+            return cached
+
     data = driver.get(reference.uri)
     if len(data) != reference.size_bytes:
         raise ExternalPayloadIntegrityError("external payload size does not match its reference")
 
     actual_sha256 = hashlib.sha256(data).hexdigest()
     if actual_sha256 != reference.sha256:
         raise ExternalPayloadIntegrityError("external payload hash does not match its reference")
+    if cache is not None:
+        cache.put(reference, data)
     return data
 
 

src/durable_workflow/serializer.py

@@ -30,6 +30,7 @@
 
 from . import _avro
 from .external_storage import (
+    ExternalPayloadCache,
     ExternalPayloadReference,
     ExternalStorageDriver,
     fetch_external_payload,
@@ -423,6 +424,7 @@ def decode_envelope(
     codec: str | None = None,
     *,
     external_storage: ExternalStorageDriver | None = None,
+    external_storage_cache: ExternalPayloadCache | None = None,
 ) -> Any:
     """Decode a value that may be a ``{codec, blob}`` envelope or a raw blob.
 
@@ -440,7 +442,7 @@ def decode_envelope(
         envelope_codec = value.get("codec")
         if envelope_codec is not None and envelope_codec != reference.codec:
             raise ValueError("external payload reference codec does not match envelope codec")
-        blob = fetch_external_payload(external_storage, reference).decode("utf-8")
+        blob = fetch_external_payload(external_storage, reference, cache=external_storage_cache).decode("utf-8")
         return decode(blob, codec=reference.codec)
     return decode(value, codec=codec)
 
@@ -450,13 +452,19 @@ def decode_envelopes(
     codec: str | None = None,
     *,
     external_storage: ExternalStorageDriver | None = None,
+    external_storage_cache: ExternalPayloadCache | None = None,
 ) -> list[Any]:
     """Decode several raw blobs or ``{codec, blob}`` envelopes in order."""
     if external_storage is not None or any(
         isinstance(value, dict) and "external_storage" in value for value in values
     ):
         return [
-            decode_envelope(value, codec=codec, external_storage=external_storage)
+            decode_envelope(
+                value,
+                codec=codec,
+                external_storage=external_storage,
+                external_storage_cache=external_storage_cache,
+            )
             for value in values
         ]
 

tests/test_external_storage.py

@@ -6,6 +6,7 @@
 from durable_workflow import serializer
 from durable_workflow.external_storage import (
     EXTERNAL_PAYLOAD_REFERENCE_SCHEMA,
+    ExternalPayloadCache,
     ExternalPayloadIntegrityError,
     ExternalPayloadReference,
     LocalFilesystemExternalStorage,
@@ -103,6 +104,74 @@ def test_fetch_external_payload_rejects_mutated_bytes(tmp_path: Path) -> None:
         fetch_external_payload(storage, reference)
 
 
+def test_fetch_external_payload_cache_reuses_verified_bytes(tmp_path: Path) -> None:
+    storage = LocalFilesystemExternalStorage(tmp_path)
+    cache = ExternalPayloadCache(max_entries=2, max_bytes=1024)
+    reference = store_external_payload(storage, b'{"stable":true}', codec="json")
+    path = Path(reference.uri.removeprefix("file://"))
+
+    assert fetch_external_payload(storage, reference, cache=cache) == b'{"stable":true}'
+    path.write_bytes(b'{"stable":false}')
+
+    assert fetch_external_payload(storage, reference, cache=cache) == b'{"stable":true}'
+    assert len(cache) == 1
+
+
+def test_fetch_external_payload_does_not_cache_failed_integrity_check(tmp_path: Path) -> None:
+    storage = LocalFilesystemExternalStorage(tmp_path)
+    cache = ExternalPayloadCache(max_entries=2, max_bytes=1024)
+    reference = store_external_payload(storage, b'{"safe":true}', codec="json")
+    path = Path(reference.uri.removeprefix("file://"))
+    path.write_bytes(b'{"safe":false}')
+
+    with pytest.raises(ExternalPayloadIntegrityError, match="size|hash"):
+        fetch_external_payload(storage, reference, cache=cache)
+
+    assert len(cache) == 0
+
+
+def test_decode_envelopes_can_share_external_payload_cache(tmp_path: Path) -> None:
+    storage = LocalFilesystemExternalStorage(tmp_path)
+    cache = ExternalPayloadCache(max_entries=2, max_bytes=1024)
+    env = serializer.external_storage_envelope(
+        {"message": "x" * 64},
+        external_storage=storage,
+        threshold_bytes=10,
+        codec="json",
+    )
+    path = Path(env["external_storage"]["uri"].removeprefix("file://"))
+
+    assert serializer.decode_envelope(env, external_storage=storage, external_storage_cache=cache) == {
+        "message": "x" * 64
+    }
+    path.write_bytes(b'{"message":"mutated"}')
+
+    assert serializer.decode_envelopes(
+        [env],
+        external_storage=storage,
+        external_storage_cache=cache,
+    ) == [{"message": "x" * 64}]
+
+
+def test_external_payload_cache_is_bounded_by_entries_and_bytes(tmp_path: Path) -> None:
+    storage = LocalFilesystemExternalStorage(tmp_path)
+    cache = ExternalPayloadCache(max_entries=1, max_bytes=20)
+    first = store_external_payload(storage, b"first", codec="json")
+    second = store_external_payload(storage, b"second", codec="json")
+
+    fetch_external_payload(storage, first, cache=cache)
+    fetch_external_payload(storage, second, cache=cache)
+
+    assert cache.get(first) is None
+    assert cache.get(second) == b"second"
+
+    too_large = store_external_payload(storage, b"x" * 21, codec="json")
+    fetch_external_payload(storage, too_large, cache=cache)
+
+    assert cache.get(too_large) is None
+    assert cache.current_bytes <= cache.max_bytes
+
+
 def test_local_storage_rejects_file_uri_outside_root(tmp_path: Path) -> None:
     storage = LocalFilesystemExternalStorage(tmp_path / "root")
     outside = tmp_path / "outside"