Fix run-journal writer-lock cache leak on session delete
`_lock_for` caches one threading.Lock per (dir, file, pid) in the module-global
`_WRITER_LOCKS`, but nothing evicted those entries: `delete_run_journal` rmtree'd
the on-disk `_run_journal/{sid}/` directory yet left the cached lock objects
behind, so a long-lived gateway leaked one entry per deleted run forever.
Delete now evicts every cached lock whose parent directory matches the removed
session (pid-independent) under `_WRITER_LOCKS_GUARD`, leaving unrelated
sessions' locks intact.
Refs #4633, #2097.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
6ab95bffa4
commit
c5b15850e5
2 files changed
+65
-1
No files matched your search
+16
-1
@@ -285,7 +285,22 @@ def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> b
|
|||||||
if not session_journal_dir.exists():
|
if not session_journal_dir.exists():
|
||||||
return False
|
return False
|
||||||
shutil.rmtree(session_journal_dir, ignore_errors=True)
|
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:
|
def stale_interrupted_event(session_id: str, run_id: str, *, after_seq: int | None = None) -> dict | None:
|
||||||
|
|||||||
@@ -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
|
||||||
Reference in new issue
Block a user