* perf(run-journal): cache unchanged run summaries * fix(run-journal): reject cache after missing-file race * run-journal cache: add st_ctime_ns to signature (close same-size mtime-preserving rewrite window) + regression test [maintainer fix on @sjungwon03 #6291] * docs(changelog): stamp #6291 run-journal summary cache --------- Co-authored-by: sjungwon03 <sjungwon03@gmail.com> Co-authored-by: nesquena-hermes <agent@nesquena-hermes>
This commit is contained in:
3 files changed
+209
-6
No files matched your search
+92
-6
@@ -10,6 +10,7 @@ import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from collections import OrderedDict
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
@@ -27,6 +28,14 @@ _WRITER_LOCKS_GUARD = threading.Lock()
|
||||
# ``_reserve_next_seq`` and ``delete_run_journal`` (which evicts stale entries).
|
||||
_SEQ_CACHE: dict[str, int] = {}
|
||||
_SEQ_CACHE_LOCK = threading.Lock()
|
||||
# Summary callers only need terminal state and the latest cursor. Re-parsing a
|
||||
# completed journal's full payload (which can include multi-megabyte tool or
|
||||
# session results) on every status/reconnect probe is needless. This process
|
||||
# cache is keyed by a complete stat identity, so it is never used after an
|
||||
# atomic replacement, append, truncate, or same-path file recreation.
|
||||
_SUMMARY_CACHE_MAX_ENTRIES = 128
|
||||
_SUMMARY_CACHE: OrderedDict[str, tuple[tuple[int, int, int, int, int], dict]] = OrderedDict()
|
||||
_SUMMARY_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"
|
||||
@@ -71,6 +80,68 @@ def _lock_for(path: Path) -> threading.Lock:
|
||||
return lock
|
||||
|
||||
|
||||
def _summary_cache_signature(path: Path) -> tuple[int, int, int, int, int] | None:
|
||||
"""Return the complete filesystem identity used for summary-cache validity.
|
||||
|
||||
Includes ``st_ctime_ns`` so a same-inode, same-size rewrite that restores the
|
||||
original ``mtime_ns`` (e.g. an atomic replace) still invalidates the cache —
|
||||
ctime advances on any metadata/content change and cannot be forged back.
|
||||
"""
|
||||
try:
|
||||
stat = path.stat()
|
||||
except OSError:
|
||||
return None
|
||||
return (
|
||||
int(stat.st_dev),
|
||||
int(stat.st_ino),
|
||||
int(stat.st_size),
|
||||
int(stat.st_mtime_ns),
|
||||
int(stat.st_ctime_ns),
|
||||
)
|
||||
|
||||
|
||||
def _get_cached_summary(path: Path) -> dict | None:
|
||||
signature = _summary_cache_signature(path)
|
||||
if signature is None:
|
||||
return None
|
||||
key = str(path)
|
||||
with _SUMMARY_CACHE_LOCK:
|
||||
cached = _SUMMARY_CACHE.get(key)
|
||||
if cached is None:
|
||||
return None
|
||||
cached_signature, summary = cached
|
||||
if cached_signature != signature:
|
||||
_SUMMARY_CACHE.pop(key, None)
|
||||
return None
|
||||
_SUMMARY_CACHE.move_to_end(key)
|
||||
return dict(summary)
|
||||
|
||||
|
||||
def _cache_summary(
|
||||
path: Path,
|
||||
summary: dict,
|
||||
*,
|
||||
expected_signature: tuple[int, int, int, int, int] | None = None,
|
||||
) -> None:
|
||||
signature = _summary_cache_signature(path)
|
||||
# The pre-read signature is an enforced TOCTOU precondition. In particular,
|
||||
# a journal created after a missing-file read has ``None -> signature`` and
|
||||
# must not cache the empty/unknown result under the new file's identity.
|
||||
if signature is None or signature != expected_signature:
|
||||
return
|
||||
key = str(path)
|
||||
with _SUMMARY_CACHE_LOCK:
|
||||
_SUMMARY_CACHE[key] = (signature, dict(summary))
|
||||
_SUMMARY_CACHE.move_to_end(key)
|
||||
while len(_SUMMARY_CACHE) > _SUMMARY_CACHE_MAX_ENTRIES:
|
||||
_SUMMARY_CACHE.popitem(last=False)
|
||||
|
||||
|
||||
def _discard_cached_summary(path: Path) -> None:
|
||||
with _SUMMARY_CACHE_LOCK:
|
||||
_SUMMARY_CACHE.pop(str(path), None)
|
||||
|
||||
|
||||
def _read_jsonl(path: Path) -> tuple[list[dict], list[dict]]:
|
||||
events: list[dict] = []
|
||||
malformed: list[dict] = []
|
||||
@@ -351,6 +422,7 @@ def append_run_event(
|
||||
fh.flush()
|
||||
if _should_fsync_event(terminal_state):
|
||||
os.fsync(fh.fileno())
|
||||
_discard_cached_summary(path)
|
||||
if created_file:
|
||||
_fsync_parent_dir(path)
|
||||
return event
|
||||
@@ -427,8 +499,15 @@ def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -
|
||||
|
||||
|
||||
def latest_run_summary(session_id: str, run_id: str, *, session_dir: Path | None = None) -> dict:
|
||||
journal = read_run_events(session_id, run_id, session_dir=session_dir)
|
||||
return _summary_from_events(session_id, run_id, journal.get("events") or [])
|
||||
path = _run_path(session_id, run_id, session_dir=session_dir)
|
||||
cached = _get_cached_summary(path)
|
||||
if cached is not None:
|
||||
return cached
|
||||
pre_read_signature = _summary_cache_signature(path)
|
||||
events, _malformed = _read_jsonl(path)
|
||||
summary = _summary_from_events(session_id, run_id, events)
|
||||
_cache_summary(path, summary, expected_signature=pre_read_signature)
|
||||
return summary
|
||||
|
||||
|
||||
def session_journal_fingerprint(session_id: str, *, session_dir: Path | None = None) -> tuple[int, float, int]:
|
||||
@@ -471,8 +550,12 @@ def find_run_summary(run_id: str, *, session_dir: Path | None = None) -> dict |
|
||||
journal_root = root / RUN_JOURNAL_DIR_NAME
|
||||
for path in journal_root.glob(f"*/{rid}.jsonl"):
|
||||
session_id = path.parent.name
|
||||
events, _malformed = _read_jsonl(path)
|
||||
summary = _summary_from_events(session_id, rid, events)
|
||||
summary = _get_cached_summary(path)
|
||||
if summary is None:
|
||||
pre_read_signature = _summary_cache_signature(path)
|
||||
events, _malformed = _read_jsonl(path)
|
||||
summary = _summary_from_events(session_id, rid, events)
|
||||
_cache_summary(path, summary, expected_signature=pre_read_signature)
|
||||
summary["path"] = str(path)
|
||||
return summary
|
||||
return None
|
||||
@@ -640,8 +723,11 @@ def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> b
|
||||
# ``_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]
|
||||
for cache_key in [entry for entry in _SEQ_CACHE if str(Path(entry).parent) == dir_key]:
|
||||
del _SEQ_CACHE[cache_key]
|
||||
with _SUMMARY_CACHE_LOCK:
|
||||
for cache_key in [entry for entry in _SUMMARY_CACHE if str(Path(entry).parent) == dir_key]:
|
||||
del _SUMMARY_CACHE[cache_key]
|
||||
return removed
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user