mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-04 20:08:40 +00:00
c8cf1bf2fb
* feat(checkpoint-cache): delta-mode checkpoint history cache with recursive compose
Read-only, invalidation-free cache for LangGraph delta-channel history
({writes, seed}) at the get_delta_channel_history choke point:
- database.checkpoint_cache config (memory|redis; max_entries 0=disabled;
redis bounded by TTL, Gateway/async only)
- memory LRU backend (copy-on-read, zero-serde hit path) and redis backend
(lazy import, degrades to all-miss on outage)
- CachedHistorySaver: recursive composition from the nearest warm ancestor
(depth budget 8), caching each level; depth-0 cold chains delegate one
inner fast-path walk. Entries keyed by immutable
(db, thread, ns, checkpoint_id, channel) — no invalidation, coherent
across workers
- provider wiring: wraps in delta mode only (async + sync), full mode
untouched; sync path is memory-only
- bench opt-in: DEERFLOW_CHECKPOINT_BENCH_HISTORY_CACHE=1
sqlite bench (500 updates, payload 2KB): write phase 2.28x at f=250,
1.32x at f=10; one delegated walk per thread cold start.
* chore(config): bump config_version to 32 for database.checkpoint_cache
The checkpoint history cache feature added the database.checkpoint_cache
section to config.example.yaml; bump the schema version so existing
deployments get the outdated-config warning and can run make config-upgrade.
* chore(helm): bump config_version to 32 in chart values and README
* fix(checkpoint-cache): purge thread history entries on delete paths
Addresses review on #4638: delete_thread/prune removed source-of-truth
checkpoints but left the thread's materialized history payloads in the
cache (memory: until LRU eviction; redis: until TTL, default 1 day) — a
data-lifecycle gap for tenant offboarding / GDPR-style erasure.
- Cache contract gains thread-scoped adelete_thread/delete_thread
(lifecycle purge, not invalidation; entries remain immutable)
- Memory backend: stem scan over the LRU map; redis: SCAN MATCH + UNLINK,
outage degrades to TTL-bounded retention without raising
- CachedHistorySaver purges on delete_thread/adelete_thread and
prune/aprune (prune rewrites chains, so pre-prune histories must go);
delete_for_runs stays delegation-only (run->thread mapping unavailable,
no in-tree callers), documented in code
- ttl_seconds description documents the residual-retention window
- Tests: thread-scoped purge on both backends, saver-level delete/prune
purge, prefix-safety (t1 vs t10), redis outage degradation, and the
pinned no-purge behavior of delete_for_runs
* fix(checkpoint-cache): stable db identity, prefix-aware sync singleton, explicit zero TTL
Addresses Copilot review on #4638:
- checkpoint_cache_db_hash now hashes the credential-free postgres
identity (host:port/database + schema): credential rotation no longer
changes the cache namespace (cold cache + orphaned keys until TTL).
Unparseable URLs fall back to the raw string.
- The sync-path memory cache singleton is also keyed by its key_prefix:
a namespace change (db identity change or operator override) recreates
the cache instead of leaving stale-prefix entries unreachable and
unpurgeable.
- ttl_seconds=0 is now an explicit, documented opt-out of redis expiry
(SET without EX; redis maxmemory policy only) instead of a silent
'ttl_seconds or None' coercion.
Tests: credential-rotation hash stability, unparseable-URL fallback,
prefix-change singleton recreation, same-prefix singleton reuse, and
zero-TTL wire behavior (ex=None).
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""Cache backend contract for checkpoint delta-history entries.
|
|
|
|
Entries are ``DeltaChannelHistory``-shaped dicts (``{"writes": [...], "seed"?}``)
|
|
keyed by immutable (database, thread, namespace, checkpoint_id, channel)
|
|
tuples. Checkpoint lineage is append-only and a checkpoint's history excludes
|
|
its own pending writes, so entries never change once written: correctness
|
|
never requires invalidation, and a shared backend is coherent across
|
|
processes without any coordination.
|
|
|
|
The only delete API is thread-scoped (``adelete_thread``/``delete_thread``),
|
|
and it exists purely for data lifecycle, not correctness: when the source
|
|
checkpoints are erased (thread deletion, tenant offboarding, GDPR-style
|
|
erasure), the cached history payloads for that thread must go too instead of
|
|
lingering until LRU eviction or TTL expiry.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from dataclasses import dataclass
|
|
from typing import Any, Protocol
|
|
|
|
CACHE_FORMAT_VERSION = 1
|
|
|
|
|
|
def make_history_key(
|
|
key_prefix: str,
|
|
thread_id: str,
|
|
checkpoint_ns: str,
|
|
checkpoint_id: str,
|
|
channel: str,
|
|
) -> str:
|
|
"""Build a collision-safe cache key.
|
|
|
|
``thread_id`` stays readable for ops debugging; the remaining components
|
|
are hashed with NUL separators so namespaces containing ':' cannot
|
|
produce ambiguous keys.
|
|
"""
|
|
digest = hashlib.sha256(f"{checkpoint_ns}\x00{checkpoint_id}\x00{channel}".encode()).hexdigest()[:24]
|
|
return f"{key_prefix}:{thread_id}:{digest}"
|
|
|
|
|
|
def thread_key_stem(key_prefix: str, thread_id: str) -> str:
|
|
"""Prefix matching every history key of one thread (see make_history_key)."""
|
|
return f"{key_prefix}:{thread_id}:"
|
|
|
|
|
|
@dataclass
|
|
class CheckpointCacheStats:
|
|
hits: int = 0
|
|
misses: int = 0
|
|
evictions: int = 0
|
|
entries: int = 0
|
|
|
|
def as_dict(self) -> dict[str, int]:
|
|
return {
|
|
"hits": self.hits,
|
|
"misses": self.misses,
|
|
"evictions": self.evictions,
|
|
"entries": self.entries,
|
|
}
|
|
|
|
|
|
class CheckpointHistoryCache(Protocol):
|
|
"""Async backend contract. Deletes are thread-scoped lifecycle purges only."""
|
|
|
|
async def aget_many(self, keys: list[str]) -> dict[str, dict[str, Any]]: ...
|
|
async def aset_many(self, entries: dict[str, dict[str, Any]]) -> None: ...
|
|
async def adelete_thread(self, key_prefix: str, thread_id: str) -> None: ...
|
|
def stats(self) -> CheckpointCacheStats: ...
|
|
async def aclose(self) -> None: ...
|
|
|
|
|
|
class SyncCheckpointHistoryCache(Protocol):
|
|
"""Sync backend contract (embedded/TUI path). Memory backend only."""
|
|
|
|
def get_many(self, keys: list[str]) -> dict[str, dict[str, Any]]: ...
|
|
def set_many(self, entries: dict[str, dict[str, Any]]) -> None: ...
|
|
def delete_thread(self, key_prefix: str, thread_id: str) -> None: ...
|
|
def stats(self) -> CheckpointCacheStats: ...
|