From caa7f21f4c4879da693bc29486dc27ec1135562e Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sun, 13 Sep 2026 17:29:16 -0700 Subject: [PATCH] fix(state): run table rebuilds as one write transaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port from qwibitai/nanoclaw#3766: hold the SQLite write lock across a schema migration so two openers cannot interleave. The SessionDB writer connection is autocommit, so _rebuild_table's RENAME / CREATE / copy / DROP each committed on its own. A sibling process (gateway + CLI opening the same state.db) whose SCHEMA_SQL bootstrap ran between RENAME and CREATE recreated the table empty; the rebuilder's CREATE then failed with "table already exists", the heal was logged as "skipped", the rows stayed in *_legacy_pk and the live table was empty — silent loss of routing / usage accounting rows. BEGIN IMMEDIATE around the whole sequence (unless the caller already owns a transaction) makes the sibling wait under its existing lock patience, and a mid-rebuild failure rolls the RENAME back. --- hermes_state_schema.py | 24 ++++++- .../hermes_state/test_rebuild_table_atomic.py | 67 +++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) create mode 100644 tests/hermes_state/test_rebuild_table_atomic.py diff --git a/hermes_state_schema.py b/hermes_state_schema.py index eafa1835ad..645556ad67 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -781,7 +781,29 @@ class SessionSchemaMixin: @staticmethod def _rebuild_table(cursor: sqlite3.Cursor, table: str, legacy_name: str, ddl: str, copy_sql: str, indexes=()) -> None: """RENAME *table* to *legacy_name*, CREATE it fresh from *ddl*, copy rows back with - *copy_sql*, DROP the legacy copy, recreate *indexes*.""" + *copy_sql*, DROP the legacy copy, recreate *indexes* — as ONE write transaction. + + The writer connection is autocommit (``isolation_level=None``), so without an explicit + BEGIN each statement commits on its own and a sibling process opening the same state.db + between RENAME and CREATE runs SCHEMA_SQL's ``CREATE TABLE IF NOT EXISTS`` first: our + CREATE then fails with "table already exists", every row is stranded in *legacy_name* + and the live table is empty. BEGIN IMMEDIATE holds the write lock for the whole rebuild + so the sibling blocks (and retries under its lock patience) instead of interleaving; + any failure rolls the RENAME back.""" + conn = cursor.connection + if conn.in_transaction: # caller already owns the transaction + SessionSchemaMixin._rebuild_table_statements(cursor, table, legacy_name, ddl, copy_sql, indexes) + return + cursor.execute("BEGIN IMMEDIATE") + try: + SessionSchemaMixin._rebuild_table_statements(cursor, table, legacy_name, ddl, copy_sql, indexes) + except BaseException: + cursor.execute("ROLLBACK") + raise + cursor.execute("COMMIT") + + @staticmethod + def _rebuild_table_statements(cursor, table, legacy_name, ddl, copy_sql, indexes) -> None: cursor.execute(f"ALTER TABLE {table} RENAME TO {legacy_name}") cursor.execute(ddl) cursor.execute(copy_sql) diff --git a/tests/hermes_state/test_rebuild_table_atomic.py b/tests/hermes_state/test_rebuild_table_atomic.py new file mode 100644 index 0000000000..6918958b52 --- /dev/null +++ b/tests/hermes_state/test_rebuild_table_atomic.py @@ -0,0 +1,67 @@ +"""``_rebuild_table`` is one write transaction (port of qwibitai/nanoclaw#3766's idea: +hold the write lock across a migration so two openers cannot interleave). + +The writer connection is autocommit, so RENAME / CREATE / copy / DROP used to commit one +by one. A sibling process opening state.db between RENAME and CREATE runs SCHEMA_SQL's +``CREATE TABLE IF NOT EXISTS`` first; the rebuilder's CREATE then fails with "table +already exists", the rows are stranded in the ``*_legacy`` copy and the live table is +empty — silent data loss on the gateway + CLI concurrent-open path. +""" + +import sqlite3 + +import pytest + +from hermes_state_schema import SessionSchemaMixin + +DDL = "CREATE TABLE t (a INTEGER, b TEXT, c INTEGER NOT NULL DEFAULT 0)" +COPY = "INSERT INTO t (a, b) SELECT a, b FROM t_legacy" + + +def _db(tmp_path): + conn = sqlite3.connect(tmp_path / "t.db", isolation_level=None, timeout=0.05) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("CREATE TABLE t (a INTEGER PRIMARY KEY, b TEXT)") + conn.executemany("INSERT INTO t VALUES (?, ?)", [(1, "x"), (2, "y")]) + return conn + + +def test_sibling_opener_between_rename_and_create_cannot_strand_rows(tmp_path): + conn = _db(tmp_path) + sibling = sqlite3.connect(tmp_path / "t.db", isolation_level=None, timeout=0.05) + seen = {} + + class InterleavingCursor(sqlite3.Cursor): + def execute(self, sql, *args): + result = super().execute(sql, *args) + if sql.startswith("ALTER TABLE t RENAME"): + # A sibling's schema bootstrap fires in the RENAME→CREATE gap. Its snapshot still + # shows the uncommitted-away ``t``, so IF NOT EXISTS is a no-op, and any real write + # must be refused (busy) while the rebuild holds the write lock — never applied. + try: + sibling.execute("CREATE TABLE IF NOT EXISTS t (a INTEGER, b TEXT, c INTEGER)") + sibling.execute("INSERT INTO t (a, b) VALUES (3, 'z')") + seen["sibling"] = "applied" + except sqlite3.OperationalError as exc: + seen["sibling"] = "busy" if ("locked" in str(exc) or "busy" in str(exc)) else str(exc) + return result + + SessionSchemaMixin._rebuild_table(conn.cursor(InterleavingCursor), "t", "t_legacy", DDL, COPY) + assert seen["sibling"] == "busy" + assert not conn.in_transaction + names = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} + assert names == {"t"} + assert conn.execute("SELECT COUNT(*) FROM t").fetchone()[0] == 2 + assert conn.execute("SELECT c FROM t WHERE a = 1").fetchone()[0] == 0 # new shape landed + sibling.close() + conn.close() + + +def test_rebuild_failure_rolls_back_rename(tmp_path): + conn = _db(tmp_path) + with pytest.raises(sqlite3.OperationalError): + SessionSchemaMixin._rebuild_table(conn.cursor(), "t", "t_legacy", DDL, "INSERT INTO nope SELECT * FROM t_legacy") + assert not conn.in_transaction + assert conn.execute("SELECT COUNT(*) FROM t").fetchone()[0] == 2 + assert conn.execute("SELECT name FROM sqlite_master WHERE name = 't_legacy'").fetchone() is None + conn.close()