Keep CLI cache rebuilds singleflight across invalidation
This commit is contained in:
Rod Boev
nesquena-hermes
parent
6a5a514e8c
commit
9130edbc31
2 changed files with 190 additions and 5 deletions
|
|
@ -49,7 +49,6 @@ _CLI_SESSIONS_CACHE_INFLIGHT: "dict[tuple, threading.Event]" = {}
|
|||
_CLI_SESSIONS_CACHE_INVALIDATION_VERSION = 0
|
||||
_CLI_SESSIONS_CACHE = {}
|
||||
# Event waits that keep stale rows visible while a rebuild is in flight.
|
||||
_CLI_SESSIONS_CACHE_WAIT_SECONDS = 0.25
|
||||
_CLI_SESSIONS_CACHE_STALE_WAIT_SECONDS = 0.10
|
||||
|
||||
# Per-file parse cache for Claude Code JSONL transcripts (#4718/#4662 phase 4).
|
||||
|
|
@ -4213,7 +4212,6 @@ def clear_cli_sessions_cache() -> None:
|
|||
global _CLI_SESSIONS_CACHE_INVALIDATION_VERSION
|
||||
_CLI_SESSIONS_CACHE_INVALIDATION_VERSION += 1
|
||||
_CLI_SESSIONS_CACHE.clear()
|
||||
_CLI_SESSIONS_CACHE_INFLIGHT.clear()
|
||||
# The sidecar-metadata projection cache is stat-keyed (self-invalidating on
|
||||
# any file change), but clear it alongside the CLI cache so an explicit
|
||||
# reset — a mutating sidebar action or test isolation — starts fully cold.
|
||||
|
|
@ -4249,6 +4247,61 @@ def _cli_sessions_cache_done(cache_key: tuple, event: threading.Event | None) ->
|
|||
event.set()
|
||||
|
||||
|
||||
def _reload_cli_sessions_after_inflight(
|
||||
*,
|
||||
cache_key: tuple,
|
||||
ttl: float,
|
||||
stale_sessions,
|
||||
stale_stamp,
|
||||
load_sessions,
|
||||
all_profiles: bool,
|
||||
db_path: str,
|
||||
) -> list:
|
||||
event, is_owner = _cli_sessions_cache_claim_rebuild(cache_key)
|
||||
if not is_owner:
|
||||
try:
|
||||
event.wait()
|
||||
except Exception:
|
||||
pass
|
||||
with _CLI_SESSIONS_CACHE_LOCK:
|
||||
cached_entry = _CLI_SESSIONS_CACHE.get(cache_key)
|
||||
if cached_entry is not None:
|
||||
if len(cached_entry) == 3:
|
||||
cached_expires_at, cached_stamp, cached_sessions = cached_entry
|
||||
else:
|
||||
cached_expires_at, cached_sessions = cached_entry
|
||||
cached_stamp = _CLI_SESSIONS_CACHE_INVALIDATION_VERSION
|
||||
if (
|
||||
cached_stamp == _CLI_SESSIONS_CACHE_INVALIDATION_VERSION
|
||||
and cached_expires_at > time.monotonic()
|
||||
):
|
||||
return _copy_cli_sessions(cached_sessions)
|
||||
if stale_sessions is not None and stale_stamp == _cli_sessions_cache_invalidation_stamp():
|
||||
return stale_sessions
|
||||
return []
|
||||
try:
|
||||
try:
|
||||
sessions = load_sessions()
|
||||
except Exception as _cli_err:
|
||||
logger.warning(
|
||||
"get_cli_sessions() failed — check state.db schema or path (%s): %s",
|
||||
"all profiles" if all_profiles else db_path, _cli_err,
|
||||
)
|
||||
if stale_sessions is not None and stale_stamp == _cli_sessions_cache_invalidation_stamp():
|
||||
return stale_sessions
|
||||
return []
|
||||
if _cli_sessions_cache_invalidation_stamp() == fallback_invalidation_stamp:
|
||||
with _CLI_SESSIONS_CACHE_LOCK:
|
||||
_CLI_SESSIONS_CACHE[cache_key] = (
|
||||
time.monotonic() + ttl,
|
||||
_CLI_SESSIONS_CACHE_INVALIDATION_VERSION,
|
||||
_copy_cli_sessions(sessions),
|
||||
)
|
||||
return _copy_cli_sessions(sessions)
|
||||
finally:
|
||||
_cli_sessions_cache_done(cache_key, event)
|
||||
|
||||
|
||||
def _cli_sessions_cache_ttl_seconds() -> float:
|
||||
# #4842: widen the freshness window while a turn is streaming so the fixed
|
||||
# ~5s streaming poll cadence doesn't force a rebuild on every poll. Paired
|
||||
|
|
@ -4876,12 +4929,11 @@ def get_cli_sessions(source_filter=None, *, all_profiles: bool = False) -> list:
|
|||
finally:
|
||||
_cli_sessions_cache_done(cache_key, event)
|
||||
try:
|
||||
timeout = (
|
||||
event.wait(
|
||||
_CLI_SESSIONS_CACHE_STALE_WAIT_SECONDS
|
||||
if stale_sessions is not None
|
||||
else _CLI_SESSIONS_CACHE_WAIT_SECONDS
|
||||
else None
|
||||
)
|
||||
event.wait(timeout)
|
||||
except Exception:
|
||||
pass
|
||||
with _CLI_SESSIONS_CACHE_LOCK:
|
||||
|
|
@ -4899,6 +4951,15 @@ def get_cli_sessions(source_filter=None, *, all_profiles: bool = False) -> list:
|
|||
return _copy_cli_sessions(cached_sessions)
|
||||
if stale_sessions is not None and stale_stamp == _cli_sessions_cache_invalidation_stamp():
|
||||
return stale_sessions
|
||||
return _reload_cli_sessions_after_inflight(
|
||||
cache_key=cache_key,
|
||||
ttl=ttl,
|
||||
stale_sessions=stale_sessions,
|
||||
stale_stamp=stale_stamp,
|
||||
load_sessions=_load_sessions,
|
||||
all_profiles=all_profiles,
|
||||
db_path=db_path,
|
||||
)
|
||||
fallback_invalidation_stamp = _cli_sessions_cache_invalidation_stamp()
|
||||
try:
|
||||
sessions = _load_sessions()
|
||||
|
|
|
|||
|
|
@ -100,6 +100,59 @@ def test_get_cli_sessions_follower_reuses_stale_rows_during_slow_rebuild(monkeyp
|
|||
assert results.get("owner") == [{"session_id": "fresh", "title": "fresh-row"}]
|
||||
|
||||
|
||||
def test_get_cli_sessions_cold_followers_join_single_rebuild(monkeypatch, tmp_path):
|
||||
hermes_home = tmp_path / "hermes"
|
||||
hermes_home.mkdir()
|
||||
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: str(hermes_home))
|
||||
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default")
|
||||
models.clear_cli_sessions_cache()
|
||||
monkeypatch.setattr(models, "_CLI_SESSIONS_CACHE_TTL_SECONDS", 60.0, raising=False)
|
||||
|
||||
owner_started = threading.Event()
|
||||
owner_block = threading.Event()
|
||||
load_count = {"value": 0}
|
||||
load_count_lock = threading.Lock()
|
||||
results = {}
|
||||
|
||||
def _blocking_loader(*_args, **_kwargs):
|
||||
with load_count_lock:
|
||||
load_count["value"] += 1
|
||||
call_number = load_count["value"]
|
||||
if call_number == 1:
|
||||
owner_started.set()
|
||||
owner_block.wait()
|
||||
return [{"session_id": "fresh", "title": "fresh-row"}]
|
||||
|
||||
monkeypatch.setattr(models, "_load_cli_sessions_uncached", _blocking_loader)
|
||||
|
||||
def _owner():
|
||||
results["owner"] = models.get_cli_sessions()
|
||||
|
||||
def _follower():
|
||||
results["follower"] = models.get_cli_sessions()
|
||||
|
||||
owner = threading.Thread(target=_owner, daemon=True)
|
||||
follower = threading.Thread(target=_follower, daemon=True)
|
||||
|
||||
owner.start()
|
||||
assert owner_started.wait(1.0), "owner did not start"
|
||||
follower.start()
|
||||
time.sleep(0.2)
|
||||
|
||||
assert load_count["value"] == 1
|
||||
assert follower.is_alive()
|
||||
|
||||
owner_block.set()
|
||||
owner.join(2.0)
|
||||
follower.join(2.0)
|
||||
|
||||
assert not owner.is_alive()
|
||||
assert not follower.is_alive()
|
||||
assert load_count["value"] == 1
|
||||
assert results.get("owner") == [{"session_id": "fresh", "title": "fresh-row"}]
|
||||
assert results.get("follower") == [{"session_id": "fresh", "title": "fresh-row"}]
|
||||
|
||||
|
||||
def test_get_cli_sessions_clear_during_rebuild_does_not_restore_stale_rows(monkeypatch, tmp_path):
|
||||
hermes_home = tmp_path / "hermes"
|
||||
hermes_home.mkdir()
|
||||
|
|
@ -142,3 +195,74 @@ def test_get_cli_sessions_clear_during_rebuild_does_not_restore_stale_rows(monke
|
|||
assert recovered == [{"session_id": "recovered", "title": "recovered-row"}]
|
||||
with models._CLI_SESSIONS_CACHE_LOCK:
|
||||
assert cache_key in models._CLI_SESSIONS_CACHE
|
||||
|
||||
|
||||
def test_get_cli_sessions_clear_during_rebuild_preserves_joiners(monkeypatch, tmp_path):
|
||||
hermes_home = tmp_path / "hermes"
|
||||
hermes_home.mkdir()
|
||||
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: str(hermes_home))
|
||||
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default")
|
||||
models.clear_cli_sessions_cache()
|
||||
monkeypatch.setattr(models, "_CLI_SESSIONS_CACHE_TTL_SECONDS", 60.0, raising=False)
|
||||
|
||||
owner_started = threading.Event()
|
||||
owner_block = threading.Event()
|
||||
second_started = threading.Event()
|
||||
second_block = threading.Event()
|
||||
load_count = {"value": 0}
|
||||
load_count_lock = threading.Lock()
|
||||
results = {}
|
||||
|
||||
def _blocking_loader(*_args, **_kwargs):
|
||||
with load_count_lock:
|
||||
load_count["value"] += 1
|
||||
call_number = load_count["value"]
|
||||
if call_number == 1:
|
||||
owner_started.set()
|
||||
owner_block.wait()
|
||||
return [{"session_id": "owner", "title": "owner-row"}]
|
||||
second_started.set()
|
||||
second_block.wait()
|
||||
return [{"session_id": "fresh", "title": "fresh-row"}]
|
||||
|
||||
monkeypatch.setattr(models, "_load_cli_sessions_uncached", _blocking_loader)
|
||||
|
||||
def _owner():
|
||||
results["owner"] = models.get_cli_sessions()
|
||||
|
||||
def _follower(name):
|
||||
results[name] = models.get_cli_sessions()
|
||||
|
||||
owner = threading.Thread(target=_owner, daemon=True)
|
||||
follower_a = threading.Thread(target=lambda: _follower("follower_a"), daemon=True)
|
||||
follower_b = threading.Thread(target=lambda: _follower("follower_b"), daemon=True)
|
||||
|
||||
owner.start()
|
||||
assert owner_started.wait(1.0), "owner did not enter rebuild"
|
||||
follower_a.start()
|
||||
models.clear_cli_sessions_cache()
|
||||
follower_b.start()
|
||||
time.sleep(0.2)
|
||||
|
||||
assert load_count["value"] == 1
|
||||
assert follower_a.is_alive()
|
||||
assert follower_b.is_alive()
|
||||
|
||||
owner_block.set()
|
||||
assert second_started.wait(1.0), "second rebuild did not start"
|
||||
time.sleep(0.2)
|
||||
|
||||
assert load_count["value"] == 2
|
||||
second_block.set()
|
||||
|
||||
owner.join(2.0)
|
||||
follower_a.join(2.0)
|
||||
follower_b.join(2.0)
|
||||
|
||||
assert not owner.is_alive()
|
||||
assert not follower_a.is_alive()
|
||||
assert not follower_b.is_alive()
|
||||
assert load_count["value"] == 2
|
||||
assert results.get("owner") == [{"session_id": "owner", "title": "owner-row"}]
|
||||
assert results.get("follower_a") == [{"session_id": "fresh", "title": "fresh-row"}]
|
||||
assert results.get("follower_b") == [{"session_id": "fresh", "title": "fresh-row"}]
|
||||
|
|
|
|||
Loading…
Reference in a new issue