Properly handle max payload length of zero (#1966)

Author: 1egoman

.changeset/chatty-oranges-clap.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Properly handle maxPayloadLength of 0, upper bound maxMessageSize

src/room/RTCEngine.test.ts

@@ -223,15 +223,18 @@ describe('RTCEngine', () => {
 
   describe('sendDataPacket', () => {
     const MAX_DATA_PACKET_SIZE = 64 * 1024 - 1; // 65535 bytes (64 KB - 1)
-    function stubConnectedEngine(engine: RTCEngine) {
+    function stubConnectedEngine(
+      engine: RTCEngine,
+      maxDataPacketSize: number = MAX_DATA_PACKET_SIZE,
+    ) {
       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),
+          getMaxPublisherMessageSize: vi.fn(() => maxDataPacketSize),
         },
       });
       return send;
@@ -257,6 +260,25 @@ describe('RTCEngine', () => {
       expect(send).not.toHaveBeenCalled();
     });
 
+    it('does not reject packets if the max data packet size is 0', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const send = stubConnectedEngine(engine, 0);
+
+      const packet = new DataPacket({
+        kind: DataPacket_Kind.RELIABLE,
+        value: {
+          case: 'user',
+          value: new UserPacket({ payload: new Uint8Array(100) }),
+        },
+      });
+
+      // Sending the packet should succeed, there isn't a size limit
+      await expect(
+        engine.sendDataPacket(packet, DataChannelKind.RELIABLE),
+      ).resolves.toBeUndefined();
+      expect(send).toHaveBeenCalledTimes(1);
+    });
+
     it('sends packets within the max data packet size', async () => {
       const engine = new RTCEngine(roomOptionDefaults);
       const send = stubConnectedEngine(engine);
@@ -275,4 +297,57 @@ describe('RTCEngine', () => {
       expect(send).toHaveBeenCalledTimes(1);
     });
   });
+
+  describe('handleDataChannelClose', () => {
+    function stubCloseEnv(
+      engine: RTCEngine,
+      { closed, publisherState }: { closed: boolean; publisherState: RTCPeerConnectionState },
+    ) {
+      const error = vi.fn();
+      Object.assign(engine as unknown as Record<string, unknown>, {
+        _isClosed: closed,
+        log: { error },
+        pcManager: {
+          publisher: { getConnectionState: () => publisherState },
+        },
+      });
+      return error;
+    }
+
+    function fireClose(engine: RTCEngine, kind: DataChannelKind) {
+      (
+        engine as unknown as {
+          handleDataChannelClose: (kind: DataChannelKind) => () => void;
+        }
+      ).handleDataChannelClose(kind)();
+    }
+
+    it('logs an error when a publisher channel closes while connected', () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const error = stubCloseEnv(engine, { closed: false, publisherState: 'connected' });
+
+      fireClose(engine, DataChannelKind.RELIABLE);
+
+      expect(error).toHaveBeenCalledOnce();
+      expect(error.mock.calls[0][0]).toContain('RELIABLE');
+    });
+
+    it('stays quiet when the engine is already closed', () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const error = stubCloseEnv(engine, { closed: true, publisherState: 'connected' });
+
+      fireClose(engine, DataChannelKind.RELIABLE);
+
+      expect(error).not.toHaveBeenCalled();
+    });
+
+    it('stays quiet when the publisher PC is no longer connected', () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const error = stubCloseEnv(engine, { closed: false, publisherState: 'closed' });
+
+      fireClose(engine, DataChannelKind.RELIABLE);
+
+      expect(error).not.toHaveBeenCalled();
+    });
+  });
 });

src/room/RTCEngine.ts

@@ -127,6 +127,11 @@ export enum DataChannelKind {
   DATA_TRACK_LOSSY = 2,
 }
 
+// Default data-channel max message size (bytes), used when the remote SDP
+// answer does not advertise an `a=max-message-size` attribute (RFC 8841).
+// `0` means "no limit".
+const DEFAULT_MAX_MESSAGE_SIZE = 64_000;
+
 /** @internal */
 export default class RTCEngine extends (EventEmitter as new () => TypedEventEmitter<EngineEventCallbacks>) {
   client: SignalClient;
@@ -864,14 +869,17 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     if (this.lossyDC) {
       this.lossyDC.onmessage = null;
       this.lossyDC.onerror = null;
+      this.lossyDC.onclose = null;
     }
     if (this.reliableDC) {
       this.reliableDC.onmessage = null;
       this.reliableDC.onerror = null;
+      this.reliableDC.onclose = null;
     }
     if (this.dataTrackDC) {
       this.dataTrackDC.onmessage = null;
       this.dataTrackDC.onerror = null;
+      this.dataTrackDC.onclose = null;
     }
 
     // create data channels
@@ -897,6 +905,11 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     this.reliableDC.onerror = this.handleDataError;
     this.dataTrackDC.onerror = this.handleDataError;
 
+    // detect unexpected publisher data channel closes
+    this.lossyDC.onclose = this.handleDataChannelClose(DataChannelKind.LOSSY);
+    this.reliableDC.onclose = this.handleDataChannelClose(DataChannelKind.RELIABLE);
+    this.dataTrackDC.onclose = this.handleDataChannelClose(DataChannelKind.DATA_TRACK_LOSSY);
+
     // set up dc buffer threshold, set to 64kB (otherwise 0 by default)
     this.lossyDC.bufferedAmountLowThreshold = 65535;
     this.reliableDC.bufferedAmountLowThreshold = 65535;
@@ -1036,6 +1049,18 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     }
   };
 
+  private handleDataChannelClose = (kind: DataChannelKind) => () => {
+    // A publisher DC closing while the session is up and the publisher PC is still
+    // connected is the signature of an oversized message having aborted the channel
+    // (see livekit/rust-sdks#1137). Surface it; do not attempt renegotiation.
+    if (!this._isClosed && this.pcManager?.publisher.getConnectionState() === 'connected') {
+      this.log.error(
+        `publisher data channel '${DataChannelKind[kind]}' closed unexpectedly`,
+        this.logContext,
+      );
+    }
+  };
+
   private handleBufferedAmountLow = (channelKind: DataChannelKind) => {
     this.updateAndEmitDCBufferStatus(channelKind);
   };
@@ -1537,9 +1562,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
     const msg = packet.toBinary() as Uint8Array<ArrayBuffer>;
 
-    const maxPublisherMessageSizeBytes = this.pcManager?.getMaxPublisherMessageSize();
+    // Clamp to the SDK default - libwebrtc advertises larger (~256 KiB)
+    // than LiveKit/pion can deliver end-to-end (~64 KiB), so we trust
+    // the answer up untilthe built in ceiling.
+    const maxPublisherMessageSizeBytes = Math.min(
+      this.pcManager?.getMaxPublisherMessageSize() ?? DEFAULT_MAX_MESSAGE_SIZE,
+      DEFAULT_MAX_MESSAGE_SIZE,
+    );
     if (
       typeof maxPublisherMessageSizeBytes !== 'undefined' &&
+      maxPublisherMessageSizeBytes !== 0 /* 0 means "no limit" */ &&
       msg.byteLength > maxPublisherMessageSizeBytes
     ) {
       throw new PublishDataError(