rpc interceptors: keep in-chain TimeoutError an application error

asyncio.wait_for raises the same TimeoutError a handler or interceptor might
raise on its own (an HTTP client timing out, say), so _run_incoming_chain
reported both as RESPONSE_TIMEOUT. The pre-existing handler wrapper had the
same conflation, but now that the deadline spans the chain it is cheap to get
right: a TimeoutError raised inside the chain is tagged (_ChainTimeoutError)
and re-raised as the original exception, which _handle_rpc_method_invocation
maps to APPLICATION_ERROR like any other handler failure; only wait_for's own
expiry becomes RESPONSE_TIMEOUT.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: davidzhao

livekit-rtc/livekit/rtc/participant.py

@@ -236,6 +236,17 @@ def disconnect_reason(
 
 
 RpcHandler = Callable[["RpcInvocationData"], Union[Awaitable[Optional[str]], Optional[str]]]
+
+
+class _ChainTimeoutError(Exception):
+    """A ``TimeoutError`` raised *inside* the incoming RPC chain (by the handler or an
+    interceptor), wrapped so it is distinguishable from the response deadline expiring."""
+
+    def __init__(self, error: asyncio.TimeoutError) -> None:
+        super().__init__(str(error))
+        self.error = error
+
+
 F = TypeVar(
     "F", bound=Callable[[RpcInvocationData], Union[Awaitable[Optional[str]], Optional[str]]]
 )
@@ -633,10 +644,24 @@ async def _run_incoming_chain(self, invocation: RpcInvocationData) -> Optional[s
         ``next`` counts against it; when it passes, the chain is cancelled and the caller
         gets ``RESPONSE_TIMEOUT``. Cancellation from outside (the room disconnecting) maps to
         ``RECIPIENT_DISCONNECTED``, as before.
+
+        A ``TimeoutError`` raised by the handler or an interceptor itself (an HTTP client
+        timing out, say) is not the response deadline: it propagates as an application
+        error rather than being reported to the caller as ``RESPONSE_TIMEOUT``.
         """
         handle = _chain_incoming(list(self._rpc_interceptors), self._invoke_rpc_handler)
+
+        async def _guarded() -> Optional[str]:
+            try:
+                return await handle(invocation)
+            except asyncio.TimeoutError as e:
+                # tag it so it cannot be mistaken for wait_for's own deadline expiry below
+                raise _ChainTimeoutError(e) from e
+
         try:
-            return await asyncio.wait_for(handle(invocation), timeout=invocation.response_timeout)
+            return await asyncio.wait_for(_guarded(), timeout=invocation.response_timeout)
+        except _ChainTimeoutError as e:
+            raise e.error from e.error.__cause__
         except asyncio.TimeoutError:
             raise RpcError._built_in(RpcError.ErrorCode.RESPONSE_TIMEOUT) from None
         except asyncio.CancelledError:

livekit-rtc/tests/test_rpc_interceptors.py

@@ -242,6 +242,36 @@ async def fast(data: RpcInvocationData) -> str:
     assert handler_ran.is_set() == (delay_position == "after_next")
 
 
+async def test_timeout_raised_inside_the_chain_is_not_the_response_deadline() -> None:
+    """An interceptor (or handler) whose own I/O times out raises TimeoutError well before
+    the response deadline. That is an application failure, not RESPONSE_TIMEOUT."""
+    lp = _participant()
+
+    class UpstreamTimesOut(rtc.RpcInterceptor):
+        async def intercept_incoming(
+            self, invocation: RpcInvocationData, next: IncomingRpcNext
+        ) -> Optional[str]:
+            raise asyncio.TimeoutError("upstream lookup timed out")
+
+    lp.add_rpc_interceptor(UpstreamTimesOut())
+    lp._rpc_handlers["m"] = lambda data: "unused"
+
+    # propagates as the original TimeoutError, which _handle_rpc_method_invocation maps to
+    # APPLICATION_ERROR like any other handler exception
+    with pytest.raises(asyncio.TimeoutError) as info:
+        await lp._run_incoming_chain(RpcInvocationData("r1", "alice", "{}", 5.0, method="m"))
+    assert not isinstance(info.value, rtc.RpcError)
+    assert str(info.value) == "upstream lookup timed out"
+
+    async def handler_times_out(data: RpcInvocationData) -> str:
+        raise asyncio.TimeoutError("db timed out")
+
+    lp2 = _participant()
+    lp2._rpc_handlers["m"] = handler_times_out
+    with pytest.raises(asyncio.TimeoutError):
+        await lp2._run_incoming_chain(RpcInvocationData("r2", "alice", "{}", 5.0, method="m"))
+
+
 async def test_outside_cancellation_maps_to_recipient_disconnected() -> None:
     lp = _participant()
     started = asyncio.Event()