fix: warn instead of rejecting when a data stream handler fails after the room disconnects

.changeset/catch-stream-handler-errors.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+A text or byte stream handler that fails after the room has disconnected (for example an `async` handler that doesn't catch the error from a stream cut short by the disconnect) is now logged as a warning instead of surfacing as an unhandled promise rejection. Handler errors while the room is connected or reconnecting are unchanged. `TextStreamHandler` and `ByteStreamHandler` are now typed as returning `void | Promise<void>`.

src/room/data-stream/incoming/IncomingDataStreamManager.test.ts

@@ -8,11 +8,13 @@ import {
   DataStream_Trailer,
   Encryption_Type,
 } from '@livekit/protocol';
-import { describe, expect, it } from 'vitest';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import log from '../../../logger';
 import {
   stubMissingCompressionStreams,
   stubMissingDeflateRawSupport,
 } from '../../../test/compressionStreamStubs';
+import { DataStreamErrorReason } from '../../errors';
 import { deflateRawCompress } from '../compression';
 import { STREAM_CHUNK_SIZE_BYTES } from '../constants';
 import IncomingDataStreamManager from './IncomingDataStreamManager';
@@ -2007,4 +2009,188 @@ describe('IncomingDataStreamManager', () => {
       expect(await readers[1].readAll()).toEqual('fresh stream');
     });
   });
+
+  describe('Stream handler errors', () => {
+    let unhandled: unknown[];
+    let vitestListeners: NodeJS.UnhandledRejectionListener[];
+    const onUnhandled = (reason: unknown) => {
+      unhandled.push(reason);
+    };
+
+    // Some cases expect an unhandled rejection, so swap out vitest's own listener (which would fail
+    // the run) for one that records them.
+    beforeEach(() => {
+      unhandled = [];
+      vitestListeners = process.listeners('unhandledRejection');
+      process.removeAllListeners('unhandledRejection');
+      process.on('unhandledRejection', onUnhandled);
+    });
+
+    afterEach(() => {
+      process.off('unhandledRejection', onUnhandled);
+      for (const listener of vitestListeners) {
+        process.on('unhandledRejection', listener);
+      }
+      vi.restoreAllMocks();
+    });
+
+    /** Drains microtasks and timers so any unhandledRejection has a chance to fire. */
+    const flushAnyUnhandledRejections = async () => {
+      await new Promise((resolve) => setTimeout(resolve, 0));
+      await new Promise((resolve) => setTimeout(resolve, 0));
+    };
+
+    /** Registers async handlers that read without a try/catch, the way most handlers are written. */
+    const registerUncaughtHandlers = (manager: IncomingDataStreamManager) => {
+      manager.registerTextStreamHandler('my-topic', async (reader) => {
+        await reader.readAll();
+      });
+      manager.registerByteStreamHandler('my-topic', async (reader) => {
+        await reader.readAll();
+      });
+    };
+
+    it('should warn instead of rejecting when a stream is cut short by the room disconnecting', async () => {
+      const warnSpy = vi.spyOn(log, 'warn').mockImplementation(() => {});
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+      registerUncaughtHandlers(manager);
+
+      manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE);
+      manager.handleDataStreamPacket(
+        chunkPacket('text-1', 0, new Uint8Array([0x68, 0x69])),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(headerPacket('byte-1', 'byteHeader'), Encryption_Type.NONE);
+      manager.handleDataStreamPacket(
+        chunkPacket('byte-1', 0, new Uint8Array([0x01])),
+        Encryption_Type.NONE,
+      );
+
+      // Room#handleDisconnect calls this while the streams are still open
+      manager.clearControllers();
+      manager.setConnected(false);
+      await flushAnyUnhandledRejections();
+
+      expect(unhandled).toEqual([]);
+      for (const streamId of ['text-1', 'byte-1']) {
+        expect(warnSpy).toHaveBeenCalledWith(
+          'data stream handler failed after the room disconnected',
+          expect.objectContaining({
+            topic: 'my-topic',
+            streamId,
+            error: expect.objectContaining({ reason: DataStreamErrorReason.AbnormalEnd }),
+          }),
+        );
+      }
+    });
+
+    it('should warn for any handler error that lands after the room disconnected', async () => {
+      const warnSpy = vi.spyOn(log, 'warn').mockImplementation(() => {});
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+
+      const handlerError = new Error('application bug');
+      let release!: () => void;
+      const released = new Promise<void>((resolve) => {
+        release = resolve;
+      });
+      manager.registerTextStreamHandler('my-topic', async () => {
+        await released;
+        throw handlerError;
+      });
+
+      manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE);
+      manager.clearControllers();
+      release();
+      await flushAnyUnhandledRejections();
+
+      expect(unhandled).toEqual([]);
+      expect(warnSpy).toHaveBeenCalledWith(
+        'data stream handler failed after the room disconnected',
+        expect.objectContaining({ error: handlerError }),
+      );
+    });
+
+    it('should still reject unhandled when the sender disconnects mid stream', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+      registerUncaughtHandlers(manager);
+
+      manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE);
+      manager.handleDataStreamPacket(
+        chunkPacket('text-1', 0, new Uint8Array([0x68, 0x69])),
+        Encryption_Type.NONE,
+      );
+      manager.validateParticipantHasNoActiveDataStreams('alice');
+      await flushAnyUnhandledRejections();
+
+      expect(unhandled).toEqual([
+        expect.objectContaining({
+          message: 'Participant alice unexpectedly disconnected in the middle of sending data',
+        }),
+      ]);
+    });
+
+    it('should still reject unhandled when a chunk is lost', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+      registerUncaughtHandlers(manager);
+
+      manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE);
+      manager.handleDataStreamPacket(
+        chunkPacket('text-1', 0, new Uint8Array([0x68])),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(
+        chunkPacket('text-1', 2, new Uint8Array([0x69])),
+        Encryption_Type.NONE,
+      );
+      await flushAnyUnhandledRejections();
+
+      expect(unhandled).toEqual([
+        expect.objectContaining({ reason: DataStreamErrorReason.Incomplete }),
+      ]);
+    });
+
+    it('should still reject unhandled while the room is reconnecting', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+
+      const handlerError = new Error('application bug');
+      let release!: () => void;
+      const released = new Promise<void>((resolve) => {
+        release = resolve;
+      });
+      manager.registerTextStreamHandler('my-topic', async () => {
+        await released;
+        throw handlerError;
+      });
+
+      manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE);
+      // A reconnect drops out of the connected state without disconnecting the room.
+      manager.setConnected(false);
+      release();
+      await flushAnyUnhandledRejections();
+
+      expect(unhandled).toEqual([handlerError]);
+    });
+
+    it('should reject unhandled again once the room connects after a disconnect', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+      manager.clearControllers();
+      manager.setConnected(false);
+      manager.setConnected(true);
+
+      const handlerError = new Error('application bug');
+      manager.registerTextStreamHandler('my-topic', async () => {
+        throw handlerError;
+      });
+      manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE);
+      await flushAnyUnhandledRejections();
+
+      expect(unhandled).toEqual([handlerError]);
+    });
+  });
 });

src/room/data-stream/incoming/IncomingDataStreamManager.ts

@@ -41,11 +41,16 @@ export default class IncomingDataStreamManager {
 
   private isConnected = false;
 
+  /** Set once the room disconnects, and cleared when it connects again. Unlike `isConnected`, this
+   * stays false while the room is reconnecting. */
+  private hasDisconnected = false;
+
   private bufferedPackets: Array<{ packet: DataPacket; encryptionType: Encryption_Type }> = [];
 
   setConnected(connected: boolean) {
     this.isConnected = connected;
     if (connected) {
+      this.hasDisconnected = false;
       this.flushBufferedPackets();
     }
   }
@@ -87,6 +92,8 @@ export default class IncomingDataStreamManager {
   }
 
   clearControllers() {
+    this.hasDisconnected = true;
+
     // Error any in flight streams so readers do not wait forever for chunks that can no longer arrive.
     const abnormalEndError = new DataStreamError(
       'Room disconnected in the middle of receiving data',
@@ -131,6 +138,32 @@ export default class IncomingDataStreamManager {
     }
   }
 
+  /**
+   * Calls a registered stream handler.
+   *
+   * Nothing awaits the handler, so an error it rejects with and doesn't catch itself surfaces as
+   * an unhandled promise rejection. That is left as is while the room is connected, so failures
+   * like a lost chunk stay loud. Once the room has disconnected, though, the session is over and
+   * whatever went wrong is out of the application's hands, so the error is logged as a warning
+   * instead.
+   */
+  private invokeStreamHandler<R extends ByteStreamReader | TextStreamReader>(
+    handler: (reader: R, participantInfo: { identity: string }) => void | Promise<void>,
+    reader: R,
+    participantInfo: { identity: string },
+  ) {
+    Promise.resolve(handler(reader, participantInfo)).catch((error: unknown) => {
+      if (!this.hasDisconnected) {
+        throw error;
+      }
+      this.log.warn('data stream handler failed after the room disconnected', {
+        topic: reader.info.topic,
+        streamId: reader.info.id,
+        error,
+      });
+    });
+  }
+
   handleDataStreamPacket(packet: DataPacket, encryptionType: Encryption_Type) {
     if (!this.isConnected) {
       this.bufferedPackets.push({ packet, encryptionType });
@@ -212,7 +245,8 @@ export default class IncomingDataStreamManager {
         const inlineContent = streamHeader.inlineContent as NonSharedUint8Array;
         if (typeof inlineContent !== 'undefined') {
           // Inline bytes are the raw payload, optionally deflate-raw compressed.
-          streamHandlerCallback(
+          this.invokeStreamHandler(
+            streamHandlerCallback,
             new ByteStreamReader(
               info,
               createInlineStream(
@@ -257,7 +291,8 @@ export default class IncomingDataStreamManager {
             }
           },
         });
-        streamHandlerCallback(
+        this.invokeStreamHandler(
+          streamHandlerCallback,
           new ByteStreamReader(
             info,
             compressed
@@ -330,7 +365,8 @@ export default class IncomingDataStreamManager {
           const content = compressed
             ? deflateRawDecompress(inlineContent, this.maxPayloadByteLength)
             : inlineContent;
-          streamHandlerCallback(
+          this.invokeStreamHandler(
+            streamHandlerCallback,
             new TextStreamReader(
               info,
               createInlineStream(streamHeader.streamId, content),
@@ -370,7 +406,8 @@ export default class IncomingDataStreamManager {
             }
           },
         });
-        streamHandlerCallback(
+        this.invokeStreamHandler(
+          streamHandlerCallback,
           new TextStreamReader(
             info,
             compressed

src/room/data-stream/incoming/StreamReader.ts

@@ -278,9 +278,9 @@ export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
 export type ByteStreamHandler = (
   reader: ByteStreamReader,
   participantInfo: { identity: string },
-) => void;
+) => void | Promise<void>;
 
 export type TextStreamHandler = (
   reader: TextStreamReader,
   participantInfo: { identity: string },
-) => void;
+) => void | Promise<void>;