Skip to content

Commit 80fc507

Browse files
committed
add cancel_reason to RpcInvocationData
in order to allow downstream interceptors/tracing to determine why the call was not successful.
1 parent 184d5a6 commit 80fc507

3 files changed

Lines changed: 72 additions & 0 deletions

File tree

‎livekit-rtc/livekit/rtc/participant.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -676,7 +676,10 @@ def _on_deadline() -> None:
676676
# only a cancel the chain accepted counts: cancel() is False when the chain has
677677
# already finished, which can happen in the same loop iteration the timer fires
678678
# while this task has not resumed yet; that result is the caller's, not a timeout
679+
invocation.cancel_reason = RpcError.ErrorCode.RESPONSE_TIMEOUT
679680
deadline_fired = chain_task.cancel()
681+
if not deadline_fired:
682+
invocation.cancel_reason = None
680683

681684
deadline = loop.call_later(invocation.response_timeout, _on_deadline)
682685
try:
@@ -691,6 +694,7 @@ def _on_deadline() -> None:
691694
except asyncio.CancelledError:
692695
# cancelled from outside: stop the chain and let it unwind before answering the
693696
# caller, but not for long; this is the path room.disconnect() waits on
697+
invocation.cancel_reason = RpcError.ErrorCode.RECIPIENT_DISCONNECTED
694698
chain_task.cancel()
695699
_, pending = await asyncio.wait([chain_task], timeout=_RPC_CANCEL_UNWIND_TIMEOUT)
696700
if pending:

‎livekit-rtc/livekit/rtc/rpc.py‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,13 +30,20 @@ class RpcInvocationData:
3030
payload (str): The payload of the request. User-definable format, typically JSON.
3131
response_timeout (float): The maximum time the caller will wait for a response.
3232
method (str): The name of the invoked RPC method.
33+
cancel_reason (Optional[RpcError.ErrorCode]): Why the SDK cancelled the handler chain,
34+
set on this object just before it does: ``RESPONSE_TIMEOUT`` when the caller's
35+
deadline passed, ``RECIPIENT_DISCONNECTED`` when the room disconnected. ``None``
36+
while the chain runs, and for a ``CancelledError`` raised inside the chain (which
37+
the caller receives as ``APPLICATION_ERROR``). Lets an interceptor unwinding from
38+
the cancellation record the outcome the caller gets.
3339
"""
3440

3541
request_id: str
3642
caller_identity: str
3743
payload: str
3844
response_timeout: float
3945
method: str = ""
46+
cancel_reason: Optional[RpcError.ErrorCode] = None
4047

4148

4249
@dataclass

‎livekit-rtc/tests/test_rpc_interceptors.py‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -444,6 +444,67 @@ async def failing_cleanup(data: RpcInvocationData) -> str:
444444
)
445445

446446

447+
class _SeesCancelReason(rtc.RpcInterceptor):
448+
"""Records what ``invocation.cancel_reason`` says while unwinding from a cancellation."""
449+
450+
def __init__(self) -> None:
451+
self.seen: list[object] = []
452+
453+
async def intercept_incoming(
454+
self, invocation: RpcInvocationData, next: IncomingRpcNext
455+
) -> Optional[str]:
456+
try:
457+
return await next(invocation)
458+
except asyncio.CancelledError:
459+
self.seen.append(invocation.cancel_reason)
460+
raise
461+
462+
463+
async def test_cancel_reason_tells_interceptors_why_the_chain_was_cancelled() -> None:
464+
"""The SDK maps a cancellation to an RpcError only after the chain has unwound, so an
465+
interceptor sees a bare CancelledError; ``cancel_reason`` on the invocation says what the
466+
caller will get: the deadline, the disconnect, or nothing for a cancel raised inside."""
467+
lp = _participant()
468+
seen = _SeesCancelReason()
469+
lp.add_rpc_interceptor(seen)
470+
started = asyncio.Event()
471+
472+
async def slow(data: RpcInvocationData) -> str:
473+
started.set()
474+
await asyncio.sleep(10)
475+
return "never"
476+
477+
async def cancels_itself(data: RpcInvocationData) -> str:
478+
raise asyncio.CancelledError()
479+
480+
lp._rpc_handlers["slow"] = slow
481+
lp._rpc_handlers["self"] = cancels_itself
482+
483+
# the caller's deadline
484+
with pytest.raises(rtc.RpcError) as info:
485+
await lp._run_incoming_chain(RpcInvocationData("r1", "alice", "{}", 0.02, method="slow"))
486+
assert info.value.code == rtc.RpcError.ErrorCode.RESPONSE_TIMEOUT
487+
assert seen.seen == [rtc.RpcError.ErrorCode.RESPONSE_TIMEOUT]
488+
489+
# the room disconnecting (the invocation task is cancelled from outside)
490+
started.clear()
491+
task = asyncio.ensure_future(
492+
lp._run_incoming_chain(RpcInvocationData("r2", "alice", "{}", 5.0, method="slow"))
493+
)
494+
await started.wait()
495+
task.cancel()
496+
with pytest.raises(rtc.RpcError) as info:
497+
await task
498+
assert info.value.code == rtc.RpcError.ErrorCode.RECIPIENT_DISCONNECTED
499+
assert seen.seen[-1] == rtc.RpcError.ErrorCode.RECIPIENT_DISCONNECTED
500+
501+
# a cancel raised inside the chain: not the SDK's doing, so no reason
502+
with pytest.raises(rtc.RpcError) as info:
503+
await lp._run_incoming_chain(RpcInvocationData("r3", "alice", "{}", 5.0, method="self"))
504+
assert info.value.code == rtc.RpcError.ErrorCode.APPLICATION_ERROR
505+
assert seen.seen[-1] is None
506+
507+
447508
async def test_handlers_returning_an_awaitable_are_awaited() -> None:
448509
"""RpcHandler admits any callable returning a payload or an awaitable of one, not only
449510
coroutine functions: a callable object with an async __call__, a sync wrapper handing

0 commit comments

Comments
 (0)