From f96aca5059ae5835f1aa175d9927b432f8389d46 Mon Sep 17 00:00:00 2001 From: NanPan <111261006+poijygfdyy@users.noreply.github.com> Date: Sun, 18 Oct 2026 10:30:14 +0800 Subject: [PATCH] fix(checkpoints): skip unserializable Redis history cache entries (#7014) * fix(checkpoints): skip Redis cache entries that cannot be serialized * style(checkpoints): format Redis serialization regression * fix(test): assert Redis serialization payload bytes * test(checkpoints): format Redis cache serialization regression * chore: sync frontend agent guidance with upstream size-compliant main * fix(checkpoints): track Redis cache serialization failures * test(checkpoints): consolidate Redis serialization fallback regression * style(tests): format Redis cache regression signature --------- Co-authored-by: Willem Jiang --- .../deerflow/runtime/checkpoint_cache/base.py | 2 + .../runtime/checkpoint_cache/redis.py | 20 +++++++- backend/tests/test_checkpoint_cache_redis.py | 47 ++++++++++++++++++- 3 files changed, 65 insertions(+), 4 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/checkpoint_cache/base.py b/backend/packages/harness/deerflow/runtime/checkpoint_cache/base.py index 0bc79e9fb3..29242207f9 100644 --- a/backend/packages/harness/deerflow/runtime/checkpoint_cache/base.py +++ b/backend/packages/harness/deerflow/runtime/checkpoint_cache/base.py @@ -51,6 +51,7 @@ class CheckpointCacheStats: misses: int = 0 evictions: int = 0 entries: int = 0 + write_serialization_failures: int = 0 def as_dict(self) -> dict[str, int]: return { @@ -58,6 +59,7 @@ class CheckpointCacheStats: "misses": self.misses, "evictions": self.evictions, "entries": self.entries, + "write_serialization_failures": self.write_serialization_failures, } diff --git a/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py b/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py index b8a25593ad..4f9b9d38f3 100644 --- a/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py +++ b/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py @@ -59,6 +59,7 @@ class RedisCheckpointHistoryCache: self._ttl = ttl_seconds if ttl_seconds > 0 else None self._hits = 0 self._misses = 0 + self._write_serialization_failures = 0 async def aget_many(self, keys: list[str]) -> dict[str, dict[str, Any]]: if not keys: @@ -100,7 +101,18 @@ class RedisCheckpointHistoryCache: try: pipe = self._client.pipeline(transaction=False) for key, entry in entries.items(): - tag, data = self._serde.dumps_typed(entry) + try: + tag, data = self._serde.dumps_typed(entry) + except Exception as exc: + self._write_serialization_failures += 1 + # Cache writes are optional; one unserializable entry must not + # abort authoritative history reads or discard other entries. + logger.warning( + "checkpoint history cache serialization failed; skipping (key=%s): %s", + key, + type(exc).__name__, + ) + continue pipe.set(key, tag.encode() + _TAG_SEPARATOR + data, ex=self._ttl) await pipe.execute() except _redis_error() as exc: @@ -130,7 +142,11 @@ class RedisCheckpointHistoryCache: logger.warning("checkpoint history cache thread purge failed; residual entries expire via TTL: %s", exc) def stats(self) -> CheckpointCacheStats: - return CheckpointCacheStats(hits=self._hits, misses=self._misses) + return CheckpointCacheStats( + hits=self._hits, + misses=self._misses, + write_serialization_failures=self._write_serialization_failures, + ) async def aclose(self) -> None: await self._client.aclose() diff --git a/backend/tests/test_checkpoint_cache_redis.py b/backend/tests/test_checkpoint_cache_redis.py index 2114bebcb0..3e1606658e 100644 --- a/backend/tests/test_checkpoint_cache_redis.py +++ b/backend/tests/test_checkpoint_cache_redis.py @@ -94,11 +94,23 @@ class _FailingRedis(_FakeRedis): raise RedisError("connection refused") -def _make_cache(monkeypatch: pytest.MonkeyPatch, fake: _FakeRedis, ttl_seconds: int = 60, **kwargs: Any): +def _make_cache( + monkeypatch: pytest.MonkeyPatch, + fake: _FakeRedis, + ttl_seconds: int = 60, + *, + serde: Any | None = None, + **kwargs: Any, +): import deerflow.runtime.checkpoint_cache.redis as redis_mod monkeypatch.setattr(redis_mod, "_create_client", lambda *a, **k: fake) - return redis_mod.RedisCheckpointHistoryCache("redis://unused", serde=JsonPlusSerializer(), ttl_seconds=ttl_seconds, **kwargs) + return redis_mod.RedisCheckpointHistoryCache( + "redis://unused", + serde=serde if serde is not None else JsonPlusSerializer(), + ttl_seconds=ttl_seconds, + **kwargs, + ) def _entry(i: int) -> dict: @@ -161,6 +173,37 @@ async def test_redis_outage_skips_write(monkeypatch: pytest.MonkeyPatch, caplog: assert "write failed" in caplog.text +@pytest.mark.anyio +async def test_redis_serialization_failure_preserves_neighbor_writes(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None: + class _SelectiveFailingSerde: + def dumps_typed(self, entry: dict) -> tuple[str, bytes]: + if entry.get("bad"): + raise TypeError("sensitive serializer detail") + return "json", b"encoded" + + fake = _FakeRedis() + cache = _make_cache(monkeypatch, fake, serde=_SelectiveFailingSerde()) + with caplog.at_level("WARNING"): + await cache.aset_many( + { + "before": {"writes": []}, + "broken": {"writes": [], "bad": True}, + "after": {"writes": []}, + } + ) + + assert fake.store == { + "before": b"json\x00encoded", + "after": b"json\x00encoded", + } + assert fake.ttls == {"before": 60, "after": 60} + assert "broken" in caplog.text + assert "TypeError" in caplog.text + assert "sensitive serializer detail" not in caplog.text + assert cache.stats().write_serialization_failures == 1 + assert cache.stats().as_dict()["write_serialization_failures"] == 1 + + @pytest.mark.anyio async def test_zero_ttl_disables_expiry_explicitly(monkeypatch: pytest.MonkeyPatch): fake = _FakeRedis()