mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-28 14:10:37 +00:00
f7d02a7515
* fix(checkpoints): reject malformed Redis history write records Validate pending-write tuples before optional Redis history entries reach delta materialization. Treat the whole malformed entry as a miss so the authoritative saver rebuilds it without dropping or inventing state. Cover actual checkpoint reads, whole-entry replacement, valid tuple/list records, neighbor preservation and warm-cache rereads. * fix(checkpoints): bind cached history writes to their channel * fix(checkpoints): keep cache warning failures best-effort Guard warning handlers across decode, ingestion, serialization, Redis outages and purge. Preserve authoritative rebuilds and valid neighboring writes, with native graph regression coverage. * docs(gateway): keep guidance below the hard size budget --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
90 lines
3.4 KiB
Python
90 lines
3.4 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 valid_history_entry(entry: Any, *, channel: str | None = None) -> bool:
|
|
"""Check replay records; seed and write values belong to the channel schema."""
|
|
if not isinstance(entry, dict) or not isinstance(entry.get("writes"), list):
|
|
return False
|
|
return all(isinstance(write, (tuple, list)) and len(write) == 3 and isinstance(write[0], str) and isinstance(write[1], str) and (channel is None or write[1] == channel) for write in entry["writes"])
|
|
|
|
|
|
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
|
|
write_serialization_failures: int = 0
|
|
|
|
def as_dict(self) -> dict[str, int]:
|
|
return {
|
|
"hits": self.hits,
|
|
"misses": self.misses,
|
|
"evictions": self.evictions,
|
|
"entries": self.entries,
|
|
"write_serialization_failures": self.write_serialization_failures,
|
|
}
|
|
|
|
|
|
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: ...
|