From 2023f6cc4d28bbb22653c19c9ae0f5611eccb3e9 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:53:20 -0700 Subject: [PATCH] fix(gateway): a Group Chat whose members can't be woken says so once, with Retry Review of #145981 (P2): after a Send is stored, a room preparation that keeps failing (a member whose profile was deleted, a full room log) retried on every driver pass forever, and the error only overwrote the process-wide driver last_error that no client renders. The user saw "sent" and no member answered. Now each room counts consecutive preparation failures (contention and ownership moves never count). The third one appends ONE member.unavailable event to the room ("Members could not be woken: "), which Desktop and the Webapp already render, using the control headroom so a full room can still say why. The room's status lists a Retry row (groups.retry, task_id prepare:) and a per-room last_error, and the driver stops preparing that room until the next Send or Retry. A failed Retry stays stalled and is the RPC's error, without a second notice. Refs #145977 --- apps/shared/src/gateway-contract.generated.ts | 3 +- apps/shared/src/gateway-contract.openrpc.json | 14 ++- gateway/hosted_rooms.py | 7 +- .../gateway/test_hosted_room_handoff_send.py | 58 +++++++++ tui_gateway/contracts/groups_bot_relay.py | 7 +- tui_gateway/hosted_room_prepare_stall.py | 110 ++++++++++++++++++ tui_gateway/hosted_room_service.py | 48 +++++++- 7 files changed, 235 insertions(+), 12 deletions(-) create mode 100644 tui_gateway/hosted_room_prepare_stall.py diff --git a/apps/shared/src/gateway-contract.generated.ts b/apps/shared/src/gateway-contract.generated.ts index fefb528d1c..41613fa880 100644 --- a/apps/shared/src/gateway-contract.generated.ts +++ b/apps/shared/src/gateway-contract.generated.ts @@ -1631,7 +1631,7 @@ export interface GroupsStateResult { room: Room driver_status?: RoomDriverStatus | null } -/** ``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt (``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``) or the driver's approval action (``kind: approval`` + run/session/approval context). ``member_activity`` lists live tasks in admission order (absent from older gateways). */ +/** ``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt (``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``) or the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled room preparation's Retry (``task_id`` ``prepare:``, ``member_id`` the members not woken). ``member_activity`` lists live tasks in admission order (absent from older gateways). ``last_error`` is this room's last preparation failure, null once one succeeds. */ export interface RoomDriverStatus { running: boolean working: boolean @@ -1639,6 +1639,7 @@ export interface RoomDriverStatus { counts: Record pending_actions: Record[] member_activity?: RoomMemberActivity[] + last_error?: string | null peer_routes: PeerRouteStatus[] } /** One live room task: ``status`` is ``queued``, ``running`` or ``stopping``. */ diff --git a/apps/shared/src/gateway-contract.openrpc.json b/apps/shared/src/gateway-contract.openrpc.json index 7d750494a5..5fd45a6dcb 100644 --- a/apps/shared/src/gateway-contract.openrpc.json +++ b/apps/shared/src/gateway-contract.openrpc.json @@ -34015,7 +34015,7 @@ }, "RoomDriverStatus": { "additionalProperties": false, - "description": "``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt\n(``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``)\nor the driver's approval action (``kind: approval`` + run/session/approval context).\n``member_activity`` lists live tasks in admission order (absent from older gateways).", + "description": "``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt\n(``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``)\nor the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled\nroom preparation's Retry (``task_id`` ``prepare:``, ``member_id`` the members not woken).\n``member_activity`` lists live tasks in admission order (absent from older gateways).\n``last_error`` is this room's last preparation failure, null once one succeeds.", "properties": { "running": { "title": "Running", @@ -34051,6 +34051,18 @@ "title": "Member Activity", "type": "array" }, + "last_error": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Last Error" + }, "peer_routes": { "items": { "$ref": "#/components/schemas/PeerRouteStatus" diff --git a/gateway/hosted_rooms.py b/gateway/hosted_rooms.py index 34f3abe4b2..01a478f952 100644 --- a/gateway/hosted_rooms.py +++ b/gateway/hosted_rooms.py @@ -956,9 +956,10 @@ def rename_room(db_path: DbPath, *, room_id: Any, event_id: Any, name: Any, now: def append_event( db_path: DbPath, *, room_id: Any, event_id: Any, kind: Any, actor: Any, payload: Any, authority_gateway_id: Any = None, authority_epoch: Any = None, now: float | None = None, - expected_latest_seq: int | None = None) -> dict[str, Any]: + expected_latest_seq: int | None = None, control_reserve: bool = False) -> dict[str, Any]: """Append one immutable event and allocate its per-room sequence atomically; repeating an ``event_id`` - with identical content returns the original, different content fails closed.""" + with identical content returns the original, different content fails closed. ``control_reserve`` lets a + bounded gateway notice use the control headroom, so a room whose log is full can still say why.""" room_id = _room_id(room_id) event_id = _event_id(event_id) if expected_latest_seq is not None: @@ -993,7 +994,7 @@ def append_event( manifest=json.loads(payload_json).get("attachments", []), now=now) event_bytes = _insert_event( conn, room, room_id, seq, event_id, kind, actor_json, authority_epoch, payload_json, now, - allow_control=kind in _CONTROL_EVENT_KINDS) + allow_control=kind in _CONTROL_EVENT_KINDS or (control_reserve and normalized_actor["kind"] == "gateway")) _fenced_update(conn, """UPDATE hosted_rooms SET next_seq=?, event_bytes=event_bytes+?, updated_at=? WHERE room_id=? AND next_seq=?""", (seq + 1, event_bytes, now, room_id, seq), RuntimeError("hosted room sequence advance lost its write fence")) diff --git a/tests/gateway/test_hosted_room_handoff_send.py b/tests/gateway/test_hosted_room_handoff_send.py index 3a6a6e8b5a..1aad21aaa2 100644 --- a/tests/gateway/test_hosted_room_handoff_send.py +++ b/tests/gateway/test_hosted_room_handoff_send.py @@ -81,6 +81,11 @@ def test_a_room_handed_to_the_gateway_takes_sends_and_wakes_its_members(tmp_path 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 + # #145977: a Desktop that journaled the refused Send retries the exact same intent; it is + # acknowledged as the stored message (so the client retires it), never stored twice. + again = service.send(room_id="room-desk", event_id="user-1", + payload={"text": "@ops inspect", "thread_id": "thread-1"}) + assert again["seq"] == sent["seq"] and _kinds(target, "room-desk").count("message.user") == 1 assert len(hosted_room_driver.list_tasks(target, room_id="room-desk", status="queued")) == 1 service.start() @@ -107,3 +112,56 @@ def test_a_send_already_stored_is_answered_as_sent_when_preparing_the_room_fails assert [sent["kind"], again["kind"]] == ["message.user", "message.user"] assert _kinds(db, "room-1").count("message.user") == 2 + + +def test_a_room_whose_members_cannot_be_woken_says_so_once_and_stops_retrying(tmp_path): + db = tmp_path / "state.db" + service, _ = _service(db) + service.create_room(room_id="room-1", name="Room", members=MEMBERS) + service.local_profiles = lambda: ("default", "dev", "ops") + service.create_room(room_id="room-2", name="Other", members=[ + MEMBERS[0], {"member_id": "dev", "profile": "dev", "handle": "dev"}]) + service.local_profiles = lambda: ("default", "dev") # the ops member's profile was deleted + attempts: list[str] = [] + real_prepare = service.prepare_room + + def counted(binding): + attempts.append(binding.room_id) + return real_prepare(binding) + + service.prepare_room = counted + sent = service.send(room_id="room-1", event_id="user-1", payload={"text": "@ops hi", "thread_id": "t"}) + assert sent["kind"] == "message.user" + + service.runtime.poll_interval_seconds = 0.02 + service.start() + try: + wait_until(lambda: "member.unavailable" in _kinds(db, "room-1"), timeout=5, what="stall notice") + settled = attempts.count("room-1") + wait_until(lambda: service.runtime.status()["cycles"] > 30, timeout=5, what="more driver passes") + finally: + assert service.stop(timeout=5.0) + assert attempts.count("room-1") == settled == 3 # bounded: no retries after the stall + events = hosted_rooms.read_events(db, room_id="room-1", since_seq=0)["events"] + notices = [e for e in events if e["kind"] == "member.unavailable"] + assert len(notices) == 1 and "ops" in notices[0]["payload"]["error"] + status = service.status("room-1") + assert "ops" in status["last_error"] and service.status("room-2")["last_error"] is None + [retry] = [a for a in status["pending_actions"] if a["task_id"].startswith("prepare:")] + + # Retry while the profile is still missing fails loudly and adds no second notice. + try: + service.retry_room_task("room-1", task_id=retry["task_id"], member_id=retry["member_id"], + execution_generation=retry["execution_generation"]) + raise AssertionError("retry of a still-broken room must fail") + except Exception as exc: + assert "ops" in str(exc) + assert _kinds(db, "room-1").count("member.unavailable") == 1 + + service.local_profiles = lambda: ("default", "dev", "ops") # restored + receipt = service.retry_room_task("room-1", task_id=retry["task_id"], member_id=retry["member_id"], + execution_generation=1) + assert receipt["identity"].task_id == retry["task_id"] + assert service.status("room-1")["last_error"] is None + assert not [a for a in service.status("room-1")["pending_actions"] if a["task_id"].startswith("prepare:")] + assert len(hosted_room_driver.list_tasks(db, room_id="room-1", status="queued")) == 1 diff --git a/tui_gateway/contracts/groups_bot_relay.py b/tui_gateway/contracts/groups_bot_relay.py index 041328af68..8cd707f0a6 100644 --- a/tui_gateway/contracts/groups_bot_relay.py +++ b/tui_gateway/contracts/groups_bot_relay.py @@ -224,8 +224,10 @@ class RoomMemberActivity(Result): class RoomDriverStatus(Result): """``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt (``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``) - or the driver's approval action (``kind: approval`` + run/session/approval context). - ``member_activity`` lists live tasks in admission order (absent from older gateways).""" + or the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled + room preparation's Retry (``task_id`` ``prepare:``, ``member_id`` the members not woken). + ``member_activity`` lists live tasks in admission order (absent from older gateways). + ``last_error`` is this room's last preparation failure, null once one succeeds.""" running: bool working: bool @@ -233,6 +235,7 @@ class RoomDriverStatus(Result): counts: dict[str, int] pending_actions: list[dict[str, JsonValue]] member_activity: list[RoomMemberActivity] = Field(default_factory=list) + last_error: str | None = None peer_routes: list[PeerRouteStatus] diff --git a/tui_gateway/hosted_room_prepare_stall.py b/tui_gateway/hosted_room_prepare_stall.py new file mode 100644 index 0000000000..41d9a766ca --- /dev/null +++ b/tui_gateway/hosted_room_prepare_stall.py @@ -0,0 +1,110 @@ +"""Bound the room driver's retries of a room preparation that keeps failing, and say so in the room. + +A Send returns the stored message even when preparing the room afterwards fails (#143239), and the +driver retries the preparation on its next passes. A failure that never clears (a member whose +profile was deleted, a full room log) would otherwise retry forever with the error only in the +gateway log: the user sees "sent" and no member ever answers. After ``STALL_ATTEMPTS`` consecutive +failures the room gets ONE ``member.unavailable`` event (Desktop and the Webapp render it), the +room's status offers Retry, and the driver stops preparing that room until the next Send or Retry. +""" + +from __future__ import annotations + +import logging +import sqlite3 +import threading +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from gateway import hosted_room_driver as driver +from gateway import hosted_rooms + +logger = logging.getLogger(__name__) + +STALL_ATTEMPTS = 3 +RETRY_TASK_PREFIX = "prepare:" +_EVENT_PREFIX = "prepare-stalled:" +_MAX_ERROR_CHARS = 300 +# Contention or an ownership move, not a broken room: never counted toward the stall. +_TRANSIENT = ( + sqlite3.OperationalError, hosted_rooms.AuthorityConflictError, hosted_rooms.EventCursorConflictError, + hosted_rooms.RoomNotFoundError, driver.LeaseHeldError, driver.StaleLeaseError, driver.RoomUnavailableError) + + +@dataclass +class _Streak: + failures: int = 0 + error: str | None = None + stalled: bool = False + + +def _describe(exc: BaseException) -> str: + text = str(exc).strip() or type(exc).__name__ + return text if len(text) <= _MAX_ERROR_CHARS else text[:_MAX_ERROR_CHARS - 1] + "…" + + +class PrepareStallTracker: + """Per-room consecutive preparation failures; thread-safe (driver room threads and RPC threads).""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._rooms: dict[str, _Streak] = {} + + def stalled(self, room_id: str) -> bool: + with self._lock: + streak = self._rooms.get(room_id) + return bool(streak and streak.stalled) + + def last_error(self, room_id: str) -> str | None: + with self._lock: + streak = self._rooms.get(room_id) + return streak.error if streak else None + + def clear(self, room_id: str) -> None: + with self._lock: + self._rooms.pop(room_id, None) + + def failed(self, room_id: str, exc: BaseException) -> bool: + """Record one failure; True exactly once, when this failure stalls the room.""" + with self._lock: + streak = self._rooms.setdefault(room_id, _Streak()) + streak.error = _describe(exc) + if isinstance(exc, _TRANSIENT) or streak.stalled: + return False + streak.failures += 1 + streak.stalled = streak.failures >= STALL_ATTEMPTS + return streak.stalled + + def stall(self, room_id: str, exc: BaseException) -> None: + """A failed explicit Retry: stalled again at once, without a second room event.""" + with self._lock: + self._rooms[room_id] = _Streak(failures=STALL_ATTEMPTS, error=_describe(exc), stalled=True) + + + +def retry_action(room_id: str, room: Mapping[str, Any]) -> dict[str, Any]: + """A stalled room's Retry row in ``groups.state``; ``member_id`` names who was not woken.""" + names = [str(m.get("handle") or m.get("member_id") or "") for m in room.get("members") or ()] + return {"kind": "retry", "member_id": ", ".join(n for n in names if n) or room_id, + "task_id": f"{RETRY_TASK_PREFIX}{room_id}", "execution_generation": 1} + + +def report_stall(db_path: Any, room_id: str, exc: BaseException) -> None: + """Append the room's one stall notice; a stall already reported last in the log is not repeated.""" + try: + room = hosted_rooms.room_state(db_path, room_id=room_id) + latest_seq = int(room["latest_seq"]) + tail = hosted_rooms.read_events( + db_path, room_id=room_id, since_seq=max(0, latest_seq - 1), limit=1)["events"] + if tail and str(tail[-1]["event_id"]).startswith(_EVENT_PREFIX): + return + gateway_id, epoch = str(room["authority_gateway_id"]), int(room["authority_epoch"]) + hosted_rooms.append_event( + db_path, room_id=room_id, event_id=f"{_EVENT_PREFIX}{latest_seq}", kind="member.unavailable", + actor={"kind": "gateway", "id": gateway_id}, + payload={"error": f"Members could not be woken: {_describe(exc)}", "reason_code": "prepare_failed"}, + authority_gateway_id=gateway_id, authority_epoch=epoch, control_reserve=True) + except Exception: + # The Retry row in the room's status still shows the stall. + logger.warning("hosted room %s: recording the stalled preparation failed", room_id, exc_info=True) diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index d5c17772f2..faafbf3434 100644 --- a/tui_gateway/hosted_room_service.py +++ b/tui_gateway/hosted_room_service.py @@ -21,6 +21,7 @@ from gateway import hosted_rooms from gateway.hosted_room_policy_checkpoint import HostedRoomPolicyCheckpoint, PolicySnapshot from gateway.hosted_room_peer import ( GatewayRoomCatalog, HostedMemberDispatch, PROTOCOL_VERSION, room_grant_needs_dispatch_refresh) +from tui_gateway import hosted_room_prepare_stall as prepare_stall from tui_gateway.hosted_room_driver import HostedRoomBinding, HostedRoomRuntime from tui_gateway.hosted_room_server_rpc import HostedRoomServerRPC from tui_gateway.hosted_room_peer_http import ( @@ -119,6 +120,7 @@ class HostedRoomService: self._peer_route_status: dict[tuple[str, str], str] = {} self.peer_routes: dict[tuple[str, str], PeerMemberRoute] = {} self.peer_clients: dict[tuple[str, str], Any] = {} + self._prepare_stalls = prepare_stall.PrepareStallTracker() try: self._load_stored_links() except Exception as exc: @@ -132,7 +134,7 @@ class HostedRoomService: self.runtime = HostedRoomRuntime( db_path=self.db_path, rooms=self.bindings, rpc=self.rpc, transport_resolver=self._resolve_member_transport, turn_lock=self._turn_lock, - prepare_room=self.prepare_room, publish_terminal=self.publish_terminal, + prepare_room=self._prepare_tracked, publish_terminal=self.publish_terminal, pending_action=self._set_pending_action, poll_interval_seconds=_HOSTED_ROOM_IDLE_FALLBACK_SECONDS, active_poll_interval_seconds=_HOSTED_ROOM_ACTIVE_POLL_SECONDS, @@ -511,8 +513,22 @@ class HostedRoomService: elif decision.status in {"settled", "bounded"}: self._append_room_status(room, decision) + def _prepare_tracked(self, binding: HostedRoomBinding, *, retry_stalled: bool = False) -> None: + """``prepare_room`` with its failures counted per room: a stalled room is skipped (not retried) + until the next Send or Retry, and is reported once in the room (hosted_room_prepare_stall).""" + stalls = self._prepare_stalls + if stalls.stalled(binding.room_id) and not retry_stalled: + return + try: + self.prepare_room(binding) + except Exception as exc: + if stalls.failed(binding.room_id, exc): + prepare_stall.report_stall(self.db_path, binding.room_id, exc) + raise + stalls.clear(binding.room_id) + def publish_terminal(self, binding: HostedRoomBinding, _task: Mapping[str, Any]) -> None: - self.prepare_room(binding) + self._prepare_tracked(binding) self.runtime.wakeup() def create_room(self, *, room_id: str, name: str, members: Any, admit: Any = None) -> dict[str, Any]: @@ -537,12 +553,14 @@ class HostedRoomService: self, room_id=room_id, event_id=event_id, payload=normalized, gateway_id=gateway_id, epoch=epoch) # 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). + # which runs the same preparation on its next passes and, if it keeps failing, says so in + # the room once. Raising would tell the client a message it can already see in the room was + # not sent (#143239). A new message starts a fresh attempt streak for a stalled room. + self._prepare_stalls.clear(room_id) 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) + self._prepare_tracked(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) @@ -575,6 +593,8 @@ class HostedRoomService: self, room_id: str, *, task_id: str, member_id: str | None = None, execution_generation: int | None = None) -> dict[str, Any]: """Retry one uncertain or deferred task only after explicit user action.""" + if task_id.startswith(prepare_stall.RETRY_TASK_PREFIX): + return self._retry_preparation(room_id, task_id) candidates = self._list_tasks(room_id, _RETRYABLE_STATUSES) task = next((c for c in candidates if c["identity"].task_id == task_id and member_id in (None, task_member_id(c)) @@ -584,6 +604,21 @@ class HostedRoomService: return self.runtime.retry_indeterminate( task["identity"], execution_generation=task["execution_generation"]) + def _retry_preparation(self, room_id: str, task_id: str) -> dict[str, Any]: + """The stalled room's Retry: prepare once now; a failure stays stalled and is the RPC's error.""" + binding = next((b for b in self.bindings() if b.room_id == room_id), None) + if (binding is None or task_id != f"{prepare_stall.RETRY_TASK_PREFIX}{room_id}" + or not self._prepare_stalls.stalled(room_id)): + raise driver.InvalidTaskTransitionError("no retryable room task matches task_id") + try: + self._prepare_tracked(binding, retry_stalled=True) + except Exception as exc: + self._prepare_stalls.stall(room_id, exc) + raise + self.runtime.wakeup() + return {"identity": driver.TaskIdentity(room_id=room_id, task_id=task_id, thread_id=task_id, turn_id=task_id), + "status": "prepared", "execution_generation": 1, "cancel_generation": 0} + def approve_room_task( self, room_id: str, *, member_id: str, task_id: str, execution_generation: int, choice: str, request_id: str | None = None) -> Mapping[str, Any]: @@ -632,6 +667,8 @@ class HostedRoomService: counts = Counter(str(task["status"]) for task in tasks) pending_actions = [ attempt_action("retry", task) for task in tasks if task["status"] in _RETRYABLE_STATUSES] + if self._prepare_stalls.stalled(room_id): + pending_actions.append(prepare_stall.retry_action(room_id, self._room(room_id))) with self._policy_lock: pending_actions.extend( dict(action) for (action_room_id, _member_id), action @@ -646,6 +683,7 @@ class HostedRoomService: or bool(counts.get("indeterminate") or counts.get("stopping")), "counts": dict(counts), "pending_actions": pending_actions, "member_activity": member_activity, + "last_error": self._prepare_stalls.last_error(room_id), "peer_routes": self._route_statuses(room_id)}