diff --git a/tests/tui_gateway/test_hosted_room_approval_lifetime.py b/tests/tui_gateway/test_hosted_room_approval_lifetime.py index d217db2c83..432fe5ff77 100644 --- a/tests/tui_gateway/test_hosted_room_approval_lifetime.py +++ b/tests/tui_gateway/test_hosted_room_approval_lifetime.py @@ -256,3 +256,63 @@ def test_stale_local_approval_cannot_resolve_replacement_request(tmp_path: Path, assert service.status("room-1")["pending_actions"][0]["request_id"] == ( "approval-B" ) + + +def test_allow_paused_after_its_live_read_cannot_answer_after_a_concurrent_stop(pending_approval, monkeypatch): + """An Allow whose live-attempt read raced another window's Stop is ordered with that Stop.""" + import threading + + service, task, _lease = pending_approval + statuses_at_dispatch = [] + + class _RecordingRPC(_FakeRPC): + def approve(self, **kwargs): + statuses_at_dispatch.append(driver.get_task(service.db_path, task["identity"])["status"]) + return super().approve(**kwargs) + + service.rpc = service.runtime.rpc = _RecordingRPC() + for hook in ("_interrupt_stopping_task", "_settle_stopping_completion", "_peer_stop_acknowledged"): + monkeypatch.setattr(service.runtime, hook, lambda *_args: False) + read, stop_progress = threading.Event(), threading.Event() + approver = threading.current_thread() + real_list_tasks = driver.list_tasks + + def list_tasks(*args, **kwargs): + rows = real_list_tasks(*args, **kwargs) + if threading.current_thread() is approver and not read.is_set(): + read.set() + assert stop_progress.wait(10), "Stop neither finished nor waited for this Allow" + return rows + + class _SignallingLock: + """Stop waiting on this lock is as far as it can get while the Allow holds it.""" + def __init__(self, lock): + self._lock = lock + + def __enter__(self): + if not self._lock.acquire(blocking=False): + if threading.current_thread() is not approver: + stop_progress.set() + self._lock.acquire() + return self + + def __exit__(self, *exc): + self._lock.release() + + monkeypatch.setattr(driver, "list_tasks", list_tasks) + service._approval_stop_lock = _SignallingLock(threading.RLock()) + stopper = threading.Thread(target=lambda: (read.wait(10), service.stop_room("room", cancel_id="stop"), + stop_progress.set())) + stopper.start() + try: + result = service.approve_room_task( + "room", member_id="ops", task_id=task["identity"].task_id, + execution_generation=task["execution_generation"], request_id="approval", choice="once") + except RuntimeError as exc: + result = str(exc) + stopper.join(10) + assert not stopper.is_alive() + # Either the member got its answer while the attempt still ran, or the Allow was refused. + assert statuses_at_dispatch in ([], ["running"]), (statuses_at_dispatch, result) + assert driver.get_task(service.db_path, task["identity"])["status"] == "stopping" + assert ("room", "ops") not in service._pending_actions diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index 4c8321ed31..4f6cc7809e 100644 --- a/tui_gateway/hosted_room_service.py +++ b/tui_gateway/hosted_room_service.py @@ -134,6 +134,10 @@ class HostedRoomService: self.discussion_limits = _discussion_limits(self.root) hosted_rooms.prune_disbanded_rooms(self.db_path) self._policy_lock = threading.RLock() + # Orders Stop with an approval's live check and dispatch: an Allow either reaches the member + # before Stop marks its attempt stopping, or sees the stop and is refused. Taken before + # ``_policy_lock``, never inside it. + self._approval_stop_lock = threading.RLock() self._pending_actions: dict[tuple[str, str], dict[str, Any]] = {} self.policy_checkpoint = HostedRoomPolicyCheckpoint( self.db_path, @@ -698,6 +702,10 @@ class HostedRoomService: def stop_room( self, room_id: str, *, cancel_id: str, require_acknowledged: bool = False) -> int: + with self._approval_stop_lock: + return self._stop_room(room_id, cancel_id=cancel_id, require_acknowledged=require_acknowledged) + + def _stop_room(self, room_id: str, *, cancel_id: str, require_acknowledged: bool) -> int: gateway_id, epoch = self._owned_authority(room_id) fence = hosted_rooms.request_room_stop( self.db_path, room_id=room_id, cancel_id=cancel_id, expected_gateway_id=gateway_id, @@ -710,6 +718,11 @@ class HostedRoomService: # A replay retains its original fence; work admitted afterward belongs to the # next message. Disband must drain all work before deleting the room. if require_acknowledged or int(task["payload"]["source_event_seq"]) < int(fence["seq"])} + # A stopped attempt's approval is spent: a concurrent Allow waiting on the lock finds nothing. + stopped = {(task["identity"].task_id, task.get("execution_generation")) for task in tasks.values()} + for key in [key for key, action in self._pending_actions.items() if key[0] == room_id + and (action.get("task_id"), action.get("execution_generation")) in stopped]: + self._pending_actions.pop(key, None) for task in tasks.values(): own_cancel_id = ( task.get("status") == "stopping" and str(task.get("cancel_id") or "")) @@ -767,33 +780,44 @@ class HostedRoomService: str(pending.get("request_id") or ""), pending.get("task_id"), int(pending.get("execution_generation") or 0), ) == (requested_approval_id, task_id, execution_generation) - live = _approvable_attempts(driver.list_tasks(self.db_path, room_id=room_id)) - with self._policy_lock: - action = self._pending_actions.get(key) - if not requested_approval_id or not matches(action): - raise RuntimeError("room approval is no longer pending") - if (member_id, task_id, execution_generation) not in live: - self._pending_actions.pop(key, None) - raise RuntimeError("room approval is no longer pending") - if choice not in {"once", "deny"}: - raise RuntimeError("room approval choice must be once or deny") + # The live check and the dispatch hold the Stop lock, so a Stop either lands first (and this + # Allow is refused) or after the member already has its answer. + with self._approval_stop_lock: + live = _approvable_attempts(driver.list_tasks(self.db_path, room_id=room_id)) + with self._policy_lock: + action = self._pending_actions.get(key) + if not requested_approval_id or not matches(action): + raise RuntimeError("room approval is no longer pending") + if (member_id, task_id, execution_generation) not in live: + self._pending_actions.pop(key, None) + raise RuntimeError("room approval is no longer pending") + if choice not in {"once", "deny"}: + raise RuntimeError("room approval choice must be once or deny") + result = self._dispatch_approval( + route, client, action, task_id=task_id, execution_generation=execution_generation, + request_id=requested_approval_id, choice=choice) + with self._policy_lock: + if matches(self._pending_actions.get(key)): + self._pending_actions.pop(key, None) + self.runtime.wakeup() + return result + + def _dispatch_approval( + self, route, client, action: Mapping[str, Any], *, task_id: str, execution_generation: int, + request_id: str, choice: str) -> Mapping[str, Any]: approve = _hook(client, "approve_receipt") if route is not None and approve is not None: result = approve( task_id=task_id, execution_generation=execution_generation, - request_id=requested_approval_id, choice=choice, grant=route.grant) + request_id=request_id, choice=choice, grant=route.grant) else: session_id = str(action.get("session_id") or "") if not session_id: raise RuntimeError("local room approval identity is unavailable") result = self.rpc.approve( - session_id=session_id, request_id=requested_approval_id, choice=choice) + session_id=session_id, request_id=request_id, choice=choice) if result is None: raise RuntimeError("room approval target is unavailable") - with self._policy_lock: - if matches(self._pending_actions.get(key)): - self._pending_actions.pop(key, None) - self.runtime.wakeup() return result def status(self, room_id: str | None = None) -> dict[str, Any]: