diff --git a/livekit-rtc/livekit/rtc/participant.py b/livekit-rtc/livekit/rtc/participant.py index 73f14a3b..dec60bd7 100644 --- a/livekit-rtc/livekit/rtc/participant.py +++ b/livekit-rtc/livekit/rtc/participant.py @@ -676,7 +676,10 @@ def _on_deadline() -> None: # only a cancel the chain accepted counts: cancel() is False when the chain has # already finished, which can happen in the same loop iteration the timer fires # while this task has not resumed yet; that result is the caller's, not a timeout + invocation.cancel_reason = RpcError.ErrorCode.RESPONSE_TIMEOUT deadline_fired = chain_task.cancel() + if not deadline_fired: + invocation.cancel_reason = None deadline = loop.call_later(invocation.response_timeout, _on_deadline) try: @@ -691,6 +694,7 @@ def _on_deadline() -> None: except asyncio.CancelledError: # cancelled from outside: stop the chain and let it unwind before answering the # caller, but not for long; this is the path room.disconnect() waits on + invocation.cancel_reason = RpcError.ErrorCode.RECIPIENT_DISCONNECTED chain_task.cancel() _, pending = await asyncio.wait([chain_task], timeout=_RPC_CANCEL_UNWIND_TIMEOUT) if pending: diff --git a/livekit-rtc/livekit/rtc/rpc.py b/livekit-rtc/livekit/rtc/rpc.py index f96b24cb..04c3f2e2 100644 --- a/livekit-rtc/livekit/rtc/rpc.py +++ b/livekit-rtc/livekit/rtc/rpc.py @@ -30,6 +30,12 @@ class RpcInvocationData: payload (str): The payload of the request. User-definable format, typically JSON. response_timeout (float): The maximum time the caller will wait for a response. method (str): The name of the invoked RPC method. + cancel_reason (Optional[RpcError.ErrorCode]): Why the SDK cancelled the handler chain, + set on this object just before it does: ``RESPONSE_TIMEOUT`` when the caller's + deadline passed, ``RECIPIENT_DISCONNECTED`` when the room disconnected. ``None`` + while the chain runs, and for a ``CancelledError`` raised inside the chain (which + the caller receives as ``APPLICATION_ERROR``). Lets an interceptor unwinding from + the cancellation record the outcome the caller gets. """ request_id: str @@ -37,6 +43,7 @@ class RpcInvocationData: payload: str response_timeout: float method: str = "" + cancel_reason: Optional[RpcError.ErrorCode] = None @dataclass diff --git a/livekit-rtc/tests/test_rpc_interceptors.py b/livekit-rtc/tests/test_rpc_interceptors.py index c762d43b..394a1b69 100644 --- a/livekit-rtc/tests/test_rpc_interceptors.py +++ b/livekit-rtc/tests/test_rpc_interceptors.py @@ -444,6 +444,67 @@ async def failing_cleanup(data: RpcInvocationData) -> str: ) +class _SeesCancelReason(rtc.RpcInterceptor): + """Records what ``invocation.cancel_reason`` says while unwinding from a cancellation.""" + + def __init__(self) -> None: + self.seen: list[object] = [] + + async def intercept_incoming( + self, invocation: RpcInvocationData, next: IncomingRpcNext + ) -> Optional[str]: + try: + return await next(invocation) + except asyncio.CancelledError: + self.seen.append(invocation.cancel_reason) + raise + + +async def test_cancel_reason_tells_interceptors_why_the_chain_was_cancelled() -> None: + """The SDK maps a cancellation to an RpcError only after the chain has unwound, so an + interceptor sees a bare CancelledError; ``cancel_reason`` on the invocation says what the + caller will get: the deadline, the disconnect, or nothing for a cancel raised inside.""" + lp = _participant() + seen = _SeesCancelReason() + lp.add_rpc_interceptor(seen) + started = asyncio.Event() + + async def slow(data: RpcInvocationData) -> str: + started.set() + await asyncio.sleep(10) + return "never" + + async def cancels_itself(data: RpcInvocationData) -> str: + raise asyncio.CancelledError() + + lp._rpc_handlers["slow"] = slow + lp._rpc_handlers["self"] = cancels_itself + + # the caller's deadline + with pytest.raises(rtc.RpcError) as info: + await lp._run_incoming_chain(RpcInvocationData("r1", "alice", "{}", 0.02, method="slow")) + assert info.value.code == rtc.RpcError.ErrorCode.RESPONSE_TIMEOUT + assert seen.seen == [rtc.RpcError.ErrorCode.RESPONSE_TIMEOUT] + + # the room disconnecting (the invocation task is cancelled from outside) + started.clear() + task = asyncio.ensure_future( + lp._run_incoming_chain(RpcInvocationData("r2", "alice", "{}", 5.0, method="slow")) + ) + await started.wait() + task.cancel() + with pytest.raises(rtc.RpcError) as info: + await task + assert info.value.code == rtc.RpcError.ErrorCode.RECIPIENT_DISCONNECTED + assert seen.seen[-1] == rtc.RpcError.ErrorCode.RECIPIENT_DISCONNECTED + + # a cancel raised inside the chain: not the SDK's doing, so no reason + with pytest.raises(rtc.RpcError) as info: + await lp._run_incoming_chain(RpcInvocationData("r3", "alice", "{}", 5.0, method="self")) + assert info.value.code == rtc.RpcError.ErrorCode.APPLICATION_ERROR + assert seen.seen[-1] is None + + async def test_handlers_returning_an_awaitable_are_awaited() -> None: """RpcHandler admits any callable returning a payload or an awaitable of one, not only coroutine functions: a callable object with an async __call__, a sync wrapper handing

AltStyle によって変換されたページ (->オリジナル) /