From 5ee439d04e34336bd860008a1edf6c8b90fa6bac Mon Sep 17 00:00:00 2001 From: stalker <122468888+stalker023reg@users.noreply.github.com> Date: Sun, 26 Jul 2026 13:04:10 +0000 Subject: [PATCH 1/2] fix(state): make malformed-DB backup copies atomic with the live-connection registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The snapshot path in session_recovery was moved onto offline_file_access when the live-connection registry landed (#71779), but the two other raw-copy sites kept the older shape: check has_live_connection(), then copy. That is a check/use race the module's own docstring warns about — a connection opened in the window between the check and the raw close() has its POSIX advisory locks cancelled, which is the documented route to "database disk image is malformed" and exactly what the registry exists to prevent. Route both remaining sites through offline_file_access, which holds the connection-lifecycle lock across the whole raw copy so connect_tracked blocks instead of slipping into the gap: * hermes_state._backup_db_file — the malformed-DB backup taken before schema surgery, called from SessionDB open-time self-heal, where the gateway, dashboard, and dispatcher share a process and hold other connections to the same state.db. * hermes_cli.kanban_db._backup_corrupt_db — the corrupt-board quarantine, which fingerprints the whole file (a second raw read) before copying. _backup_db_file keeps a registry-less fallback: in constrained embeds where hermes_cli.sqlite_safe_read can't be imported, there is no tracked connection to endanger, so the best-effort copy still runs. This is a different axis than #69609, which serializes the schema surgery across processes with a flock sidecar: that guards two repairers against each other, while this guards the backup copy against a connection opened in the same process. Neither change touches the other's lines. Tested on: Linux (pytest). --- hermes_cli/kanban_db.py | 23 +++- hermes_state.py | 47 +++++-- tests/test_raw_copy_offline_guard.py | 178 +++++++++++++++++++++++++++ 3 files changed, 234 insertions(+), 14 deletions(-) create mode 100644 tests/test_raw_copy_offline_guard.py diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 75f922d0565c5..3226bc565044d 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1759,9 +1759,18 @@ def _backup_corrupt_db(path: Path) -> Optional[Path]: # risk -- so REFUSE rather than warn-and-proceed. Losing a forensic copy # is strictly better than corrupting the live database we are trying to # rescue. - from hermes_cli.sqlite_safe_read import has_live_connection + # + # The refusal and the raw reads run as one atomic step under + # ``offline_file_access``: a bare has_live_connection() check followed by + # the fingerprint/copy is a check/use race -- a connection opened in the + # gap (connect_tracked blocks on the same lock instead) would have its + # POSIX locks cancelled by our close(). + from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access - if has_live_connection(resolved): + try: + with offline_file_access(resolved, what="quarantine"): + return _backup_corrupt_db_locked(resolved, parent, base_name) + except LiveConnectionError: _log.error( "refusing to quarantine %s: a connection to it is still open in " "this process, and fingerprinting the file would cancel that " @@ -1769,6 +1778,16 @@ def _backup_corrupt_db(path: Path) -> Optional[Path]: resolved, ) return None + + +def _backup_corrupt_db_locked( + resolved: Path, parent: Path, base_name: str +) -> Optional[Path]: + """Fingerprint + copy body of ``_backup_corrupt_db``. + + Must be called inside ``offline_file_access(resolved)`` -- every byte-level + read here is only safe while the connection-lifecycle lock is held. + """ digest = hashlib.sha256() try: with resolved.open("rb") as handle: diff --git a/hermes_state.py b/hermes_state.py index 32fcffac585f6..ac24a9a59835d 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -801,28 +801,51 @@ def _backup_db_file(db_path: Path) -> Optional[Path]: import shutil try: - from hermes_cli.sqlite_safe_read import has_live_connection - except ImportError: - has_live_connection = None # type: ignore[assignment] - - if has_live_connection is not None and has_live_connection(db_path): - logger.error( - "Refusing to raw-copy %s for backup: a connection to it is still " - "open in this process and the copy would cancel that connection's " - "POSIX locks. Close all SessionDB handles first.", - db_path, + from hermes_cli.sqlite_safe_read import ( + LiveConnectionError, + offline_file_access, ) - return None + except ImportError: + LiveConnectionError = None # type: ignore[assignment] + offline_file_access = None # type: ignore[assignment] stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = db_path.with_name(f"{db_path.name}.malformed-backup-{stamp}") - try: + + def _copy_all() -> Path: shutil.copy2(db_path, backup_path) for suffix in ("-wal", "-shm"): sidecar = db_path.with_name(db_path.name + suffix) if sidecar.exists(): shutil.copy2(sidecar, backup_path.with_name(backup_path.name + suffix)) return backup_path + + if offline_file_access is None: + # Constrained embed path: sqlite_safe_read (and with it the whole + # live-connection registry) is unavailable, so there is nothing to + # guard against — no connection in this process can be tracked. + try: + return _copy_all() + except Exception as exc: # pragma: no cover - best effort + logger.warning("Could not back up malformed DB %s: %s", db_path, exc) + return None + + # The registry check and the raw copy must be one atomic step: a bare + # has_live_connection() followed by the copy is a check/use race — a + # connection opened in the gap has its POSIX locks cancelled by the + # copy's close() (see offline_file_access in hermes_cli.sqlite_safe_read, + # which exists for exactly this pattern). + try: + with offline_file_access(db_path, what="back up"): + return _copy_all() + except LiveConnectionError: + logger.error( + "Refusing to raw-copy %s for backup: a connection to it is still " + "open in this process and the copy would cancel that connection's " + "POSIX locks. Close all SessionDB handles first.", + db_path, + ) + return None except Exception as exc: # pragma: no cover - best effort logger.warning("Could not back up malformed DB %s: %s", db_path, exc) return None diff --git a/tests/test_raw_copy_offline_guard.py b/tests/test_raw_copy_offline_guard.py new file mode 100644 index 0000000000000..2e80f05b57058 --- /dev/null +++ b/tests/test_raw_copy_offline_guard.py @@ -0,0 +1,178 @@ +"""The raw-copy quarantine/backup paths hold the connection-lifecycle lock. + +``offline_file_access`` (hermes_cli.sqlite_safe_read) exists because checking +``has_live_connection()`` and *then* doing raw file I/O is a check/use race: a +connection opened in the window between the two has its POSIX advisory locks +cancelled by the raw ``close()`` -- the exact corruption route the registry +guards against. The snapshot path in ``session_recovery`` was converted to the +context manager when it landed; these tests pin the other two raw-copy sites +to the same contract: + +* ``hermes_state._backup_db_file`` (malformed-DB backup before schema surgery) +* ``hermes_cli.kanban_db._backup_corrupt_db`` (corrupt-board quarantine) + +Each site gets the same pair: a live connection means refusal, and a +connection attempted MID-COPY blocks on the lifecycle lock until the copy is +done rather than slipping into the gap. +""" + +from __future__ import annotations + +import shutil +import sqlite3 +import threading + +import pytest + +import hermes_state +from hermes_cli import kanban_db +from hermes_cli import sqlite_safe_read +from hermes_cli.sqlite_safe_read import connect_tracked + + +@pytest.fixture(autouse=True) +def _clean_registry(): + """Each test starts and ends with an empty live-connection registry. + + The registry is process-global; a connection leaked by one test would make + ``offline_file_access`` in the next raise ``LiveConnectionError`` and + silently skip the copy under test. + """ + with sqlite_safe_read._live_lock: + sqlite_safe_read._live_connections.clear() + yield + with sqlite_safe_read._live_lock: + sqlite_safe_read._live_connections.clear() + + +@pytest.fixture +def db_file(tmp_path): + path = tmp_path / "state.db" + conn = sqlite3.connect(str(path)) + conn.execute("CREATE TABLE t (x)") + conn.commit() + conn.close() + (tmp_path / "state.db-wal").write_bytes(b"wal") + return path + + +def _connect_attempt_during(monkeypatch, db_path): + """Patch ``shutil.copy2`` so a tracked connect races the first copy. + + The verdict is taken INSIDE the patched copy, while the raw I/O is still + in flight -- once the site under test returns, its guard has released the + lifecycle lock and the queued connect is free to land, so a check after + the return would race the very thing it measures. + + Returns (thread, holder, verdict): ``verdict["landed_during_copy"]`` says + whether the connect got through mid-copy; ``holder`` collects the + connection so the test can close it. + """ + real_copy2 = shutil.copy2 + started = threading.Event() + holder: list[sqlite3.Connection] = [] + verdict: dict = {"landed_during_copy": None} + + def _racing_connect(): + started.set() + conn = connect_tracked(str(db_path), check_same_thread=False) + holder.append(conn) + + thread = threading.Thread(target=_racing_connect, daemon=True) + fired = threading.Event() + + def _copy2_with_race(src, dst, *a, **kw): + if not fired.is_set(): + fired.set() + thread.start() + started.wait(timeout=5) + # Give the connector every chance to slip in if nothing blocks it. + thread.join(timeout=0.5) + verdict["landed_during_copy"] = bool(holder) + return real_copy2(src, dst, *a, **kw) + + # Both call sites bind the stdlib module object (kanban_db at module + # level, hermes_state via a function-local ``import shutil``), so one + # patch on the module covers both. + monkeypatch.setattr(shutil, "copy2", _copy2_with_race) + return thread, holder, verdict + + +# --------------------------------------------------------------------------- +# hermes_state._backup_db_file +# --------------------------------------------------------------------------- + +def test_backup_db_file_refuses_with_live_connection(db_file): + conn = connect_tracked(str(db_file)) + try: + assert hermes_state._backup_db_file(db_file) is None + assert not list(db_file.parent.glob("*.malformed-backup-*")) + finally: + conn.close() + + +def test_backup_db_file_copy_is_atomic_with_the_registry(db_file, monkeypatch): + """A connect attempted mid-copy must wait, not land in the gap.""" + thread, holder, verdict = _connect_attempt_during(monkeypatch, db_file) + + result = hermes_state._backup_db_file(db_file) + + assert result is not None and result.exists() + assert verdict["landed_during_copy"] is False, ( + "a connection was opened while the raw copy was in flight -- its " + "POSIX locks would be cancelled by the copy's close()" + ) + thread.join(timeout=5) + assert holder, "the queued connect must succeed once the copy is done" + holder[0].close() + + +def test_backup_db_file_still_copies_without_the_registry(db_file, monkeypatch): + """Constrained embeds without hermes_cli keep the best-effort copy.""" + import builtins + + real_import = builtins.__import__ + + def _no_safe_read(name, *args, **kwargs): + if name == "hermes_cli.sqlite_safe_read": + raise ImportError("embed path") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", _no_safe_read) + result = hermes_state._backup_db_file(db_file) + assert result is not None and result.exists() + assert result.with_name(result.name + "-wal").exists() + + +# --------------------------------------------------------------------------- +# hermes_cli.kanban_db._backup_corrupt_db +# --------------------------------------------------------------------------- + +def test_backup_corrupt_db_refuses_with_live_connection(db_file): + conn = connect_tracked(str(db_file)) + try: + assert kanban_db._backup_corrupt_db(db_file) is None + assert not list(db_file.parent.glob("*.corrupt.*.bak")) + finally: + conn.close() + + +def test_backup_corrupt_db_copy_is_atomic_with_the_registry(db_file, monkeypatch): + thread, holder, verdict = _connect_attempt_during(monkeypatch, db_file) + + result = kanban_db._backup_corrupt_db(db_file) + + assert result is not None and result.exists() + assert verdict["landed_during_copy"] is False, ( + "a connection was opened while the quarantine fingerprint/copy was " + "in flight -- its POSIX locks would be cancelled by our close()" + ) + thread.join(timeout=5) + assert holder, "the queued connect must succeed once the quarantine is done" + holder[0].close() + + +def test_backup_corrupt_db_still_copies_sidecars(db_file): + result = kanban_db._backup_corrupt_db(db_file) + assert result is not None and result.exists() + assert result.with_name(result.name + "-wal").exists() From 432b7108cd85295c606fd18b0e824713112f1b22 Mon Sep 17 00:00:00 2001 From: stalker <122468888+stalker023reg@users.noreply.github.com> Date: Mon, 3 Aug 2026 21:56:02 +0300 Subject: [PATCH 2/2] test(state): replace timing-based join with park/release event protocol --- tests/test_raw_copy_offline_guard.py | 184 +++++++++++++++++++-------- 1 file changed, 134 insertions(+), 50 deletions(-) diff --git a/tests/test_raw_copy_offline_guard.py b/tests/test_raw_copy_offline_guard.py index 2e80f05b57058..badfcb5f303dd 100644 --- a/tests/test_raw_copy_offline_guard.py +++ b/tests/test_raw_copy_offline_guard.py @@ -57,45 +57,63 @@ def db_file(tmp_path): def _connect_attempt_during(monkeypatch, db_path): - """Patch ``shutil.copy2`` so a tracked connect races the first copy. + """Patch ``shutil.copy2`` to probe lock ordering, not scheduling timing. - The verdict is taken INSIDE the patched copy, while the raw I/O is still - in flight -- once the site under test returns, its guard has released the - lifecycle lock and the queued connect is free to land, so a check after - the return would race the very thing it measures. + The fixture parks the copy thread inside the patched ``copy2`` using a + park/release event pair, starts the connector thread, waits until the + connector has actually called into ``connect_tracked``, then checks whether + the connection opened while the copy still holds the lifecycle lock. Only + after that does it release the copy so both threads can finish cleanly. - Returns (thread, holder, verdict): ``verdict["landed_during_copy"]`` says - whether the connect got through mid-copy; ``holder`` collects the - connection so the test can close it. + This is scheduling-independent: the assertion is about lock ordering + (``connect_tracked`` must block while the copy holds the lock), not about + which thread the scheduler happens to resume first within a fixed timeout. + + Returns ``(copier, connector, events)`` where ``events`` is a namespace + with ``inside_copy``, ``release_copy``, ``connect_attempted``, + ``connection_opened``, and ``holder`` (list collecting the opened conn). """ real_copy2 = shutil.copy2 - started = threading.Event() + inside_copy = threading.Event() + release_copy = threading.Event() + connect_attempted = threading.Event() + connection_opened = threading.Event() holder: list[sqlite3.Connection] = [] - verdict: dict = {"landed_during_copy": None} - - def _racing_connect(): - started.set() - conn = connect_tracked(str(db_path), check_same_thread=False) - holder.append(conn) - - thread = threading.Thread(target=_racing_connect, daemon=True) + errors: list[str] = [] fired = threading.Event() - def _copy2_with_race(src, dst, *a, **kw): + def _slow_copy2(src, dst, *a, **kw): + result = real_copy2(src, dst, *a, **kw) if not fired.is_set(): fired.set() - thread.start() - started.wait(timeout=5) - # Give the connector every chance to slip in if nothing blocks it. - thread.join(timeout=0.5) - verdict["landed_during_copy"] = bool(holder) - return real_copy2(src, dst, *a, **kw) + inside_copy.set() + release_copy.wait(timeout=30) + return result - # Both call sites bind the stdlib module object (kanban_db at module - # level, hermes_state via a function-local ``import shutil``), so one - # patch on the module covers both. - monkeypatch.setattr(shutil, "copy2", _copy2_with_race) - return thread, holder, verdict + def _racing_connect(): + # Signal immediately *before* the blocking call so a timed + # "still blocked" assertion cannot pass merely because this thread + # had not been scheduled yet. + connect_attempted.set() + try: + conn = connect_tracked(str(db_path), check_same_thread=False) + connection_opened.set() + holder.append(conn) + except Exception as exc: # pragma: no cover + errors.append(f"connect failed: {exc}") + + monkeypatch.setattr(shutil, "copy2", _slow_copy2) + + import types + ns = types.SimpleNamespace( + inside_copy=inside_copy, + release_copy=release_copy, + connect_attempted=connect_attempted, + connection_opened=connection_opened, + holder=holder, + errors=errors, + ) + return ns # --------------------------------------------------------------------------- @@ -112,19 +130,54 @@ def test_backup_db_file_refuses_with_live_connection(db_file): def test_backup_db_file_copy_is_atomic_with_the_registry(db_file, monkeypatch): - """A connect attempted mid-copy must wait, not land in the gap.""" - thread, holder, verdict = _connect_attempt_during(monkeypatch, db_file) + """A connect attempted mid-copy must block on the lock, not slip in the gap.""" + ev = _connect_attempt_during(monkeypatch, db_file) - result = hermes_state._backup_db_file(db_file) - - assert result is not None and result.exists() - assert verdict["landed_during_copy"] is False, ( - "a connection was opened while the raw copy was in flight -- its " - "POSIX locks would be cancelled by the copy's close()" + copier = threading.Thread( + target=lambda: hermes_state._backup_db_file(db_file), daemon=True ) - thread.join(timeout=5) - assert holder, "the queued connect must succeed once the copy is done" - holder[0].close() + connector = threading.Thread(target=lambda: None, daemon=True) # replaced below + + connect_attempted = ev.connect_attempted + connection_opened = ev.connection_opened + + def _do_connect(): + connect_attempted.set() + try: + conn = connect_tracked(str(db_file), check_same_thread=False) + connection_opened.set() + ev.holder.append(conn) + except Exception as exc: # pragma: no cover + ev.errors.append(f"connect failed: {exc}") + + connector = threading.Thread(target=_do_connect, daemon=True) + + try: + copier.start() + assert ev.inside_copy.wait(30), "copy never reached the patched operation" + + connector.start() + assert ev.connect_attempted.wait(30), "connector thread never started" + # The connector is at the lock. While the copy holds it the connection + # must not open. + assert not ev.connection_opened.wait(1.0), ( + "connect_tracked() completed while the raw copy was in flight -- " + "its POSIX locks would be cancelled by the copy's close()" + ) + + ev.release_copy.set() + # Once the copy releases the lock the connection must open promptly. + assert ev.connection_opened.wait(30), ( + "connect_tracked() never completed after the copy released the lock" + ) + finally: + ev.release_copy.set() + copier.join(30) + connector.join(30) + + assert not ev.errors, ev.errors[0] + assert ev.holder, "the queued connect must succeed once the copy is done" + ev.holder[0].close() def test_backup_db_file_still_copies_without_the_registry(db_file, monkeypatch): @@ -158,18 +211,49 @@ def test_backup_corrupt_db_refuses_with_live_connection(db_file): def test_backup_corrupt_db_copy_is_atomic_with_the_registry(db_file, monkeypatch): - thread, holder, verdict = _connect_attempt_during(monkeypatch, db_file) + """A connect attempted mid-quarantine must block, not land in the gap.""" + ev = _connect_attempt_during(monkeypatch, db_file) + + connect_attempted = ev.connect_attempted + connection_opened = ev.connection_opened + + def _do_connect(): + connect_attempted.set() + try: + conn = connect_tracked(str(db_file), check_same_thread=False) + connection_opened.set() + ev.holder.append(conn) + except Exception as exc: # pragma: no cover + ev.errors.append(f"connect failed: {exc}") + + copier = threading.Thread( + target=lambda: kanban_db._backup_corrupt_db(db_file), daemon=True + ) + connector = threading.Thread(target=_do_connect, daemon=True) - result = kanban_db._backup_corrupt_db(db_file) + try: + copier.start() + assert ev.inside_copy.wait(30), "copy never reached the patched operation" + + connector.start() + assert ev.connect_attempted.wait(30), "connector thread never started" + assert not ev.connection_opened.wait(1.0), ( + "connect_tracked() completed while the quarantine fingerprint/copy " + "was in flight -- its POSIX locks would be cancelled by our close()" + ) + + ev.release_copy.set() + assert ev.connection_opened.wait(30), ( + "connect_tracked() never completed after the quarantine released the lock" + ) + finally: + ev.release_copy.set() + copier.join(30) + connector.join(30) - assert result is not None and result.exists() - assert verdict["landed_during_copy"] is False, ( - "a connection was opened while the quarantine fingerprint/copy was " - "in flight -- its POSIX locks would be cancelled by our close()" - ) - thread.join(timeout=5) - assert holder, "the queued connect must succeed once the quarantine is done" - holder[0].close() + assert not ev.errors, ev.errors[0] + assert ev.holder, "the queued connect must succeed once the quarantine is done" + ev.holder[0].close() def test_backup_corrupt_db_still_copies_sidecars(db_file):