rpc interceptors: record deadline expiry independently of the unwinding error

wait_for reported whatever the chain raised while being cancelled; an
interceptor that catches the cancellation and raises during cleanup could turn
a deadline expiry into APPLICATION_ERROR (or, via the TimeoutError tag, back
into a mislabelled application timeout).

The chain now runs as its own future under a call_later deadline that sets a
flag before cancelling it. Once the flag is set, anything the chain raises
while unwinding is reported as RESPONSE_TIMEOUT; a CancelledError without the
flag is external cancellation and stays RECIPIENT_DISCONNECTED; any exception
before the deadline, TimeoutError included, propagates as an application
error. The _ChainTimeoutError tag is no longer needed.

Tests: cleanup raising TimeoutError and RuntimeError after the deadline both
map to RESPONSE_TIMEOUT; the existing deadline, in-chain timeout and outside
cancellation cases still hold.

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

Author: davidzhao

livekit-rtc/livekit/rtc/participant.py

@@ -238,15 +238,6 @@ 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]]]
 )
@@ -642,30 +633,46 @@ async def _run_incoming_chain(self, invocation: RpcInvocationData) -> Optional[s
 
         The deadline covers the whole chain, so time an interceptor spends before or after
         ``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.
+        gets ``RESPONSE_TIMEOUT``, whatever the chain raises while unwinding. 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``.
+        A ``TimeoutError`` raised by the handler or an interceptor itself before the
+        deadline (an HTTP client timing out, say) is not the response deadline: it
+        propagates as an application error rather than being reported as
+        ``RESPONSE_TIMEOUT``.
         """
         handle = _chain_incoming(list(self._rpc_interceptors), self._invoke_rpc_handler)
+        loop = asyncio.get_running_loop()
+        # ensure_future: `next` continuations are typed as Awaitable, not Coroutine
+        chain_task: asyncio.Future[Optional[str]] = asyncio.ensure_future(handle(invocation))
 
-        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
+        # the deadline is recorded independently of whatever the chain raises while it
+        # unwinds, so a TimeoutError (or anything else) from cancellation cleanup cannot be
+        # mistaken for an application failure, nor a genuine in-chain timeout for the deadline
+        deadline_fired = False
 
+        def _on_deadline() -> None:
+            nonlocal deadline_fired
+            deadline_fired = True
+            chain_task.cancel()
+
+        deadline = loop.call_later(invocation.response_timeout, _on_deadline)
         try:
-            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
+            return await chain_task
         except asyncio.CancelledError:
+            if deadline_fired:
+                raise RpcError._built_in(RpcError.ErrorCode.RESPONSE_TIMEOUT) from None
+            # cancelled from outside: awaiting propagated the cancel into the chain; let it
+            # finish unwinding before answering the caller
+            if not chain_task.done():
+                await asyncio.wait([chain_task])
             raise RpcError._built_in(RpcError.ErrorCode.RECIPIENT_DISCONNECTED) from None
+        except Exception:
+            if deadline_fired:
+                raise RpcError._built_in(RpcError.ErrorCode.RESPONSE_TIMEOUT) from None
+            raise
+        finally:
+            deadline.cancel()
 
     async def _invoke_rpc_handler(self, invocation: RpcInvocationData) -> Optional[str]:
         """Run the registered handler for ``invocation`` (the innermost step of the chain).

livekit-rtc/tests/test_rpc_interceptors.py

@@ -272,6 +272,35 @@ async def handler_times_out(data: RpcInvocationData) -> str:
         await lp2._run_incoming_chain(RpcInvocationData("r2", "alice", "{}", 5.0, method="m"))
 
 
+@pytest.mark.parametrize("cleanup_error", [asyncio.TimeoutError, RuntimeError])
+async def test_cleanup_errors_after_the_deadline_are_still_a_timeout(
+    cleanup_error: type[BaseException],
+) -> None:
+    """Once the deadline fired, whatever the chain raises while unwinding is not the
+    caller's business: they get RESPONSE_TIMEOUT."""
+    lp = _participant()
+
+    class RaisesOnCancel(rtc.RpcInterceptor):
+        async def intercept_incoming(
+            self, invocation: RpcInvocationData, next: IncomingRpcNext
+        ) -> Optional[str]:
+            try:
+                return await next(invocation)
+            except asyncio.CancelledError:
+                raise cleanup_error("cleanup failed while cancelling") from None
+
+    async def slow(data: RpcInvocationData) -> str:
+        await asyncio.sleep(10)
+        return "never"
+
+    lp.add_rpc_interceptor(RaisesOnCancel())
+    lp._rpc_handlers["slow"] = slow
+
+    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
+
+
 async def test_outside_cancellation_maps_to_recipient_disconnected() -> None:
     lp = _participant()
     started = asyncio.Event()