diff --git a/README.md b/README.md index 50db98b04b..0ca2916713 100644 --- a/README.md +++ b/README.md @@ -498,7 +498,9 @@ 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. +different cache settings. Redis thread cleanup treats a configured `key_prefix` +as literal text, including glob metacharacters, so deletion stays within that +namespace and thread. > [!TIP] > On Linux, if Docker-based commands fail with `permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock`, add your user to the `docker` group and re-login before retrying. See [CONTRIBUTING.md](CONTRIBUTING.md#linux-docker-daemon-permission-denied) for the full fix. diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 6b72f4fb1e..f3cd8925e3 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -1,3 +1,10 @@ +### Redis Checkpoint Cache Purge + +Thread purge escapes Redis glob metacharacters in the complete literal key stem +before SCAN, and checks returned byte/string keys against that stem before UNLINK. +Do not interpret operator `key_prefix` text as a pattern or widen thread deletion +to sibling namespaces. Keep the cursor loop and outage/TTL fallback unchanged. + ### Stream Bridge Heartbeats Memory/Redis bridges keep startup-only `stream_bridge.heartbeat_interval_seconds`; explicit `subscribe(..., heartbeat_interval=...)` overrides it. Provider contexts drain owned cache/bridge/checkpointer/Store teardown across cancellation. `close_agent_stream()` shields close to completion: preserve counts, defer host cancellation, balance repeats, keep active errors, log attached close failures at callers, map close cancellation to failure, never time out. diff --git a/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py b/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py index 4671d7a3c7..b8a25593ad 100644 --- a/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py +++ b/backend/packages/harness/deerflow/runtime/checkpoint_cache/redis.py @@ -112,10 +112,16 @@ class RedisCheckpointHistoryCache: TTL-bounded residual retention; the source-of-truth delete already happened, so this never raises.""" stem = thread_key_stem(key_prefix, thread_id) + # The configured namespace is a literal key prefix, not a Redis glob. + pattern = "".join("\\" + char if char in "\\*?[]" else char for char in stem) + "*" + # redis-py encodes SCAN results with the configured connection encoder. + # A fixed UTF-8 fence silently skips owned keys for other encodings. + encoded_stem = self._client.connection_pool.get_encoder().encode(stem) try: cursor = 0 while True: - cursor, keys = await self._client.scan(cursor=cursor, match=stem + "*", count=500) + cursor, keys = await self._client.scan(cursor=cursor, match=pattern, count=500) + keys = [key for key in keys if key.startswith(encoded_stem if isinstance(key, bytes) else stem)] if keys: await self._client.unlink(*keys) if cursor == 0: diff --git a/backend/tests/test_checkpoint_cache_redis.py b/backend/tests/test_checkpoint_cache_redis.py index b3ca9702c5..2114bebcb0 100644 --- a/backend/tests/test_checkpoint_cache_redis.py +++ b/backend/tests/test_checkpoint_cache_redis.py @@ -13,6 +13,16 @@ from deerflow.runtime.checkpoint_cache.provider import ( ) +class _FakeEncoder: + def encode(self, value: str) -> bytes: + return value.encode() + + +class _FakeConnectionPool: + def get_encoder(self) -> _FakeEncoder: + return _FakeEncoder() + + class _FakeRedis: """Minimal async redis stand-in: mget / set / pipeline / scan / unlink.""" @@ -20,6 +30,7 @@ class _FakeRedis: self.store: dict[str, bytes] = {} self.ttls: dict[str, int | None] = {} self.unlinked: list[tuple[str, ...]] = [] + self.connection_pool = _FakeConnectionPool() async def mget(self, keys: list[str]) -> list[bytes | None]: return [self.store.get(k) for k in keys] diff --git a/backend/tests/test_checkpoint_cache_redis_literal_purge.py b/backend/tests/test_checkpoint_cache_redis_literal_purge.py new file mode 100644 index 0000000000..528d7bf694 --- /dev/null +++ b/backend/tests/test_checkpoint_cache_redis_literal_purge.py @@ -0,0 +1,78 @@ +"""Thread deletion treats configured cache namespaces as literal Redis keys.""" + +from unittest.mock import AsyncMock, Mock + +import pytest + +from deerflow.runtime.checkpoint_cache.base import make_history_key +from deerflow.runtime.checkpoint_cache.redis import RedisCheckpointHistoryCache + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ("prefix", "thread_id", "sibling_thread_id", "pattern"), + [ + ("tenant*", "t1", "t10", r"tenant\*:t1:*"), + ("tenant?", "t1", "t10", r"tenant\?:t1:*"), + ("tenant[ab]", "t1", "t10", r"tenant\[ab\]:t1:*"), + (r"tenant\name", "t1", "t10", r"tenant\\name:t1:*"), + ("plain", "t1", "t10", "plain:t1:*"), + ("plain", "t1*", "t10", r"plain:t1\*:*"), + ], +) +async def test_purge_escapes_namespace_and_filters_results(prefix, thread_id, sibling_thread_id, pattern): + # Exercise the actual purge method without needing a running Redis server. + # The expected SCAN pattern is a wire-level Redis glob contract; returned + # keys also include a foreign namespace to pin the final deletion fence. + cache = RedisCheckpointHistoryCache.__new__(RedisCheckpointHistoryCache) + client = AsyncMock() + cache._client = client + client.connection_pool = Mock() + client.connection_pool.get_encoder.return_value.encode.side_effect = lambda value: value.encode("utf-8") + owned = make_history_key(prefix, thread_id, "", "c1", "messages").encode() + foreign = make_history_key("tenant-other", thread_id, "", "c1", "messages").encode() + sibling = make_history_key(prefix, sibling_thread_id, "", "c1", "messages").encode() + + async def scan(*, cursor, match, count): + assert match == pattern + assert count == 500 + # Mixed byte/string results and multiple pages exercise redis clients + # with either response mode and preserve the scan cursor protocol. + if cursor == 0: + return 7, [owned, foreign, sibling] + assert cursor == 7 + return 0, [owned.decode()] + + client.scan.side_effect = scan + await cache.adelete_thread(prefix, thread_id) + assert client.unlink.await_args_list[0].args == (owned,) + assert client.unlink.await_args_list[1].args == (owned.decode(),) + + +@pytest.mark.anyio +async def test_purge_does_not_unlink_empty_filtered_page(): + cache = RedisCheckpointHistoryCache.__new__(RedisCheckpointHistoryCache) + cache._client = AsyncMock() + cache._client.connection_pool = Mock() + cache._client.connection_pool.get_encoder.return_value.encode.side_effect = lambda value: value.encode("utf-8") + cache._client.scan.return_value = (0, [b"other:t1:history"]) + await cache.adelete_thread("owned", "t1") + cache._client.unlink.assert_not_awaited() + + +@pytest.mark.anyio +async def test_purge_uses_client_encoding_for_byte_key_fence(): + """A non-UTF-8 Redis client must still unlink its own non-ASCII keys.""" + cache = RedisCheckpointHistoryCache.__new__(RedisCheckpointHistoryCache) + client = AsyncMock() + cache._client = client + client.connection_pool = Mock() + client.connection_pool.get_encoder.return_value.encode.side_effect = lambda value: value.encode("latin-1") + owned = make_history_key("café", "t1", "", "c1", "messages").encode("latin-1") + sibling = make_history_key("café", "t10", "", "c1", "messages").encode("latin-1") + client.scan.return_value = (0, [owned, sibling]) + + await cache.adelete_thread("café", "t1") + + client.scan.assert_awaited_once_with(cursor=0, match="café:t1:*", count=500) + client.unlink.assert_awaited_once_with(owned)