fix

Author: davidzhao

livekit-rtc/livekit/rtc/participant.py

@@ -673,8 +673,10 @@ async def _run_incoming_chain(self, invocation: RpcInvocationData) -> Optional[s
 
         def _on_deadline() -> None:
             nonlocal deadline_fired
-            deadline_fired = True
-            chain_task.cancel()
+            # 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
+            deadline_fired = chain_task.cancel()
 
         deadline = loop.call_later(invocation.response_timeout, _on_deadline)
         try:
@@ -709,6 +711,16 @@ def _on_deadline() -> None:
             # whatever the chain raised while unwinding is not the caller's business
             _observe_unwind(invocation.method, chain_task)
             raise RpcError._built_in(RpcError.ErrorCode.RESPONSE_TIMEOUT)
+        if chain_task.cancelled():
+            # nothing here cancelled it: an interceptor or the handler let a CancelledError
+            # of its own escape. That is a failure of the handler, and it must still be
+            # answered; result() would re-raise the CancelledError past the response code
+            # and leave the caller waiting for its timeout.
+            logger.warning(
+                "RPC handler for %s raised CancelledError; returning APPLICATION_ERROR",
+                invocation.method,
+            )
+            raise RpcError._built_in(RpcError.ErrorCode.APPLICATION_ERROR)
         # the chain's own outcome: its exception, if any, propagates unchanged
         return chain_task.result()
 

livekit-rtc/tests/test_rpc_interceptors.py

@@ -302,6 +302,37 @@ async def slow(data: RpcInvocationData) -> str:
     assert info.value.code == rtc.RpcError.ErrorCode.RESPONSE_TIMEOUT
 
 
+async def test_chain_finishing_as_the_deadline_fires_keeps_its_result() -> None:
+    """The deadline timer can run in the same loop iteration the chain completes in, before
+    the waiting task resumes: its cancel is rejected (the chain is done) and the result is
+    the caller's. A zero deadline with an immediate handler pins that ordering."""
+    lp = _participant()
+    lp._rpc_handlers["fast"] = lambda data: "ok"
+
+    result = await lp._run_incoming_chain(
+        RpcInvocationData("r1", "alice", "{}", 0.0, method="fast")
+    )
+    assert result == "ok"
+
+
+async def test_cancelled_error_escaping_the_chain_is_an_application_error(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A CancelledError that an interceptor or the handler lets escape on its own is not a
+    deadline nor a disconnect; the caller must still get an answer rather than wait out its
+    timeout because the cancellation slipped past the response code."""
+    lp = _participant()
+
+    async def leaks_cancel(data: RpcInvocationData) -> str:
+        raise asyncio.CancelledError()
+
+    lp._rpc_handlers["leaky"] = leaks_cancel
+    with pytest.raises(rtc.RpcError) as info:
+        await lp._run_incoming_chain(RpcInvocationData("r1", "alice", "{}", 5.0, method="leaky"))
+    assert info.value.code == rtc.RpcError.ErrorCode.APPLICATION_ERROR
+    assert any("leaky" in r.getMessage() for r in caplog.records)
+
+
 async def test_outside_cancellation_maps_to_recipient_disconnected() -> None:
     lp = _participant()
     started = asyncio.Event()