add cancel_reason to RpcInvocationData

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:
@@ -690,8 +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
-            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(
@@ -703,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()
 

livekit-rtc/livekit/rtc/rpc.py

@@ -30,13 +30,20 @@ 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
     caller_identity: str
     payload: str
     response_timeout: float
     method: str = ""
+    cancel_reason: Optional[RpcError.ErrorCode] = None
 
 
 @dataclass

livekit-rtc/tests/test_rpc_interceptors.py

@@ -444,6 +444,102 @@ 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
+
+
+@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