From c2a3b9ce58f1300b5236d4278b7deee4d2878938 Mon Sep 17 00:00:00 2001 From: liuhao1024 Date: Sat, 13 Jun 2026 12:10:05 +0800 Subject: [PATCH] fix(state): use PASSIVE checkpoint for periodic WAL flush to prevent B-tree corruption TRUNCATE checkpoint every 50 writes causes B-tree corruption on large databases (65K+ pages) due to the exclusive-lock I/O pressure from checkpointing thousands of frames at once. Switch periodic _try_wal_checkpoint() to PASSIVE mode which does not require an exclusive lock and cannot corrupt pages under I/O pressure. Keep TRUNCATE in close() and pre-VACUUM paths where it is safe (infrequent, controlled conditions). Also replace silent `except Exception: pass` with logged warnings for checkpoint failures so operators can detect early corruption signals. Fixes #45383 --- hermes_state.py | 41 ++++---- tests/test_wal_checkpoint_strategy.py | 138 ++++++++++++++++++++++++++ 2 files changed, 158 insertions(+), 21 deletions(-) create mode 100644 tests/test_wal_checkpoint_strategy.py diff --git a/hermes_state.py b/hermes_state.py index 7181266fe1ca..eafefc2ea976 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -986,7 +986,7 @@ class SessionDB: _WRITE_MAX_RETRIES = 15 _WRITE_RETRY_MIN_S = 0.020 # 20ms _WRITE_RETRY_MAX_S = 0.150 # 150ms - # Attempt a PASSIVE WAL checkpoint every N successful writes. + # Attempt a WAL checkpoint every N successful writes (PASSIVE mode). _CHECKPOINT_EVERY_N_WRITES = 50 # Merge fragmented FTS5 segments every N successful writes. The message # triggers append one segment per insert; left unmaintained these grow @@ -1301,35 +1301,34 @@ class SessionDB: ) def _try_wal_checkpoint(self) -> None: - """Best-effort TRUNCATE WAL checkpoint. Never raises. + """Best-effort PASSIVE WAL checkpoint. Never raises. - Flushes committed WAL frames back into the main DB file and - truncates the WAL file to zero bytes. Keeps the WAL from - growing unbounded when many processes hold persistent - connections. + Flushes committed WAL frames back into the main DB file without + requiring an exclusive lock. PASSIVE is safe for frequent + periodic use because it does not block concurrent writers and + cannot corrupt B-tree pages under I/O pressure. - PASSIVE checkpoint was previously used here, but it never - truncates the WAL file — the file stays at its high-water - mark until an explicit TRUNCATE is called (which only - happened inside the infrequent vacuum()). + PASSIVE does not truncate the WAL file — it stays at its + high-water mark. WAL truncation happens in :meth:`close` + (TRUNCATE) and pre-VACUUM checkpoints, which run infrequently + under controlled conditions. - TRUNCATE may block writers briefly while checkpointing, but - _try_wal_checkpoint is called off the hot path (every 50 - writes) and already runs under ``self._lock``, so the - additional hold time is negligible. + Previous TRUNCATE strategy caused B-tree corruption on large + databases (65K+ pages) due to the exclusive-lock I/O pressure + from checkpointing thousands of frames at once (issue #45383). """ try: with self._lock: result = self._conn.execute( - "PRAGMA wal_checkpoint(TRUNCATE)" + "PRAGMA wal_checkpoint(PASSIVE)" ).fetchone() if result and result[1] > 0: logger.debug( "WAL checkpoint: %d/%d pages checkpointed", result[2], result[1], ) - except Exception: - pass # Best effort — never fatal. + except Exception as exc: + logger.warning("WAL checkpoint (PASSIVE) failed: %s", exc) def _try_optimize_fts(self) -> None: """Best-effort FTS5 segment merge. Never raises. @@ -1357,8 +1356,8 @@ class SessionDB: if self._conn: try: self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") - except Exception: - pass + except Exception as exc: + logger.debug("WAL checkpoint (TRUNCATE) at close failed: %s", exc) self._conn.close() self._conn = None @@ -7015,8 +7014,8 @@ class SessionDB: # Best-effort WAL checkpoint first, then VACUUM. try: self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") - except Exception: - pass + except Exception as exc: + logger.debug("WAL checkpoint (TRUNCATE) before VACUUM failed: %s", exc) self._conn.execute("VACUUM") return optimized diff --git a/tests/test_wal_checkpoint_strategy.py b/tests/test_wal_checkpoint_strategy.py new file mode 100644 index 000000000000..880aa4b5cc2d --- /dev/null +++ b/tests/test_wal_checkpoint_strategy.py @@ -0,0 +1,138 @@ +"""Tests for SessionDB WAL checkpoint strategy (issue #45383). + +Verifies that periodic checkpoints use PASSIVE mode (safe for large DBs) +while close() and pre-VACUUM paths still use TRUNCATE. +""" + +import sqlite3 +import logging +from unittest.mock import MagicMock, patch + +import pytest + +from hermes_state import SessionDB + + +@pytest.fixture() +def db(tmp_path): + """Create a SessionDB with a temp database file.""" + db_path = tmp_path / "test_state.db" + session_db = SessionDB(db_path=db_path) + yield session_db + try: + session_db.close() + except Exception: + pass + + +class TestTryWalCheckpointPassive: + """_try_wal_checkpoint() should use PASSIVE mode for periodic use.""" + + def test_checkpoint_uses_passive_mode(self, db): + """PASSIVE checkpoint does not require exclusive lock — safe for large DBs.""" + # Capture the real connection's execute before mocking + real_conn = db._conn + execute_calls = [] + + def tracking_execute(sql, *args, **kwargs): + execute_calls.append(sql) + return real_conn.execute(sql, *args, **kwargs) + + # sqlite3.Connection.execute is read-only (C extension) — replace _conn + mock_conn = MagicMock() + mock_conn.execute.side_effect = tracking_execute + mock_conn.fetchone.return_value = None + db._conn = mock_conn + + db._try_wal_checkpoint() + + passive_calls = [c for c in execute_calls if "wal_checkpoint(PASSIVE)" in c] + truncate_calls = [c for c in execute_calls if "wal_checkpoint(TRUNCATE)" in c] + assert len(passive_calls) == 1, ( + f"Expected 1 PASSIVE checkpoint call, got {len(passive_calls)}" + ) + assert len(truncate_calls) == 0, ( + "Periodic checkpoint should NOT use TRUNCATE" + ) + + def test_checkpoint_logs_warning_on_failure(self, db, caplog): + """Failed PASSIVE checkpoint logs a warning instead of silent pass.""" + mock_conn = MagicMock() + mock_conn.execute.side_effect = sqlite3.OperationalError("disk I/O error") + db._conn = mock_conn + + with caplog.at_level(logging.WARNING): + db._try_wal_checkpoint() + + assert any("WAL checkpoint (PASSIVE) failed" in r.message for r in caplog.records), ( + f"Expected warning log about PASSIVE checkpoint failure, got: {caplog.text}" + ) + + def test_checkpoint_returns_result_on_success(self, db): + """Successful PASSIVE checkpoint does not raise.""" + db._try_wal_checkpoint() + + +class TestCloseUsesTruncate: + """close() should still use TRUNCATE to shrink WAL on shutdown.""" + + def test_close_uses_truncate_mode(self, db): + """TRUNCATE at close is safe — no concurrent writers during shutdown.""" + real_conn = db._conn + execute_calls = [] + + def tracking_execute(sql, *args, **kwargs): + execute_calls.append(sql) + return real_conn.execute(sql, *args, **kwargs) + + mock_conn = MagicMock() + mock_conn.execute.side_effect = tracking_execute + db._conn = mock_conn + + db.close() + + truncate_calls = [c for c in execute_calls if "wal_checkpoint(TRUNCATE)" in c] + assert len(truncate_calls) == 1, ( + f"Expected 1 TRUNCATE checkpoint at close, got {len(truncate_calls)}" + ) + + def test_close_logs_debug_on_failure(self, db, caplog): + """Failed TRUNCATE at close logs debug (not warning — close is best-effort).""" + mock_conn = MagicMock() + mock_conn.execute.side_effect = sqlite3.OperationalError("database is locked") + db._conn = mock_conn + + with caplog.at_level(logging.DEBUG): + db.close() + + assert any("WAL checkpoint (TRUNCATE) at close failed" in r.message for r in caplog.records), ( + f"Expected debug log about TRUNCATE failure at close, got: {caplog.text}" + ) + + +class TestCheckpointFrequency: + """Checkpoint triggers every N writes.""" + + def test_checkpoint_triggers_at_interval(self, db): + """_try_wal_checkpoint is called every _CHECKPOINT_EVERY_N_WRITES writes.""" + call_count = [0] + original = db._try_wal_checkpoint + + def counting_checkpoint(): + call_count[0] += 1 + original() + + db._try_wal_checkpoint = counting_checkpoint + + # Write exactly _CHECKPOINT_EVERY_N_WRITES sessions to trigger one checkpoint + n = db._CHECKPOINT_EVERY_N_WRITES + import time as _time + for i in range(n): + db._execute_write(lambda conn, _i=i: conn.execute( + "INSERT INTO sessions (id, source, started_at) VALUES (?, ?, ?)", + (f"sess_{_i}", "test", _time.time()), + )) + + assert call_count[0] == 1, ( + f"Expected 1 checkpoint after {n} writes, got {call_count[0]}" + )