diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index 05ce8dde70..2355a90348 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -274,10 +274,12 @@ class SessionTranscriptMixin: # DB write outside the retry lock so other sessions can append. while True: - # Spooled backlog (cap eviction or a stalled session) is older than ``msg``: replay it - # first so recovery keeps transcript order; a still-dead DB just fails both. - self._drain_spooled_drops(session_id) try: + # Spooled backlog (cap eviction, a stalled session, a boot hold-back) is older than + # ``msg``: replay it first. While any of it stays on disk ``msg`` stays queued, since + # writing it now would give it a lower row id than that backlog for good. + if self._drain_spooled_drops(session_id): + raise RuntimeError(f"older spooled transcript rows for {session_id} still pending") self._append_transcript_message(session_id, msg) except Exception as exc: from hermes_state import StateDbCorruptError, StateDbReplacedError @@ -371,12 +373,13 @@ class SessionTranscriptMixin: self._dirty_transcripts.pop(queue_session_id, None) return spooled - def _drain_spooled_drops(self, session_id: str) -> None: - """Replay cap-dropped spooled transcript messages after DB recovery. Best-effort: replay - failures keep the spool files for the next successful flush; nothing here may raise.""" + def _drain_spooled_drops(self, session_id: str) -> bool: + """Replay cap-dropped spooled transcript messages after DB recovery; True while some of them + are still on disk. Best-effort: replay failures keep the spool files for the next successful + flush; nothing here may raise.""" spooled_sessions = getattr(self, "_spooled_drop_sessions", None) if not spooled_sessions or session_id not in spooled_sessions: - return + return False try: from gateway.shutdown_flush import drain_transcript_spool # Inside an outage the append that follows logs/escalates the same failure; the @@ -389,8 +392,10 @@ class SessionTranscriptMixin: ) if not remaining: spooled_sessions.discard(session_id) + return bool(remaining) except Exception as exc: logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc) + return False def spooled_drop_sessions(self) -> set: """The live set of sessions whose next transcript write drains the spool first. Boot recovery diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 35c33f102b..8cbf5f03c7 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -193,3 +193,38 @@ def test_held_back_files_drain_before_that_sessions_next_live_write(flush_dir, m assert _contents(mock_db) == ["old0", "old0", "old1", "new-live"] assert not list(flush_dir.glob("*.json")) + + +def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, monkeypatch): + """If the pre-write drain fails again, the live row must stay queued in memory: writing it + while the older files are still on disk gives it a lower row id than they will get.""" + import hermes_state_registry + from gateway.run import _recover_pending_flushes + from gateway.session import SessionStore + + _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "old0"}, ts=100, seq=0) + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "old1"}, ts=100, seq=1) + mock_db = MagicMock() + mock_db.append_message.side_effect = [ + RuntimeError("still locked at boot"), RuntimeError("still locked at drain"), 1, 1, 1, 1] + monkeypatch.setattr(hermes_state_registry, "acquire", lambda: mock_db) + monkeypatch.setattr(hermes_state_registry, "release_or_close", lambda _db: None) + store = object.__new__(SessionStore) + store._db = mock_db + store._transcript_retry_lock = threading.Lock() + store._dirty_transcripts = {} + store._transcript_append_failures = {} + store._fts_rebuild_last_attempt_at = time.monotonic() + + runner = SimpleNamespace(config=SimpleNamespace(multiplex_profiles=False), session_store=store) + assert _recover_pending_flushes(runner) == 0 + store.append_to_transcript("sess-1", {"role": "user", "content": "new-live"}) + assert _contents(mock_db) == ["old0", "old0"] + assert len(list(flush_dir.glob("*.json"))) == 2 + + store.append_to_transcript("sess-1", {"role": "user", "content": "new-live-2"}) + assert _contents(mock_db) == ["old0", "old0", "old0", "old1", "new-live", "new-live-2"] + assert not list(flush_dir.glob("*.json")) + assert not store._dirty_transcripts