diff --git a/README.md b/README.md index fab3074612..b24b097e4c 100644 --- a/README.md +++ b/README.md @@ -2484,6 +2484,12 @@ DeerFlow can be used as an embedded Python library without running the full HTTP For database-backed run events, deleting a run preserves its thread's sequence watermark. Thread deletion removes that watermark once no events remain, allowing a recreated thread to restart at sequence 1. Owner-scoped deletion preserves the watermark when another owner's events remain. +Single-process JSONL event storage also retains the thread sequence watermark +across run deletion and restarts, so clients using `after_seq` do not miss later +messages. Keep `runs/.seq-watermark` with the run files when backing up this +backend. New or replaced watermarks inherit the deleted run file's permission bits. +Deleting the complete thread removes the watermark and resets allocation. + Thread IDs may be supplied by callers and do not have to be UUIDs. Explicit IDs must contain 1–64 ASCII letters, digits, hyphens, or underscores (`^[A-Za-z0-9_-]{1,64}$`). DeerFlow generates a UUID only when `thread_id` is diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 5d73e24b7b..a699fe54e9 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -120,6 +120,17 @@ idempotent writes, LF/CRLF, blank lines, and malformed records. Reads and deletes treat a run ID writes reject as an unknown run (routes pass URL IDs through); writes still raise. +**JSONL sequence watermark:** before deleting a non-empty run, save the thread's +allocation floor with atomic replacement of `runs/.seq-watermark`, off-loop under +the existing mutation fence. Recovery takes the maximum of that floor and surviving +run files, including legacy directories without a watermark. Copy the deleted run's +permission bits onto the temporary watermark before publication, preserving both +shared-read and restrictive modes. Failed permission setup or publication must leave +the run intact; unreadable/corrupt watermarks fail closed. Complete +thread deletion removes the watermark only after deleting all run files, including +when no events remain. This does not add multi-process JSONL support. Coverage: +`tests/test_jsonl_event_store_seq_watermark.py` and the mutation cancellation suite. + **Targeted run-event attribution** (`runtime/events/store/`): `RunEventStore.find_latest_ai_message_run_ids()` has a complete-or-error contract. Its default implementation walks `list_messages()` backward in diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 47dac36a5b..a30a63d5e2 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -33,6 +33,7 @@ import json import logging import os import re +import tempfile import weakref from collections.abc import Callable, Coroutine, Iterator from datetime import UTC, datetime @@ -134,7 +135,7 @@ class JsonlRunEventStore(RunEventStore): def _compute_max_seq(self, thread_id: str) -> int: """Scan all run files for a thread and return the current max seq (blocking I/O).""" - max_seq = 0 + max_seq = self._read_seq_watermark(thread_id) thread_dir = self._thread_dir(thread_id) if thread_dir.exists(): for f in thread_dir.glob("*.jsonl"): @@ -142,6 +143,37 @@ class JsonlRunEventStore(RunEventStore): max_seq = max(max_seq, record.get("seq", 0)) return max_seq + def _read_seq_watermark(self, thread_id: str) -> int: + path = self._thread_dir(thread_id) / ".seq-watermark" + try: + seq = int(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return 0 + if seq < 0: + raise ValueError(f"Invalid JSONL sequence watermark in {path}") + return seq + + def _save_seq_watermark(self, thread_id: str, seq: int, run_id: str) -> None: + """Publish the allocation floor before deleting records (blocking I/O).""" + if seq <= self._read_seq_watermark(thread_id): + return + path = self._thread_dir(thread_id) / ".seq-watermark" + run_mode = self._run_file(thread_id, run_id).stat().st_mode & 0o777 + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, prefix=".seq-", suffix=".tmp", delete=False) as file: + temporary = Path(file.name) + file.write(str(seq)) + # Match the run's access policy before atomically publishing the floor. + temporary.chmod(run_mode) + temporary.replace(path) + finally: + if temporary is not None: + try: + temporary.unlink(missing_ok=True) + except OSError: + logger.warning("Could not remove JSONL watermark temporary file %s", temporary, exc_info=True) + async def _ensure_seq_loaded(self, thread_id: str) -> None: """Load max seq from existing files into the in-memory counter (non-blocking).""" if thread_id in self._seq_counters: @@ -189,6 +221,8 @@ class JsonlRunEventStore(RunEventStore): if thread_dir.exists(): for f in thread_dir.glob("*.jsonl"): f.unlink() + # Reset only after every run file has been removed successfully. + (thread_dir / ".seq-watermark").unlink(missing_ok=True) def _delete_run_file(self, thread_id: str, run_id: str) -> None: path = self._existing_run_file(thread_id, run_id) @@ -477,6 +511,9 @@ class JsonlRunEventStore(RunEventStore): async def mutate(): events = await asyncio.to_thread(self._read_run_events, thread_id, run_id) count = len(events) + if count: + await self._ensure_seq_loaded(thread_id) + await asyncio.to_thread(self._save_seq_watermark, thread_id, self._seq_counters[thread_id], run_id) await asyncio.to_thread(self._delete_run_file, thread_id, run_id) return count diff --git a/backend/tests/test_jsonl_event_store_cancellation.py b/backend/tests/test_jsonl_event_store_cancellation.py index c39637b2b4..edca017ec8 100644 --- a/backend/tests/test_jsonl_event_store_cancellation.py +++ b/backend/tests/test_jsonl_event_store_cancellation.py @@ -41,6 +41,36 @@ async def _checkpoint(): await asyncio.sleep(0) +@pytest.mark.anyio +async def test_cancelled_watermark_publication_drains_before_thread_delete(tmp_path, monkeypatch): + store = JsonlRunEventStore(tmp_path) + await store.put(**_event(content="baseline")) + paused = _PausedIO(store._save_seq_watermark) + monkeypatch.setattr(store, "_save_seq_watermark", paused) + pending = asyncio.create_task(store.delete_by_run("t1", "r1")) + deletion = None + try: + await asyncio.wait_for(paused.entered.wait(), 5) + pending.cancel() + await _checkpoint() + pending.cancel() + deletion = asyncio.create_task(store.delete_by_thread("t1")) + await _checkpoint() + assert not pending.done() + assert not deletion.done() + paused.release.set() + with pytest.raises(asyncio.CancelledError): + await pending + assert await deletion == 0 + assert not (tmp_path / "threads" / "t1" / "runs" / ".seq-watermark").exists() + new = await JsonlRunEventStore(tmp_path).put(**_event("r2")) + assert new["seq"] == 1 + finally: + paused.release.set() + await asyncio.gather(pending, *([deletion] if deletion is not None else []), return_exceptions=True) + await asyncio.wait_for(paused.finished.wait(), 5) + + @pytest.mark.anyio @pytest.mark.parametrize("method", ["put", "put_if_absent", "put_batch"]) @pytest.mark.parametrize("cancellations", [1, 3]) diff --git a/backend/tests/test_jsonl_event_store_seq_watermark.py b/backend/tests/test_jsonl_event_store_seq_watermark.py new file mode 100644 index 0000000000..836eb3b527 --- /dev/null +++ b/backend/tests/test_jsonl_event_store_seq_watermark.py @@ -0,0 +1,185 @@ +"""Durable cursor allocation after JSONL run deletion and store reopening.""" + +import os +from pathlib import Path + +import pytest + +from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + +def event(run_id): + return {"thread_id": "t1", "run_id": run_id, "event_type": "human_message", "category": "message", "content": run_id} + + +async def write(store, run_id, method): + if method == "put_batch": + return (await store.put_batch([event(run_id)]))[0] + if method == "put_if_absent": + return (await store.put_if_absent(**event(run_id)))[0] + return await store.put(**event(run_id)) + + +@pytest.mark.anyio +@pytest.mark.parametrize("method", ["put", "put_batch", "put_if_absent"]) +@pytest.mark.parametrize("reopen_before_delete", [False, True]) +@pytest.mark.parametrize("delete_all", [False, True]) +async def test_run_deletion_preserves_cursor_across_reopen(tmp_path, method, reopen_before_delete, delete_all): + store = JsonlRunEventStore(tmp_path) + await write(store, "r1", method) + last = await write(store, "r2", method) + assert last["seq"] == 2 + if reopen_before_delete: + store = JsonlRunEventStore(tmp_path) + assert await store.delete_by_run("t1", "r2") == 1 + if delete_all: + assert await store.delete_by_run("t1", "r1") == 1 + reopened = JsonlRunEventStore(tmp_path) + new = await write(reopened, "r3", method) + assert new["seq"] == 3 + assert [row["seq"] for row in await reopened.list_messages("t1", after_seq=last["seq"])] == [3] + + +@pytest.mark.anyio +async def test_thread_deletion_resets_watermark_after_all_runs_deleted(tmp_path): + store = JsonlRunEventStore(tmp_path) + await store.put(**event("r1")) + await store.delete_by_run("t1", "r1") + assert await JsonlRunEventStore(tmp_path).delete_by_thread("t1") == 0 + recreated = await JsonlRunEventStore(tmp_path).put(**event("r2")) + assert recreated["seq"] == 1 + + +@pytest.mark.anyio +async def test_missing_run_deletion_does_not_create_storage(tmp_path): + store = JsonlRunEventStore(tmp_path) + assert await store.delete_by_run("t1", "missing") == 0 + assert not (tmp_path / "threads").exists() + + +@pytest.mark.anyio +async def test_legacy_directory_and_surviving_higher_seq(tmp_path): + runs = tmp_path / "threads" / "t1" / "runs" + runs.mkdir(parents=True) + (runs / "legacy.jsonl").write_text('{"seq":9,"run_id":"legacy","category":"message"}\n', encoding="utf-8") + store = JsonlRunEventStore(tmp_path) + assert await store.delete_by_run("t1", "legacy") == 1 + assert (await store.put(**event("r1")))["seq"] == 10 + # The persisted floor is 9; recovery must also consider newer run records. + assert (await JsonlRunEventStore(tmp_path).put(**event("r2")))["seq"] == 11 + + +@pytest.mark.anyio +async def test_deleting_older_run_cannot_lower_watermark(tmp_path): + store = JsonlRunEventStore(tmp_path) + await store.put(**event("r1")) + await store.put(**event("r2")) + await store.delete_by_run("t1", "r2") + await JsonlRunEventStore(tmp_path).delete_by_run("t1", "r1") + assert (await JsonlRunEventStore(tmp_path).put(**event("r3")))["seq"] == 3 + + +@pytest.mark.anyio +@pytest.mark.parametrize("invalid", ["garbage", "-1"]) +async def test_invalid_watermark_fails_closed(tmp_path, invalid): + runs = tmp_path / "threads" / "t1" / "runs" + runs.mkdir(parents=True) + (runs / ".seq-watermark").write_text(invalid, encoding="utf-8") + with pytest.raises(ValueError): + await JsonlRunEventStore(tmp_path).put(**event("r1")) + assert not (runs / "r1.jsonl").exists() + + +@pytest.mark.anyio +async def test_failed_watermark_publish_keeps_previous_floor_and_run(tmp_path, monkeypatch): + store = JsonlRunEventStore(tmp_path) + await store.put(**event("r1")) + await store.delete_by_run("t1", "r1") + await store.put(**event("r2")) + replace = Path.replace + + def fail_publish(path, target): + if Path(target).name == ".seq-watermark": + raise OSError("synthetic watermark publication failure") + return replace(path, target) + + monkeypatch.setattr(Path, "replace", fail_publish) + with pytest.raises(OSError, match="publication failure"): + await store.delete_by_run("t1", "r2") + assert len(await store.list_events("t1", "r2")) == 1 + runs = tmp_path / "threads" / "t1" / "runs" + assert (runs / ".seq-watermark").read_text(encoding="utf-8") == "1" + assert not list(runs.glob(".seq-*.tmp")) + assert (await JsonlRunEventStore(tmp_path).put(**event("r3")))["seq"] == 3 + + +@pytest.mark.anyio +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits are not supported on Windows") +@pytest.mark.parametrize("run_mode", [0o600, 0o640, 0o644]) +@pytest.mark.parametrize("replace_existing", [False, True]) +async def test_watermark_matches_run_permissions_before_publication(tmp_path, monkeypatch, run_mode, replace_existing): + store = JsonlRunEventStore(tmp_path) + if replace_existing: + await store.put(**event("r0")) + await store.delete_by_run("t1", "r0") + last = await store.put(**event("r1")) + runs = tmp_path / "threads" / "t1" / "runs" + (runs / "r1.jsonl").chmod(run_mode) + replace = Path.replace + + def check_permissions(path, target): + if Path(target).name == ".seq-watermark": + assert path.stat().st_mode & 0o777 == run_mode + return replace(path, target) + + monkeypatch.setattr(Path, "replace", check_permissions) + assert await store.delete_by_run("t1", "r1") == 1 + watermark = runs / ".seq-watermark" + assert watermark.stat().st_mode & 0o777 == run_mode + assert watermark.read_text(encoding="utf-8") == str(last["seq"]) + assert not (runs / "r1.jsonl").exists() + assert (await JsonlRunEventStore(tmp_path).put(**event("r2")))["seq"] == last["seq"] + 1 + + +@pytest.mark.anyio +@pytest.mark.parametrize("replace_existing", [False, True]) +async def test_failed_watermark_permissions_keep_previous_floor_and_run(tmp_path, monkeypatch, replace_existing): + store = JsonlRunEventStore(tmp_path) + if replace_existing: + await store.put(**event("r1")) + await store.delete_by_run("t1", "r1") + last = await store.put(**event("r2")) + chmod = Path.chmod + + def fail_permissions(path, mode, **kwargs): + if path.name.startswith(".seq-") and path.suffix == ".tmp": + raise OSError("synthetic watermark permission failure") + return chmod(path, mode, **kwargs) + + monkeypatch.setattr(Path, "chmod", fail_permissions) + with pytest.raises(OSError, match="permission failure"): + await store.delete_by_run("t1", "r2") + assert len(await store.list_events("t1", "r2")) == 1 + runs = tmp_path / "threads" / "t1" / "runs" + watermark = runs / ".seq-watermark" + if replace_existing: + assert watermark.read_text(encoding="utf-8") == "1" + else: + assert not watermark.exists() + assert not list(runs.glob(".seq-*.tmp")) + assert (await JsonlRunEventStore(tmp_path).put(**event("r3")))["seq"] == last["seq"] + 1 + + +@pytest.mark.anyio +async def test_failed_run_unlink_keeps_safe_allocation_floor(tmp_path, monkeypatch): + store = JsonlRunEventStore(tmp_path) + await store.put(**event("r1")) + + def fail_delete(*args): + raise OSError("synthetic unlink failure") + + monkeypatch.setattr(store, "_delete_run_file", fail_delete) + with pytest.raises(OSError, match="unlink failure"): + await store.delete_by_run("t1", "r1") + assert len(await store.list_messages("t1")) == 1 + assert (await JsonlRunEventStore(tmp_path).put(**event("r2")))["seq"] == 2