diff --git a/livekit-portal/src/portal.rs b/livekit-portal/src/portal.rs index e934912..8cd4462 100644 --- a/livekit-portal/src/portal.rs +++ b/livekit-portal/src/portal.rs @@ -301,6 +301,15 @@ pub struct Portal { // attribute-change and participant-connect events can update operators, // robot_identity, and active_operator without taking a Portal-level lock. controller: Arc, + + // Handle to the tokio runtime `connect()` ran on. Captured because + // `register_rpc_method` can be called from a foreign thread with no + // runtime context (e.g. a binding's asyncio loop); registering an RPC + // method on a live LocalParticipant triggers publisher negotiation, + // which `tokio::spawn`s and would otherwise panic with "no reactor + // running". We enter this handle around that call. `None` before the + // first connect. + runtime_handle: Mutex>, } impl Portal { @@ -379,11 +388,15 @@ impl Portal { rpc_handlers: Arc::new(Mutex::new(HashMap::new())), local_participant: Arc::new(Mutex::new(None)), controller: Arc::new(ControllerState::new()), + runtime_handle: Mutex::new(None), } } pub async fn connect(&self, url: &str, token: &str) -> PortalResult<()> { let _lifecycle = self.lifecycle.lock().await; + // Capture the runtime we're running on so `register_rpc_method` can + // enter it when called later from a non-runtime (foreign) thread. + *self.runtime_handle.lock() = Some(tokio::runtime::Handle::current()); if self.conn.lock().room.is_some() { return Err(PortalError::AlreadyConnected); } @@ -852,7 +865,20 @@ impl Portal { map.insert(method.to_string(), handler.clone()); } if let Some(lp) = self.local_participant.lock().clone() { - register_handler_on(&lp, method.to_string(), handler); + // The SDK's `register_rpc_method` kicks off publisher negotiation, + // which `tokio::spawn`s internally. If we were called from a + // foreign thread with no runtime context (a binding's asyncio + // loop), that spawn panics. Enter the runtime `connect()` ran on + // so the spawn lands on it. The handle is always present here + // (an LP exists only after a successful connect set it), but fall + // back to a bare call rather than panicking if it somehow isn't. + match self.runtime_handle.lock().clone() { + Some(handle) => { + let _guard = handle.enter(); + register_handler_on(&lp, method.to_string(), handler); + } + None => register_handler_on(&lp, method.to_string(), handler), + } } } diff --git a/python/packages/livekit-portal/tests/integration/conftest.py b/python/packages/livekit-portal/tests/integration/conftest.py index f3a7632..e786e0f 100644 --- a/python/packages/livekit-portal/tests/integration/conftest.py +++ b/python/packages/livekit-portal/tests/integration/conftest.py @@ -44,6 +44,7 @@ "test_multi_operator.py", "test_action_subscription.py", "test_webrtc_codecs.py", + "test_rpc.py", ] ) diff --git a/python/packages/livekit-portal/tests/integration/test_rpc.py b/python/packages/livekit-portal/tests/integration/test_rpc.py new file mode 100644 index 0000000..7aa9379 --- /dev/null +++ b/python/packages/livekit-portal/tests/integration/test_rpc.py @@ -0,0 +1,99 @@ +"""RPC integration tests. + +Covers handler registration both before and after `connect()`, plus error +propagation. The post-connect case is the regression guard: registering an +RPC handler on a live participant triggers SDK publisher negotiation, which +spawns onto the tokio runtime. When `register_rpc_method` is called from a +binding's asyncio thread (no runtime context), that spawn used to panic with +"there is no reactor running". The core now enters the runtime captured at +connect time around the registration. + +Skipped automatically without `LIVEKIT_URL` (see conftest). +""" +from __future__ import annotations + +import asyncio +import os + +import pytest + +from integration.conftest import URL, _make_token +from livekit.portal import ( + DType, + Operator, + OperatorConfig, + PortalError, + Robot, + RobotConfig, + RpcError, +) + +pytestmark = pytest.mark.asyncio + + +async def test_rpc_register_after_connect(pair): + """Regression: registering a handler AFTER connect must not panic and the + method must be invocable from the peer.""" + await pair.start() + + async def handler(data): + return f"pong:{data.payload}" + + pair.robot.register_rpc_method("ping", handler) + await asyncio.sleep(0.1) + + result = await pair.operator.perform_rpc( + "ping", "hi", destination=pair.robot.local_identity() + ) + assert result == "pong:hi" + + +async def test_rpc_register_before_connect(): + """The pre-connect path keeps working: handlers registered before connect + are applied on connect and invocable afterwards. Built manually rather + than via the `pair` fixture so the robot exists before `connect()`. + """ + room = f"rpc-pre-{os.urandom(4).hex()}" + robot_cfg = RobotConfig(room) + robot_cfg.add_state_typed([("j", DType.F32)]) + operator_cfg = OperatorConfig(room) + operator_cfg.add_state_typed([("j", DType.F32)]) + robot = Robot(robot_cfg) + operator = Operator(operator_cfg) + + async def handler(data): + return f"echo:{data.payload}" + + robot.register_rpc_method("echo", handler) # before connect + + try: + await robot.connect(URL, _make_token("robot", room)) + await asyncio.sleep(0.2) + await operator.connect(URL, _make_token("operator", room)) + await asyncio.sleep(0.2) + result = await operator.perform_rpc( + "echo", "x", destination=robot.local_identity() + ) + assert result == "echo:x" + finally: + for side in (operator, robot): + try: + await side.disconnect() + except Exception: # noqa: BLE001 + pass + + +async def test_rpc_handler_error_propagates(pair): + """An application error raised by the handler surfaces on the caller.""" + await pair.start() + + async def handler(data): + raise RpcError.Error(code=1234, message="boom", data=None) + + pair.robot.register_rpc_method("fail", handler) + await asyncio.sleep(0.1) + + with pytest.raises(PortalError): + await pair.operator.perform_rpc( + "fail", "", destination=pair.robot.local_identity() + )