From 0ac2b3a3fab1e3a6f7e3174ef2f981496eb42910 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Wed, 8 Jul 2026 23:39:16 +0200 Subject: [PATCH] Cache next-seq per run journal path to avoid O(n^2) appends MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit append_run_event() without an explicit seq called _next_seq(path), which re-read and re-parsed the entire journal file on every append — O(n) per append, O(n^2) over a run. Direct callers therefore paid a full file reparse per event; only RunJournalWriter avoided it via its own in-memory counter. Introduce a module-level _SEQ_CACHE keyed by run-journal file path. The per-path _lock_for(path) serializes same-path reserve->append so seqs stay monotonic and file order matches; both RunJournalWriter and the free append_run_event now draw from this one cache (_reserve_next_seq), so seqs stay gapless even when both write the same path — the writer's private counter is removed to keep a single source of truth. An explicit caller-supplied seq advances the cache (_note_assigned_seq) so it can never be re-issued. delete_run_journal() now also evicts cache entries for the removed session so a run re-created at the same path restarts numbering at 1. Guard every structural _SEQ_CACHE access (reserve/note/evict) with a dedicated _SEQ_CACHE_LOCK. The eviction sweep in delete_run_journal iterates the whole dict; the per-path locks the append path holds do NOT serialize against it, so a concurrent append on another session inserted a fresh key mid-iteration and raised "dictionary changed size during iteration" (the caller swallows it, but the eviction then aborts and leaks stale seqs, so a re-created run resumed a stale number instead of restarting at 1). A stress test races deletes against cross-session appends and asserts no such error. Rollback: revert this commit; _next_seq(path) re-read behaviour returns. No on-disk format change, so existing journals stay compatible. Co-Authored-By: Claude Opus 4.8 (1M context) --- api/run_journal.py | 77 +++++++++++- tests/test_run_journal_seq_cache.py | 175 ++++++++++++++++++++++++++++ 2 files changed, 247 insertions(+), 5 deletions(-) create mode 100644 tests/test_run_journal_seq_cache.py diff --git a/api/run_journal.py b/api/run_journal.py index 5098a133..81e92e75 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -17,6 +17,16 @@ RUN_JOURNAL_DIR_NAME = "_run_journal" _SAFE_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+$") _WRITER_LOCKS: dict[tuple[str, str, str], threading.Lock] = {} _WRITER_LOCKS_GUARD = threading.Lock() +# Next-seq to assign per run-journal file path, kept in memory so repeat appends +# to the same run do not re-parse the whole file on every call. The per-path +# ``_lock_for(path)`` serializes same-path reserve→append so seqs stay monotonic +# and file order matches; ``_SEQ_CACHE_LOCK`` (below) additionally guards every +# *structural* access to the dict (reserve/note/evict) so ``delete_run_journal`` +# can iterate + drop keys while a concurrent append on ANOTHER path inserts one, +# without a ``dictionary changed size during iteration`` crash. See +# ``_reserve_next_seq`` and ``delete_run_journal`` (which evicts stale entries). +_SEQ_CACHE: dict[str, int] = {} +_SEQ_CACHE_LOCK = threading.Lock() _TERMINAL_SSE_EVENTS = {"done", "cancel", "apperror", "error", "stream_end"} _FSYNC_MODE_ENV = "HERMES_WEBUI_RUN_JOURNAL_FSYNC" _FSYNC_MODE_EAGER = "eager" @@ -81,6 +91,49 @@ def _next_seq(path: Path) -> int: return (max(seqs) + 1) if seqs else 1 +def _reserve_next_seq(path: Path) -> int: + """Reserve and return the next seq for ``path``, advancing the in-memory cache. + + Callers MUST hold ``_lock_for(path)``. The first append per path in this + process seeds the cache from ``_next_seq(path)`` (one file read); every later + append is a pure in-memory increment, avoiding the O(n) re-parse that + re-reading the whole journal on every append caused (O(n^2) over a run). + Because ``RunJournalWriter`` and the free ``append_run_event`` share this one + cache under the same per-path lock, their seqs stay monotonic and gapless + even when both write the same path. ``_SEQ_CACHE_LOCK`` additionally makes the + dict get+set atomic against a concurrent cross-path eviction. + """ + key = str(path) + with _SEQ_CACHE_LOCK: + nxt = _SEQ_CACHE.get(key) + if nxt is not None: + _SEQ_CACHE[key] = nxt + 1 + return nxt + # Cache miss: seed from disk WITHOUT holding the module-global lock, so a + # slow first-access file read for one path can't block every other path's + # cache ops. The caller holds the per-path lock, so only one thread per path + # can reach this branch — no double-seed, and no same-path writer can race + # the value in between. + seeded = _next_seq(path) + with _SEQ_CACHE_LOCK: + _SEQ_CACHE[key] = seeded + 1 + return seeded + + +def _note_assigned_seq(path: Path, seq: int) -> None: + """Keep the cache at least one past an explicitly-supplied ``seq``. + + Callers MUST hold ``_lock_for(path)``. When an append carries a caller-chosen + ``seq`` rather than drawing from the cache, advance the cache so a later + cache-based append on the same path cannot re-issue an already-used seq. + """ + key = str(path) + nxt = int(seq) + 1 + with _SEQ_CACHE_LOCK: + if _SEQ_CACHE.get(key, 0) < nxt: + _SEQ_CACHE[key] = nxt + + def _terminal_state_for_event(event_name: str, payload) -> str | None: name = str(event_name or "") if name == "done" or name == "stream_end": @@ -145,7 +198,11 @@ def append_run_event( if not event_name: raise ValueError("event_name is required") with _lock_for(path): - assigned_seq = int(seq) if seq is not None else _next_seq(path) + if seq is not None: + assigned_seq = int(seq) + _note_assigned_seq(path, assigned_seq) + else: + assigned_seq = _reserve_next_seq(path) terminal_state = _terminal_state_for_event(event_name, payload) event = { "version": 1, @@ -183,13 +240,13 @@ class RunJournalWriter: self.session_dir = Path(session_dir) if session_dir is not None else None self._path = _run_path(self.session_id, self.run_id, session_dir=self.session_dir) self._lock = _lock_for(self._path) - with self._lock: - self._next_seq = _next_seq(self._path) def append_sse_event(self, event_name: str, payload=None) -> dict: + # Draw from the shared module-level seq cache under the per-path lock so + # this writer and any direct append_run_event() call on the same path + # agree on one monotonic, gapless sequence. with self._lock: - seq = self._next_seq - self._next_seq += 1 + seq = _reserve_next_seq(self._path) return append_run_event( self.session_id, self.run_id, @@ -300,6 +357,16 @@ def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> b with _WRITER_LOCKS_GUARD: for key in [k for k in _WRITER_LOCKS if k[0] == dir_key]: del _WRITER_LOCKS[key] + # Drop cached next-seq entries for the removed runs too. Every run file + # for this session lives directly under ``session_journal_dir``, so its + # cache key's parent dir matches. Without this, a run re-created at the + # same path would resume the stale cached seq instead of restarting at 1. + # Hold ``_SEQ_CACHE_LOCK`` — the SAME mutex ``_reserve_next_seq``/ + # ``_note_assigned_seq`` take — so a concurrent append on another path + # cannot mutate the dict mid-iteration (``dictionary changed size``). + with _SEQ_CACHE_LOCK: + for key in [k for k in _SEQ_CACHE if str(Path(k).parent) == dir_key]: + del _SEQ_CACHE[key] return removed diff --git a/tests/test_run_journal_seq_cache.py b/tests/test_run_journal_seq_cache.py new file mode 100644 index 00000000..23a18fb3 --- /dev/null +++ b/tests/test_run_journal_seq_cache.py @@ -0,0 +1,175 @@ +"""Regression tests for the in-memory run-journal seq cache. + +Repeat ``append_run_event`` calls used to re-read (and re-parse) the entire +journal file on every append via ``_next_seq``, which is O(n) per append and +O(n^2) over a run. The cache seeds once per path and then increments in memory +while staying consistent with ``RunJournalWriter`` (both share one cache under +the same per-path lock). +""" +import threading + +import pytest # noqa: F401 # top-level import keeps pytest collection unambiguous + +from api import run_journal + + +def test_append_run_event_seeds_seq_once_and_stays_gapless(tmp_path, monkeypatch): + calls = {"next_seq": 0, "read_jsonl": 0} + real_next_seq = run_journal._next_seq + real_read_jsonl = run_journal._read_jsonl + + def counting_next_seq(path): + calls["next_seq"] += 1 + return real_next_seq(path) + + def counting_read_jsonl(path): + calls["read_jsonl"] += 1 + return real_read_jsonl(path) + + monkeypatch.setattr(run_journal, "_next_seq", counting_next_seq) + monkeypatch.setattr(run_journal, "_read_jsonl", counting_read_jsonl) + + n = 25 + seqs = [ + run_journal.append_run_event( + "sess_cache", "run_cache", "token", {"text": str(i)}, session_dir=tmp_path + )["seq"] + for i in range(n) + ] + + assert seqs == list(range(1, n + 1)) + # Seeded from the file exactly once; every later append is in-memory only. + assert calls["next_seq"] == 1 + assert calls["read_jsonl"] <= 1 + + +def test_writer_and_free_function_share_one_gapless_sequence(tmp_path): + writer = run_journal.RunJournalWriter("sess_shared", "run_shared", session_dir=tmp_path) + a = writer.append_sse_event("token", {"text": "a"}) + b = run_journal.append_run_event( + "sess_shared", "run_shared", "token", {"text": "b"}, session_dir=tmp_path + ) + c = writer.append_sse_event("token", {"text": "c"}) + d = run_journal.append_run_event( + "sess_shared", "run_shared", "done", {"session": {}}, session_dir=tmp_path + ) + + assert [a["seq"], b["seq"], c["seq"], d["seq"]] == [1, 2, 3, 4] + + journal = run_journal.read_run_events("sess_shared", "run_shared", session_dir=tmp_path) + file_seqs = sorted(event["seq"] for event in journal["events"]) + assert file_seqs == [1, 2, 3, 4] + + +def test_explicit_seq_keeps_cache_from_reissuing(tmp_path): + # A caller-supplied seq must push the cache forward so a later cache append + # does not collide with it. + run_journal.append_run_event( + "sess_expl", "run_expl", "token", {"text": "x"}, session_dir=tmp_path, seq=5 + ) + nxt = run_journal.append_run_event( + "sess_expl", "run_expl", "token", {"text": "y"}, session_dir=tmp_path + ) + assert nxt["seq"] == 6 + + +def test_delete_evicts_seq_cache_so_recreated_run_restarts(tmp_path): + run_journal.append_run_event( + "sess_del", "run_del", "token", {"text": "one"}, session_dir=tmp_path + ) + run_journal.append_run_event( + "sess_del", "run_del", "token", {"text": "two"}, session_dir=tmp_path + ) + + assert run_journal.delete_run_journal("sess_del", session_dir=tmp_path) is True + + restarted = run_journal.append_run_event( + "sess_del", "run_del", "token", {"text": "fresh"}, session_dir=tmp_path + ) + assert restarted["seq"] == 1 + + +def test_delete_evicts_seq_cache_concurrently_without_crash(tmp_path): + """delete_run_journal must evict _SEQ_CACHE under a shared lock. + + The eviction iterates the whole ``_SEQ_CACHE`` to drop the deleted session's + keys. It ran outside any mutex the append path holds, so a concurrent append + on ANOTHER session — which inserts a fresh key — mutated the dict mid-iteration + and raised ``RuntimeError: dictionary changed size during iteration``. Both + paths now take ``_SEQ_CACHE_LOCK``, so the eviction and inserts serialize. + """ + # Seed the cache with many keys so an eviction sweep iterates a wide dict, + # widening the window for a concurrent insert to collide. + for s in range(60): + run_journal.append_run_event( + f"sess_seed{s}", "run", "token", {"text": "x"}, session_dir=tmp_path + ) + + errors: list[BaseException] = [] + errors_lock = threading.Lock() + stop = threading.Event() + + def deleter(): + i = 0 + while not stop.is_set(): + sid = f"sess_del{i}" + try: + run_journal.append_run_event( + sid, "run", "token", {"text": "d"}, session_dir=tmp_path + ) + run_journal.delete_run_journal(sid, session_dir=tmp_path) + except BaseException as exc: # noqa: BLE001 - recorded for the assert + with errors_lock: + errors.append(exc) + i += 1 + + def inserter(base): + i = 0 + while not stop.is_set(): + try: + # Each append to a brand-new session inserts a fresh cache key, + # racing the deleter's eviction comprehension. + run_journal.append_run_event( + f"sess_ins{base}_{i}", "run", "token", {"text": "i"}, + session_dir=tmp_path, + ) + except BaseException as exc: # noqa: BLE001 - recorded for the assert + with errors_lock: + errors.append(exc) + i += 1 + + workers = [threading.Thread(target=deleter)] + workers += [threading.Thread(target=inserter, args=(b,)) for b in range(4)] + for w in workers: + w.start() + # Let them contend briefly, then wind down. + for _ in range(200): + run_journal.delete_run_journal("sess_seed0", session_dir=tmp_path) + stop.set() + for w in workers: + w.join(timeout=10.0) + + assert not any(w.is_alive() for w in workers), "worker threads did not finish" + assert not errors, f"eviction raced an insert: {errors[:3]}" + + +def test_concurrent_appends_produce_unique_gapless_seqs(tmp_path): + threads = [] + results: list[int] = [] + results_lock = threading.Lock() + + def worker(i): + event = run_journal.append_run_event( + "sess_conc", "run_conc", "token", {"text": str(i)}, session_dir=tmp_path + ) + with results_lock: + results.append(event["seq"]) + + for i in range(40): + threads.append(threading.Thread(target=worker, args=(i,))) + for t in threads: + t.start() + for t in threads: + t.join() + + assert sorted(results) == list(range(1, 41))