From 2e1867c6e5fc2848f92e18453519643fdb96a6dc Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 17 Jul 2026 03:55:22 -0400 Subject: [PATCH 1/2] fix(#6108): keep session_search alive through transient state.db lock races --- api/streaming.py | 16 ++++- tests/test_issue3548_sessiondb_self_heal.py | 65 ++++++++++++++++++--- 2 files changed, 72 insertions(+), 9 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index 7ef3dd99..a34c7c24 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -10,6 +10,7 @@ import logging import mimetypes import os import queue +import random import re import shlex import sys @@ -6333,7 +6334,20 @@ def _build_session_db_for_stream(state_db_path): """ try: from hermes_state import SessionDB - return SessionDB(db_path=state_db_path) + _attempts = 3 + _last_error = None + for _attempt in range(_attempts): + try: + return SessionDB(db_path=state_db_path) + except Exception as _db_err: + _last_error = _db_err + if _attempt < _attempts - 1: + print( + f"[webui] WARNING: SessionDB init attempt {_attempt + 1}/{_attempts} failed, retrying: {_db_err}", + flush=True, + ) + time.sleep(0.05 * (2 ** _attempt) + random.uniform(0, 0.05)) + raise _last_error except Exception as _db_err: print(f"[webui] WARNING: SessionDB init failed - session_search will be unavailable: {_db_err}", flush=True) return None diff --git a/tests/test_issue3548_sessiondb_self_heal.py b/tests/test_issue3548_sessiondb_self_heal.py index 9241eb08..8add20f4 100644 --- a/tests/test_issue3548_sessiondb_self_heal.py +++ b/tests/test_issue3548_sessiondb_self_heal.py @@ -3,6 +3,7 @@ from __future__ import annotations from pathlib import Path +import sqlite3 import sys import types from unittest import mock @@ -22,28 +23,76 @@ def test_session_db_helper_uses_request_state_db_path(): fake_state = types.ModuleType("hermes_state") fake_state.SessionDB = FakeSessionDB - with mock.patch.dict(sys.modules, {"hermes_state": fake_state}): + with ( + mock.patch.dict(sys.modules, {"hermes_state": fake_state}), + mock.patch.object(streaming.time, "sleep") as sleep, + ): state_db_path = Path("/tmp/profile") / "state.db" db = streaming._build_session_db_for_stream(state_db_path) assert db is not None assert calls["db_path"] == state_db_path assert isinstance(db, FakeSessionDB) + sleep.assert_not_called() -def test_session_db_helper_returns_none_when_constructor_fails(): +def test_session_db_helper_retries_transient_constructor_failure(): import api.streaming as streaming - def failing_session_db(db_path=None): - raise RuntimeError("SessionDB unavailable") - + state_db_path = Path("/tmp/profile/state.db") + created = mock.Mock(name="session_db") fake_state = types.ModuleType("hermes_state") - fake_state.SessionDB = mock.Mock(side_effect=failing_session_db) + fake_random = mock.Mock() + fake_random.uniform.return_value = 0.0 + fake_state.SessionDB = mock.Mock( + side_effect=[ + sqlite3.OperationalError("database is locked"), + sqlite3.OperationalError("database is locked"), + created, + ] + ) - with mock.patch.dict(sys.modules, {"hermes_state": fake_state}): - db = streaming._build_session_db_for_stream(Path("/tmp/profile/state.db")) + with ( + mock.patch.dict(sys.modules, {"hermes_state": fake_state}), + mock.patch.object(streaming, "random", fake_random, create=True), + mock.patch.object(streaming.time, "sleep") as sleep, + ): + db = streaming._build_session_db_for_stream(state_db_path) + + assert db is created + assert fake_state.SessionDB.call_args_list == [ + mock.call(db_path=state_db_path), + mock.call(db_path=state_db_path), + mock.call(db_path=state_db_path), + ] + assert sleep.call_args_list == [mock.call(0.05), mock.call(0.1)] + + +def test_session_db_helper_returns_none_after_exhausted_retries(): + import api.streaming as streaming + + state_db_path = Path("/tmp/profile/state.db") + fake_state = types.ModuleType("hermes_state") + fake_random = mock.Mock() + fake_random.uniform.return_value = 0.0 + fake_state.SessionDB = mock.Mock( + side_effect=sqlite3.OperationalError("database is locked") + ) + + with ( + mock.patch.dict(sys.modules, {"hermes_state": fake_state}), + mock.patch.object(streaming, "random", fake_random, create=True), + mock.patch.object(streaming.time, "sleep") as sleep, + ): + db = streaming._build_session_db_for_stream(state_db_path) assert db is None + assert fake_state.SessionDB.call_args_list == [ + mock.call(db_path=state_db_path), + mock.call(db_path=state_db_path), + mock.call(db_path=state_db_path), + ] + assert sleep.call_args_list == [mock.call(0.05), mock.call(0.1)] def test_self_heal_session_db_handle_is_replaced_safely(): From 73305706c666db022ef55d3a45f78c210876465c Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 17 Jul 2026 05:42:00 -0400 Subject: [PATCH 2/2] fix(#6108): bound SessionDB retries to transient sqlite contention --- api/streaming.py | 10 ++++- tests/test_issue3548_sessiondb_self_heal.py | 44 ++++++++++++++++++++- tests/test_sprint42.py | 9 +++++ 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index a34c7c24..540b44bb 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -12,6 +12,7 @@ import os import queue import random import re +import sqlite3 import shlex import sys import subprocess @@ -6339,7 +6340,12 @@ def _build_session_db_for_stream(state_db_path): for _attempt in range(_attempts): try: return SessionDB(db_path=state_db_path) - except Exception as _db_err: + except sqlite3.OperationalError as _db_err: + _db_err_text = str(_db_err).lower() + if not ( + "locked" in _db_err_text or "busy" in _db_err_text + ): + raise _last_error = _db_err if _attempt < _attempts - 1: print( @@ -6347,7 +6353,7 @@ def _build_session_db_for_stream(state_db_path): flush=True, ) time.sleep(0.05 * (2 ** _attempt) + random.uniform(0, 0.05)) - raise _last_error + raise _last_error or RuntimeError("SessionDB construction exhausted all attempts") except Exception as _db_err: print(f"[webui] WARNING: SessionDB init failed - session_search will be unavailable: {_db_err}", flush=True) return None diff --git a/tests/test_issue3548_sessiondb_self_heal.py b/tests/test_issue3548_sessiondb_self_heal.py index 8add20f4..b79470d0 100644 --- a/tests/test_issue3548_sessiondb_self_heal.py +++ b/tests/test_issue3548_sessiondb_self_heal.py @@ -47,7 +47,7 @@ def test_session_db_helper_retries_transient_constructor_failure(): fake_state.SessionDB = mock.Mock( side_effect=[ sqlite3.OperationalError("database is locked"), - sqlite3.OperationalError("database is locked"), + sqlite3.OperationalError("database is busy"), created, ] ) @@ -95,6 +95,48 @@ def test_session_db_helper_returns_none_after_exhausted_retries(): assert sleep.call_args_list == [mock.call(0.05), mock.call(0.1)] +def test_session_db_helper_does_not_retry_permanent_constructor_failure(): + import api.streaming as streaming + + state_db_path = Path("/tmp/profile/state.db") + permanent_error = TypeError("unsupported SessionDB argument") + fake_state = types.ModuleType("hermes_state") + fake_state.SessionDB = mock.Mock(side_effect=permanent_error) + + with ( + mock.patch.dict(sys.modules, {"hermes_state": fake_state}), + mock.patch.object(streaming, "random", mock.Mock(), create=True) as random, + mock.patch.object(streaming.time, "sleep") as sleep, + ): + db = streaming._build_session_db_for_stream(state_db_path) + + assert db is None + fake_state.SessionDB.assert_called_once_with(db_path=state_db_path) + random.uniform.assert_not_called() + sleep.assert_not_called() + + +def test_session_db_helper_does_not_retry_noncontention_operational_error(): + import api.streaming as streaming + + state_db_path = Path("/tmp/profile/state.db") + permanent_error = sqlite3.OperationalError("locking protocol") + fake_state = types.ModuleType("hermes_state") + fake_state.SessionDB = mock.Mock(side_effect=permanent_error) + + with ( + mock.patch.dict(sys.modules, {"hermes_state": fake_state}), + mock.patch.object(streaming, "random", mock.Mock(), create=True) as random, + mock.patch.object(streaming.time, "sleep") as sleep, + ): + db = streaming._build_session_db_for_stream(state_db_path) + + assert db is None + fake_state.SessionDB.assert_called_once_with(db_path=state_db_path) + random.uniform.assert_not_called() + sleep.assert_not_called() + + def test_self_heal_session_db_handle_is_replaced_safely(): import api.streaming as streaming diff --git a/tests/test_sprint42.py b/tests/test_sprint42.py index 17a28ddc..a0930603 100644 --- a/tests/test_sprint42.py +++ b/tests/test_sprint42.py @@ -71,6 +71,15 @@ class TestSessionDBInjection(unittest.TestCase): "SessionDB init helper must use try/except for non-fatal error handling", ) + def test_sessiondb_retry_only_targets_transient_sqlite_errors(self): + """Permanent constructor errors must leave the retry loop immediately.""" + helper_start = STREAMING_PY.find("def _build_session_db_for_stream") + helper_end = STREAMING_PY.find("\n\ndef _attempt_credential_self_heal", helper_start) + helper_src = STREAMING_PY[helper_start:helper_end] + self.assertIn("except sqlite3.OperationalError as _db_err", helper_src) + self.assertIn('"locked" in _db_err_text or "busy" in _db_err_text', helper_src) + self.assertIn("raise _last_error or RuntimeError", helper_src) + def test_sessiondb_failure_logs_warning(self): """A failure initializing SessionDB must print a WARNING (not silently drop the error).""" self.assertIn(