From 0192ca28306018161d253df87ce4f972857c3f17 Mon Sep 17 00:00:00 2001 From: HexLab98 Date: Tue, 1 Sep 2026 22:59:41 +0900 Subject: [PATCH] fix(recovery): copy delivery_obligations during session salvage The lazy gateway outbox was missing from the recovery inventory, so a verified salvage could drop owed replies even when the rows were still readable. Initialize the destination schema and copy the table. --- hermes_cli/session_lost_and_found.py | 6 +++ hermes_cli/session_recovery.py | 69 ++++++++++++++++++++++++++-- 2 files changed, 70 insertions(+), 5 deletions(-) diff --git a/hermes_cli/session_lost_and_found.py b/hermes_cli/session_lost_and_found.py index 90d8acba9a..97a407e9f4 100644 --- a/hermes_cli/session_lost_and_found.py +++ b/hermes_cli/session_lost_and_found.py @@ -334,11 +334,17 @@ def _copy_direct_tables( "compression_locks", "gateway_routing", "async_delegations", + "delivery_obligations", ): source_columns = _table_columns(lf_conn, table) if not source_columns: continue dest_columns = _table_columns(dest, table) + if table == "delivery_obligations" and not dest_columns: + from gateway.delivery_ledger import _initialize_schema + + _initialize_schema(dest) + dest_columns = _table_columns(dest, table) columns = [c for c in dest_columns if c in source_columns] if not columns: continue diff --git a/hermes_cli/session_recovery.py b/hermes_cli/session_recovery.py index 6d5f5e8bf8..a06328a733 100644 --- a/hermes_cli/session_recovery.py +++ b/hermes_cli/session_recovery.py @@ -45,6 +45,20 @@ _TOPIC_TABLES = ( "telegram_dm_topic_bindings", ) +# Durable gateway outbox. Created lazily by gateway.delivery_ledger, so a +# fresh SessionDB destination does not have the table until we initialize it. +# Omitting it from the copy inventory drops owed replies (#100313, #86236). +_AUXILIARY_TABLES = ( + "delivery_obligations", +) + +_INVENTORY_TABLES = ( + *_CANONICAL_TABLES, + "state_meta", + *_TOPIC_TABLES, + *_AUXILIARY_TABLES, +) + # These values describe derived indexes or the schema that owns an optional # table. A fresh destination must generate them from its own current schema. _GENERATED_META_KEYS = frozenset({ @@ -305,7 +319,7 @@ def _inspect_connection(conn: sqlite3.Connection) -> dict[str, Any]: # A damaged journal pragma must not block rows that are still readable. report["warnings"].append(f"journal mode: {exc}") - for table in (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES): + for table in _INVENTORY_TABLES: report["tables"][table] = _table_inventory(conn, table) for required in ("sessions", "messages"): @@ -385,6 +399,23 @@ def inspect_session_database( temp_dir.cleanup() +def _ensure_auxiliary_destination_schema( + destination: sqlite3.Connection, + table: str, +) -> None: + """Create a lazy auxiliary table on the recovered destination. + + Recovery initializes the destination through base ``SessionDB``, which + does not create gateway-owned tables. Copying into a missing dest table + would report ``missing`` / ``no compatible columns`` and drop the rows. + """ + + if table == "delivery_obligations": + from gateway.delivery_ledger import _initialize_schema + + _initialize_schema(destination) + + def _copy_table( source: sqlite3.Connection, destination: sqlite3.Connection, @@ -1243,7 +1274,7 @@ def _verify_recovered_database( ) counts: dict[str, int] = {} - for table in (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES): + for table in _INVENTORY_TABLES: columns = _table_columns(conn, table) if columns: counts[table] = int( @@ -1251,7 +1282,7 @@ def _verify_recovered_database( ) verification["table_counts"] = counts - for table in ("sessions", "messages"): + for table in ("sessions", "messages", *_AUXILIARY_TABLES): expected = expected_counts.get(table) if expected is not None and counts.get(table) != expected: message = ( @@ -1662,6 +1693,27 @@ def recover_session_database( progress_cb=progress_cb, source_rows=table_inspection.get("rows"), ) + + for table in _AUXILIARY_TABLES: + table_inspection = inspection["tables"][table] + if not table_inspection.get("available"): + copy_report[table] = { + "status": "missing", + "copied_rows": 0, + } + continue + _ensure_auxiliary_destination_schema(destination_conn, table) + copy_function = ( + _copy_table_salvage if allow_partial else _copy_table + ) + copy_report[table] = copy_function( + source_conn, + destination_conn, + table, + chunk_size=chunk_size, + progress_cb=progress_cb, + source_rows=table_inspection.get("rows"), + ) orphan_cleanup = ( _cleanup_partial_orphans(destination_conn) if allow_partial @@ -1678,8 +1730,15 @@ def recover_session_database( verification = _verify_recovered_database( output, expected_counts={ - table: inspection["tables"][table].get("rows") - for table in _CANONICAL_TABLES + **{ + table: inspection["tables"][table].get("rows") + for table in _CANONICAL_TABLES + }, + **{ + table: inspection["tables"][table].get("rows") + for table in _AUXILIARY_TABLES + if inspection["tables"].get(table, {}).get("available") + }, }, copy_report=copy_report, allow_partial=allow_partial,