From d9e6341e5bfbbaf1423d338b6098c30aef4cfa1b Mon Sep 17 00:00:00 2001 From: SumEn <1824963386@qq.com> Date: Sun, 25 Oct 2026 22:00:14 +0800 Subject: [PATCH] fix(checkpoints): isolate cached history payloads from state readers (#7259) * fix(checkpoints): isolate cached history payloads from state readers Copy nested snapshot seeds and pending-write payloads at both cache boundaries. Keep user copy callbacks outside the registry lock so readers cannot poison later checkpoint materialization or deadlock by re-entering the cache. Cover sync/async state accessor reads, nested message payloads, disabled-cache controls and callback lock ownership. * docs(checkpoints): document synchronous history copy costs --------- Co-authored-by: Willem Jiang --- README.md | 3 +- .../harness/deerflow/runtime/AGENTS.md | 13 ++ .../runtime/checkpoint_cache/memory.py | 23 ++-- ...test_checkpoint_cache_payload_isolation.py | 116 ++++++++++++++++++ 4 files changed, 143 insertions(+), 12 deletions(-) create mode 100644 backend/tests/test_checkpoint_cache_payload_isolation.py diff --git a/README.md b/README.md index 6b825816d4..f4c78bb76b 100644 --- a/README.md +++ b/README.md @@ -564,7 +564,8 @@ caches materialized checkpoint histories: `type` is `memory` (default) or Gateway/async-only; the sync TUI/embedded path supports `memory` only. The cache is performance-only — results are identical with it disabled — so it is never frozen and workers sharing one checkpoint database may safely run -different cache settings. Redis thread cleanup treats a configured `key_prefix` +different cache settings. Returned state snapshots have independent nested values, +so editing one cannot change later checkpoint reads. Redis thread cleanup treats a configured `key_prefix` as literal text, including glob metacharacters, so deletion stays within that namespace and thread. diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index db64630729..dc751f8939 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -4,6 +4,19 @@ The memory history cache is shared by sync saver wrappers. Its threading lock serializes lookup/LRU promotion, population/eviction, thread purge, close, and statistics snapshots. Keep cache critical sections free of awaits and saver I/O; immutable histories do not make the mutable LRU registry thread-safe. +Memory entries own deep copies of write payloads and snapshot seeds on ingest +and read. Clone payloads outside the registry lock: user-defined copy callbacks +can re-enter the cache. Preserve sync/async state isolation through +`CheckpointStateAccessor`; tests live in `tests/test_checkpoint_cache_payload_isolation.py`. + +Copies run synchronously on the caller's thread, including `aget_many` and +`aset_many` on the event loop. Each ingest or cache hit copies the whole +entry, so CPU and allocation costs grow with payload size. A warm scan of +N checkpoints can repeatedly copy growing snapshot seeds, reaching O(N²) +total copied data. The LRU entry cap bounds entry count, not payload bytes +or scan cost. Profile long-thread history reads before offloading copies +to worker threads; preserve payload isolation and copy callbacks outside +the registry lock. ### Redis Checkpoint Cache Purge diff --git a/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py b/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py index def37d3c5d..5c97c13c4c 100644 --- a/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py +++ b/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py @@ -4,17 +4,15 @@ from __future__ import annotations import threading from collections import OrderedDict +from copy import deepcopy from typing import Any from deerflow.runtime.checkpoint_cache.base import CheckpointCacheStats, thread_key_stem def _copy_entry(entry: dict[str, Any]) -> dict[str, Any]: - """Copy-on-read/write: fresh writes list; seed shared (never mutated in place).""" - copied: dict[str, Any] = {"writes": list(entry["writes"])} - if "seed" in entry: - copied["seed"] = entry["seed"] - return copied + """Own nested write payloads and snapshot seeds on both sides of the cache.""" + return deepcopy(entry) class MemoryCheckpointHistoryCache: @@ -42,15 +40,18 @@ class MemoryCheckpointHistoryCache: continue self._data.move_to_end(key) self._hits += 1 - found[key] = _copy_entry(entry) - return found + found[key] = entry + # Entries are privately owned and never changed after publication. Copy + # outside the registry lock: user payloads can implement __deepcopy__. + return {key: _copy_entry(entry) for key, entry in found.items()} def set_many(self, entries: dict[str, dict[str, Any]]) -> None: + if not self.enabled: + return + copied = {key: _copy_entry(entry) for key, entry in entries.items()} with self._lock: - if not self.enabled: - return - for key, entry in entries.items(): - self._data[key] = _copy_entry(entry) + for key, entry in copied.items(): + self._data[key] = entry self._data.move_to_end(key) while len(self._data) > self._max_entries: self._data.popitem(last=False) diff --git a/backend/tests/test_checkpoint_cache_payload_isolation.py b/backend/tests/test_checkpoint_cache_payload_isolation.py new file mode 100644 index 0000000000..1ec62068a1 --- /dev/null +++ b/backend/tests/test_checkpoint_cache_payload_isolation.py @@ -0,0 +1,116 @@ +"""Returned checkpoint snapshots must never mutate cached history payloads.""" + +from copy import deepcopy +from typing import Annotated, TypedDict + +import pytest +from langgraph.channels import DeltaChannel +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.graph import StateGraph + +from deerflow.runtime.checkpoint_cache.memory import MemoryCheckpointHistoryCache +from deerflow.runtime.checkpoint_state import CheckpointStateAccessor +from deerflow.runtime.checkpointer.cached_saver import CachedHistorySaver + + +def _fold(state, writes): + result = list(state) + for batch in writes: + result.extend(batch) + return result + + +def _stack(frequency, enabled): + class State(TypedDict): + items: Annotated[list[dict], DeltaChannel(_fold, snapshot_frequency=frequency)] + + inner = InMemorySaver() + cache = MemoryCheckpointHistoryCache(max_entries=32 if enabled else 0) + saver = CachedHistorySaver(inner, cache, key_prefix="payload-isolation") + builder = StateGraph(State) + for step in range(4): + builder.add_node(str(step), lambda state: {"items": [{"nested": {"value": "saved"}}]}) + if step: + builder.add_edge(str(step - 1), str(step)) + builder.set_entry_point("0") + builder.set_finish_point("3") + graph = builder.compile(checkpointer=saver) + return graph, CheckpointStateAccessor.bind(graph, saver, mode="delta"), cache + + +def _mutate(snapshot): + original = deepcopy(snapshot.values) + snapshot.values["items"][0]["nested"]["value"] = "tampered" + snapshot.values["items"].append({"nested": {"value": "injected"}}) + return original + + +@pytest.mark.parametrize("frequency", [2, 100], ids=["snapshot-seed", "pending-writes"]) +@pytest.mark.parametrize("enabled", [True, False], ids=["cached", "uncached-control"]) +def test_sync_graph_snapshots_are_isolated(frequency, enabled): + graph, accessor, cache = _stack(frequency, enabled) + config = {"configurable": {"thread_id": "sync-payload-isolation"}} + graph.invoke({"items": [{"nested": {"value": "input"}}]}, config) + history = accessor.history(config) + checked = 0 + for checkpoint in history: + snapshot = accessor.get(checkpoint.config) + if not snapshot.values.get("items"): + continue + original = _mutate(snapshot) + assert accessor.get(checkpoint.config).values == original + checked += 1 + assert checked >= 4 + assert cache.stats().hits > 0 if enabled else cache.stats().entries == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("frequency", [2, 100], ids=["snapshot-seed", "pending-writes"]) +@pytest.mark.parametrize("enabled", [True, False], ids=["cached", "uncached-control"]) +async def test_async_graph_snapshots_are_isolated(frequency, enabled): + graph, accessor, cache = _stack(frequency, enabled) + config = {"configurable": {"thread_id": "async-payload-isolation"}} + await graph.ainvoke({"items": [{"nested": {"value": "input"}}]}, config) + history = await accessor.ahistory(config) + checked = 0 + for checkpoint in history: + snapshot = await accessor.aget(checkpoint.config) + if not snapshot.values.get("items"): + continue + original = _mutate(snapshot) + assert (await accessor.aget(checkpoint.config)).values == original + checked += 1 + assert checked >= 4 + assert cache.stats().hits > 0 if enabled else cache.stats().entries == 0 + + +@pytest.mark.parametrize("direction", ["ingest", "read"]) +def test_cache_copies_nested_write_payloads_and_snapshot_seeds(direction): + from langchain_core.messages import HumanMessage + from langgraph.checkpoint.serde.types import _DeltaSnapshot + + cache = MemoryCheckpointHistoryCache() + seed = _DeltaSnapshot([HumanMessage(content="seed", id="seed", additional_kwargs={"nested": ["original"]})]) + entry = {"seed": seed, "writes": [("task", "messages", [HumanMessage(content="write", id="write", additional_kwargs={"nested": ["original"]})])]} + original = deepcopy(entry) + cache.set_many({"key": entry}) + exposed = entry if direction == "ingest" else cache.get_many(["key"])["key"] + exposed["seed"].value[0].additional_kwargs["nested"].append("corrupted") + exposed["writes"][0][2][0].content = "corrupted" + assert cache.get_many(["key"])["key"] == original + + +def test_payload_copying_runs_outside_registry_lock(): + cache = MemoryCheckpointHistoryCache() + copies = [] + + class Probe: + def __deepcopy__(self, memo): + assert cache._lock.acquire(blocking=False), "copy callback ran under the cache lock" + cache._lock.release() + copies.append(True) + return Probe() + + cache.set_many({"key": {"writes": [("task", "channel", Probe())]}}) + cache.get_many(["key"]) + assert len(copies) == 2