From 9d434d850133549ceba98657f30b2920d550ed7a Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 28 Oct 2026 12:12:06 +1300 Subject: [PATCH 1/3] fix: catch errors thrown by data stream handlers Registered text/byte stream handlers were invoked without awaiting or catching their result. An async handler that iterated its reader without a try/catch therefore produced an unhandled promise rejection whenever the stream ended abnormally (sender or local room disconnecting mid stream), and a synchronous throw escaped out of packet handling. Invoke handlers through a wrapper that catches both: AbnormalEnd DataStreamErrors are logged at debug, anything else at error. Co-Authored-By: Claude Opus 5.5 --- .changeset/catch-stream-handler-errors.md | 5 + .../IncomingDataStreamManager.test.ts | 132 +++++++++++++++++- .../incoming/IncomingDataStreamManager.ts | 41 +++++- src/room/data-stream/incoming/StreamReader.ts | 4 +- 4 files changed, 175 insertions(+), 7 deletions(-) create mode 100644 .changeset/catch-stream-handler-errors.md diff --git a/.changeset/catch-stream-handler-errors.md b/.changeset/catch-stream-handler-errors.md new file mode 100644 index 0000000000..4a49940de2 --- /dev/null +++ b/.changeset/catch-stream-handler-errors.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Errors thrown or rejected by a registered text or byte stream handler no longer surface as unhandled promise rejections. Previously an `async` handler that did not wrap its reads in `try`/`catch` produced an uncatchable unhandled rejection whenever a stream ended abnormally (the sender or the local room disconnecting mid stream). Abnormal ends are now logged at debug level and any other handler error is logged as an error. `TextStreamHandler` and `ByteStreamHandler` are now typed as returning `void | Promise`. diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts index b2ca7c75fa..e175777a2c 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts @@ -8,7 +8,8 @@ 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, @@ -2007,4 +2008,133 @@ describe('IncomingDataStreamManager', () => { expect(await readers[1].readAll()).toEqual('fresh stream'); }); }); + + describe('Stream handler errors', () => { + let unhandled: unknown[]; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + + beforeEach(() => { + unhandled = []; + process.on('unhandledRejection', onUnhandled); + }); + + afterEach(() => { + process.off('unhandledRejection', onUnhandled); + vi.restoreAllMocks(); + }); + + /** Drains microtasks and timers so any unhandledRejection has a chance to fire. */ + const flush = async () => { + await new Promise((resolve) => setTimeout(resolve, 0)); + await new Promise((resolve) => setTimeout(resolve, 0)); + }; + + it('should not surface an unhandled rejection when an async text handler does not catch a remote disconnect', async () => { + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + const manager = new IncomingDataStreamManager(); + manager.setConnected(true); + + const received: string[] = []; + manager.registerTextStreamHandler('my-topic', async (reader) => { + // No try/catch, the way most handlers are written. + for await (const chunk of reader) { + received.push(chunk); + } + }); + + manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE); + manager.handleDataStreamPacket( + chunkPacket('text-1', 0, new TextEncoder().encode('hello')), + Encryption_Type.NONE, + ); + await flush(); + manager.validateParticipantHasNoActiveDataStreams('alice'); + await flush(); + + expect(received).toEqual(['hello']); + expect(unhandled).toEqual([]); + expect(debugSpy).toHaveBeenCalledWith( + 'data stream handler did not handle an abnormally ended stream', + expect.objectContaining({ topic: 'my-topic', streamId: 'text-1' }), + ); + }); + + it('should not surface an unhandled rejection when an async byte handler does not catch a local disconnect', async () => { + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + const manager = new IncomingDataStreamManager(); + manager.setConnected(true); + + manager.registerByteStreamHandler('my-topic', async (reader) => { + await reader.readAll(); + }); + + 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 stream is still open + manager.clearControllers(); + await flush(); + + expect(unhandled).toEqual([]); + expect(debugSpy).toHaveBeenCalledWith( + 'data stream handler did not handle an abnormally ended stream', + expect.objectContaining({ topic: 'my-topic', streamId: 'byte-1' }), + ); + }); + + it('should log an error when a handler rejects for another reason', async () => { + const errorSpy = vi.spyOn(log, 'error').mockImplementation(() => {}); + const manager = new IncomingDataStreamManager(); + 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 flush(); + + expect(unhandled).toEqual([]); + expect(errorSpy).toHaveBeenCalledWith('unhandled error in data stream handler', { + topic: 'my-topic', + streamId: 'text-1', + error: handlerError, + }); + }); + + it('should not throw out of packet handling when a handler throws synchronously', async () => { + const errorSpy = vi.spyOn(log, 'error').mockImplementation(() => {}); + const manager = new IncomingDataStreamManager(); + manager.setConnected(true); + + const handlerError = new Error('sync application bug'); + manager.registerByteStreamHandler('my-topic', () => { + throw handlerError; + }); + + expect(() => + manager.handleDataStreamPacket(headerPacket('byte-1', 'byteHeader'), Encryption_Type.NONE), + ).not.toThrow(); + await flush(); + + expect(unhandled).toEqual([]); + expect(errorSpy).toHaveBeenCalledWith( + 'unhandled error in data stream handler', + expect.objectContaining({ error: handlerError }), + ); + + // The stream is still registered, so its later packets are handled normally. + expect(() => + manager.handleDataStreamPacket( + chunkPacket('byte-1', 0, new Uint8Array([0x01])), + Encryption_Type.NONE, + ), + ).not.toThrow(); + }); + }); }); diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.ts index 0cdb397b02..b2903cf45f 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.ts @@ -131,6 +131,35 @@ export default class IncomingDataStreamManager { } } + /** + * Calls a registered stream handler and catches anything it throws or rejects with. + * + * Handlers are usually `async` functions that iterate the reader, and an incoming stream can + * error at any point (e.g. the sender or the local room disconnects mid stream). Nothing awaits + * the handler, so an error it doesn't catch itself would otherwise surface as an unhandled + * promise rejection that application code has no way to observe. + */ + private invokeStreamHandler( + handler: (reader: R, participantInfo: { identity: string }) => void | Promise, + reader: R, + participantInfo: { identity: string }, + ) { + const onError = (error: unknown) => { + const context = { topic: reader.info.topic, streamId: reader.info.id, error }; + if (error instanceof DataStreamError && error.reason === DataStreamErrorReason.AbnormalEnd) { + // The stream was cut short by a disconnect, which is expected churn. + this.log.debug('data stream handler did not handle an abnormally ended stream', context); + } else { + this.log.error('unhandled error in data stream handler', context); + } + }; + try { + Promise.resolve(handler(reader, participantInfo)).catch(onError); + } catch (error) { + onError(error); + } + } + handleDataStreamPacket(packet: DataPacket, encryptionType: Encryption_Type) { if (!this.isConnected) { this.bufferedPackets.push({ packet, encryptionType }); @@ -212,7 +241,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 +287,8 @@ export default class IncomingDataStreamManager { } }, }); - streamHandlerCallback( + this.invokeStreamHandler( + streamHandlerCallback, new ByteStreamReader( info, compressed @@ -330,7 +361,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 +402,8 @@ export default class IncomingDataStreamManager { } }, }); - streamHandlerCallback( + this.invokeStreamHandler( + streamHandlerCallback, new TextStreamReader( info, compressed diff --git a/src/room/data-stream/incoming/StreamReader.ts b/src/room/data-stream/incoming/StreamReader.ts index 28f63cee0e..88eff5c30c 100644 --- a/src/room/data-stream/incoming/StreamReader.ts +++ b/src/room/data-stream/incoming/StreamReader.ts @@ -278,9 +278,9 @@ export class TextStreamReader extends BaseStreamReader { export type ByteStreamHandler = ( reader: ByteStreamReader, participantInfo: { identity: string }, -) => void; +) => void | Promise; export type TextStreamHandler = ( reader: TextStreamReader, participantInfo: { identity: string }, -) => void; +) => void | Promise; From 90730159e6ee9fd9dc5fb65aafc35bf01c9d56d4 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 28 Oct 2026 12:37:40 +1300 Subject: [PATCH 2/3] fix: only downgrade data stream handler errors after the room disconnects Rather than catching every handler error, re-throw anything that happens while the room is connected (or reconnecting) so it still surfaces as an unhandled rejection, e.g. a lost chunk or a remote participant leaving mid stream. Only errors that land after the room has disconnected are logged as a warning, since the session is over and the failure is out of the application's hands. A dedicated flag, set by clearControllers() and cleared by setConnected(true), tracks the disconnect so a reconnect isn't mistaken for one. Co-Authored-By: Claude Opus 5.5 --- .changeset/catch-stream-handler-errors.md | 2 +- .../IncomingDataStreamManager.test.ts | 174 ++++++++++++------ .../incoming/IncomingDataStreamManager.ts | 40 ++-- 3 files changed, 138 insertions(+), 78 deletions(-) diff --git a/.changeset/catch-stream-handler-errors.md b/.changeset/catch-stream-handler-errors.md index 4a49940de2..b57781f283 100644 --- a/.changeset/catch-stream-handler-errors.md +++ b/.changeset/catch-stream-handler-errors.md @@ -2,4 +2,4 @@ 'livekit-client': patch --- -Errors thrown or rejected by a registered text or byte stream handler no longer surface as unhandled promise rejections. Previously an `async` handler that did not wrap its reads in `try`/`catch` produced an uncatchable unhandled rejection whenever a stream ended abnormally (the sender or the local room disconnecting mid stream). Abnormal ends are now logged at debug level and any other handler error is logged as an error. `TextStreamHandler` and `ByteStreamHandler` are now typed as returning `void | Promise`. +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`. diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts index e175777a2c..7652648b2f 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts @@ -14,6 +14,7 @@ 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'; @@ -2011,17 +2012,25 @@ describe('IncomingDataStreamManager', () => { 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(); }); @@ -2031,110 +2040,157 @@ describe('IncomingDataStreamManager', () => { await new Promise((resolve) => setTimeout(resolve, 0)); }; - it('should not surface an unhandled rejection when an async text handler does not catch a remote disconnect', async () => { - const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); - const manager = new IncomingDataStreamManager(); - manager.setConnected(true); - - const received: string[] = []; + /** 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) => { - // No try/catch, the way most handlers are written. - for await (const chunk of reader) { - received.push(chunk); - } + 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 TextEncoder().encode('hello')), + chunkPacket('text-1', 0, new Uint8Array([0x68, 0x69])), Encryption_Type.NONE, ); - await flush(); - manager.validateParticipantHasNoActiveDataStreams('alice'); + 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 flush(); - expect(received).toEqual(['hello']); expect(unhandled).toEqual([]); - expect(debugSpy).toHaveBeenCalledWith( - 'data stream handler did not handle an abnormally ended stream', - expect.objectContaining({ topic: 'my-topic', streamId: 'text-1' }), - ); + 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 not surface an unhandled rejection when an async byte handler does not catch a local disconnect', async () => { - const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + 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); - manager.registerByteStreamHandler('my-topic', async (reader) => { - await reader.readAll(); + const handlerError = new Error('application bug'); + let release!: () => void; + const released = new Promise((resolve) => { + release = resolve; + }); + manager.registerTextStreamHandler('my-topic', async () => { + await released; + throw handlerError; }); - manager.handleDataStreamPacket(headerPacket('byte-1', 'byteHeader'), Encryption_Type.NONE); + manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE); + manager.clearControllers(); + release(); + await flush(); + + 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('byte-1', 0, new Uint8Array([0x01])), + chunkPacket('text-1', 0, new Uint8Array([0x68, 0x69])), Encryption_Type.NONE, ); - // Room#handleDisconnect calls this while the stream is still open - manager.clearControllers(); + manager.validateParticipantHasNoActiveDataStreams('alice'); await flush(); - expect(unhandled).toEqual([]); - expect(debugSpy).toHaveBeenCalledWith( - 'data stream handler did not handle an abnormally ended stream', - expect.objectContaining({ topic: 'my-topic', streamId: 'byte-1' }), + 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 flush(); + + expect(unhandled).toEqual([ + expect.objectContaining({ reason: DataStreamErrorReason.Incomplete }), + ]); }); - it('should log an error when a handler rejects for another reason', async () => { - const errorSpy = vi.spyOn(log, 'error').mockImplementation(() => {}); + 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((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 flush(); - expect(unhandled).toEqual([]); - expect(errorSpy).toHaveBeenCalledWith('unhandled error in data stream handler', { - topic: 'my-topic', - streamId: 'text-1', - error: handlerError, - }); + expect(unhandled).toEqual([handlerError]); }); - it('should not throw out of packet handling when a handler throws synchronously', async () => { - const errorSpy = vi.spyOn(log, 'error').mockImplementation(() => {}); + 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('sync application bug'); - manager.registerByteStreamHandler('my-topic', () => { + const handlerError = new Error('application bug'); + manager.registerTextStreamHandler('my-topic', async () => { throw handlerError; }); - - expect(() => - manager.handleDataStreamPacket(headerPacket('byte-1', 'byteHeader'), Encryption_Type.NONE), - ).not.toThrow(); + manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE); await flush(); - expect(unhandled).toEqual([]); - expect(errorSpy).toHaveBeenCalledWith( - 'unhandled error in data stream handler', - expect.objectContaining({ error: handlerError }), - ); - - // The stream is still registered, so its later packets are handled normally. - expect(() => - manager.handleDataStreamPacket( - chunkPacket('byte-1', 0, new Uint8Array([0x01])), - Encryption_Type.NONE, - ), - ).not.toThrow(); + expect(unhandled).toEqual([handlerError]); }); }); }); diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.ts index b2903cf45f..41c85af9e5 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.ts +++ b/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', @@ -132,32 +139,29 @@ export default class IncomingDataStreamManager { } /** - * Calls a registered stream handler and catches anything it throws or rejects with. + * Calls a registered stream handler. * - * Handlers are usually `async` functions that iterate the reader, and an incoming stream can - * error at any point (e.g. the sender or the local room disconnects mid stream). Nothing awaits - * the handler, so an error it doesn't catch itself would otherwise surface as an unhandled - * promise rejection that application code has no way to observe. + * 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( handler: (reader: R, participantInfo: { identity: string }) => void | Promise, reader: R, participantInfo: { identity: string }, ) { - const onError = (error: unknown) => { - const context = { topic: reader.info.topic, streamId: reader.info.id, error }; - if (error instanceof DataStreamError && error.reason === DataStreamErrorReason.AbnormalEnd) { - // The stream was cut short by a disconnect, which is expected churn. - this.log.debug('data stream handler did not handle an abnormally ended stream', context); - } else { - this.log.error('unhandled error in data stream handler', context); + Promise.resolve(handler(reader, participantInfo)).catch((error: unknown) => { + if (!this.hasDisconnected) { + throw error; } - }; - try { - Promise.resolve(handler(reader, participantInfo)).catch(onError); - } catch (error) { - onError(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) { From d4f2950764a8a1524c6c6c8092163904739fa744 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 28 Oct 2026 12:46:35 +1300 Subject: [PATCH 3/3] test: rename flush helper to flushAnyUnhandledRejections Co-Authored-By: Claude Opus 5.5 --- .../incoming/IncomingDataStreamManager.test.ts | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts index 7652648b2f..129bba51c8 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts @@ -2035,7 +2035,7 @@ describe('IncomingDataStreamManager', () => { }); /** Drains microtasks and timers so any unhandledRejection has a chance to fire. */ - const flush = async () => { + const flushAnyUnhandledRejections = async () => { await new Promise((resolve) => setTimeout(resolve, 0)); await new Promise((resolve) => setTimeout(resolve, 0)); }; @@ -2070,7 +2070,7 @@ describe('IncomingDataStreamManager', () => { // Room#handleDisconnect calls this while the streams are still open manager.clearControllers(); manager.setConnected(false); - await flush(); + await flushAnyUnhandledRejections(); expect(unhandled).toEqual([]); for (const streamId of ['text-1', 'byte-1']) { @@ -2103,7 +2103,7 @@ describe('IncomingDataStreamManager', () => { manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE); manager.clearControllers(); release(); - await flush(); + await flushAnyUnhandledRejections(); expect(unhandled).toEqual([]); expect(warnSpy).toHaveBeenCalledWith( @@ -2123,7 +2123,7 @@ describe('IncomingDataStreamManager', () => { Encryption_Type.NONE, ); manager.validateParticipantHasNoActiveDataStreams('alice'); - await flush(); + await flushAnyUnhandledRejections(); expect(unhandled).toEqual([ expect.objectContaining({ @@ -2146,7 +2146,7 @@ describe('IncomingDataStreamManager', () => { chunkPacket('text-1', 2, new Uint8Array([0x69])), Encryption_Type.NONE, ); - await flush(); + await flushAnyUnhandledRejections(); expect(unhandled).toEqual([ expect.objectContaining({ reason: DataStreamErrorReason.Incomplete }), @@ -2171,7 +2171,7 @@ describe('IncomingDataStreamManager', () => { // A reconnect drops out of the connected state without disconnecting the room. manager.setConnected(false); release(); - await flush(); + await flushAnyUnhandledRejections(); expect(unhandled).toEqual([handlerError]); }); @@ -2188,7 +2188,7 @@ describe('IncomingDataStreamManager', () => { throw handlerError; }); manager.handleDataStreamPacket(headerPacket('text-1', 'textHeader'), Encryption_Type.NONE); - await flush(); + await flushAnyUnhandledRejections(); expect(unhandled).toEqual([handlerError]); });