From 6af548a1f73f26bca54dd2e23f0d278c37f3ba6c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 30 Aug 2026 20:10:25 -0700 Subject: [PATCH] =?UTF-8?q?feat(bot-mode):=20Group=20Chats=20survive=20the?= =?UTF-8?q?=20authority=20gateway=20dying=20=E2=80=94=20log=20replication?= =?UTF-8?q?=20and=20fenced=20takeover?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every participant gateway can now keep a durable copy of a hosted room's ordered log and continue the room when its authority host is gone: - gateway/hosted_room_replicas.py: replica store in root state.db. ingest_page() persists authority-stamped groups.log pages idempotently, refusing sequence gaps and authority-epoch regressions. promote_replica() continues the room locally at epoch+1 with a lineage-proving authority.claimed event; the stale owner is fenced everywhere the claim replicates. demote_room() lets a returning stale authority fence itself (authority.lost) upon observing a newer epoch, killing split-brain writes. - tui_gateway/methods_groups.py: groups.replicate / groups.replica_state / groups.promote / groups.demote RPC surface. Promotion requires confirm=true — storage decides HOW takeover is atomic and provable, the caller (user action now, lease/quorum driver later) decides WHEN it is safe, matching the boundary blessed on #97681. Validation: 20 new tests incl. a full failover round-trip (A hosts, B replicates incrementally, A dies, B promotes with complete history, A returns demoted and fenced); 69 total across the hosted-rooms area; E2E with two real gateway stores and real install identities. --- gateway/hosted_room_replicas.py | 570 ++++++++++++++++++ tests/gateway/test_hosted_room_replicas.py | 296 +++++++++ .../test_groups_replication_methods.py | 173 ++++++ tui_gateway/methods_groups.py | 102 ++++ 4 files changed, 1141 insertions(+) create mode 100644 gateway/hosted_room_replicas.py create mode 100644 tests/gateway/test_hosted_room_replicas.py create mode 100644 tests/tui_gateway/test_groups_replication_methods.py diff --git a/gateway/hosted_room_replicas.py b/gateway/hosted_room_replicas.py new file mode 100644 index 0000000000000..26b87080eda0e --- /dev/null +++ b/gateway/hosted_room_replicas.py @@ -0,0 +1,570 @@ +"""Replica store and takeover primitives for hosted Group Chat rooms. + +The authority gateway owns a room's ordered log in ``gateway/hosted_rooms.py``. +This module gives every OTHER participant gateway a durable local copy of that +log, and the fenced primitives to continue the room when the authority host +dies: + +- ``ingest_page()`` persists replay pages (``groups.log`` output, which carries + the room's authority stamp) idempotently, refusing sequence gaps and + authority-epoch regressions. +- ``promote_replica()`` instantiates the replicated log as a locally-owned + hosted room at ``epoch + 1`` with a lineage-proving ``authority.claimed`` + event, so a surviving participant can resume the room. +- ``demote_room()`` fences a returning stale authority: presented with proof of + a newer epoch, the local room records ``authority.lost`` and stops being + authoritative. + +Storage primitives only: none of these decide *when* takeover is safe. The +caller (an explicit user action today; a lease/quorum driver later) must +establish that the previous owner can no longer commit before promoting. +""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from pathlib import Path +from typing import Any + +from gateway.hosted_rooms import ( + MAX_ACTOR_ID_CHARS, + MAX_EVENT_JSON_BYTES, + MAX_ROOM_ID_CHARS, + HostedRoomError, + RoomConflictError, + _canonical_json, + _connect, + _transaction, + _validate_identifier, + _validate_members, + _validate_room_name, + local_authority_gateway_id, +) + +MAX_REPLICA_ROOMS = 256 +MAX_REPLICA_EVENT_BYTES = 256 * 1024 * 1024 + + +class ReplicaError(HostedRoomError): + """Base class for invalid or conflicting replica operations.""" + + +class ReplicaGapError(ReplicaError): + """A page does not start at the replica's next expected sequence.""" + + +class ReplicaEpochRegressionError(ReplicaError): + """A page or demotion carries an older authority epoch than stored.""" + + +def _initialize_replica_schema(conn: sqlite3.Connection) -> None: + conn.execute( + """CREATE TABLE IF NOT EXISTS hosted_room_replicas ( + room_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + members_json TEXT NOT NULL, + authority_gateway_id TEXT NOT NULL, + authority_epoch INTEGER NOT NULL CHECK (authority_epoch >= 1), + last_seq INTEGER NOT NULL DEFAULT 0 CHECK (last_seq >= 0), + latest_seq INTEGER NOT NULL DEFAULT 0, + event_bytes INTEGER NOT NULL DEFAULT 0, + created_at REAL NOT NULL, + updated_at REAL NOT NULL + )""" + ) + conn.execute( + """CREATE TABLE IF NOT EXISTS hosted_room_replica_events ( + room_id TEXT NOT NULL, + seq INTEGER NOT NULL CHECK (seq >= 1), + event_id TEXT NOT NULL, + kind TEXT NOT NULL, + actor_json TEXT NOT NULL, + authority_epoch INTEGER, + payload_json TEXT NOT NULL, + created_at REAL NOT NULL, + PRIMARY KEY (room_id, seq) + )""" + ) + + +def _replica_transaction(db_path: Path | str): + return _transaction(db_path, immediate=True) + + +def _ensure_schema(db_path: Path | str) -> None: + conn = _connect(db_path) + try: + with conn: + _initialize_replica_schema(conn) + finally: + conn.close() + + +def _event_bytes(event: dict[str, Any]) -> int: + return ( + len(str(event["event_id"]).encode("utf-8")) + + len(str(event["kind"]).encode("utf-8")) + + len( + json.dumps( + event["actor"], ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + ) + + len( + json.dumps( + event["payload"], ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + ) + ) + + +def _validate_page(page: Any) -> tuple[list[dict[str, Any]], dict[str, Any]]: + if not isinstance(page, dict): + raise ReplicaError("page must be an object") + events = page.get("events") + authority = page.get("authority") + if not isinstance(events, list): + raise ReplicaError("page.events must be a list") + if not isinstance(authority, dict): + raise ReplicaError("page.authority is required for replication") + gateway_id = _validate_identifier( + authority.get("gateway_id"), + label="page.authority.gateway_id", + max_chars=MAX_ACTOR_ID_CHARS, + ) + epoch = authority.get("epoch") + if isinstance(epoch, bool) or not isinstance(epoch, int) or epoch < 1: + raise ReplicaError("page.authority.epoch must be a positive integer") + previous_seq: int | None = None + for event in events: + if not isinstance(event, dict): + raise ReplicaError("page events must be objects") + seq = event.get("seq") + if isinstance(seq, bool) or not isinstance(seq, int) or seq < 1: + raise ReplicaError("event.seq must be a positive integer") + if previous_seq is not None and seq != previous_seq + 1: + raise ReplicaGapError("page events must be contiguous") + previous_seq = seq + for field in ("event_id", "kind"): + if not isinstance(event.get(field), str) or not event[field]: + raise ReplicaError(f"event.{field} must be a non-empty string") + if not isinstance(event.get("actor"), dict): + raise ReplicaError("event.actor must be an object") + if "payload" not in event: + raise ReplicaError("event.payload is required") + return events, {"gateway_id": gateway_id, "epoch": epoch} + + +def ingest_page( + db_path: Path | str, + *, + room_id: Any, + room_name: Any, + members: Any, + page: Any, + now: float | None = None, +) -> dict[str, Any]: + """Persist one replay page for ``room_id``; idempotent, gap- and + epoch-regression-safe. + + ``page`` is the verbatim result of the authority's ``groups.log`` call + (``read_events()``), whose ``authority`` stamp proves lineage. + """ + room_id = _validate_identifier( + room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS + ) + room_name = _validate_room_name(room_name) + _, members_json = _validate_members(members) + events, authority = _validate_page(page) + now = time.time() if now is None else float(now) + _ensure_schema(db_path) + + with _replica_transaction(db_path) as conn: + _initialize_replica_schema(conn) + row = conn.execute( + """SELECT authority_gateway_id, authority_epoch, last_seq, + latest_seq, event_bytes + FROM hosted_room_replicas WHERE room_id=?""", + (room_id,), + ).fetchone() + if row is None: + count = conn.execute( + "SELECT COUNT(*) FROM hosted_room_replicas" + ).fetchone()[0] + if int(count) >= MAX_REPLICA_ROOMS: + raise ReplicaError("replica room capacity exhausted") + stored_epoch = 0 + last_seq = 0 + stored_bytes = 0 + else: + stored_epoch = int(row["authority_epoch"]) + last_seq = int(row["last_seq"]) + stored_bytes = int(row["event_bytes"]) + + if authority["epoch"] < stored_epoch: + raise ReplicaEpochRegressionError( + "page authority epoch is older than the stored replica epoch" + ) + + new_events = [e for e in events if int(e["seq"]) > last_seq] + if new_events and int(new_events[0]["seq"]) != last_seq + 1: + raise ReplicaGapError( + "page skips sequences the replica has not stored" + ) + added_bytes = 0 + for event in new_events: + size = _event_bytes(event) + if stored_bytes + added_bytes + size > MAX_REPLICA_EVENT_BYTES: + raise ReplicaError("replica event storage exhausted") + actor_json = _canonical_json( + event["actor"], label="actor", max_bytes=4 * 1024 + ) + payload_json = _canonical_json( + event["payload"], label="payload", max_bytes=MAX_EVENT_JSON_BYTES + ) + epoch_value = event.get("authority_epoch") + conn.execute( + """INSERT INTO hosted_room_replica_events + (room_id, seq, event_id, kind, actor_json, authority_epoch, + payload_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + ( + room_id, + int(event["seq"]), + event["event_id"], + event["kind"], + actor_json, + epoch_value, + payload_json, + float(event.get("created_at") or now), + ), + ) + added_bytes += size + new_last = int(new_events[-1]["seq"]) if new_events else last_seq + latest_seq = page.get("latest_seq") + if isinstance(latest_seq, bool) or not isinstance(latest_seq, int): + latest_seq = new_last + if row is None: + conn.execute( + """INSERT INTO hosted_room_replicas + (room_id, name, members_json, authority_gateway_id, + authority_epoch, last_seq, latest_seq, event_bytes, + created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + room_id, + room_name, + members_json, + authority["gateway_id"], + authority["epoch"], + new_last, + max(latest_seq, new_last), + added_bytes, + now, + now, + ), + ) + else: + conn.execute( + """UPDATE hosted_room_replicas + SET name=?, members_json=?, authority_gateway_id=?, + authority_epoch=?, last_seq=?, latest_seq=?, + event_bytes=event_bytes+?, updated_at=? + WHERE room_id=?""", + ( + room_name, + members_json, + authority["gateway_id"], + authority["epoch"], + new_last, + max(latest_seq, new_last), + added_bytes, + now, + room_id, + ), + ) + return { + "room_id": room_id, + "stored_seq": new_last, + "ingested": len(new_events), + "authority": authority, + "caught_up": new_last >= max(latest_seq, new_last), + } + + +def replica_state(db_path: Path | str, *, room_id: Any) -> dict[str, Any]: + """Return the stored replica's coverage and authority lineage.""" + room_id = _validate_identifier( + room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS + ) + _ensure_schema(db_path) + with _replica_transaction(db_path) as conn: + _initialize_replica_schema(conn) + row = conn.execute( + """SELECT room_id, name, members_json, authority_gateway_id, + authority_epoch, last_seq, latest_seq, event_bytes, + created_at, updated_at + FROM hosted_room_replicas WHERE room_id=?""", + (room_id,), + ).fetchone() + if row is None: + raise ReplicaError("replica not found") + return { + "room_id": row["room_id"], + "name": row["name"], + "members": json.loads(row["members_json"]), + "authority": { + "gateway_id": row["authority_gateway_id"], + "epoch": int(row["authority_epoch"]), + }, + "last_seq": int(row["last_seq"]), + "latest_seq": int(row["latest_seq"]), + "event_bytes": int(row["event_bytes"]), + "created_at": float(row["created_at"]), + "updated_at": float(row["updated_at"]), + } + + +def promote_replica( + db_path: Path | str, + *, + room_id: Any, + reason: Any = "authority-unreachable", + now: float | None = None, +) -> dict[str, Any]: + """Continue a replicated room on THIS gateway at ``epoch + 1``. + + Copies the replica's log into the authoritative store, appends a lineage- + proving ``authority.claimed`` event, and returns the new room state. The + old authority is fenced everywhere the claim replicates: its epoch is now + stale and every fenced primitive rejects it. + + The caller decides that takeover is safe (the previous owner can no longer + commit). This primitive only makes the takeover atomic and provable. + """ + room_id = _validate_identifier( + room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS + ) + if not isinstance(reason, str) or not reason or len(reason) > 200: + raise ReplicaError("reason must be a non-empty string of at most 200 chars") + now = time.time() if now is None else float(now) + local_gateway = local_authority_gateway_id() + _ensure_schema(db_path) + + with _replica_transaction(db_path) as conn: + _initialize_replica_schema(conn) + replica = conn.execute( + """SELECT room_id, name, members_json, authority_gateway_id, + authority_epoch, last_seq, event_bytes + FROM hosted_room_replicas WHERE room_id=?""", + (room_id,), + ).fetchone() + if replica is None: + raise ReplicaError("replica not found") + if replica["authority_gateway_id"] == local_gateway: + raise ReplicaError("this gateway already holds the room authority") + if conn.execute( + "SELECT 1 FROM hosted_rooms WHERE room_id=?", (room_id,) + ).fetchone(): + raise RoomConflictError( + "room_id already exists in the local authoritative store" + ) + if conn.execute( + "SELECT 1 FROM hosted_room_retired_ids WHERE room_id=?", + (room_id,), + ).fetchone(): + raise RoomConflictError("room_id belongs to a disbanded room") + + previous_gateway = str(replica["authority_gateway_id"]) + previous_epoch = int(replica["authority_epoch"]) + target_epoch = previous_epoch + 1 + last_seq = int(replica["last_seq"]) + claim_seq = last_seq + 1 + claim_event_id = f"system:authority-claimed:{target_epoch}" + claim_actor_json = _canonical_json( + {"kind": "system", "id": "authority-control"}, + label="actor", + max_bytes=4 * 1024, + ) + claim_payload_json = _canonical_json( + { + "previous_gateway_id": previous_gateway, + "authority_gateway_id": local_gateway, + "authority_epoch": target_epoch, + "promoted_from_replica": True, + "reason": reason, + }, + label="payload", + max_bytes=MAX_EVENT_JSON_BYTES, + ) + claim_bytes = ( + len(claim_event_id.encode("utf-8")) + + len(b"authority.claimed") + + len(claim_actor_json.encode("utf-8")) + + len(claim_payload_json.encode("utf-8")) + ) + + conn.execute( + """INSERT INTO hosted_rooms + (room_id, name, members_json, authority_gateway_id, + authority_epoch, next_seq, event_bytes, revision, + created_at, updated_at, disbanded_at) + VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?, ?, NULL)""", + ( + room_id, + replica["name"], + replica["members_json"], + local_gateway, + target_epoch, + claim_seq + 1, + int(replica["event_bytes"]) + claim_bytes, + now, + now, + ), + ) + conn.execute( + """INSERT INTO hosted_room_events + (room_id, seq, event_id, kind, actor_json, authority_epoch, + payload_json, created_at) + SELECT room_id, seq, event_id, kind, actor_json, + authority_epoch, payload_json, created_at + FROM hosted_room_replica_events WHERE room_id=?""", + (room_id,), + ) + conn.execute( + """INSERT INTO hosted_room_events + (room_id, seq, event_id, kind, actor_json, authority_epoch, + payload_json, created_at) + VALUES (?, ?, ?, 'authority.claimed', ?, ?, ?, ?)""", + ( + room_id, + claim_seq, + claim_event_id, + claim_actor_json, + target_epoch, + claim_payload_json, + now, + ), + ) + conn.execute( + "DELETE FROM hosted_room_replica_events WHERE room_id=?", (room_id,) + ) + conn.execute( + "DELETE FROM hosted_room_replicas WHERE room_id=?", (room_id,) + ) + return { + "room_id": room_id, + "authority_gateway_id": local_gateway, + "authority_epoch": target_epoch, + "previous_gateway_id": previous_gateway, + "previous_epoch": previous_epoch, + "claim_seq": claim_seq, + "latest_seq": claim_seq, + } + + +def demote_room( + db_path: Path | str, + *, + room_id: Any, + observed_gateway_id: Any, + observed_epoch: Any, + now: float | None = None, +) -> dict[str, Any]: + """Fence THIS gateway's stale room authority against a proven newer epoch. + + Called when a returning gateway observes (via a replicated + ``authority.claimed`` event or a transport rejection) that another gateway + now owns the room at a higher epoch. Appends ``authority.lost`` and adopts + the observed lineage so no further local sends can be committed at the + stale epoch. Idempotent for repeated observations of the same lineage. + """ + room_id = _validate_identifier( + room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS + ) + observed_gateway_id = _validate_identifier( + observed_gateway_id, + label="observed_gateway_id", + max_chars=MAX_ACTOR_ID_CHARS, + ) + if ( + isinstance(observed_epoch, bool) + or not isinstance(observed_epoch, int) + or observed_epoch < 1 + ): + raise ReplicaError("observed_epoch must be a positive integer") + now = time.time() if now is None else float(now) + local_gateway = local_authority_gateway_id() + + with _replica_transaction(db_path) as conn: + row = conn.execute( + """SELECT authority_gateway_id, authority_epoch, next_seq + FROM hosted_rooms WHERE room_id=? AND disbanded_at IS NULL""", + (room_id,), + ).fetchone() + if row is None: + raise ReplicaError("room not found in the local authoritative store") + current_gateway = str(row["authority_gateway_id"]) + current_epoch = int(row["authority_epoch"]) + if ( + current_gateway == observed_gateway_id + and current_epoch == observed_epoch + ): + return { + "room_id": room_id, + "authority_gateway_id": current_gateway, + "authority_epoch": current_epoch, + "idempotent": True, + } + if observed_epoch <= current_epoch: + raise ReplicaEpochRegressionError( + "observed epoch does not supersede the stored authority" + ) + if current_gateway != local_gateway: + raise ReplicaError( + "room is not locally authoritative; nothing to demote" + ) + seq = int(row["next_seq"]) + lost_actor_json = _canonical_json( + {"kind": "system", "id": "authority-control"}, + label="actor", + max_bytes=4 * 1024, + ) + lost_payload_json = _canonical_json( + { + "previous_gateway_id": current_gateway, + "authority_gateway_id": observed_gateway_id, + "authority_epoch": observed_epoch, + }, + label="payload", + max_bytes=MAX_EVENT_JSON_BYTES, + ) + conn.execute( + """INSERT INTO hosted_room_events + (room_id, seq, event_id, kind, actor_json, authority_epoch, + payload_json, created_at) + VALUES (?, ?, ?, 'authority.lost', ?, ?, ?, ?)""", + ( + room_id, + seq, + f"system:authority-lost:{observed_epoch}", + lost_actor_json, + observed_epoch, + lost_payload_json, + now, + ), + ) + conn.execute( + """UPDATE hosted_rooms + SET authority_gateway_id=?, authority_epoch=?, + next_seq=next_seq+1, revision=revision+1, updated_at=? + WHERE room_id=?""", + (observed_gateway_id, observed_epoch, now, room_id), + ) + return { + "room_id": room_id, + "authority_gateway_id": observed_gateway_id, + "authority_epoch": observed_epoch, + "idempotent": False, + } diff --git a/tests/gateway/test_hosted_room_replicas.py b/tests/gateway/test_hosted_room_replicas.py new file mode 100644 index 0000000000000..cfe550168378d --- /dev/null +++ b/tests/gateway/test_hosted_room_replicas.py @@ -0,0 +1,296 @@ +"""Tests for gateway/hosted_room_replicas.py — replica ingest, promotion, and +stale-authority demotion for hosted Group Chat rooms.""" + +import json + +import pytest + +import gateway.hosted_room_replicas as replicas +import gateway.hosted_rooms as rooms + +USER = {"kind": "user", "id": "tek"} +MEMBERS = [{"kind": "bot", "id": "planner"}, {"kind": "bot", "id": "coder"}] + +AUTH_A = "install:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +AUTH_B = "install:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + + +def _authority_db(tmp_path, name="authority.db"): + return tmp_path / name + + +def _replica_db(tmp_path, name="replica.db"): + return tmp_path / name + + +def _seed_room(db, *, gateway_id=AUTH_A, n_events=3, room_id="room-1"): + rooms.create_room( + db, + room_id=room_id, + name="Field Room", + members=MEMBERS, + authority_gateway_id=gateway_id, + ) + for index in range(n_events): + rooms.append_event( + db, + room_id=room_id, + event_id=f"e{index}", + kind="message.user", + actor=USER, + payload={"text": f"msg {index} 😀"}, + authority_gateway_id=gateway_id, + authority_epoch=1, + ) + return rooms.read_events(db, room_id=room_id, since_seq=0, limit=100) + + +def test_ingest_page_persists_events_and_lineage(tmp_path): + page = _seed_room(_authority_db(tmp_path)) + rdb = _replica_db(tmp_path) + result = replicas.ingest_page( + rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page + ) + assert result["ingested"] == 3 + assert result["stored_seq"] == 3 + assert result["caught_up"] is True + state = replicas.replica_state(rdb, room_id="room-1") + assert state["last_seq"] == 3 + assert state["authority"] == page["authority"] + assert state["members"] == MEMBERS + + +def test_ingest_page_is_idempotent(tmp_path): + page = _seed_room(_authority_db(tmp_path)) + rdb = _replica_db(tmp_path) + replicas.ingest_page( + rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page + ) + again = replicas.ingest_page( + rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page + ) + assert again["ingested"] == 0 + assert again["stored_seq"] == 3 + + +def test_ingest_rejects_sequence_gap(tmp_path): + adb = _authority_db(tmp_path) + _seed_room(adb, n_events=5) + later = rooms.read_events(adb, room_id="room-1", since_seq=2, limit=100) + rdb = _replica_db(tmp_path) + with pytest.raises(replicas.ReplicaGapError): + replicas.ingest_page( + rdb, + room_id="room-1", + room_name="Field Room", + members=MEMBERS, + page=later, + ) + + +def test_ingest_rejects_epoch_regression(tmp_path): + page = _seed_room(_authority_db(tmp_path)) + rdb = _replica_db(tmp_path) + newer = json.loads(json.dumps(page)) + newer["authority"]["epoch"] = 3 + replicas.ingest_page( + rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=newer + ) + stale = json.loads(json.dumps(page)) + stale["authority"]["epoch"] = 2 + with pytest.raises(replicas.ReplicaEpochRegressionError): + replicas.ingest_page( + rdb, + room_id="room-1", + room_name="Field Room", + members=MEMBERS, + page=stale, + ) + + +def test_ingest_requires_authority_stamp(tmp_path): + page = _seed_room(_authority_db(tmp_path)) + page.pop("authority") + with pytest.raises(replicas.ReplicaError): + replicas.ingest_page( + _replica_db(tmp_path), + room_id="room-1", + room_name="Field Room", + members=MEMBERS, + page=page, + ) + + +def test_promote_replica_continues_room_at_next_epoch(tmp_path, monkeypatch): + page = _seed_room(_authority_db(tmp_path)) + rdb = _replica_db(tmp_path) + replicas.ingest_page( + rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page + ) + monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_B) + + promoted = replicas.promote_replica(rdb, room_id="room-1") + assert promoted["authority_gateway_id"] == AUTH_B + assert promoted["authority_epoch"] == 2 + assert promoted["previous_gateway_id"] == AUTH_A + assert promoted["claim_seq"] == 4 + + # The room is now locally authoritative with the full history + claim. + replay = rooms.read_events(rdb, room_id="room-1", since_seq=0, limit=100) + assert [e["seq"] for e in replay["events"]] == [1, 2, 3, 4] + claim = replay["events"][-1] + assert claim["kind"] == "authority.claimed" + assert claim["payload"]["previous_gateway_id"] == AUTH_A + assert claim["payload"]["authority_epoch"] == 2 + assert replay["authority"] == {"gateway_id": AUTH_B, "epoch": 2} + + # New work continues under the new epoch. + rooms.append_event( + rdb, + room_id="room-1", + event_id="post-takeover", + kind="message.user", + actor=USER, + payload={"text": "continuing"}, + authority_gateway_id=AUTH_B, + authority_epoch=2, + ) + + # The old authority's identity/epoch is fenced out. + with pytest.raises(rooms.HostedRoomError): + rooms.append_event( + rdb, + room_id="room-1", + event_id="stale-write", + kind="message.user", + actor=USER, + payload={"text": "stale"}, + authority_gateway_id=AUTH_A, + authority_epoch=1, + ) + + # Replica bookkeeping is consumed by promotion. + with pytest.raises(replicas.ReplicaError): + replicas.replica_state(rdb, room_id="room-1") + + +def test_promote_refuses_when_room_exists_locally(tmp_path, monkeypatch): + db = _authority_db(tmp_path) + page = _seed_room(db) + # Same DB also holds a replica row for the same id — conflict must win. + replicas.ingest_page( + db, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page + ) + monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_B) + with pytest.raises(rooms.RoomConflictError): + replicas.promote_replica(db, room_id="room-1") + + +def test_promote_refuses_when_already_authority(tmp_path, monkeypatch): + page = _seed_room(_authority_db(tmp_path)) + rdb = _replica_db(tmp_path) + replicas.ingest_page( + rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page + ) + monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A) + with pytest.raises(replicas.ReplicaError): + replicas.promote_replica(rdb, room_id="room-1") + + +def test_demote_fences_stale_local_authority(tmp_path, monkeypatch): + adb = _authority_db(tmp_path) + _seed_room(adb) + monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A) + + result = replicas.demote_room( + adb, room_id="room-1", observed_gateway_id=AUTH_B, observed_epoch=2 + ) + assert result["idempotent"] is False + assert result["authority_gateway_id"] == AUTH_B + assert result["authority_epoch"] == 2 + + replay = rooms.read_events(adb, room_id="room-1", since_seq=0, limit=100) + lost = replay["events"][-1] + assert lost["kind"] == "authority.lost" + assert lost["payload"]["authority_gateway_id"] == AUTH_B + assert replay["authority"] == {"gateway_id": AUTH_B, "epoch": 2} + + # Local sends at the stale identity/epoch are now rejected. + with pytest.raises(rooms.HostedRoomError): + rooms.append_event( + adb, + room_id="room-1", + event_id="after-demote", + kind="message.user", + actor=USER, + payload={"text": "stale"}, + authority_gateway_id=AUTH_A, + authority_epoch=1, + ) + + # Repeating the same observation is idempotent. + again = replicas.demote_room( + adb, room_id="room-1", observed_gateway_id=AUTH_B, observed_epoch=2 + ) + assert again["idempotent"] is True + + +def test_demote_rejects_non_superseding_epoch(tmp_path, monkeypatch): + adb = _authority_db(tmp_path) + _seed_room(adb) + monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A) + with pytest.raises(replicas.ReplicaEpochRegressionError): + replicas.demote_room( + adb, room_id="room-1", observed_gateway_id=AUTH_B, observed_epoch=1 + ) + + +def test_full_failover_round_trip(tmp_path, monkeypatch): + """Authority A hosts, replica B follows, A dies, B promotes, A returns + and is fenced + demoted; the room's history survives intact throughout.""" + adb = _authority_db(tmp_path) + rdb = _replica_db(tmp_path) + page = _seed_room(adb, n_events=4) + replicas.ingest_page( + rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page + ) + + # A "dies"; B takes over. + monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_B) + promoted = replicas.promote_replica(rdb, room_id="room-1") + rooms.append_event( + rdb, + room_id="room-1", + event_id="b-work", + kind="message.user", + actor=USER, + payload={"text": "work continues on B"}, + authority_gateway_id=AUTH_B, + authority_epoch=promoted["authority_epoch"], + ) + + # A comes back, observes B's claim, and fences itself. + monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A) + replicas.demote_room( + adb, + room_id="room-1", + observed_gateway_id=AUTH_B, + observed_epoch=promoted["authority_epoch"], + ) + with pytest.raises(rooms.HostedRoomError): + rooms.append_event( + adb, + room_id="room-1", + event_id="a-stale", + kind="message.user", + actor=USER, + payload={"text": "split brain attempt"}, + authority_gateway_id=AUTH_A, + authority_epoch=1, + ) + + # B's room holds the complete history: 4 original + claim + new work. + replay = rooms.read_events(rdb, room_id="room-1", since_seq=0, limit=100) + kinds = [e["kind"] for e in replay["events"]] + assert kinds == ["message.user"] * 4 + ["authority.claimed", "message.user"] + assert replay["authority"]["gateway_id"] == AUTH_B diff --git a/tests/tui_gateway/test_groups_replication_methods.py b/tests/tui_gateway/test_groups_replication_methods.py new file mode 100644 index 0000000000000..9101a8cbbe53c --- /dev/null +++ b/tests/tui_gateway/test_groups_replication_methods.py @@ -0,0 +1,173 @@ +"""Tests for the ``groups.replicate`` / ``groups.promote`` / ``groups.demote`` +JSON-RPC surface — cross-gateway room durability.""" + +from __future__ import annotations + +import pytest + +import tui_gateway.server as srv + +MEMBERS = [{"kind": "bot", "id": "planner"}] + + +@pytest.fixture +def home(tmp_path, monkeypatch): + path = tmp_path / ".hermes" + path.mkdir() + monkeypatch.setenv("HERMES_HOME", str(path)) + return path + + +def _result(envelope): + assert "error" not in envelope, envelope + return envelope["result"] + + +def _error(envelope): + assert "error" in envelope, envelope + return envelope["error"] + + +def _authority_page(tmp_path, gateway_id="install:" + "a" * 32, n=3): + """Build a real room + log on a SEPARATE 'remote authority' DB and return + its replay page, as a replicating client would fetch via groups.log.""" + from gateway import hosted_rooms as rooms + + db = tmp_path / "remote-authority.db" + rooms.create_room( + db, + room_id="room-1", + name="Field Room", + members=MEMBERS, + authority_gateway_id=gateway_id, + ) + for index in range(n): + rooms.append_event( + db, + room_id="room-1", + event_id=f"e{index}", + kind="message.user", + actor={"kind": "user", "id": "tek"}, + payload={"text": f"msg {index}"}, + authority_gateway_id=gateway_id, + authority_epoch=1, + ) + return rooms.read_events(db, room_id="room-1", since_seq=0, limit=100) + + +def test_capabilities_advertise_replication(home): + result = _result(srv._methods["groups.capabilities"](1, {})) + assert "log_replication" in result["features"] + assert "authority_takeover" in result["features"] + for name in ( + "groups.replicate", + "groups.replica_state", + "groups.promote", + "groups.demote", + ): + assert name in result["methods"] + assert name in srv._LONG_HANDLERS + + +def test_replicate_then_state_roundtrip(home, tmp_path): + page = _authority_page(tmp_path) + result = _result( + srv._methods["groups.replicate"]( + 1, + { + "room_id": "room-1", + "room_name": "Field Room", + "members": MEMBERS, + "page": page, + }, + ) + ) + assert result["ingested"] == 3 + state = _result(srv._methods["groups.replica_state"](2, {"room_id": "room-1"})) + assert state["last_seq"] == 3 + assert state["authority"] == page["authority"] + + +def test_promote_requires_confirm_and_takes_over(home, tmp_path): + page = _authority_page(tmp_path) + _result( + srv._methods["groups.replicate"]( + 1, + { + "room_id": "room-1", + "room_name": "Field Room", + "members": MEMBERS, + "page": page, + }, + ) + ) + + refused = _error(srv._methods["groups.promote"](2, {"room_id": "room-1"})) + assert refused["code"] == 4118 + + promoted = _result( + srv._methods["groups.promote"](3, {"room_id": "room-1", "confirm": True}) + ) + assert promoted["authority_epoch"] == 2 + assert promoted["previous_gateway_id"] == page["authority"]["gateway_id"] + + # The room is now hosted locally with full history + claim event. + log = _result(srv._methods["groups.log"](4, {"room_id": "room-1"})) + kinds = [event["kind"] for event in log["events"]] + assert kinds == ["message.user"] * 3 + ["authority.claimed"] + assert log["authority"]["epoch"] == 2 + + +def test_demote_fences_local_room_against_newer_epoch(home): + from gateway.hosted_rooms import local_authority_gateway_id + + _result( + srv._methods["groups.create"]( + 1, + {"room_id": "room-1", "name": "Local room", "members": MEMBERS}, + ) + ) + observed_gateway = "install:" + "b" * 32 + result = _result( + srv._methods["groups.demote"]( + 2, + { + "room_id": "room-1", + "observed_gateway_id": observed_gateway, + "observed_epoch": 2, + }, + ) + ) + assert result["idempotent"] is False + assert result["authority_gateway_id"] == observed_gateway + + # Local sends at the stale authority now fail. + envelope = srv._methods["groups.send"]( + 3, + { + "room_id": "room-1", + "event_id": "stale-send", + "actor": {"kind": "user", "id": "tek"}, + "payload": {"text": "should fence"}, + }, + ) + assert "error" in envelope + assert local_authority_gateway_id() != observed_gateway + + +def test_replicate_rejects_gapped_page(home, tmp_path): + from gateway import hosted_rooms as rooms + + _authority_page(tmp_path, n=5) + db = tmp_path / "remote-authority.db" + gapped = rooms.read_events(db, room_id="room-1", since_seq=2, limit=100) + envelope = srv._methods["groups.replicate"]( + 1, + { + "room_id": "room-1", + "room_name": "Field Room", + "members": MEMBERS, + "page": gapped, + }, + ) + assert _error(envelope)["code"] == 4116 diff --git a/tui_gateway/methods_groups.py b/tui_gateway/methods_groups.py index c2b6ecb4e9f30..9c014d797c44e 100644 --- a/tui_gateway/methods_groups.py +++ b/tui_gateway/methods_groups.py @@ -19,6 +19,10 @@ "groups.send", "groups.log", "groups.disband", + "groups.replicate", + "groups.replica_state", + "groups.promote", + "groups.demote", }) @@ -46,6 +50,8 @@ def _(rid, params: dict) -> dict: "replayable_disband", "typed_events", "actor_identity", + "log_replication", + "authority_takeover", ], "methods": [ "groups.capabilities", @@ -55,6 +61,10 @@ def _(rid, params: dict) -> dict: "groups.send", "groups.log", "groups.disband", + "groups.replicate", + "groups.replica_state", + "groups.promote", + "groups.demote", ], "max_log_limit": MAX_LOG_LIMIT, }, @@ -262,5 +272,97 @@ def _(rid, params: dict) -> dict: return _err(rid, 5113, str(exc)) +@method("groups.replicate") +def _(rid, params: dict) -> dict: + """Persist one authority-stamped replay page into the local replica store. + + ``page`` is the verbatim ``groups.log`` result read from the room's + authority gateway; ingest is idempotent and refuses sequence gaps and + authority-epoch regressions. + """ + from gateway.hosted_room_replicas import ReplicaError, ingest_page + from gateway.hosted_rooms import default_db_path + + try: + result = ingest_page( + default_db_path(), + room_id=params.get("room_id"), + room_name=params.get("room_name"), + members=params.get("members"), + page=params.get("page"), + ) + return _ok(rid, result) + except ReplicaError as exc: + return _err(rid, 4116, str(exc)) + except Exception as exc: + return _err(rid, 5116, str(exc)) + + +@method("groups.replica_state") +def _(rid, params: dict) -> dict: + """Report the local replica's coverage and authority lineage.""" + from gateway.hosted_room_replicas import ReplicaError, replica_state + from gateway.hosted_rooms import default_db_path + + try: + return _ok(rid, replica_state(default_db_path(), room_id=params.get("room_id"))) + except ReplicaError as exc: + return _err(rid, 4117, str(exc)) + except Exception as exc: + return _err(rid, 5117, str(exc)) + + +@method("groups.promote") +def _(rid, params: dict) -> dict: + """Continue a replicated room on THIS gateway at ``epoch + 1``. + + Requires ``confirm: true`` — the caller asserts the previous authority can + no longer commit (explicit user action; a lease/quorum driver later). + """ + from gateway.hosted_room_replicas import ReplicaError, promote_replica + from gateway.hosted_rooms import HostedRoomError, default_db_path + + if params.get("confirm") is not True: + return _err( + rid, + 4118, + "promotion requires confirm=true acknowledging the previous " + "authority can no longer commit", + ) + try: + result = promote_replica( + default_db_path(), + room_id=params.get("room_id"), + reason=params.get("reason", "authority-unreachable"), + ) + return _ok(rid, result) + except ReplicaError as exc: + return _err(rid, 4118, str(exc)) + except HostedRoomError as exc: + return _err(rid, 4118, str(exc)) + except Exception as exc: + return _err(rid, 5118, str(exc)) + + +@method("groups.demote") +def _(rid, params: dict) -> dict: + """Fence this gateway's stale room authority against a proven newer epoch.""" + from gateway.hosted_room_replicas import ReplicaError, demote_room + from gateway.hosted_rooms import default_db_path + + try: + result = demote_room( + default_db_path(), + room_id=params.get("room_id"), + observed_gateway_id=params.get("observed_gateway_id"), + observed_epoch=params.get("observed_epoch"), + ) + return _ok(rid, result) + except ReplicaError as exc: + return _err(rid, 4119, str(exc)) + except Exception as exc: + return _err(rid, 5119, str(exc)) + + def register(server) -> None: _registry.install(server)