feat: migrate rpc to use data streams for sending rpc requests

Author: 1egoman

src/room/Room.ts

@@ -1853,7 +1853,6 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
         rpc.id,
         rpc.method,
         rpc.payload,
-        rpc.compressedPayload,
         rpc.responseTimeoutMs,
         rpc.version,
         () => this.remoteParticipants.has(packet.participantIdentity),

src/room/rpc.test.ts

@@ -48,7 +48,6 @@ describe('RpcClientManager', () => {
     expect(packet.value.value.id).toStrictEqual(requestId);
     expect(packet.value.value.method).toStrictEqual('test-method');
     expect(packet.value.value.payload).toStrictEqual('request-payload');
-    expect(packet.value.value.compressedPayload).toStrictEqual(new Uint8Array());
 
     rpcClientManager.handleIncomingRpcAck(requestId);
 
@@ -192,7 +191,6 @@ describe('RpcServerManager', () => {
       requestId,
       'test-method',
       'request payload',
-      new Uint8Array(),
       responseTimeoutMs,
       1,
       () => true,
@@ -225,7 +223,6 @@ describe('RpcServerManager', () => {
       'test-request-id',
       methodName,
       'test payload',
-      new Uint8Array(),
       5000,
       1,
       () => true,
@@ -262,7 +259,6 @@ describe('RpcServerManager', () => {
       'test-error-request-id',
       methodName,
       'test payload',
-      new Uint8Array(),
       5000,
       1,
       () => true,
@@ -300,7 +296,6 @@ describe('RpcServerManager', () => {
       'test-rpc-error-request-id',
       methodName,
       'test payload',
-      new Uint8Array(),
       5000,
       1,
       () => true,

src/room/rpc/RpcClientManager.ts

@@ -11,7 +11,6 @@ import { EngineEvent } from '../events';
 import type Participant from '../participant/Participant';
 import { Future, compareVersions } from '../utils';
 import {
-  DATA_STREAM_MIN_BYTES,
   MAX_LEGACY_PAYLOAD_BYTES,
   type PerformRpcParams,
   RPC_DATA_STREAM_TOPIC,
@@ -20,9 +19,7 @@ import {
   RPC_REQUEST_RESPONSE_TIMEOUT_MS_ATTR,
   RpcError,
   byteLength,
-  gzipCompress,
   gzipCompressToWriter,
-  gzipDecompress,
   gzipDecompressFromReader,
 } from './utils';
 
@@ -151,70 +148,38 @@ export default class RpcClientManager {
     responseTimeout: number,
     remoteClientProtocol: number,
   ) {
-    const payloadBytes = byteLength(payload);
-
-    let mode: 'regular' | 'compressed' | 'compressed-data-stream' = 'regular';
     if (remoteClientProtocol >= CLIENT_PROTOCOL_GZIP_RPC) {
-      mode = 'compressed';
-    }
-    if (mode === 'compressed' && payloadBytes > DATA_STREAM_MIN_BYTES) {
-      mode = 'compressed-data-stream';
+      // Send payload as a compressed data stream
+      const writer = await this.outgoingDataStreamManager.streamBytes({
+        topic: RPC_DATA_STREAM_TOPIC,
+        destinationIdentities: [destinationIdentity],
+        mimeType: 'application/octet-stream',
+        attributes: {
+          [RPC_REQUEST_ID_ATTR]: requestId,
+          [RPC_REQUEST_METHOD_ATTR]: method,
+          [RPC_REQUEST_RESPONSE_TIMEOUT_MS_ATTR]: `${responseTimeout}`,
+        },
+      });
+      await gzipCompressToWriter(payload, writer);
+      await writer.close();
+      return;
     }
 
-    switch (mode) {
-      case 'compressed-data-stream': {
-        // Large payload: create the data stream tagged with the request ID,
-        // send the RPC request with empty payload/compressedPayload, then
-        // stream compressed chunks for lower TTFB
-        const writer = await this.outgoingDataStreamManager.streamBytes({
-          topic: RPC_DATA_STREAM_TOPIC,
-          destinationIdentities: [destinationIdentity],
-          mimeType: 'application/octet-stream',
-          attributes: {
-            [RPC_REQUEST_ID_ATTR]: requestId,
-            [RPC_REQUEST_METHOD_ATTR]: method,
-            [RPC_REQUEST_RESPONSE_TIMEOUT_MS_ATTR]: `${responseTimeout}`,
-          },
-        });
-        await gzipCompressToWriter(payload, writer);
-        await writer.close();
-        return;
-      }
-
-      case 'compressed':
-        // Medium payload: compress inline
-        const compressedPayload = await gzipCompress(payload);
-        await this.sendRpcRequestPacket(
-          destinationIdentity,
-          requestId,
-          method,
-          '',
-          compressedPayload,
-          responseTimeout,
-        );
-        break;
-
-      case 'regular':
-      default:
-        // Small payload: just include the payload directly, uncompressed
-        await this.sendRpcRequestPacket(
-          destinationIdentity,
-          requestId,
-          method,
-          payload,
-          undefined,
-          responseTimeout,
-        );
-        break;
-    }
+    // Legacy client: send uncompressed payload inline
+    await this.sendRpcRequestPacket(
+      destinationIdentity,
+      requestId,
+      method,
+      payload,
+      responseTimeout,
+    );
   }
 
   private async sendRpcRequestPacket(
     destinationIdentity: string,
     requestId: string,
     method: string,
-    payload: string | undefined,
-    compressedPayload: Uint8Array | undefined,
+    payload: string,
     responseTimeout: number,
   ) {
     const packet = new DataPacket({
@@ -225,8 +190,7 @@ export default class RpcClientManager {
         value: new RpcRequest({
           id: requestId,
           method,
-          payload: payload ?? '',
-          compressedPayload: compressedPayload ?? new Uint8Array(),
+          payload,
           responseTimeoutMs: responseTimeout,
           version: 1,
         }),
@@ -251,22 +215,6 @@ export default class RpcClientManager {
             return true;
           }
 
-          case 'compressedPayload': {
-            let payload;
-            try {
-              payload = await gzipDecompress(rpcResponse.value.value);
-            } catch (e) {
-              this.log.error('Failed to decompress RPC response', e);
-              this.handleIncomingRpcResponseFailure(
-                rpcResponse.requestId,
-                RpcError.builtIn('APPLICATION_ERROR'),
-              );
-              return true;
-            }
-            this.handleIncomingRpcResponseSuccess(rpcResponse.requestId, payload);
-            return true;
-          }
-
           case 'error': {
             const error = RpcError.fromProto(rpcResponse.value.value);
             this.handleIncomingRpcResponseFailure(rpcResponse.requestId, error);