From d9cfa8b1d25bc499a7bb5266796088de45e6a6b5 Mon Sep 17 00:00:00 2001 From: JoelTowell Date: Tue, 4 Aug 2026 21:10:22 +1000 Subject: [PATCH 1/6] Add TextDecoder fallback without fatal option when TextDecoder doesn't support it --- src/room/data-stream/incoming/StreamReader.ts | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/room/data-stream/incoming/StreamReader.ts b/src/room/data-stream/incoming/StreamReader.ts index bd700285f0..81e6a191e7 100644 --- a/src/room/data-stream/incoming/StreamReader.ts +++ b/src/room/data-stream/incoming/StreamReader.ts @@ -204,7 +204,17 @@ export class TextStreamReader extends BaseStreamReader { // Suppress unhandled rejection on reader.closed — errors are // already propagated through reader.read() to the consumer. reader.closed.catch(() => {}); - const decoder = new TextDecoder('utf-8', { fatal: true }); + + const getTextDecoder = (): TextDecoder => { + // Fallback for runtimes with partial TextDecoder support. + try { + return new TextDecoder('utf-8', { fatal: true }); + } catch { + return new TextDecoder('utf-8'); + } + }; + + const decoder = getTextDecoder(); const signal = this.signal; const cleanup = () => { From 784f1b60b3fb101d42bee0d5043a40d675a08535 Mon Sep 17 00:00:00 2001 From: JoelTowell Date: Tue, 4 Aug 2026 21:11:12 +1000 Subject: [PATCH 2/6] add regression test --- .../data-stream/incoming/StreamReader.test.ts | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 src/room/data-stream/incoming/StreamReader.test.ts diff --git a/src/room/data-stream/incoming/StreamReader.test.ts b/src/room/data-stream/incoming/StreamReader.test.ts new file mode 100644 index 0000000000..9480954e50 --- /dev/null +++ b/src/room/data-stream/incoming/StreamReader.test.ts @@ -0,0 +1,60 @@ +import { DataStream_Chunk, Encryption_Type } from '@livekit/protocol'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { TextStreamInfo } from '../../types'; +import { TextStreamReader } from './StreamReader'; + +describe('TextStreamReader', () => { + const OriginalTextDecoder = globalThis.TextDecoder; + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('falls back when TextDecoder does not support fatal mode', async () => { + const constructorArguments: Array<{ encoding?: string; options?: TextDecoderOptions }> = []; + + vi.stubGlobal( + 'TextDecoder', + class extends OriginalTextDecoder { + constructor(encoding?: string, options?: TextDecoderOptions) { + constructorArguments.push({ encoding, options }); + + if (options?.fatal) throw new TypeError('fatal flag not supported'); + + super(encoding, options); + } + }, + ); + + const info: TextStreamInfo = { + id: 'stream-1', + mimeType: 'text/plain', + topic: 'test', + timestamp: 0, + encryptionType: Encryption_Type.NONE, + }; + + const text = 'This is a stream.'; + + const textStream = new ReadableStream({ + start(controller) { + controller.enqueue( + new DataStream_Chunk({ + streamId: info.id, + chunkIndex: 0n, + content: new TextEncoder().encode(text), + }), + ); + controller.close(); + }, + }); + + const reader = new TextStreamReader(info, textStream); + + expect(await reader.readAll()).toBe(text); + expect(constructorArguments).toEqual([ + { encoding: 'utf-8', options: { fatal: true } }, + { encoding: 'utf-8', options: undefined }, + ]); + }); +}); From e0550c16bed3a4ec7fa2d8d44ef615c9b1da140a Mon Sep 17 00:00:00 2001 From: JoelTowell Date: Tue, 4 Aug 2026 21:54:00 +1000 Subject: [PATCH 3/6] add changeset --- .changeset/fancy-terms-walk.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/fancy-terms-walk.md diff --git a/.changeset/fancy-terms-walk.md b/.changeset/fancy-terms-walk.md new file mode 100644 index 0000000000..9fdd9114d1 --- /dev/null +++ b/.changeset/fancy-terms-walk.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Add TextDecoder fallback inside TextStreamReader when fatal option is not supported From 81bad1bc951e1934d414af7248875003930e0b9c Mon Sep 17 00:00:00 2001 From: JoelTowell Date: Tue, 4 Aug 2026 22:35:05 +1000 Subject: [PATCH 4/6] Factor into utils and use in IncomingDataStreamManager as well --- .../data-stream/incoming/IncomingDataStreamManager.ts | 9 +++++++-- src/room/data-stream/incoming/StreamReader.ts | 11 +---------- src/room/utils.ts | 9 +++++++++ 3 files changed, 17 insertions(+), 12 deletions(-) diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.ts index 5e10b0aae6..155c6ba893 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.ts @@ -10,7 +10,12 @@ import log from '../../../logger'; import { type NonSharedUint8Array } from '../../../type-polyfills/non-shared-typed-arrays'; import { DataStreamError, DataStreamErrorReason } from '../../errors'; import { type ByteStreamInfo, type StreamController, type TextStreamInfo } from '../../types'; -import { bigIntToNumber, isCompressionStreamSupported, numberToBigInt } from '../../utils'; +import { + bigIntToNumber, + getTextDecoder, + isCompressionStreamSupported, + numberToBigInt, +} from '../../utils'; import { deflateRawDecompress, inflateRawTransform } from '../compression'; import { DEFAULT_MAX_PAYLOAD_BYTE_LENGTH } from '../constants'; import { @@ -535,7 +540,7 @@ function bytesToChunks(streamId: string): TransformStream { - const decoder = new TextDecoder('utf-8', { fatal: true }); + const decoder = getTextDecoder(); const encoder = new TextEncoder(); let outIndex = 0; diff --git a/src/room/data-stream/incoming/StreamReader.ts b/src/room/data-stream/incoming/StreamReader.ts index 81e6a191e7..b9fbd28671 100644 --- a/src/room/data-stream/incoming/StreamReader.ts +++ b/src/room/data-stream/incoming/StreamReader.ts @@ -1,7 +1,7 @@ import type { DataStream_Chunk } from '@livekit/protocol'; import { DataStreamError, DataStreamErrorReason } from '../../errors'; import type { BaseStreamInfo, ByteStreamInfo, TextStreamInfo } from '../../types'; -import { bigIntToNumber } from '../../utils'; +import { bigIntToNumber, getTextDecoder } from '../../utils'; export type BaseStreamReaderReadAllOpts = { /** An AbortSignal can be used to terminate reads early. */ @@ -205,15 +205,6 @@ export class TextStreamReader extends BaseStreamReader { // already propagated through reader.read() to the consumer. reader.closed.catch(() => {}); - const getTextDecoder = (): TextDecoder => { - // Fallback for runtimes with partial TextDecoder support. - try { - return new TextDecoder('utf-8', { fatal: true }); - } catch { - return new TextDecoder('utf-8'); - } - }; - const decoder = getTextDecoder(); const signal = this.signal; diff --git a/src/room/utils.ts b/src/room/utils.ts index 3ad2e3798f..b57db789e5 100644 --- a/src/room/utils.ts +++ b/src/room/utils.ts @@ -880,3 +880,12 @@ export function extractTrackSid( return mediaTrack.id; } } + +/** Fallback for runtimes with partial TextDecoder support. */ +export function getTextDecoder(): TextDecoder { + try { + return new TextDecoder('utf-8', { fatal: true }); + } catch { + return new TextDecoder('utf-8'); + } +} From 01e9813c460249ff403a4fb7e6f75e43268e98eb Mon Sep 17 00:00:00 2001 From: JoelTowell Date: Tue, 4 Aug 2026 22:41:37 +1000 Subject: [PATCH 5/6] add test for IncomingDataStreamManager --- .../IncomingDataStreamManager.test.ts | 46 ++++++++++++++++++- .../data-stream/incoming/StreamReader.test.ts | 2 +- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts index 672bdaaab5..9c67804087 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts @@ -8,7 +8,7 @@ import { DataStream_Trailer, Encryption_Type, } from '@livekit/protocol'; -import { describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { deflateRawCompress } from '../compression'; import { STREAM_CHUNK_SIZE_BYTES } from '../constants'; import IncomingDataStreamManager from './IncomingDataStreamManager'; @@ -1389,6 +1389,50 @@ describe('IncomingDataStreamManager', () => { Promise.race([readerPromise, Promise.resolve('still pending')]), ).resolves.toStrictEqual('still pending'); }); + + describe('TextDecoder without fatal support', () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('falls back when decoding a compressed text stream', async () => { + const OriginalTextDecoder = globalThis.TextDecoder; + + vi.stubGlobal( + 'TextDecoder', + class extends OriginalTextDecoder { + constructor(encoding?: string, options?: TextDecoderOptions) { + if (options?.fatal) throw new TypeError('fatal flag not supported'); + super(encoding, options); + } + }, + ); + + const manager = new IncomingDataStreamManager(); + manager.setConnected(true); + + const readerPromise = new Promise((resolve) => { + manager.registerTextStreamHandler('my-topic', (reader) => resolve(reader)); + }); + + const streamId = crypto.randomUUID(); + const text = 'hello world'; + const compressed = await deflateRawCompress(new TextEncoder().encode(text)); + + manager.handleDataStreamPacket( + headerPacket(streamId, 'textHeader', { + totalLength: BigInt(text.length), + compression: DataStream_CompressionType.DEFLATE_RAW, + }), + Encryption_Type.NONE, + ); + manager.handleDataStreamPacket(chunkPacket(streamId, 0, compressed), Encryption_Type.NONE); + manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE); + + const reader = await readerPromise; + expect(await reader.readAll()).toBe(text); + }); + }); }); describe('Receive-side protocol edge cases', () => { diff --git a/src/room/data-stream/incoming/StreamReader.test.ts b/src/room/data-stream/incoming/StreamReader.test.ts index 9480954e50..6952600f7f 100644 --- a/src/room/data-stream/incoming/StreamReader.test.ts +++ b/src/room/data-stream/incoming/StreamReader.test.ts @@ -34,7 +34,7 @@ describe('TextStreamReader', () => { encryptionType: Encryption_Type.NONE, }; - const text = 'This is a stream.'; + const text = 'hello world'; const textStream = new ReadableStream({ start(controller) { From a48e60cf2e98ce7e237cec640776c4d38fc00220 Mon Sep 17 00:00:00 2001 From: JoelTowell Date: Wed, 5 Aug 2026 01:08:28 +1000 Subject: [PATCH 6/6] revert changes, stop passing fatal option to TextDecoder --- .changeset/fancy-terms-walk.md | 2 +- .../IncomingDataStreamManager.test.ts | 46 +------------- .../incoming/IncomingDataStreamManager.ts | 9 +-- .../data-stream/incoming/StreamReader.test.ts | 60 ------------------- src/room/data-stream/incoming/StreamReader.ts | 5 +- src/room/utils.ts | 9 --- 6 files changed, 6 insertions(+), 125 deletions(-) delete mode 100644 src/room/data-stream/incoming/StreamReader.test.ts diff --git a/.changeset/fancy-terms-walk.md b/.changeset/fancy-terms-walk.md index 9fdd9114d1..83b354af7c 100644 --- a/.changeset/fancy-terms-walk.md +++ b/.changeset/fancy-terms-walk.md @@ -2,4 +2,4 @@ 'livekit-client': patch --- -Add TextDecoder fallback inside TextStreamReader when fatal option is not supported +Decode text data streams without TextDecoder fatal mode for broader runtime support diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts index 9c67804087..672bdaaab5 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts @@ -8,7 +8,7 @@ import { DataStream_Trailer, Encryption_Type, } from '@livekit/protocol'; -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { describe, expect, it } from 'vitest'; import { deflateRawCompress } from '../compression'; import { STREAM_CHUNK_SIZE_BYTES } from '../constants'; import IncomingDataStreamManager from './IncomingDataStreamManager'; @@ -1389,50 +1389,6 @@ describe('IncomingDataStreamManager', () => { Promise.race([readerPromise, Promise.resolve('still pending')]), ).resolves.toStrictEqual('still pending'); }); - - describe('TextDecoder without fatal support', () => { - afterEach(() => { - vi.unstubAllGlobals(); - }); - - it('falls back when decoding a compressed text stream', async () => { - const OriginalTextDecoder = globalThis.TextDecoder; - - vi.stubGlobal( - 'TextDecoder', - class extends OriginalTextDecoder { - constructor(encoding?: string, options?: TextDecoderOptions) { - if (options?.fatal) throw new TypeError('fatal flag not supported'); - super(encoding, options); - } - }, - ); - - const manager = new IncomingDataStreamManager(); - manager.setConnected(true); - - const readerPromise = new Promise((resolve) => { - manager.registerTextStreamHandler('my-topic', (reader) => resolve(reader)); - }); - - const streamId = crypto.randomUUID(); - const text = 'hello world'; - const compressed = await deflateRawCompress(new TextEncoder().encode(text)); - - manager.handleDataStreamPacket( - headerPacket(streamId, 'textHeader', { - totalLength: BigInt(text.length), - compression: DataStream_CompressionType.DEFLATE_RAW, - }), - Encryption_Type.NONE, - ); - manager.handleDataStreamPacket(chunkPacket(streamId, 0, compressed), Encryption_Type.NONE); - manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE); - - const reader = await readerPromise; - expect(await reader.readAll()).toBe(text); - }); - }); }); describe('Receive-side protocol edge cases', () => { diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.ts index 155c6ba893..93f11bda68 100644 --- a/src/room/data-stream/incoming/IncomingDataStreamManager.ts +++ b/src/room/data-stream/incoming/IncomingDataStreamManager.ts @@ -10,12 +10,7 @@ import log from '../../../logger'; import { type NonSharedUint8Array } from '../../../type-polyfills/non-shared-typed-arrays'; import { DataStreamError, DataStreamErrorReason } from '../../errors'; import { type ByteStreamInfo, type StreamController, type TextStreamInfo } from '../../types'; -import { - bigIntToNumber, - getTextDecoder, - isCompressionStreamSupported, - numberToBigInt, -} from '../../utils'; +import { bigIntToNumber, isCompressionStreamSupported, numberToBigInt } from '../../utils'; import { deflateRawDecompress, inflateRawTransform } from '../compression'; import { DEFAULT_MAX_PAYLOAD_BYTE_LENGTH } from '../constants'; import { @@ -540,7 +535,7 @@ function bytesToChunks(streamId: string): TransformStream { - const decoder = getTextDecoder(); + const decoder = new TextDecoder('utf-8'); const encoder = new TextEncoder(); let outIndex = 0; diff --git a/src/room/data-stream/incoming/StreamReader.test.ts b/src/room/data-stream/incoming/StreamReader.test.ts deleted file mode 100644 index 6952600f7f..0000000000 --- a/src/room/data-stream/incoming/StreamReader.test.ts +++ /dev/null @@ -1,60 +0,0 @@ -import { DataStream_Chunk, Encryption_Type } from '@livekit/protocol'; -import { afterEach, describe, expect, it, vi } from 'vitest'; -import type { TextStreamInfo } from '../../types'; -import { TextStreamReader } from './StreamReader'; - -describe('TextStreamReader', () => { - const OriginalTextDecoder = globalThis.TextDecoder; - - afterEach(() => { - vi.unstubAllGlobals(); - }); - - it('falls back when TextDecoder does not support fatal mode', async () => { - const constructorArguments: Array<{ encoding?: string; options?: TextDecoderOptions }> = []; - - vi.stubGlobal( - 'TextDecoder', - class extends OriginalTextDecoder { - constructor(encoding?: string, options?: TextDecoderOptions) { - constructorArguments.push({ encoding, options }); - - if (options?.fatal) throw new TypeError('fatal flag not supported'); - - super(encoding, options); - } - }, - ); - - const info: TextStreamInfo = { - id: 'stream-1', - mimeType: 'text/plain', - topic: 'test', - timestamp: 0, - encryptionType: Encryption_Type.NONE, - }; - - const text = 'hello world'; - - const textStream = new ReadableStream({ - start(controller) { - controller.enqueue( - new DataStream_Chunk({ - streamId: info.id, - chunkIndex: 0n, - content: new TextEncoder().encode(text), - }), - ); - controller.close(); - }, - }); - - const reader = new TextStreamReader(info, textStream); - - expect(await reader.readAll()).toBe(text); - expect(constructorArguments).toEqual([ - { encoding: 'utf-8', options: { fatal: true } }, - { encoding: 'utf-8', options: undefined }, - ]); - }); -}); diff --git a/src/room/data-stream/incoming/StreamReader.ts b/src/room/data-stream/incoming/StreamReader.ts index b9fbd28671..1bb0d206a0 100644 --- a/src/room/data-stream/incoming/StreamReader.ts +++ b/src/room/data-stream/incoming/StreamReader.ts @@ -1,7 +1,7 @@ import type { DataStream_Chunk } from '@livekit/protocol'; import { DataStreamError, DataStreamErrorReason } from '../../errors'; import type { BaseStreamInfo, ByteStreamInfo, TextStreamInfo } from '../../types'; -import { bigIntToNumber, getTextDecoder } from '../../utils'; +import { bigIntToNumber } from '../../utils'; export type BaseStreamReaderReadAllOpts = { /** An AbortSignal can be used to terminate reads early. */ @@ -204,8 +204,7 @@ export class TextStreamReader extends BaseStreamReader { // Suppress unhandled rejection on reader.closed — errors are // already propagated through reader.read() to the consumer. reader.closed.catch(() => {}); - - const decoder = getTextDecoder(); + const decoder = new TextDecoder('utf-8'); const signal = this.signal; const cleanup = () => { diff --git a/src/room/utils.ts b/src/room/utils.ts index b57db789e5..3ad2e3798f 100644 --- a/src/room/utils.ts +++ b/src/room/utils.ts @@ -880,12 +880,3 @@ export function extractTrackSid( return mediaTrack.id; } } - -/** Fallback for runtimes with partial TextDecoder support. */ -export function getTextDecoder(): TextDecoder { - try { - return new TextDecoder('utf-8', { fatal: true }); - } catch { - return new TextDecoder('utf-8'); - } -}