From 2a81f913a510c524a4532253ee19bd2ea9bd7069 Mon Sep 17 00:00:00 2001 From: Daoyuan Li <94409450+DaoyuanLi2816@users.noreply.github.com> Date: Mon, 5 Oct 2026 17:09:15 -0700 Subject: [PATCH] fix(projects): synchronize document character-count cache (#6315) --- README.md | 3 + .../harness/deerflow/projects/documents.py | 18 ++- backend/tests/AGENTS.md | 6 + .../tests/test_project_document_char_cache.py | 128 ++++++++++++++++++ 4 files changed, 148 insertions(+), 7 deletions(-) create mode 100644 backend/tests/test_project_document_char_cache.py diff --git a/README.md b/README.md index 698640fcd7..691fb1fb2f 100644 --- a/README.md +++ b/README.md @@ -2363,6 +2363,9 @@ run from the pinned snapshot (capped by `projects.shelf_index_max_entries` and `projects.shelf_index_max_bytes`), and the agent can page the shelf and read documents with the `list_project_documents` and `read_project_document` tools. +Document character counts use a bounded, thread-safe cache; concurrent reads +do not serialize full file scans behind its cache lock. + ### Archive read semantics Archiving a project freezes writes but keeps reads. Threads in an archived diff --git a/backend/packages/harness/deerflow/projects/documents.py b/backend/packages/harness/deerflow/projects/documents.py index 2bec301d01..3a99a692a9 100644 --- a/backend/packages/harness/deerflow/projects/documents.py +++ b/backend/packages/harness/deerflow/projects/documents.py @@ -484,6 +484,7 @@ _TEXT_READ_CHUNK = 65536 #: an entry until LRU eviction — no invalidation hook needed. _CHAR_COUNT_CACHE: OrderedDict[tuple[str, str], int] = OrderedDict() _CHAR_COUNT_CACHE_MAX = 256 +_CHAR_COUNT_CACHE_LOCK = threading.Lock() def _read_text_window(path: Path, *, offset: int, limit: int) -> str: @@ -528,14 +529,17 @@ def _count_text_chars(path: Path) -> int: def _cached_char_count(document_id: str, sha256: str, path: Path) -> int: """Worker-thread: character count of an immutable document, cached by content identity.""" key = (document_id, sha256) - cached = _CHAR_COUNT_CACHE.get(key) - if cached is not None: - _CHAR_COUNT_CACHE.move_to_end(key) - return cached + with _CHAR_COUNT_CACHE_LOCK: + cached = _CHAR_COUNT_CACHE.get(key) + if cached is not None: + _CHAR_COUNT_CACHE.move_to_end(key) + return cached + # Only cache metadata is serialized; scans of unrelated files stay concurrent. total = _count_text_chars(path) - _CHAR_COUNT_CACHE[key] = total - while len(_CHAR_COUNT_CACHE) > _CHAR_COUNT_CACHE_MAX: - _CHAR_COUNT_CACHE.popitem(last=False) + with _CHAR_COUNT_CACHE_LOCK: + _CHAR_COUNT_CACHE[key] = total + while len(_CHAR_COUNT_CACHE) > _CHAR_COUNT_CACHE_MAX: + _CHAR_COUNT_CACHE.popitem(last=False) return total diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index fdf9b90e9a..a28c45110e 100644 --- a/backend/tests/AGENTS.md +++ b/backend/tests/AGENTS.md @@ -77,6 +77,12 @@ the old operation starts does not catch SQLite SELECT/ORM-flush races. Keep the old completion timestamp within its original lease so expiry cannot mask a missing token fence; always drain paused tasks and restore session patches. +## Project document cache + +`test_project_document_char_cache.py` pauses real file-IO workers between cache +lookup and promotion to exercise concurrent eviction. Keep cache metadata +operations atomic, and verify that a full scan does not hold the cache lock. + ## Executor starvation tests `test_executor_starvation.py` covers the deterministic starvation semantics from RFC #4560: diff --git a/backend/tests/test_project_document_char_cache.py b/backend/tests/test_project_document_char_cache.py new file mode 100644 index 0000000000..c558a3230f --- /dev/null +++ b/backend/tests/test_project_document_char_cache.py @@ -0,0 +1,128 @@ +"""Character-count cache invariants across real file-IO worker threads.""" + +from __future__ import annotations + +import asyncio +import threading +from collections import OrderedDict + +import pytest + +from deerflow.projects import documents + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def cache(monkeypatch): + cache = OrderedDict() + monkeypatch.setattr(documents, "_CHAR_COUNT_CACHE", cache) + monkeypatch.setattr(documents, "_CHAR_COUNT_CACHE_MAX", 2) + return cache + + +async def test_hit_promotion_cannot_race_with_eviction(tmp_path, monkeypatch): + hit_ready = threading.Event() + writer_progress = threading.Event() + release_hit = threading.Event() + hot_key = ("hot", "sha-hot") + + class PausingCache(OrderedDict): + def get(self, key, default=None): + value = super().get(key, default) + if key == hot_key: + hit_ready.set() + assert release_hit.wait(5), "cache hit was not released" + return value + + class ObservedLock: + def __init__(self): + self.lock = threading.Lock() + + def __enter__(self): + if hit_ready.is_set(): + # Signal before acquiring: a correct implementation holds the + # lock across the paused lookup and subsequent promotion. + writer_progress.set() + self.lock.acquire() + + def __exit__(self, *exc): + self.lock.release() + + cache = PausingCache({hot_key: 8}) + monkeypatch.setattr(documents, "_CHAR_COUNT_CACHE", cache) + monkeypatch.setattr(documents, "_CHAR_COUNT_CACHE_MAX", 1) + monkeypatch.setattr(documents, "_CHAR_COUNT_CACHE_LOCK", ObservedLock(), raising=False) + hot_path = tmp_path / "hot.txt" + hot_path.write_text("abcdefgh", encoding="utf-8") + other_path = tmp_path / "other.txt" + other_path.write_text("other", encoding="utf-8") + + async def insert_other(): + try: + return await documents.document_char_count(document_id="other", sha256="sha-other", path=other_path) + finally: + # Without the lock, insertion completes and evicts the paused hit. + writer_progress.set() + + tasks = [asyncio.create_task(documents.document_char_count(document_id=hot_key[0], sha256=hot_key[1], path=hot_path))] + try: + assert await asyncio.to_thread(hit_ready.wait, 5), "cache hit did not start" + tasks.append(asyncio.create_task(insert_other())) + assert await asyncio.to_thread(writer_progress.wait, 5), "competing cache operation did not start" + release_hit.set() + assert await asyncio.gather(*tasks) == [8, 5] + assert list(cache.items()) == [(("other", "sha-other"), 5)] + finally: + release_hit.set() + await asyncio.gather(*tasks, return_exceptions=True) + + +async def test_full_scan_does_not_hold_cache_lock(tmp_path, monkeypatch, cache): + cache[("hot", "sha-hot")] = 8 + scan_started = threading.Event() + release_scan = threading.Event() + slow_path = tmp_path / "slow.txt" + slow_path.write_text("slow scan", encoding="utf-8") + real_count = documents._count_text_chars + + def paused_count(path): + scan_started.set() + assert release_scan.wait(5), "file scan was not released" + return real_count(path) + + monkeypatch.setattr(documents, "_count_text_chars", paused_count) + tasks = [asyncio.create_task(documents.document_char_count(document_id="slow", sha256="sha-slow", path=slow_path))] + try: + assert await asyncio.to_thread(scan_started.wait, 5), "file scan did not start" + tasks.append(asyncio.create_task(documents.document_char_count(document_id="hot", sha256="sha-hot", path=tmp_path / "unused.txt"))) + assert await asyncio.wait_for(asyncio.shield(tasks[-1]), 5) == 8 + release_scan.set() + assert await tasks[0] == 9 + finally: + release_scan.set() + await asyncio.gather(*tasks, return_exceptions=True) + + +async def test_zero_counts_content_identity_and_lru_bound(tmp_path, monkeypatch, cache): + calls = [] + real_count = documents._count_text_chars + + def counted(path): + calls.append(path) + return real_count(path) + + monkeypatch.setattr(documents, "_count_text_chars", counted) + empty = tmp_path / "empty.txt" + empty.write_text("", encoding="utf-8") + text = tmp_path / "text.txt" + text.write_text("\u4f60\u597d", encoding="utf-8") + assert await documents.document_char_count(document_id="a", sha256="empty", path=empty) == 0 + assert await documents.document_char_count(document_id="b", sha256="text", path=text) == 2 + assert await documents.document_char_count(document_id="a", sha256="empty", path=empty) == 0 + assert calls == [empty, text] + assert await documents.document_char_count(document_id="c", sha256="text", path=text) == 2 + assert list(cache) == [("a", "empty"), ("c", "text")] + assert await documents.document_char_count(document_id="a", sha256="changed", path=text) == 2 + assert list(cache) == [("c", "text"), ("a", "changed")] + assert len(cache) == 2