Fixes

Author: bcherry

README.md

@@ -141,12 +141,12 @@ The participant who implements the method and will receive its calls must first
 
 ```python
 @room.local_participant.rpc_method("greet")
-async def handle_greet(request_id: str, caller_identity: str, payload: str, response_timeout_ms: int):
+async def handle_greet(request_id: str, caller_identity: str, payload: str, response_timeout: float):
     print(f"Received greeting from {caller_identity}: {payload}")
     return f"Hello, {caller_identity}!"
 ```
 
-In addition to the payload, your handler will also receive `response_timeout_ms`, which informs you the maximum time available to return a response. If you are unable to respond in time, the call will result in an error on the caller's side.
+In addition to the payload, your handler will also receive `response_timeout`, which informs you the maximum time available to return a response. If you are unable to respond in time, the call will result in an error on the caller's side.
 
 #### Performing an RPC request
 
@@ -164,7 +164,7 @@ except Exception as e:
   print(f"RPC call failed: {e}")
 ```
 
-You may find it useful to adjust the `response_timeout_ms` parameter, which indicates the amount of time you will wait for a response. We recommend keeping this value as low as possible while still satisfying the constraints of your application.
+You may find it useful to adjust the `response_timeout` parameter, which indicates the amount of time you will wait for a response. We recommend keeping this value as low as possible while still satisfying the constraints of your application.
 
 #### Errors
 

examples/rpc.py

@@ -3,6 +3,7 @@
 import json
 import asyncio
 from dotenv import load_dotenv
+import time
 
 load_dotenv(dotenv_path=".env.local", override=False)
 LIVEKIT_API_KEY = os.getenv("LIVEKIT_API_KEY")
@@ -15,44 +16,74 @@
 
 
 async def main():
-    room_name = f"rpc-test-{os.urandom(4).hex()}"
-
-    print(f"Connecting participants to room: {room_name}")
-
-    callers_room, greeters_room, math_genius_room = await asyncio.gather(
-        connect_participant("caller", room_name),
-        connect_participant("greeter", room_name),
-        connect_participant("math-genius", room_name),
-    )
-
-    register_receiver_methods(greeters_room, math_genius_room)
-
+    rooms = []  # Keep track of all rooms for cleanup
     try:
-        print("\n\nRunning greeting example...")
-        await asyncio.gather(perform_greeting(callers_room))
-    except Exception as error:
-        print("Error:", error)
+        room_name = f"rpc-test-{os.urandom(4).hex()}"
+        print(f"Connecting participants to room: {room_name}")
 
-    try:
-        print("\n\nRunning error handling example...")
-        await perform_divide(callers_room)
-    except Exception as error:
-        print("Error:", error)
-
-    try:
-        print("\n\nRunning math example...")
-        await perform_square_root(callers_room)
-        await asyncio.sleep(2)
-        await perform_quantum_hypergeometric_series(callers_room)
-    except Exception as error:
-        print("Error:", error)
-
-    print("\n\nParticipants done, disconnecting...")
-    await callers_room.disconnect()
-    await greeters_room.disconnect()
-    await math_genius_room.disconnect()
-
-    print("Participants disconnected. Example completed.")
+        callers_room, greeters_room, math_genius_room = await asyncio.gather(
+            connect_participant("caller", room_name),
+            connect_participant("greeter", room_name),
+            connect_participant("math-genius", room_name),
+        )
+        rooms = [callers_room, greeters_room, math_genius_room]
+
+        register_receiver_methods(greeters_room, math_genius_room)
+
+        try:
+            print("\n\nRunning greeting example...")
+            await asyncio.gather(perform_greeting(callers_room))
+        except Exception as error:
+            print("Error:", error)
+
+        try:
+            print("\n\nRunning error handling example...")
+            await perform_divide(callers_room)
+        except Exception as error:
+            print("Error:", error)
+
+        try:
+            print("\n\nRunning math example...")
+            await perform_square_root(callers_room)
+            await asyncio.sleep(2)
+            await perform_quantum_hypergeometric_series(callers_room)
+        except Exception as error:
+            print("Error:", error)
+
+        try:
+            print("\n\nRunning long calculation with timeout...")
+            await asyncio.create_task(perform_long_calculation(callers_room))
+        except Exception as error:
+            print("Error:", error)
+            
+        try:
+            print("\n\nRunning long calculation with disconnect...")
+            # Start the long calculation
+            long_calc_task = asyncio.create_task(perform_long_calculation(callers_room))
+            # Wait a bit then disconnect the math genius
+            await asyncio.sleep(5)
+            print("\nDisconnecting math genius early...")
+            await math_genius_room.disconnect()
+            # Wait for the calculation to fail
+            await long_calc_task
+        except Exception as error:
+            print("Error:", error)
+
+        print("\n\nParticipants done, disconnecting remaining participants...")
+        await callers_room.disconnect()
+        await greeters_room.disconnect()
+
+        print("Participants disconnected. Example completed.")
+
+    except KeyboardInterrupt:
+        print("\nReceived interrupt signal, cleaning up...")
+    except Exception as e:
+        print(f"Unexpected error: {e}")
+    finally:
+        # Clean up all rooms
+        print("Disconnecting all participants...")
+        await asyncio.gather(*(room.disconnect() for room in rooms), return_exceptions=True)
+        print("Cleanup complete")
 
 
 def register_receiver_methods(greeters_room: rtc.Room, math_genius_room: rtc.Room):
@@ -61,7 +92,7 @@ async def arrival_method(
         request_id: str,
         caller_identity: str,
         payload: str,
-        response_timeout_ms: int,
+        response_timeout: float,
     ):
         print(f'[Greeter] Oh {caller_identity} arrived and said "{payload}"')
         await asyncio.sleep(2)
@@ -72,12 +103,12 @@ async def square_root_method(
         request_id: str,
         caller_identity: str,
         payload: str,
-        response_timeout_ms: int,
+        response_timeout: float,
     ):
         json_data = json.loads(payload)
         number = json_data["number"]
         print(
-            f"[Math Genius] I guess {caller_identity} wants the square root of {number}. I've only got {response_timeout_ms / 1000} seconds to respond but I think I can pull it off."
+            f"[Math Genius] I guess {caller_identity} wants the square root of {number}. I've only got {response_timeout} seconds to respond but I think I can pull it off."
         )
 
         print("[Math Genius] *doing math*…")
@@ -92,7 +123,7 @@ def divide_method(
         request_id: str,
         caller_identity: str,
         payload: str,
-        response_timeout_ms: int,
+        response_timeout: float,
     ):
         json_data = json.loads(payload)
         dividend = json_data["dividend"]
@@ -104,6 +135,18 @@ def divide_method(
         result = dividend / divisor
         return json.dumps({"result": result})
 
+    @math_genius_room.local_participant.rpc_method("long-calculation")
+    async def long_calculation_method(
+        request_id: str,
+        caller_identity: str,
+        payload: str,
+        response_timeout: float,
+    ):
+        print(f"[Math Genius] Starting a very long calculation for {caller_identity}")
+        print(f"[Math Genius] This will take 30 seconds even though you're only giving me {response_timeout} seconds")
+        await asyncio.sleep(30)
+        return json.dumps({"result": "Calculation complete!"})
+
 
 async def perform_greeting(room: rtc.Room):
     print("[Caller] Letting the greeter know that I've arrived")
@@ -168,6 +211,28 @@ async def perform_divide(room: rtc.Room):
         print(f"[Caller] RPC call failed with unexpected error: {error}")
 
 
+async def perform_long_calculation(room: rtc.Room):
+    print("[Caller] Giving the math genius 10s to complete a long calculation")
+    try:
+        response = await room.local_participant.perform_rpc(
+            "math-genius", 
+            "long-calculation", 
+            json.dumps({}),
+            response_timeout=10
+        )
+        parsed_response = json.loads(response)
+        print(f"[Caller] Result: {parsed_response['result']}")
+    except rtc.RpcError as error:
+        if error.code == rtc.RpcError.ErrorCode.RESPONSE_TIMEOUT:
+            print("[Caller] Math genius took too long to respond")
+        elif error.code == rtc.RpcError.ErrorCode.RECIPIENT_DISCONNECTED:
+            print("[Caller] Math genius disconnected before response was received")
+        else:
+            print(f"[Caller] Unexpected RPC error: {error}")
+    except Exception as error:
+        print(f"[Caller] Unexpected error: {error}")
+
+
 def create_token(identity: str, room_name: str):
     token = (
         api.AccessToken(LIVEKIT_API_KEY, LIVEKIT_API_SECRET)
@@ -216,4 +281,7 @@ def _on_participant_connected(participant: rtc.RemoteParticipant):
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("\nProgram terminated by user")

livekit-rtc/livekit/rtc/participant.py

@@ -248,7 +248,7 @@ async def perform_rpc(
         destination_identity: str,
         method: str,
         payload: str,
-        response_timeout_ms: Optional[int] = None,
+        response_timeout: Optional[float] = None,
     ) -> str:
         """
         Initiate an RPC call to a remote participant.
@@ -257,7 +257,7 @@ async def perform_rpc(
             destination_identity (str): The `identity` of the destination participant
             method (str): The method name to call
             payload (str): The method payload
-            response_timeout_ms (Optional[int]): Timeout for receiving a response after initial connection
+            response_timeout (Optional[float]): Timeout for receiving a response after initial connection
 
         Returns:
             str: The response payload
@@ -270,8 +270,8 @@ async def perform_rpc(
         req.perform_rpc.destination_identity = destination_identity
         req.perform_rpc.method = method
         req.perform_rpc.payload = payload
-        if response_timeout_ms is not None:
-            req.perform_rpc.response_timeout_ms = response_timeout_ms
+        if response_timeout is not None:
+            req.perform_rpc.response_timeout_ms = int(response_timeout * 1000)
 
         queue = FfiClient.instance.queue.subscribe()
         try:
@@ -307,7 +307,7 @@ def register_rpc_method(
             RpcError: On failure. Details in `message`.
 
         Example:
-            async def greet_handler(request_id: str, caller_identity: str, payload: str, response_timeout_ms: int) -> str:
+            async def greet_handler(request_id: str, caller_identity: str, payload: str, response_timeout: float) -> str:
                 print(f"Received greeting from {caller_identity}: {payload}")
                 return f"Hello, {caller_identity}!"
 
@@ -317,10 +317,10 @@ async def greet_handler(request_id: str, caller_identity: str, payload: str, res
         - `request_id`: A unique identifier for this RPC request
         - `caller_identity`: The identity of the RemoteParticipant who initiated the RPC call
         - `payload`: The data sent by the caller (as a string)
-        - `response_timeout_ms`: The maximum time available to return a response
+        - `response_timeout`: The maximum time available to return a response
 
         The handler should return a string or a coroutine that resolves to a string.
-        If unable to respond within `response_timeout_ms`, the request will result in an error on the caller's side.
+        If unable to respond within `response_timeout`, the request will result in an error on the caller's side.
 
         You may raise errors of type `RpcError` with a string `message` in the handler,
         and they will be received on the caller's side with the message intact.
@@ -343,7 +343,7 @@ def rpc_method(self, method: str):
 
         Example:
             @local_participant.rpc_method("greet")
-            async def greet_handler(request_id: str, caller_identity: str, payload: str, response_timeout_ms: int) -> str:
+            async def greet_handler(request_id: str, caller_identity: str, payload: str, response_timeout: float) -> str:
                 print(f"Received greeting from {caller_identity}: {payload}")
                 return f"Hello, {caller_identity}!"
 
@@ -379,7 +379,7 @@ async def _handle_rpc_method_invocation(
         request_id: str,
         caller_identity: str,
         payload: str,
-        response_timeout_ms: int,
+        response_timeout: float,
     ) -> None:
         response_error: Optional[RpcError] = None
         response_payload: Optional[str] = None
@@ -391,12 +391,24 @@ async def _handle_rpc_method_invocation(
         else:
             try:
                 if asyncio.iscoroutinefunction(handler):
-                    response_payload = await handler(
-                        request_id, caller_identity, payload, response_timeout_ms
-                    )
+                    async def run_handler():
+                        try:
+                            return await handler(
+                                request_id, caller_identity, payload, response_timeout
+                            )
+                        except asyncio.CancelledError:
+                            # This will be caught by the outer try-except if it's due to timeout
+                            raise
+
+                    try:
+                        response_payload = await asyncio.wait_for(run_handler(), timeout=response_timeout)
+                    except asyncio.TimeoutError:
+                        raise RpcError._built_in(RpcError.ErrorCode.RESPONSE_TIMEOUT)
+                    except asyncio.CancelledError:
+                        raise RpcError._built_in(RpcError.ErrorCode.RECIPIENT_DISCONNECTED)
                 else:
                     response_payload = handler(
-                        request_id, caller_identity, payload, response_timeout_ms
+                        request_id, caller_identity, payload, response_timeout
                     )
             except RpcError as error:
                 response_error = error
@@ -589,3 +601,5 @@ def track_publications(self) -> Mapping[str, RemoteTrackPublication]:
 
     def __repr__(self) -> str:
         return f"rtc.RemoteParticipant(sid={self.sid}, identity={self.identity}, name={self.name})"
+
+

livekit-rtc/livekit/rtc/room.py

@@ -400,10 +400,14 @@ async def disconnect(self) -> None:
         """Disconnects from the room."""
         if not self.isconnected():
             return
+        
+        if self._rpc_invocation_tasks:
+            for task in self._rpc_invocation_tasks:
+                task.cancel()
+            await asyncio.gather(*self._rpc_invocation_tasks, return_exceptions=True)
 
         req = proto_ffi.FfiRequest()
         req.disconnect.room_handle = self._ffi_handle.handle  # type: ignore
-
         queue = FfiClient.instance.queue.subscribe()
         try:
             resp = FfiClient.instance.request(req)
@@ -412,12 +416,6 @@ async def disconnect(self) -> None:
             )
         finally:
             FfiClient.instance.queue.unsubscribe(queue)
-
-        if self._rpc_invocation_tasks:
-            for task in self._rpc_invocation_tasks:
-                task.cancel()
-            await asyncio.gather(*self._rpc_invocation_tasks, return_exceptions=True)
-
         await self._task
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
@@ -457,7 +455,7 @@ def _on_rpc_method_invocation(self, rpc_invocation: RpcMethodInvocationEvent):
                     rpc_invocation.request_id,
                     rpc_invocation.caller_identity,
                     rpc_invocation.payload,
-                    rpc_invocation.response_timeout_ms,
+                    rpc_invocation.response_timeout_ms / 1000.0,
                 )
             )
             self._rpc_invocation_tasks.add(task)