diff --git a/api/run_journal.py b/api/run_journal.py index 1853de05..5098a133 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -285,7 +285,22 @@ def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> b if not session_journal_dir.exists(): return False shutil.rmtree(session_journal_dir, ignore_errors=True) - return not session_journal_dir.exists() + removed = not session_journal_dir.exists() + # Evict any writer locks the removed runs left behind. `_lock_for` keys are + # ``(str(path.parent), path.name, pid)`` and every run file for this session + # lives directly under ``session_journal_dir``, so drop all keys whose parent + # dir matches — pid-independent — to keep `_WRITER_LOCKS` from growing forever. + # Guard on confirmed removal: `rmtree(ignore_errors=True)` can silently leave + # the directory (locked files on Windows, permission transients). If the files + # still exist their locks are still live — evicting them would hand a later + # `_lock_for` caller a brand-new Lock, breaking mutual exclusion with a writer + # still holding the old one. + if removed: + dir_key = str(session_journal_dir) + with _WRITER_LOCKS_GUARD: + for key in [k for k in _WRITER_LOCKS if k[0] == dir_key]: + del _WRITER_LOCKS[key] + return removed def stale_interrupted_event(session_id: str, run_id: str, *, after_seq: int | None = None) -> dict | None: diff --git a/tests/test_run_journal_writer_lock_evict.py b/tests/test_run_journal_writer_lock_evict.py new file mode 100644 index 00000000..3ef57ce9 --- /dev/null +++ b/tests/test_run_journal_writer_lock_evict.py @@ -0,0 +1,49 @@ +"""Regression tests for #4633/#2097 — deleting a run journal must evict its writer lock. + +`_lock_for` lazily creates a `threading.Lock` per ``(dir, file, pid)`` key and +caches it in the module-global `_WRITER_LOCKS`, but nothing ever removed those +entries: `delete_run_journal` rmtree'd the on-disk directory yet left the lock +objects behind, so a long-lived gateway leaked one entry per deleted run +forever. These tests pin that delete now drops the matching cache entries while +leaving unrelated sessions' locks intact. +""" +import api.run_journal as run_journal +from api.run_journal import RunJournalWriter, delete_run_journal + + +def _keys_for(session_dir, sid): + dir_key = str(session_dir / run_journal.RUN_JOURNAL_DIR_NAME / sid) + return [k for k in run_journal._WRITER_LOCKS if k[0] == dir_key] + + +def test_delete_run_journal_evicts_writer_locks(tmp_path): + writer = RunJournalWriter("sid-del", "run-1", session_dir=tmp_path) + writer.append_sse_event("token", {"text": "hello"}) + # Constructing the writer + appending populated the per-run lock cache. + assert _keys_for(tmp_path, "sid-del") + + assert delete_run_journal("sid-del", session_dir=tmp_path) is True + + # The cached lock(s) for the deleted session are gone — no unbounded climb. + assert _keys_for(tmp_path, "sid-del") == [] + + +def test_delete_run_journal_keeps_other_sessions_locks(tmp_path): + RunJournalWriter("sid-keep", "run-k", session_dir=tmp_path).append_sse_event("token", {"text": "k"}) + RunJournalWriter("sid-del", "run-d", session_dir=tmp_path).append_sse_event("token", {"text": "d"}) + + delete_run_journal("sid-del", session_dir=tmp_path) + + assert _keys_for(tmp_path, "sid-del") == [] + assert _keys_for(tmp_path, "sid-keep"), "unrelated session's lock must survive" + + +def test_delete_run_journal_noop_leaves_cache_untouched(tmp_path): + RunJournalWriter("sid-keep", "run-k", session_dir=tmp_path).append_sse_event("token", {"text": "k"}) + before = _keys_for(tmp_path, "sid-keep") + + # Missing/invalid ids never touch the cache. + assert delete_run_journal("nope", session_dir=tmp_path) is False + assert delete_run_journal("../escape", session_dir=tmp_path) is False + + assert _keys_for(tmp_path, "sid-keep") == before