Support for large RPC messages using data streams

.changeset/chubby-buckets-drop.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Add new RPC protocol updates to support infinite payload length in requests / responses

examples/rpc/rpc-demo.ts

@@ -36,6 +36,13 @@ async function main() {
     console.error('Error:', error);
   }
 
+  try {
+    console.log('\n\nRunning send long info example...');
+    await Promise.all([performSendVeryLongInfo(callersRoom)]);
+  } catch (error) {
+    console.error('Error:', error);
+  }
+
   try {
     console.log('\n\nRunning error handling example...');
     await Promise.all([performDivision(callersRoom)]);
@@ -85,6 +92,18 @@ const registerReceiverMethods = async (greetersRoom: Room, mathGeniusRoom: Room)
     },
   );
 
+  await greetersRoom.registerRpcMethod(
+    'exchanging-long-info',
+    // eslint-disable-next-line @typescript-eslint/no-unused-vars
+    async (data: RpcInvocationData) => {
+      console.log(
+        `[Greeter] ${data.callerIdentity} has arrived and said that its long info is "${data.payload}"`,
+      );
+      await new Promise((resolve) => setTimeout(resolve, 2000));
+      return new Array<string>(20_000).fill('Y').join('');
+    },
+  );
+
   await mathGeniusRoom.registerRpcMethod('square-root', async (data: RpcInvocationData) => {
     const jsonData = JSON.parse(data.payload);
     const number = jsonData.number;
@@ -136,6 +155,21 @@ const performGreeting = async (room: Room): Promise<void> => {
   }
 };
 
+const performSendVeryLongInfo = async (room: Room): Promise<void> => {
+  console.log('[Caller] Sending the greeter a very long message');
+  try {
+    const response = await room.localParticipant.performRpc({
+      destinationIdentity: 'greeter',
+      method: 'exchanging-long-info',
+      payload: new Array<string>(20_000).fill('X').join(''),
+    });
+    console.log(`[Caller] The greeter's long info is: "${response}"`);
+  } catch (error) {
+    console.error('[Caller] RPC call failed:', error);
+    throw error;
+  }
+};
+
 const performDisconnection = async (room: Room): Promise<void> => {
   console.log('[Caller] Checking back in on the greeter...');
   try {

src/room/RTCEngine.ts

@@ -25,7 +25,6 @@ import {
   Room as RoomModel,
   RoomMovedResponse,
   RpcAck,
-  RpcResponse,
   ServerInfo,
   SessionDescription,
   SignalTarget,
@@ -74,7 +73,6 @@ import {
   UnexpectedConnectionState,
 } from './errors';
 import { EngineEvent } from './events';
-import { RpcError } from './rpc';
 import CriticalTimers from './timers';
 import type LocalTrack from './track/LocalTrack';
 import type LocalTrackPublication from './track/LocalTrackPublication';
@@ -1391,30 +1389,6 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     });
   };
 
-  /** @internal */
-  async publishRpcResponse(
-    destinationIdentity: string,
-    requestId: string,
-    payload: string | null,
-    error: RpcError | null,
-  ) {
-    const packet = new DataPacket({
-      destinationIdentities: [destinationIdentity],
-      kind: DataPacket_Kind.RELIABLE,
-      value: {
-        case: 'rpcResponse',
-        value: new RpcResponse({
-          requestId,
-          value: error
-            ? { case: 'error', value: error.toProto() }
-            : { case: 'payload', value: payload ?? '' },
-        }),
-      },
-    });
-
-    await this.sendDataPacket(packet, DataChannelKind.RELIABLE);
-  }
-
   /** @internal */
   async publishRpcAck(destinationIdentity: string, requestId: string) {
     const packet = new DataPacket({

src/room/Room.ts

@@ -45,6 +45,7 @@ import type {
 } from '../options';
 import TypedPromise from '../utils/TypedPromise';
 import { getBrowser } from '../utils/browserParser';
+import { CLIENT_PROTOCOL_DEFAULT } from '../version';
 import { BackOffStrategy } from './BackOffStrategy';
 import DeviceManager from './DeviceManager';
 import RTCEngine, { DataChannelKind } from './RTCEngine';
@@ -78,7 +79,14 @@ import LocalParticipant from './participant/LocalParticipant';
 import Participant from './participant/Participant';
 import { type ConnectionQuality, ParticipantKind } from './participant/Participant';
 import RemoteParticipant from './participant/RemoteParticipant';
-import { MAX_PAYLOAD_BYTES, RpcError, type RpcInvocationData, byteLength } from './rpc';
+import {
+  RPC_REQUEST_DATA_STREAM_TOPIC,
+  RPC_RESPONSE_DATA_STREAM_TOPIC,
+  RpcClientManager,
+  RpcError,
+  type RpcInvocationData,
+  RpcServerManager,
+} from './rpc';
 import CriticalTimers from './timers';
 import LocalAudioTrack from './track/LocalAudioTrack';
 import type LocalTrack from './track/LocalTrack';
@@ -216,7 +224,9 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
 
   private outgoingDataTrackManager: OutgoingDataTrackManager;
 
-  private rpcHandlers: Map<string, (data: RpcInvocationData) => Promise<string>> = new Map();
+  private rpcClientManager: RpcClientManager;
+
+  private rpcServerManager: RpcServerManager;
 
   get hasE2EESetup(): boolean {
     return this.e2eeManager !== undefined;
@@ -294,15 +304,36 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
         this.engine.sendLossyBytes(bytes, DataChannelKind.DATA_TRACK_LOSSY, 'wait');
       });
 
+    this.registerRpcDataStreamHandler();
+
+    this.rpcClientManager = new RpcClientManager(
+      this.log,
+      this.outgoingDataStreamManager,
+      this.getRemoteParticipantClientProtocol,
+      () => this.engine.latestJoinResponse?.serverInfo?.version,
+    );
+    this.rpcClientManager.on('sendDataPacket', ({ packet }) => {
+      this.engine?.sendDataPacket(packet, DataChannelKind.RELIABLE);
+    });
+    this.rpcServerManager = new RpcServerManager(
+      this.log,
+      this.outgoingDataStreamManager,
+      this.getRemoteParticipantClientProtocol,
+    );
+    this.rpcServerManager.on('sendDataPacket', ({ packet }) => {
+      this.engine?.sendDataPacket(packet, DataChannelKind.RELIABLE);
+    });
+
     this.disconnectLock = new Mutex();
     this.localParticipant = new LocalParticipant(
       '',
       '',
       this.engine,
       this.options,
-      this.rpcHandlers,
       this.outgoingDataStreamManager,
       this.outgoingDataTrackManager,
+      this.rpcClientManager,
+      this.rpcServerManager,
     );
 
     if (this.options.e2ee || this.options.encryption) {
@@ -391,12 +422,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
    * Other errors thrown in your handler will not be transmitted as-is, and will instead arrive to the caller as `1500` ("Application Error").
    */
   registerRpcMethod(method: string, handler: (data: RpcInvocationData) => Promise<string>) {
-    if (this.rpcHandlers.has(method)) {
-      throw Error(
-        `RPC handler already registered for method ${method}, unregisterRpcMethod before trying to register again`,
-      );
-    }
-    this.rpcHandlers.set(method, handler);
+    this.rpcServerManager.registerRpcMethod(method, handler);
   }
 
   /**
@@ -405,7 +431,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
    * @param method - The name of the RPC method to unregister
    */
   unregisterRpcMethod(method: string) {
-    this.rpcHandlers.delete(method);
+    this.rpcServerManager.unregisterRpcMethod(method);
   }
 
   /**
@@ -1805,7 +1831,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     });
     this.emit(RoomEvent.ParticipantDisconnected, participant);
     participant.setDisconnected();
-    this.localParticipant?.handleParticipantDisconnected(participant.identity);
+    this.rpcClientManager.handleParticipantDisconnected(participant.identity);
   }
 
   // updates are sent only when there's a change to speaker ordering
@@ -1949,14 +1975,31 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
       this.handleDataStream(packet, encryptionType);
     } else if (packet.value.case === 'rpcRequest') {
       const rpc = packet.value.value;
-      this.handleIncomingRpcRequest(
-        packet.participantIdentity,
-        rpc.id,
-        rpc.method,
-        rpc.payload,
-        rpc.responseTimeoutMs,
-        rpc.version,
-      );
+      this.rpcServerManager.handleIncomingRpcRequest(packet.participantIdentity, rpc);
+    } else if (packet.value.case === 'rpcResponse') {
+      const rpcResponse = packet.value.value;
+      switch (rpcResponse.value.case) {
+        case 'payload':
+          this.rpcClientManager.handleIncomingRpcResponseSuccess(
+            rpcResponse.requestId,
+            rpcResponse.value.value,
+          );
+          break;
+        case 'error':
+          this.rpcClientManager.handleIncomingRpcResponseFailure(
+            rpcResponse.requestId,
+            RpcError.fromProto(rpcResponse.value.value),
+          );
+          break;
+        default:
+          this.log.warn(
+            `Unknown rpcResponse.value.case: ${rpcResponse.value.case}`,
+            this.logContext,
+          );
+          break;
+      }
+    } else if (packet.value.case === 'rpcAck') {
+      this.rpcClientManager.handleIncomingRpcAck(packet.value.value.requestId);
     }
   };
 
@@ -2020,68 +2063,6 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     this.incomingDataStreamManager.handleDataStreamPacket(packet, encryptionType);
   };
 
-  private async handleIncomingRpcRequest(
-    callerIdentity: string,
-    requestId: string,
-    method: string,
-    payload: string,
-    responseTimeout: number,
-    version: number,
-  ) {
-    await this.engine.publishRpcAck(callerIdentity, requestId);
-
-    if (version !== 1) {
-      await this.engine.publishRpcResponse(
-        callerIdentity,
-        requestId,
-        null,
-        RpcError.builtIn('UNSUPPORTED_VERSION'),
-      );
-      return;
-    }
-
-    const handler = this.rpcHandlers.get(method);
-
-    if (!handler) {
-      await this.engine.publishRpcResponse(
-        callerIdentity,
-        requestId,
-        null,
-        RpcError.builtIn('UNSUPPORTED_METHOD'),
-      );
-      return;
-    }
-
-    let responseError: RpcError | null = null;
-    let responsePayload: string | null = null;
-
-    try {
-      const response = await handler({
-        requestId,
-        callerIdentity,
-        payload,
-        responseTimeout,
-      });
-      if (byteLength(response) > MAX_PAYLOAD_BYTES) {
-        responseError = RpcError.builtIn('RESPONSE_PAYLOAD_TOO_LARGE');
-        this.log.warn(`RPC Response payload too large for ${method}`);
-      } else {
-        responsePayload = response;
-      }
-    } catch (error) {
-      if (error instanceof RpcError) {
-        responseError = error;
-      } else {
-        this.log.warn(
-          `Uncaught error returned by RPC handler for ${method}. Returning APPLICATION_ERROR instead.`,
-          error,
-        );
-        responseError = RpcError.builtIn('APPLICATION_ERROR');
-      }
-    }
-    await this.engine.publishRpcResponse(callerIdentity, requestId, responsePayload, responseError);
-  }
-
   bufferedSegments: Map<string, TranscriptionSegmentModel> = new Map();
 
   private handleAudioPlaybackStarted = () => {
@@ -2427,6 +2408,27 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     }
   }
 
+  private getRemoteParticipantClientProtocol = (identity: Participant['identity']) => {
+    return this.remoteParticipants.get(identity)?.clientProtocol ?? CLIENT_PROTOCOL_DEFAULT;
+  };
+
+  private registerRpcDataStreamHandler() {
+    this.incomingDataStreamManager.registerTextStreamHandler(
+      RPC_REQUEST_DATA_STREAM_TOPIC,
+      async (reader, { identity }) => {
+        const attributes = reader.info.attributes ?? {};
+        await this.rpcServerManager.handleIncomingDataStream(reader, identity, attributes);
+      },
+    );
+    this.incomingDataStreamManager.registerTextStreamHandler(
+      RPC_RESPONSE_DATA_STREAM_TOPIC,
+      async (reader) => {
+        const attributes = reader.info.attributes ?? {};
+        await this.rpcClientManager.handleIncomingDataStream(reader, attributes);
+      },
+    );
+  }
+
   private registerConnectionReconcile() {
     this.clearConnectionReconcile();
     let consecutiveFailures = 0;