From 00cb8895d4bb60aef24b282cde968745b6f3977f Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Wed, 16 Sep 2026 14:30:39 -0700 Subject: [PATCH] fix(gateway): pre-isolation room import copies per room, reads pre-actor layouts, warns once Review follow-up on the one-shot state.db -> shared-state.db import (#109775): - A room id present in both stores is skipped as a unit. Grafting the legacy events under the store's own room left next_seq behind MAX(seq) and every later append_event collided on (room_id, seq). - Columns only the target has are selected with the same defaults the in-place column migration applies ('legacy' authority, epoch 1, legacy actor json), and room-scoped tables use a plain INSERT under a savepoint. INSERT OR IGNORE used to swallow the NOT NULL violations of a pre-Sep-2 layout, drop every row and still record the marker with rooms=0. event_bytes is backfilled for the copied rooms. - An unreadable or refused legacy store is remembered per process, so the warning fires once and the retry happens on the next start instead of on every poll. - The import lives in gateway/hosted_rooms_legacy_import.py; the facade keeps only the _connect seam. --- gateway/hosted_rooms.py | 148 ++++++-------------------- gateway/hosted_rooms_legacy_import.py | 143 +++++++++++++++++++++++++ tests/gateway/test_hosted_rooms.py | 54 ++++++++++ 3 files changed, 227 insertions(+), 118 deletions(-) create mode 100644 gateway/hosted_rooms_legacy_import.py diff --git a/gateway/hosted_rooms.py b/gateway/hosted_rooms.py index 9de2dd1306..36f03ba2ae 100644 --- a/gateway/hosted_rooms.py +++ b/gateway/hosted_rooms.py @@ -9,7 +9,6 @@ from __future__ import annotations import hashlib import json -import logging import re import sqlite3 from contextlib import closing @@ -46,8 +45,6 @@ CONTROL_EVENT_COUNT_RESERVE = 64 CONTROL_EVENT_BYTE_RESERVE = 1024 * 1024 _JOURNAL_MODE_LOCK_RETRIES = 8 -logger = logging.getLogger(__name__) - _EVENT_KIND_RE = re.compile(r"^[a-z][a-z0-9_.-]*$") _CONTROL_EVENT_KINDS = frozenset({"authority.claimed", "authority.lost", "room.disbanded", "room.stop_requested"}) _EVENT_KINDS_BY_ACTOR = { @@ -339,28 +336,15 @@ def _migrate_remote_run_schema(conn: sqlite3.Connection) -> None: # Draft builds before the actor contract carried no identity. Preserve their inert replay rows explicitly # as legacy system events rather than guessing a user or Bot author. _LEGACY_ACTOR_JSON = _system_actor_json("legacy").replace("'", "''") -# (table, column, ddl) applied in this exact order; each table's PRAGMA is read on first use. -_LEGACY_COLUMN_DDL = ( - ("hosted_rooms", "authority_gateway_id", - "ALTER TABLE hosted_rooms ADD COLUMN authority_gateway_id TEXT NOT NULL DEFAULT 'legacy'"), - ("hosted_rooms", "authority_epoch", - "ALTER TABLE hosted_rooms ADD COLUMN authority_epoch INTEGER NOT NULL DEFAULT 1"), - ("hosted_rooms", "event_bytes", "ALTER TABLE hosted_rooms ADD COLUMN event_bytes INTEGER NOT NULL DEFAULT 0"), - ("hosted_room_events", "actor_json", - "ALTER TABLE hosted_room_events " f"ADD COLUMN actor_json TEXT NOT NULL DEFAULT '{_LEGACY_ACTOR_JSON}'"), - ("hosted_room_events", "authority_epoch", "ALTER TABLE hosted_room_events ADD COLUMN authority_epoch INTEGER")) - - -def _migrate_legacy_columns(conn: sqlite3.Connection) -> None: - """Add columns draft schemas lacked; backfill event_bytes when first introduced.""" - columns: dict[str, frozenset[str]] = {} - for table, column, ddl in _LEGACY_COLUMN_DDL: - if table not in columns: - columns[table] = table_columns(conn, table) - if column not in columns[table]: - conn.execute(ddl) - if "event_bytes" not in columns["hosted_rooms"]: - conn.execute("""UPDATE hosted_rooms +# (table, column, declaration, default literal) applied in this exact order; each table's PRAGMA is read on +# first use. The default is also what the pre-isolation import selects for a source that predates the column. +_LEGACY_COLUMNS = ( + ("hosted_rooms", "authority_gateway_id", "TEXT NOT NULL", "'legacy'"), + ("hosted_rooms", "authority_epoch", "INTEGER NOT NULL", "1"), + ("hosted_rooms", "event_bytes", "INTEGER NOT NULL", "0"), + ("hosted_room_events", "actor_json", "TEXT NOT NULL", f"'{_LEGACY_ACTOR_JSON}'"), + ("hosted_room_events", "authority_epoch", "INTEGER", None)) +_EVENT_BYTES_BACKFILL = """UPDATE hosted_rooms SET event_bytes=COALESCE(( SELECT SUM( length(CAST(event_id AS BLOB)) + @@ -370,7 +354,20 @@ def _migrate_legacy_columns(conn: sqlite3.Connection) -> None: ) FROM hosted_room_events WHERE hosted_room_events.room_id=hosted_rooms.room_id - ), 0)""") + ), 0) WHERE {where}""" + + +def _migrate_legacy_columns(conn: sqlite3.Connection) -> None: + """Add columns draft schemas lacked; backfill event_bytes when first introduced.""" + columns: dict[str, frozenset[str]] = {} + for table, column, declaration, default in _LEGACY_COLUMNS: + if table not in columns: + columns[table] = table_columns(conn, table) + if column not in columns[table]: + conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {declaration}" + + (f" DEFAULT {default}" if default is not None else "")) + if "event_bytes" not in columns["hosted_rooms"]: + conn.execute(_EVENT_BYTES_BACKFILL.format(where="1")) def _initialize_schema(conn: sqlite3.Connection) -> None: @@ -425,104 +422,19 @@ def local_authority_gateway_id() -> str: return _actor_id(f"install:{install_id}", "authority_gateway_id") -# --- pre-isolation import ------------------------------------------------------- -# ``0e422e0ece`` repointed the room store from the root ``state.db`` to ``shared-state.db`` without -# moving the rows it already held, so an install that had rooms started with an empty coordination -# set and every pre-existing room resolved to "hosted room not found" (#109775). The first open -# after the upgrade copies the rows across once; the marker row keeps that a one-shot step, because -# a purge in THIS store must never be undone by re-importing rows the legacy store still holds. -_LEGACY_SOURCE_NAME = "state.db" -_LEGACY_MARKER_TABLE = "hosted_room_legacy_imports" -# Liveness state, never copied: a lease is a ~15s heartbeat plus a process generation, so a copied -# lease names a process that is gone. The driver claims a fresh one instead. -_LEGACY_IMPORT_SKIP = frozenset({"hosted_room_driver_leases"}) - - -def _legacy_source_path(db_path: Path) -> Path | None: - """The pre-isolation store for ``db_path``, or ``None`` when this database has no predecessor. - - Only the shared coordination database has one: callers that pass any other path (older - layouts, tests) own that file directly. - """ - return db_path.with_name(_LEGACY_SOURCE_NAME) if db_path.name == "shared-state.db" else None - - -def _legacy_import_settled(conn: sqlite3.Connection, db_path: Path) -> bool: - """True once the pre-isolation import for this database has been recorded (or never applies).""" - if _legacy_source_path(db_path) is None: - return True - return table_exists(conn, _LEGACY_MARKER_TABLE) and conn.execute( - f"SELECT 1 FROM {_LEGACY_MARKER_TABLE} WHERE source=?", (_LEGACY_SOURCE_NAME,)).fetchone() is not None - - -def _copy_legacy_rows(target: sqlite3.Connection, source: Path) -> int: - """Copy the ``hosted_room*`` rows ``target`` is missing from ``source``; returns rooms copied. - - ``INSERT OR IGNORE`` means rows this store already has always win. A table this store has not - created yet (the driver, policy and replica schemas are initialized by their own modules) is - created from the source's own DDL so its rows survive the upgrade too. - """ - copied = 0 - with closing(sqlite3.connect(f"file:{source}?mode=ro", uri=True, timeout=10)) as legacy: - names = [str(row[0]) for row in legacy.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'hosted_room*'")] - # Parents first: hosted_room_events carries a foreign key into hosted_rooms. - for name in sorted(names, key=lambda name: (name != "hosted_rooms", name)): - if name in _LEGACY_IMPORT_SKIP or name == _LEGACY_MARKER_TABLE or name.endswith(("_next", "_migrating")): - continue - if not table_exists(target, name): - target.execute(str(legacy.execute( - "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (name,)).fetchone()[0])) - target_columns = table_columns(target, name) - # Source PRAGMA order (a frozenset would not be deterministic); columns this store - # lacks are dropped, ones it added take their DDL default. - columns = [str(row[1]) for row in legacy.execute(f"PRAGMA table_info({name})") - if str(row[1]) in target_columns] - if not columns: - continue - cursor = target.executemany( - f"INSERT OR IGNORE INTO {name} ({', '.join(columns)}) VALUES ({', '.join('?' * len(columns))})", - legacy.execute(f"SELECT {', '.join(columns)} FROM {name}")) - if name == "hosted_rooms": - copied = max(0, cursor.rowcount) - return copied - - -def _import_legacy_rooms(conn: sqlite3.Connection, db_path: Path) -> None: - """Copy the pre-isolation rows in once, then record the marker; never fails the open. - - The copy runs inside the caller's schema transaction, so a crash leaves either both the - copied rows and the marker or neither. - """ - if _legacy_import_settled(conn, db_path): - return - source = _legacy_source_path(db_path) - try: - copied = _copy_legacy_rows(conn, source) if source.is_file() else 0 - except (OSError, sqlite3.Error) as exc: - # A locked or unreadable legacy store must not take hosted rooms down with it: skip this - # open, leave the marker unset, retry on the next one. - logger.warning("hosted rooms: could not import the pre-isolation %s (%s); will retry", source, exc) - return - conn.execute( - f"CREATE TABLE IF NOT EXISTS {_LEGACY_MARKER_TABLE} (" - "source TEXT PRIMARY KEY, imported_at REAL NOT NULL, rooms INTEGER NOT NULL)") - conn.execute( - f"INSERT OR IGNORE INTO {_LEGACY_MARKER_TABLE} (source, imported_at, rooms) VALUES (?, ?, ?)", - (_LEGACY_SOURCE_NAME, _now(None), copied)) - if copied: - logger.info("hosted rooms: imported %d pre-isolation Group Chat(s) from %s", copied, source) - - def _store_ready(conn: sqlite3.Connection, db_path: Path) -> bool: """The store can serve rooms once its schema is current and the pre-isolation import has run.""" - return _schema_is_current(conn) and _legacy_import_settled(conn, db_path) + from gateway.hosted_rooms_legacy_import import settled + + return _schema_is_current(conn) and settled(conn, db_path) def _initialize_store(conn: sqlite3.Connection, db_path: Path) -> None: - """Create or migrate the schema, then copy the pre-isolation rows in.""" + """Create or migrate the schema, then copy the pre-isolation rows in (#109775).""" + from gateway.hosted_rooms_legacy_import import import_legacy_rooms + _initialize_schema(conn) - _import_legacy_rooms(conn, db_path) + import_legacy_rooms(conn, db_path) def _connect(db_path: DbPath) -> sqlite3.Connection: diff --git a/gateway/hosted_rooms_legacy_import.py b/gateway/hosted_rooms_legacy_import.py new file mode 100644 index 0000000000..3aae609e07 --- /dev/null +++ b/gateway/hosted_rooms_legacy_import.py @@ -0,0 +1,143 @@ +"""One-shot import of the pre-isolation ``hosted_room*`` rows into ``shared-state.db``. + +``0e422e0ece`` repointed the room store from the root ``state.db`` to ``shared-state.db`` without +moving the rows it already held, so an install that had rooms started with an empty coordination +set and every pre-existing room resolved to "hosted room not found" (#109775). The first open after +the upgrade copies the rows across once; the marker row keeps that a one-shot step, because a purge +in THIS store must never be undone by re-importing rows the legacy store still holds. +""" + +from __future__ import annotations + +import logging +import sqlite3 +from contextlib import closing +from pathlib import Path + +from gateway.hosted_rooms_common import clock, table_columns, table_exists + +logger = logging.getLogger(__name__) + +SOURCE_NAME = "state.db" +MARKER_TABLE = "hosted_room_legacy_imports" +# Liveness state, never copied: a lease is a ~15s heartbeat plus a process generation, so a copied +# lease names a process that is gone. The driver claims a fresh one instead. +_SKIP_TABLES = frozenset({"hosted_room_driver_leases"}) +# Sources this process could not import (unreadable file, rows the target refused). Every store open +# re-checks readiness, so without this a broken legacy file would re-run the copy and re-warn on +# every poll; the retry happens on the next process start instead. +_failed_sources: set[Path] = set() + + +def source_path(db_path: Path) -> Path | None: + """The pre-isolation store for ``db_path``, or ``None`` when this database has no predecessor. + + Only the shared coordination database has one: callers that pass any other path (older + layouts, tests) own that file directly. + """ + return db_path.with_name(SOURCE_NAME) if db_path.name == "shared-state.db" else None + + +def settled(conn: sqlite3.Connection, db_path: Path) -> bool: + """True once the import for this database has been recorded, given up on for this process, or never applies.""" + source = source_path(db_path) + if source is None or source in _failed_sources: + return True + return table_exists(conn, MARKER_TABLE) and conn.execute( + f"SELECT 1 FROM {MARKER_TABLE} WHERE source=?", (SOURCE_NAME,)).fetchone() is not None + + +def _select_expressions(legacy: sqlite3.Connection, target: sqlite3.Connection, name: str) -> list[tuple[str, str]]: + """``(target column, source expression)`` pairs for copying ``name``. + + Columns only the target has take the same default the in-place column migration applies, so a + legacy layout from before the actor/authority columns imports instead of tripping NOT NULL. + Columns only the source has are dropped; columns the target added without a default are left + to its DDL default. + """ + from gateway.hosted_rooms import _LEGACY_COLUMNS + + defaults = {(table, column): default for table, column, _, default in _LEGACY_COLUMNS if default is not None} + source_columns = [str(row[1]) for row in legacy.execute(f"PRAGMA table_info({name})")] + target_columns = table_columns(target, name) + pairs = [(column, column) for column in source_columns if column in target_columns] + pairs.extend((column, default) for (table, column), default in defaults.items() + if table == name and column in target_columns and column not in source_columns) + return pairs + + +def _copy_rows(target: sqlite3.Connection, source: Path) -> int: + """Copy every room ``target`` does not have yet, with all of its child rows; returns rooms copied. + + A room id present in both stores is skipped as a unit: grafting the legacy events under this + store's room would leave ``next_seq`` behind ``MAX(seq)`` and every later append colliding. + Room-scoped tables therefore use a plain INSERT — a row the target refuses raises and aborts + the whole import rather than being dropped in silence. A table this store has not created yet + (the driver, policy and replica schemas are initialized by their own modules) is created from + the source's own DDL so its rows survive the upgrade too. + """ + from gateway.hosted_rooms import _EVENT_BYTES_BACKFILL + + copied_rooms: list[str] = [] + existing = {str(row[0]) for row in target.execute("SELECT room_id FROM hosted_rooms")} + with closing(sqlite3.connect(f"file:{source}?mode=ro", uri=True, timeout=10)) as legacy: + names = [str(row[0]) for row in legacy.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'hosted_room*'")] + # Parents first: hosted_room_events carries a foreign key into hosted_rooms. + for name in sorted(names, key=lambda name: (name != "hosted_rooms", name)): + if name in _SKIP_TABLES or name == MARKER_TABLE or name.endswith(("_next", "_migrating")): + continue + if not table_exists(target, name): + target.execute(str(legacy.execute( + "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (name,)).fetchone()[0])) + pairs = _select_expressions(legacy, target, name) + if not pairs: + continue + columns = [column for column, _ in pairs] + rows = legacy.execute(f"SELECT {', '.join(expr for _, expr in pairs)} FROM {name}") + if "room_id" in columns: + room_index = columns.index("room_id") + rows = (row for row in rows if str(row[room_index]) not in existing) + verb = "INSERT" + else: + verb = "INSERT OR IGNORE" + if name == "hosted_rooms": + rows = list(rows) + copied_rooms = [str(row[room_index]) for row in rows] + target.executemany( + f"{verb} INTO {name} ({', '.join(columns)}) VALUES ({', '.join('?' * len(columns))})", rows) + if copied_rooms and "event_bytes" not in table_columns(legacy, "hosted_rooms"): + target.execute( + _EVENT_BYTES_BACKFILL.format(where=f"room_id IN ({', '.join('?' * len(copied_rooms))})"), copied_rooms) + return len(copied_rooms) + + +def import_legacy_rooms(conn: sqlite3.Connection, db_path: Path) -> None: + """Copy the pre-isolation rows in once, then record the marker; never fails the open. + + The copy runs inside the caller's schema transaction under its own savepoint, so a crash or a + refused row leaves either both the copied rows and the marker or neither. + """ + source = source_path(db_path) + if source is None or settled(conn, db_path): + return + conn.execute("SAVEPOINT legacy_import") + try: + copied = _copy_rows(conn, source) if source.is_file() else 0 + except (OSError, sqlite3.Error) as exc: + # A locked, unreadable or incompatible legacy store must not take hosted rooms down with + # it: drop the partial copy, leave the marker unset, retry on the next process start. + conn.execute("ROLLBACK TO legacy_import") + conn.execute("RELEASE legacy_import") + _failed_sources.add(source) + logger.warning("hosted rooms: could not import the pre-isolation %s (%s); will retry on the next start", + source, exc) + return + conn.execute( + f"CREATE TABLE IF NOT EXISTS {MARKER_TABLE} (source TEXT PRIMARY KEY, imported_at REAL NOT NULL, rooms INTEGER NOT NULL)") + conn.execute( + f"INSERT OR IGNORE INTO {MARKER_TABLE} (source, imported_at, rooms) VALUES (?, ?, ?)", + (SOURCE_NAME, clock(None), copied)) + conn.execute("RELEASE legacy_import") + if copied: + logger.info("hosted rooms: imported %d pre-isolation Group Chat(s) from %s", copied, source) diff --git a/tests/gateway/test_hosted_rooms.py b/tests/gateway/test_hosted_rooms.py index 3946f8fc80..04d080218c 100644 --- a/tests/gateway/test_hosted_rooms.py +++ b/tests/gateway/test_hosted_rooms.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import logging import sqlite3 from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor @@ -1435,3 +1436,56 @@ def test_legacy_import_is_a_one_shot_and_skips_driver_liveness_state(tmp_path): with sqlite3.connect(store) as conn: conn.execute("DELETE FROM hosted_rooms") assert rooms.list_rooms(store) == [] + + +def test_legacy_import_skips_a_room_this_store_already_owns_as_a_unit(tmp_path): + """A room id present in both stores keeps THIS store's history intact and appendable. + + Grafting only the non-colliding legacy events under the store's own room left ``next_seq`` + behind ``MAX(seq)``, so every later append collided on (room_id, seq). + """ + legacy = tmp_path / "state.db" + _create(legacy, room_id="same") + for index in range(5): + _append(legacy, room_id="same", event_id=f"legacy-{index}", kind="message.user", actor=USER, + payload={"text": str(index)}, now=11 + index) + # The store already has its own "same" before the import runs (a room re-created after "not found"). + own = tmp_path / "scratch.db" + _create(own, room_id="same") + _append(own, room_id="same", event_id="own-1", kind="message.user", actor=USER, payload={"text": "s"}, now=11) + own.rename(tmp_path / "shared-state.db") + store = tmp_path / "shared-state.db" + + assert [event["event_id"] for event in rooms.read_events(store, room_id="same")["events"]] == ["own-1"] + _append(store, room_id="same", event_id="own-2", kind="message.user", actor=USER, payload={"text": "t"}, now=30) + assert rooms.room_state(store, room_id="same")["latest_seq"] == 2 + + +def test_legacy_import_reads_layouts_from_before_the_actor_and_authority_columns(tmp_path): + """A legacy store without authority_gateway_id/actor_json imports with the migration's defaults. + + ``INSERT OR IGNORE`` used to swallow the NOT NULL violations, drop every row and still record + the marker with rooms=0, losing the rooms permanently. + """ + _create_pre_actor_database(str(tmp_path / "state.db")) + store = tmp_path / "shared-state.db" + + assert _read_legacy_state(str(store)) == ("legacy", 1) + assert [event["event_id"] for event in rooms.read_events(store, room_id="room-1")["events"]] == ["legacy-event"] + with sqlite3.connect(store) as conn: + assert conn.execute("SELECT rooms FROM hosted_room_legacy_imports").fetchone() == (1,) + assert conn.execute("SELECT event_bytes FROM hosted_rooms").fetchone()[0] > 0 + + +def test_unreadable_legacy_store_is_reported_once_per_process(tmp_path, caplog): + """A corrupt legacy file leaves the marker unset but does not re-warn on every poll.""" + (tmp_path / "state.db").write_bytes(b"not a sqlite file" * 100) + store = tmp_path / "shared-state.db" + + with caplog.at_level(logging.WARNING, logger="gateway.hosted_rooms_legacy_import"): + for _ in range(4): + assert rooms.list_rooms(store) == [] + assert len([record for record in caplog.records if "could not import" in record.message]) == 1 + with sqlite3.connect(store) as conn: + assert not conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='hosted_room_legacy_imports'").fetchone()