Add 64k max data packet size (#1962)

Author: 1egoman

.changeset/large-pillows-grow.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+Enforce a maximum size of the sdp defined maxMessageSize on individual data packets sent via `publishData`

src/room/PCTransport.ts

@@ -498,6 +498,10 @@ export default class PCTransport extends (EventEmitter as new () => TypedEmitter
     return this.pc.getStats();
   }
 
+  getMaxMessageSize() {
+    return this._pc?.sctp?.maxMessageSize;
+  }
+
   async getConnectedAddress(): Promise<string | undefined> {
     if (!this._pc) {
       return;

src/room/PCTransportManager.ts

@@ -310,6 +310,10 @@ export class PCTransportManager {
     return matchingTransceiver?.mid;
   }
 
+  getMaxPublisherMessageSize() {
+    return this.publisher.getMaxMessageSize();
+  }
+
   addPublisherTrack(track: MediaStreamTrack) {
     return this.publisher.addTrack(track);
   }

src/room/RTCEngine.test.ts

@@ -1,6 +1,8 @@
+import { DataPacket, DataPacket_Kind, UserPacket } from '@livekit/protocol';
 import { afterEach, describe, expect, it, vi } from 'vitest';
-import RTCEngine from './RTCEngine';
+import RTCEngine, { DataChannelKind } from './RTCEngine';
 import { roomOptionDefaults } from './defaults';
+import { PublishDataError } from './errors';
 
 describe('RTCEngine', () => {
   const originalRTCRtpSender = window.RTCRtpSender;
@@ -218,4 +220,59 @@ describe('RTCEngine', () => {
     expect((sender as unknown as { transform: unknown }).transform).toBe(transform);
     expect(createEncodedStreams).not.toHaveBeenCalled();
   });
+
+  describe('sendDataPacket', () => {
+    const MAX_DATA_PACKET_SIZE = 64 * 1024 - 1; // 65535 bytes (64 KB - 1)
+    function stubConnectedEngine(engine: RTCEngine) {
+      const send = vi.fn();
+      Object.assign(engine as unknown as Record<string, unknown>, {
+        ensurePublisherConnected: vi.fn().mockResolvedValue(undefined),
+        waitForBufferStatusLow: vi.fn().mockResolvedValue(undefined),
+        updateAndEmitDCBufferStatus: vi.fn(),
+        dataChannelForKind: vi.fn(() => ({ send })),
+        pcManager: {
+          getMaxPublisherMessageSize: vi.fn(() => MAX_DATA_PACKET_SIZE),
+        },
+      });
+      return send;
+    }
+
+    it('rejects packets larger than the max data packet size', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const send = stubConnectedEngine(engine);
+
+      // The serialized packet includes protobuf framing on top of the payload, so a payload at the
+      // limit is already guaranteed to exceed it once serialized.
+      const packet = new DataPacket({
+        kind: DataPacket_Kind.RELIABLE,
+        value: {
+          case: 'user',
+          value: new UserPacket({ payload: new Uint8Array(MAX_DATA_PACKET_SIZE) }),
+        },
+      });
+
+      await expect(engine.sendDataPacket(packet, DataChannelKind.RELIABLE)).rejects.toBeInstanceOf(
+        PublishDataError,
+      );
+      expect(send).not.toHaveBeenCalled();
+    });
+
+    it('sends packets within the max data packet size', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const send = stubConnectedEngine(engine);
+
+      const packet = new DataPacket({
+        kind: DataPacket_Kind.RELIABLE,
+        value: {
+          case: 'user',
+          value: new UserPacket({ payload: new Uint8Array(1024) }),
+        },
+      });
+
+      await expect(
+        engine.sendDataPacket(packet, DataChannelKind.RELIABLE),
+      ).resolves.toBeUndefined();
+      expect(send).toHaveBeenCalledTimes(1);
+    });
+  });
 });

src/room/RTCEngine.ts

@@ -73,6 +73,7 @@ import {
   ConnectionError,
   ConnectionErrorReason,
   NegotiationError,
+  PublishDataError,
   SignalReconnectError,
   TrackInvalidError,
   UnexpectedConnectionState,
@@ -108,6 +109,7 @@ const leaveReconnect = 'leave-reconnect';
 const reliabeReceiveStateTTL = 30_000;
 const lossyDataChannelBufferThresholdMin = 8 * 1024;
 const lossyDataChannelBufferThresholdMax = 256 * 1024;
+
 const initialMediaSectionsAudio = 3;
 const initialMediaSectionsVideo = 3;
 
@@ -1535,6 +1537,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
     const msg = packet.toBinary() as Uint8Array<ArrayBuffer>;
 
+    const maxPublisherMessageSizeBytes = this.pcManager?.getMaxPublisherMessageSize();
+    if (
+      typeof maxPublisherMessageSizeBytes !== 'undefined' &&
+      msg.byteLength > maxPublisherMessageSizeBytes
+    ) {
+      throw new PublishDataError(
+        `cannot publish data packet larger than ${maxPublisherMessageSizeBytes} bytes (got ${msg.byteLength})`,
+      );
+    }
+
     switch (kind) {
       case DataChannelKind.LOSSY:
       case DataChannelKind.DATA_TRACK_LOSSY: