From 90c2fbaf0d374dec76ad75e397dfeb8e3e7dfc47 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:03:31 -0700 Subject: [PATCH] fix(gateway): Group Chats handed from the standalone backend take Sends again A room the gateway took over from the standalone backend (hosted_rooms_gateway_import) refused every groups.send with invalid_params after storing the message, and no member was woken. The handoff commits two authority transfers: the standalone copy's fence (authority.lost, install -> install:gateway-store, epoch 2) and the gateway store's claim (authority.claimed, gateway-store -> install, epoch 3). The history reader and the work-record lineage took span boundaries from authority.claimed rows only, so the chain started at the gateway store and the epoch-3 claim did not extend it: "authority claim does not extend its predecessor" from every prepare_room, on Send and on each driver pass. authority.lost is the outgoing owner's committed record of the same transfer (same system actor, same previous/next/epoch payload), so both readers now count it as a span boundary under the same checks. The chain reads install@1 -> gateway-store@2 -> install@3. Nothing is rewritten: rooms users already imported (state.db epoch 3, shared-state.db epoch 2) work as they are. A fork (a claim whose previous owner is not the recorded holder, a skipped epoch) is still refused. Send also stops reporting a stored message as failed: once the user event commits it is the Send's result, and a failure preparing the room afterwards is logged and left to the room driver, which runs the same preparation on its next pass (#143239). --- gateway/hosted_room_authority_history.py | 13 +- gateway/hosted_room_work_lineage.py | 4 +- .../gateway/test_hosted_room_handoff_send.py | 111 ++++++++++++++++++ tui_gateway/hosted_room_service.py | 14 ++- 4 files changed, 133 insertions(+), 9 deletions(-) create mode 100644 tests/gateway/test_hosted_room_handoff_send.py diff --git a/gateway/hosted_room_authority_history.py b/gateway/hosted_room_authority_history.py index 3f53d8f19a..201a83ddbc 100644 --- a/gateway/hosted_room_authority_history.py +++ b/gateway/hosted_room_authority_history.py @@ -14,6 +14,13 @@ from typing import Any from gateway.hosted_rooms_common import identifier MAX_AUTHORITY_SPANS = 1024 +#: Committed authority transfers, each the first event of its span. ``authority.claimed`` is the new +#: owner's record; ``authority.lost`` is the record the outgoing owner commits when it is fenced to +#: a proven newer epoch (``hosted_room_replicas.demote_room``; the standalone -> gateway handoff in +#: ``hosted_rooms_gateway_import``). Both carry the same previous/next/epoch payload and are +#: checked alike, so a store whose log holds its own fence followed by a later claim (a handed-off +#: room, a returning gateway that reclaims) reads as one unbroken chain. +TRANSFER_KINDS = ("authority.claimed", "authority.lost") class AuthorityHistoryError(ValueError): @@ -77,14 +84,14 @@ def origin_gateway_id(room: dict[str, Any]) -> str: def read_history_locked(conn, room_id: str, *, gateway_id: str, epoch: int) -> list[dict[str, Any]] | None: - """Read committed claims within the caller's room-state transaction. + """Read committed authority transfers within the caller's room-state transaction. Legacy adoption is left on its existing path; it is not retroactively certified as a history of physical-host transfers. """ rows = conn.execute( - """SELECT seq,authority_epoch,actor_json,payload_json FROM hosted_room_events - WHERE room_id=? AND kind='authority.claimed' ORDER BY seq LIMIT ?""", + f"""SELECT seq,authority_epoch,actor_json,payload_json FROM hosted_room_events + WHERE room_id=? AND kind IN {TRANSFER_KINDS} ORDER BY seq LIMIT ?""", (room_id, MAX_AUTHORITY_SPANS), ).fetchall() if len(rows) >= MAX_AUTHORITY_SPANS: diff --git a/gateway/hosted_room_work_lineage.py b/gateway/hosted_room_work_lineage.py index c9c00d6eec..10d24064b6 100644 --- a/gateway/hosted_room_work_lineage.py +++ b/gateway/hosted_room_work_lineage.py @@ -1,5 +1,5 @@ """Validate passive evidence against committed authority spans, never elect an owner.""" -from gateway.hosted_room_authority_history import at_sequence, read_history_locked, validate_history +from gateway.hosted_room_authority_history import TRANSFER_KINDS, at_sequence, read_history_locked, validate_history def source_prefix_locked(conn, room_id, authority, seq): @@ -17,7 +17,7 @@ def source_prefix_locked(conn, room_id, authority, seq): boundary = span.epoch > 1 and count == span.from_seq if (event["seq"] != count or event["authority_epoch"] != span.epoch or (actor.get("kind") == "gateway" and actor.get("id") != span.gateway_id) - or (event["kind"] == "authority.claimed") != boundary): + or (event["kind"] in TRANSFER_KINDS) != boundary): raise WorkRecordPrefixError("work record history prefix is not verified") if count != seq: raise WorkRecordPrefixError("work record history prefix is unavailable") diff --git a/tests/gateway/test_hosted_room_handoff_send.py b/tests/gateway/test_hosted_room_handoff_send.py new file mode 100644 index 0000000000..3dd4fa8399 --- /dev/null +++ b/tests/gateway/test_hosted_room_handoff_send.py @@ -0,0 +1,111 @@ +"""A Group Chat the gateway took over keeps answering, and a stored Send is never reported as failed. + +#145960: rooms handed from the standalone backend to the gateway (``hosted_rooms_gateway_import``) +refused every Send with ``invalid_params`` after storing the message, and no member was woken. +#143239: any failure after the user event committed came back to the client as a failed Send. +""" + +from __future__ import annotations + +import sqlite3 +import threading +import time +from pathlib import Path +from types import SimpleNamespace + +from gateway import hosted_room_driver, hosted_rooms +from gateway.hosted_rooms_gateway_import import handoff_gateway_id, import_standalone_rooms +from tui_gateway.hosted_room_service import HostedRoomService + +MEMBERS = [{"member_id": "default", "profile": "default", "handle": "hermes"}, + {"member_id": "ops", "profile": "ops", "handle": "ops"}] + + +class _RPC: + def __init__(self) -> None: + self.sessions, self.submits = {}, [] + + def resolve_exact(self, *, profile, title, source): + return self.sessions.get((profile, title)) + + def create(self, *, profile, title, source): + self.sessions[(profile, title)] = session = {"session_id": f"{profile}-session", "title": title} + return session + + def resume(self, *, profile, session_id, source): + return {"session_id": session_id} + + def submit(self, **kwargs): + self.submits.append(kwargs["profile"]) + kwargs["on_terminal"]({"status": "settled", "text": f"reply from {kwargs['profile']}"}) + return {"accepted": True} + + def history(self, **kwargs): + return [] + + def info(self, **kwargs): + return {"active": False, "task_id": None} + + +def _service(db_path: Path) -> tuple[HostedRoomService, _RPC]: + service = HostedRoomService( + SimpleNamespace(_methods={}, _sessions={}, _sessions_lock=threading.Lock()), db_path=db_path) + service.rpc = service.runtime.rpc = rpc = _RPC() + service.local_profiles = lambda: ("default", "ops") + return service, rpc + + +def _kinds(db: Path, room_id: str) -> list[str]: + return [e["kind"] for e in hosted_rooms.read_events(db, room_id=room_id, since_seq=0)["events"]] + + +def _authority(db: Path, room_id: str) -> tuple[str, int]: + with sqlite3.connect(db) as conn: + return conn.execute("SELECT authority_gateway_id, authority_epoch FROM hosted_rooms WHERE room_id=?", + (room_id,)).fetchone() + + +def test_a_room_handed_to_the_gateway_takes_sends_and_wakes_its_members(tmp_path): + source, target = tmp_path / "shared-state.db", tmp_path / "state.db" + with sqlite3.connect(target) as conn: # the gateway's SessionDB table the owner claim lives in + conn.execute("CREATE TABLE state_meta (key TEXT PRIMARY KEY, value TEXT)") + local = hosted_rooms.local_authority_gateway_id() + hosted_rooms.create_room(source, room_id="room-desk", name="Desk", members=MEMBERS, + authority_gateway_id=local) + assert import_standalone_rooms(target) == 1 + # The stores every room the shipped import handed over is in (#145960's report): nothing to migrate. + assert _authority(source, "room-desk") == (handoff_gateway_id(), 2) + assert _authority(target, "room-desk") == (local, 3) + + service, rpc = _service(target) + sent = service.send(room_id="room-desk", event_id="user-1", + payload={"text": "@ops inspect", "thread_id": "thread-1"}) + assert sent["kind"] == "message.user" and sent["authority_epoch"] == 3 + assert len(hosted_room_driver.list_tasks(target, room_id="room-desk", status="queued")) == 1 + + service.start() + try: + deadline = time.monotonic() + 5 + while "message.member" not in _kinds(target, "room-desk") and time.monotonic() < deadline: + time.sleep(0.01) + finally: + assert service.stop(timeout=5.0) + assert rpc.submits == ["ops"] + assert _kinds(target, "room-desk").count("message.member") == 1 + + +def test_a_send_already_stored_is_answered_as_sent_when_preparing_the_room_fails(tmp_path): + db = tmp_path / "state.db" + service, _ = _service(db) + service.create_room(room_id="room-1", name="Room", members=MEMBERS) + + def storage_full(binding): + raise hosted_rooms.HostedRoomError("Group Chat storage is full on this host.") + + service.prepare_room = storage_full + sent = service.send(room_id="room-1", event_id="user-1", payload={"text": "hello", "thread_id": "t"}) + service.bindings = lambda: () # the binding list has not caught up with the room yet + again = service.send(room_id="room-1", event_id="user-2", payload={"text": "again", "thread_id": "t"}) + + assert [sent["kind"], again["kind"]] == ["message.user", "message.user"] + assert _kinds(db, "room-1").count("message.user") == 2 diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index 34f7e599bd..d5c17772f2 100644 --- a/tui_gateway/hosted_room_service.py +++ b/tui_gateway/hosted_room_service.py @@ -536,10 +536,16 @@ class HostedRoomService: event = append_user_event( self, room_id=room_id, event_id=event_id, payload=normalized, gateway_id=gateway_id, epoch=epoch) - binding = next((b for b in self.bindings() if b.room_id == room_id), None) - if binding is None: - raise hosted_rooms.RoomNotFoundError("hosted room not found") - self.prepare_room(binding) + # The committed event is the Send's result: a failure from here on is the room driver's, + # which runs the same preparation on its next pass and records the error there. Raising + # would tell the client a message it can already see in the room was not sent (#143239). + try: + binding = next((b for b in self.bindings() if b.room_id == room_id), None) + if binding is not None: + self.prepare_room(binding) + except Exception: + logger.warning("hosted room %s: preparing the room after a stored Send failed; " + "the room driver retries it", room_id, exc_info=True) self.runtime.wakeup() return event