From 480c7097b52b0e80cf785e7a2b6cd18b3e367c75 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 21 Oct 2026 10:37:13 +1300 Subject: [PATCH 1/4] fix: ensure text/byte stream controller(s) are dropped when the stream is cancelled (ie, out of order packet delivery) --- .../incoming/IncomingDataStreamManager.ts | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.ts index b05ca78132..0cdb397b02 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.ts @@ -246,6 +246,16 @@ export default class IncomingDataStreamManager { sendingParticipantIdentity: participantIdentity, }); }, + cancel: () => { + // The consumer side of the pipe chain errored (e.g. `ensureOrderedChunks` saw a gap) or + // was cancelled, which cancels this source stream. Unregister it so later chunks and the + // trailer for this stream are dropped rather than hitting a closed controller. + if ( + this.byteStreamControllers.get(streamHeader.streamId)?.controller === streamController + ) { + this.byteStreamControllers.delete(streamHeader.streamId); + } + }, }); streamHandlerCallback( new ByteStreamReader( @@ -349,6 +359,16 @@ export default class IncomingDataStreamManager { sendingParticipantIdentity: participantIdentity, }); }, + cancel: () => { + // The consumer side of the pipe chain errored (e.g. `ensureOrderedChunks` saw a gap) or + // was cancelled, which cancels this source stream. Unregister it so later chunks and the + // trailer for this stream are dropped rather than hitting a closed controller. + if ( + this.textStreamControllers.get(streamHeader.streamId)?.controller === streamController + ) { + this.textStreamControllers.delete(streamHeader.streamId); + } + }, }); streamHandlerCallback( new TextStreamReader( From 08dfcbed1cd93bd7e9114d0b27192973e1c55029 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 21 Oct 2026 10:38:24 +1300 Subject: [PATCH 2/4] fix: add tests to assert stream controllers are dropped from maps --- .../IncomingDataStreamManager.test.ts | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts index c973973ad5..9cfd7f0001 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts @@ -1262,6 +1262,18 @@ describe('IncomingDataStreamManager', () => { const reader = await readerPromise; await expect(reader.readAll()).rejects.toThrow('Missing chunk(s)'); + + // The errored stream must be unregistered, so packets still in flight for it are dropped + // instead of throwing on a closed controller (#2166). + expect(() => + manager.handleDataStreamPacket( + chunkPacket(streamId, 3, new Uint8Array([1])), + Encryption_Type.NONE, + ), + ).not.toThrow(); + expect(() => + manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE), + ).not.toThrow(); }); it('should error on a gap in chunk indices on an uncompressed text stream', async () => { @@ -1296,6 +1308,18 @@ describe('IncomingDataStreamManager', () => { const reader = await readerPromise; await expect(reader.readAll()).rejects.toThrow('Missing chunk(s)'); + + // The errored stream must be unregistered, so packets still in flight for it are dropped + // instead of throwing on a closed controller (#2166). + expect(() => + manager.handleDataStreamPacket( + chunkPacket(streamId, 3, new Uint8Array([1])), + Encryption_Type.NONE, + ), + ).not.toThrow(); + expect(() => + manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE), + ).not.toThrow(); }); it('should error on a gap in chunk indices on an uncompressed byte stream', async () => { @@ -1329,6 +1353,18 @@ describe('IncomingDataStreamManager', () => { const reader = await readerPromise; await expect(reader.readAll()).rejects.toThrow('Missing chunk(s)'); + + // The errored stream must be unregistered, so packets still in flight for it are dropped + // instead of throwing on a closed controller (#2166). + expect(() => + manager.handleDataStreamPacket( + chunkPacket(streamId, 3, new Uint8Array([1])), + Encryption_Type.NONE, + ), + ).not.toThrow(); + expect(() => + manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE), + ).not.toThrow(); }); it('should drop a duplicate chunk index on an uncompressed text stream and still decode', async () => { From 323677b0e82aac07ccb072bf5fd769c151b0ef3d Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 21 Oct 2026 10:47:04 +1300 Subject: [PATCH 3/4] fix: add changeset --- .changeset/olive-cities-enjoy.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/olive-cities-enjoy.md diff --git a/.changeset/olive-cities-enjoy.md b/.changeset/olive-cities-enjoy.md new file mode 100644 index 0000000000..22df7d9d65 --- /dev/null +++ b/.changeset/olive-cities-enjoy.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Fixes an issue where if a data stream runs into a terminal error, any packets delivered afterwards will be treated as if the stream doesn't exist (and not throw a sync error) From 27aa5b846f0b4844daacc6bd4ec69a9a12f4cdca Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 21 Oct 2026 10:52:02 +1300 Subject: [PATCH 4/4] fix: add regression test for exact specific issue --- .../IncomingDataStreamManager.test.ts | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts index 9cfd7f0001..b2d602b281 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts @@ -1831,5 +1831,58 @@ describe('IncomingDataStreamManager', () => { const reader = await readerPromise; await expect(reader.readAll()).rejects.toThrow('Encryption type mismatch'); }); + + it('should treat packets for a stream that errored on a chunk index gap as an unknown stream (#2166)', async () => { + const manager = new IncomingDataStreamManager(); + manager.setConnected(true); + + const readers: Array = []; + manager.registerTextStreamHandler('my-topic', (reader) => readers.push(reader)); + + const streamId = crypto.randomUUID(); + manager.handleDataStreamPacket( + headerPacket(streamId, 'textHeader', { compression: DataStream_CompressionType.NONE }), + Encryption_Type.NONE, + ); + manager.handleDataStreamPacket( + chunkPacket(streamId, 0, new TextEncoder().encode('first')), + Encryption_Type.NONE, + ); + // Skip chunk index 1, which errors the stream. + manager.handleDataStreamPacket( + chunkPacket(streamId, 2, new TextEncoder().encode('third')), + Encryption_Type.NONE, + ); + expect(readers).toHaveLength(1); + await expect(readers[0].readAll()).rejects.toThrow('Missing chunk(s)'); + + // Packets still in flight for the errored stream are dropped, just like packets for a stream + // that was never opened, rather than throwing on a closed controller. + expect(() => + manager.handleDataStreamPacket( + chunkPacket(streamId, 3, new TextEncoder().encode('fourth')), + Encryption_Type.NONE, + ), + ).not.toThrow(); + expect(() => + manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE), + ).not.toThrow(); + expect(readers).toHaveLength(1); + + // No state is left behind for the stream id: a header reusing it opens a brand new stream + // instead of failing with `AlreadyOpened`, and that stream reads cleanly from chunk 0. + manager.handleDataStreamPacket( + headerPacket(streamId, 'textHeader', { compression: DataStream_CompressionType.NONE }), + Encryption_Type.NONE, + ); + manager.handleDataStreamPacket( + chunkPacket(streamId, 0, new TextEncoder().encode('fresh stream')), + Encryption_Type.NONE, + ); + manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE); + + expect(readers).toHaveLength(2); + expect(await readers[1].readAll()).toEqual('fresh stream'); + }); }); });