diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d8ddd752..7eb4e92d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,10 @@ ## [Unreleased] +### Fixed + +- **Streaming-time session index updates now hold the global session lock for less work.** `_write_session_index()` now snapshots in-memory session rows under `LOCK`, but reads/parses the existing `_index.json`, sorts, serializes, and writes the updated index outside that lock. This reduces avoidable `/api/session` and `/api/sessions` contention while a running stream saves session metadata, addressing the lock-window half of #4918 without changing the active-stream lifecycle. + ## [v0.51.653] — 2026-06-25 — Release XI (gateway approval failures stay actionable instead of dead-ending) ### Fixed diff --git a/api/models.py b/api/models.py index c03ab0581..fd04296b5 100644 --- a/api/models.py +++ b/api/models.py @@ -260,8 +260,9 @@ def _write_session_index(updates=None, *, session_dir: Path | None = None, sessi the existing index — O(1) for single-session changes. When *updates* is None, a full rebuild is performed (used on startup / first call). - LOCK protects in-memory state snapshots and payload construction only; - disk I/O (write/flush/fsync/replace) always runs outside LOCK. + LOCK protects only in-memory session snapshots. JSON parsing, payload + construction, and disk I/O run outside LOCK so active-stream saves do not + block ordinary session reads longer than necessary. """ session_dir = session_dir or SESSION_DIR session_index_file = session_index_file or SESSION_INDEX_FILE @@ -293,13 +294,16 @@ def _write_session_index(updates=None, *, session_dir: Path | None = None, sessi logger.debug("Failed to load session from %s", p) entries = list(entry_map.values()) + existing_ids = set(entry_map.keys()) with LOCK: - existing_ids = set(entry_map.keys()) - for s in SESSIONS.values(): - if s.session_id not in existing_ids: - entries.append(s.compact()) - entries.sort(key=lambda s: s.get('updated_at', 0), reverse=True) - _payload = json.dumps(entries, ensure_ascii=False, indent=2) + in_memory_entries = [ + s.compact() + for s in SESSIONS.values() + if s.session_id not in existing_ids + ] + entries.extend(in_memory_entries) + entries.sort(key=lambda s: s.get('updated_at', 0), reverse=True) + _payload = json.dumps(entries, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: @@ -323,29 +327,30 @@ def _write_session_index(updates=None, *, session_dir: Path | None = None, sessi # Avoid N filesystem exists() checks under LOCK by collecting # on-disk IDs once before entering the critical section. on_disk_ids = _persisted_session_ids_snapshot() + existing = json.loads(session_index_file.read_text(encoding='utf-8')) + if not isinstance(existing, list): + raise ValueError("session index must be a list") with LOCK: - existing = json.loads(session_index_file.read_text(encoding='utf-8')) in_memory_ids = set(SESSIONS.keys()) - - existing = [ - e for e in existing - if (e.get('session_id') in in_memory_ids or e.get('session_id') in on_disk_ids) - ] - - # Build lookup of updated entries updated_map = {s.session_id: s.compact() for s in updates} - existing_ids = {e.get('session_id') for e in existing} - # Add any updated entries not yet in the index - for sid, entry in updated_map.items(): - if sid not in existing_ids: - existing.append(entry) - # Replace matching entries in-place - for i, e in enumerate(existing): - sid = e.get('session_id') - if sid in updated_map: - existing[i] = updated_map[sid] - existing.sort(key=lambda s: s.get('updated_at', 0), reverse=True) - _payload = json.dumps(existing, ensure_ascii=False, indent=2) + + existing = [ + e for e in existing + if (e.get('session_id') in in_memory_ids or e.get('session_id') in on_disk_ids) + ] + + existing_ids = {e.get('session_id') for e in existing} + # Add any updated entries not yet in the index. + for sid, entry in updated_map.items(): + if sid not in existing_ids: + existing.append(entry) + # Replace matching entries in-place. + for i, e in enumerate(existing): + sid = e.get('session_id') + if sid in updated_map: + existing[i] = updated_map[sid] + existing.sort(key=lambda s: s.get('updated_at', 0), reverse=True) + _payload = json.dumps(existing, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: diff --git a/tests/test_session_index_lock_window.py b/tests/test_session_index_lock_window.py new file mode 100644 index 000000000..e40a94b29 --- /dev/null +++ b/tests/test_session_index_lock_window.py @@ -0,0 +1,71 @@ +import collections +import json +import threading + + +class RecordingLock: + def __init__(self): + self._lock = threading.RLock() + self.depth = 0 + + def __enter__(self): + self._lock.acquire() + self.depth += 1 + return self + + def __exit__(self, exc_type, exc, tb): + self.depth -= 1 + self._lock.release() + + @property + def held(self): + return self.depth > 0 + + +def test_session_index_fast_path_keeps_json_work_outside_global_lock(monkeypatch, tmp_path): + import api.models as models + + session_dir = tmp_path / "sessions" + session_dir.mkdir() + index_file = session_dir / "_index.json" + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file) + + old = models.Session(session_id="idx_old", title="Old", updated_at=1.0) + updated = models.Session(session_id="idx_updated", title="Updated", updated_at=20.0) + (session_dir / "idx_old.json").write_text("{}", encoding="utf-8") + (session_dir / "idx_updated.json").write_text("{}", encoding="utf-8") + index_file.write_text( + json.dumps( + [ + old.compact(), + models.Session(session_id="idx_updated", title="Stale", updated_at=2.0).compact(), + ] + ), + encoding="utf-8", + ) + + lock = RecordingLock() + monkeypatch.setattr(models, "LOCK", lock) + monkeypatch.setattr(models, "SESSIONS", collections.OrderedDict()) + + original_loads = models.json.loads + original_dumps = models.json.dumps + + def loads_outside_lock(*args, **kwargs): + assert not lock.held + return original_loads(*args, **kwargs) + + def dumps_outside_lock(*args, **kwargs): + assert not lock.held + return original_dumps(*args, **kwargs) + + monkeypatch.setattr(models.json, "loads", loads_outside_lock) + monkeypatch.setattr(models.json, "dumps", dumps_outside_lock) + + models._write_session_index(updates=[updated]) + + rows = json.loads(index_file.read_text(encoding="utf-8")) + assert [row["session_id"] for row in rows] == ["idx_updated", "idx_old"] + assert rows[0]["title"] == "Updated" + assert rows[0]["updated_at"] == 20.0