From 80fc507c0177a37bb278558faad910e7a622f4e6 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 7 Sep 2026 23:36:45 -0700 Subject: [PATCH 1/2] add cancel_reason to RpcInvocationData in order to allow downstream interceptors/tracing to determine why the call was not successful. --- livekit-rtc/livekit/rtc/participant.py | 4 ++ livekit-rtc/livekit/rtc/rpc.py | 7 +++ livekit-rtc/tests/test_rpc_interceptors.py | 61 ++++++++++++++++++++++ 3 files changed, 72 insertions(+) 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 From 48cb5bf824db8c6132b4285c2f18794276cfd1e8 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 11 Sep 2026 15:45:51 -0700 Subject: [PATCH 2/2] fix attribution when deadline exceeded during cancel --- livekit-rtc/livekit/rtc/participant.py | 17 ++++++++--- livekit-rtc/tests/test_rpc_interceptors.py | 35 ++++++++++++++++++++++ 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/livekit-rtc/livekit/rtc/participant.py b/livekit-rtc/livekit/rtc/participant.py index dec60bd7..37c54110 100644 --- a/livekit-rtc/livekit/rtc/participant.py +++ b/livekit-rtc/livekit/rtc/participant.py @@ -693,9 +693,14 @@ def _on_deadline() -> None: await asyncio.wait([chain_task]) 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() + # caller, but not for long; this is the path room.disconnect() waits on. + # First cause wins: if the deadline already cancelled the chain, the interceptors + # are unwinding with RESPONSE_TIMEOUT and the caller gets that too, and the timer + # is cancelled here so it cannot fire into the wait below and flip the reason + deadline.cancel() + if not deadline_fired: + invocation.cancel_reason = RpcError.ErrorCode.RECIPIENT_DISCONNECTED + chain_task.cancel() _, pending = await asyncio.wait([chain_task], timeout=_RPC_CANCEL_UNWIND_TIMEOUT) if pending: logger.warning( @@ -707,7 +712,11 @@ def _on_deadline() -> None: chain_task.add_done_callback(functools.partial(_observe_unwind, invocation.method)) else: _observe_unwind(invocation.method, chain_task) - raise RpcError._built_in(RpcError.ErrorCode.RECIPIENT_DISCONNECTED) from None + raise RpcError._built_in( + RpcError.ErrorCode.RESPONSE_TIMEOUT + if deadline_fired + else RpcError.ErrorCode.RECIPIENT_DISCONNECTED + ) from None finally: deadline.cancel() diff --git a/livekit-rtc/tests/test_rpc_interceptors.py b/livekit-rtc/tests/test_rpc_interceptors.py index 394a1b69..50cc3319 100644 --- a/livekit-rtc/tests/test_rpc_interceptors.py +++ b/livekit-rtc/tests/test_rpc_interceptors.py @@ -505,6 +505,41 @@ async def cancels_itself(data: RpcInvocationData) -> str: assert seen.seen[-1] is None +@pytest.mark.parametrize("order", ["same_turn", "deadline_first", "disconnect_first"]) +async def test_deadline_and_disconnect_race_agree_on_the_reason(order: str) -> None: + """When the caller's deadline and a disconnect land close together, whichever cancelled + the chain first decides both what the interceptors saw and what the caller gets; the two + never diverge.""" + 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" + + lp._rpc_handlers["slow"] = slow + timeout = 5.0 if order == "disconnect_first" else 0.0 + task = asyncio.ensure_future( + lp._run_incoming_chain(RpcInvocationData("r1", "alice", "{}", timeout, method="slow")) + ) + await started.wait() + if order == "deadline_first": + await asyncio.sleep(0.02) # the zero deadline has fired by now + task.cancel() # the room disconnecting (same loop turn as the timer in "same_turn") + + with pytest.raises(rtc.RpcError) as info: + await task + assert seen.seen, "the interceptor never saw the cancellation" + assert seen.seen[-1] == info.value.code + if order == "deadline_first": + assert info.value.code == rtc.RpcError.ErrorCode.RESPONSE_TIMEOUT + if order == "disconnect_first": + assert info.value.code == rtc.RpcError.ErrorCode.RECIPIENT_DISCONNECTED + + 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