hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(gateway): pre-isolation room import copies per room, reads pre-actor layouts, warns once

Browse Source 
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.
This commit is contained in:
teknium1
2026-09-16 14:30:39 -07:00
committed by Teknium
parent d62036acb8
commit 00cb8895d4
3 changed files with 227 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
gateway/hosted_rooms.py
View File

@@ -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:

143
gateway/hosted_rooms_legacy_import.py Normal file
View File

@@ -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)

54
tests/gateway/test_hosted_rooms.py
View File

@@ -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()