mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(checkpoints): treat corrupt Redis history cache entries as misses (#6592)
* fix(checkpoint): treat malformed Redis history cache entries as misses * test(checkpoint): cover malformed Redis history cache entries * style(checkpoint): format corrupt cache regression test * test(checkpoint): cover non-dict cache entries and pinned Ruff formatting * fix(checkpoint): include cache key in malformed-entry diagnostics * docs: trim inherited agent guidance below size limit --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
f71999d99c
commit
29d654e86d
16
AGENTS.md
16
AGENTS.md
@ -34,15 +34,13 @@ Nginx is the single public entry: it proxies `/api/*` to the Gateway, rewriting
|
||||
HTML and configured textual assets, deliberately leaving SSE, fonts, images, audio, and
|
||||
video uncompressed at the proxy layer.
|
||||
|
||||
Both compose files publish that entry as `"${BIND_HOST:-127.0.0.1}:${PORT:-2026}:2026"`
|
||||
— **loopback by default**, matching the README's documented deployment model; a bare
|
||||
`"${PORT}:2026"` binds `0.0.0.0`, which does not. The root `PORT` value is Docker ingress
|
||||
configuration only; local orchestration pins Next.js to `3000` so loading `.env` cannot
|
||||
make `make dev` wait on the wrong port. Nginx listening `default_server` on IPv4+IPv6 and
|
||||
the Gateway binding `0.0.0.0:8001` are container-internal on purpose: the published nginx
|
||||
port is the entire external surface. Any new published port needs an explicit bind
|
||||
address; `backend/tests/test_compose_default_bind_host.py` pins this for every service in
|
||||
both compose files.
|
||||
Both Compose files publish nginx at `"${BIND_HOST:-127.0.0.1}:${PORT:-2026}:2026"`
|
||||
(README's **loopback default**); bare `"${PORT}:2026"` binds `0.0.0.0`. Root `PORT`
|
||||
controls Docker ingress only; local Next.js stays on `3000` so `.env` cannot
|
||||
redirect `make dev`'s port wait. Nginx's IPv4+IPv6 `default_server` and Gateway's
|
||||
`0.0.0.0:8001` bind are container-internal; only published nginx is external.
|
||||
Require explicit bind addresses for new published ports;
|
||||
`backend/tests/test_compose_default_bind_host.py` pins every service in both files.
|
||||
|
||||
## Repository Map
|
||||
|
||||
|
||||
@ -75,9 +75,23 @@ class RedisCheckpointHistoryCache:
|
||||
if raw is None:
|
||||
self._misses += 1
|
||||
continue
|
||||
try:
|
||||
tag, payload = raw.split(_TAG_SEPARATOR, 1)
|
||||
entry = self._serde.loads_typed((tag.decode(), payload))
|
||||
if not isinstance(entry, dict) or not isinstance(entry.get("writes"), list):
|
||||
raise ValueError("invalid checkpoint history entry shape")
|
||||
except Exception as exc:
|
||||
# Cached history is optional. A malformed entry must not prevent
|
||||
# rebuilding history from authoritative checkpoints.
|
||||
logger.warning(
|
||||
"checkpoint history cache decode failed; treating as miss (key=%s): %s",
|
||||
key,
|
||||
type(exc).__name__,
|
||||
)
|
||||
self._misses += 1
|
||||
continue
|
||||
self._hits += 1
|
||||
tag, payload = raw.split(_TAG_SEPARATOR, 1)
|
||||
found[key] = self._serde.loads_typed((tag.decode(), payload))
|
||||
found[key] = entry
|
||||
return found
|
||||
|
||||
async def aset_many(self, entries: dict[str, dict[str, Any]]) -> None:
|
||||
|
||||
@ -0,0 +1,58 @@
|
||||
"""Malformed optional Redis checkpoint cache entries must degrade to misses."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.runtime.checkpoint_cache import redis as redis_cache
|
||||
|
||||
|
||||
class _RedisClient:
|
||||
def __init__(self, values: dict[str, bytes | None]) -> None:
|
||||
self.values = values
|
||||
|
||||
async def mget(self, keys: list[str]) -> list[bytes | None]:
|
||||
return [self.values.get(key) for key in keys]
|
||||
|
||||
|
||||
class _Serde:
|
||||
def loads_typed(self, value: tuple[str, bytes]) -> Any:
|
||||
tag, payload = value
|
||||
if tag != "json":
|
||||
raise ValueError("unknown serializer")
|
||||
if payload == b"bad":
|
||||
raise ValueError("invalid serialized payload")
|
||||
if payload == b"not-a-dict":
|
||||
return ["writes"]
|
||||
if payload == b"wrong-shape":
|
||||
return {"writes": "not a list"}
|
||||
if payload == b"ok":
|
||||
return {"writes": [], "seed": None}
|
||||
raise AssertionError(f"unexpected payload: {payload!r}")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"corrupt",
|
||||
[
|
||||
b"missing-separator",
|
||||
b"\xff\x00ok",
|
||||
b"json\x00bad",
|
||||
b"json\x00wrong-shape",
|
||||
b"json\x00not-a-dict",
|
||||
],
|
||||
ids=["missing-separator", "invalid-tag-encoding", "serde-failure", "wrong-shape", "not-a-dict"],
|
||||
)
|
||||
async def test_corrupt_entry_is_miss_while_valid_entry_is_preserved(monkeypatch: pytest.MonkeyPatch, corrupt: bytes) -> None:
|
||||
client = _RedisClient({"good": b"json\x00ok", "bad": corrupt, "absent": None})
|
||||
monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: client)
|
||||
cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=_Serde(), ttl_seconds=60)
|
||||
|
||||
found = await cache.aget_many(["good", "bad", "absent"])
|
||||
|
||||
assert found == {"good": {"writes": [], "seed": None}}
|
||||
stats = cache.stats()
|
||||
assert stats.hits == 1
|
||||
assert stats.misses == 2
|
||||
Loading…
x
Reference in New Issue
Block a user