Skip to content

Navigation Menu

Sign in
Sign up

add cancel_reason to RpcInvocationData #809

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
davidzhao wants to merge 1 commit into main
base: main
Choose a base branch
Loading
from dz/rpc-cancellation-reason
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions livekit-rtc/livekit/rtc/participant.py
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

@devin-ai-integration devin-ai-integration Bot Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Concurrent disconnect records wrong reason

When a room disconnect races the response deadline, cancel_reason reports RESPONSE_TIMEOUT while the caller receives RECIPIENT_DISCONNECTED. The deadline cancels the chain before the outer task handles the disconnect and replaces the reason.

Prompt for agents 
Resolve cancellation precedence in LocalParticipant._run_incoming_chain so RpcInvocationData.cancel_reason observed during interceptor unwinding always matches the RpcError returned to the caller. A deadline callback can currently cancel the chain and expose RESPONSE_TIMEOUT before concurrent cancellation of the outer invocation enters the disconnect branch, which later returns RECIPIENT_DISCONNECTED. Add a deterministic race test covering both events in the same loop turn and coordinate the selected outcome before exposing the reason to the chain.
Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

deadline_fired = chain_task.cancel()
if not deadline_fired:
invocation.cancel_reason = None

deadline = loop.call_later(invocation.response_timeout, _on_deadline)
try:
Expand All @@ -691,6 +694,7 @@ def _on_deadline() -> None:
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
invocation.cancel_reason = RpcError.ErrorCode.RECIPIENT_DISCONNECTED
chain_task.cancel()
_, pending = await asyncio.wait([chain_task], timeout=_RPC_CANCEL_UNWIND_TIMEOUT)
if pending:
Expand Down
7 changes: 7 additions & 0 deletions livekit-rtc/livekit/rtc/rpc.py
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
61 changes: 61 additions & 0 deletions livekit-rtc/tests/test_rpc_interceptors.py
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -444,6 +444,67 @@ 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


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
Expand Down
Loading

AltStyle によって変換されたページ (->オリジナル) /