fix: error on uncompressed datastreams when chunks are missing

.changeset/seven-buckets-help.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+fix: error on datastreams when chunks are missing

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

@@ -1237,6 +1237,171 @@ describe('IncomingDataStreamManager', () => {
       await expect(reader.readAll()).rejects.toThrow('Missing chunk(s)');
     });
 
+    it('should error on a gap in chunk indices on an uncompressed text stream', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+
+      const readerPromise = new Promise<TextStreamReader>((resolve) => {
+        manager.registerTextStreamHandler('my-topic', (reader) => resolve(reader));
+      });
+
+      const streamId = crypto.randomUUID();
+      const text = randomText(3_000);
+      const textBytes = new TextEncoder().encode(text);
+      const split = Math.floor(textBytes.length / 3);
+
+      manager.handleDataStreamPacket(
+        headerPacket(streamId, 'textHeader', {
+          totalLength: BigInt(textBytes.length),
+          compression: DataStream_CompressionType.NONE,
+        }),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(
+        chunkPacket(streamId, 0, textBytes.slice(0, split)),
+        Encryption_Type.NONE,
+      );
+      // Skip chunk index 1 entirely — a gap means the payload cannot be reassembled in order.
+      manager.handleDataStreamPacket(
+        chunkPacket(streamId, 2, textBytes.slice(split)),
+        Encryption_Type.NONE,
+      );
+
+      const reader = await readerPromise;
+      await expect(reader.readAll()).rejects.toThrow('Missing chunk(s)');
+    });
+
+    it('should error on a gap in chunk indices on an uncompressed byte stream', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+
+      const readerPromise = new Promise<ByteStreamReader>((resolve) => {
+        manager.registerByteStreamHandler('my-topic', (reader) => resolve(reader));
+      });
+
+      const streamId = crypto.randomUUID();
+      const bytes = randomBytes(3_000);
+      const split = Math.floor(bytes.length / 3);
+
+      manager.handleDataStreamPacket(
+        headerPacket(streamId, 'byteHeader', {
+          totalLength: BigInt(bytes.length),
+          compression: DataStream_CompressionType.NONE,
+        }),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(
+        chunkPacket(streamId, 0, bytes.slice(0, split)),
+        Encryption_Type.NONE,
+      );
+      // Skip chunk index 1 entirely.
+      manager.handleDataStreamPacket(
+        chunkPacket(streamId, 2, bytes.slice(split)),
+        Encryption_Type.NONE,
+      );
+
+      const reader = await readerPromise;
+      await expect(reader.readAll()).rejects.toThrow('Missing chunk(s)');
+    });
+
+    it('should drop a duplicate chunk index on an uncompressed text stream and still decode', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+
+      const readerPromise = new Promise<TextStreamReader>((resolve) => {
+        manager.registerTextStreamHandler('my-topic', (reader) => resolve(reader));
+      });
+
+      const streamId = crypto.randomUUID();
+      const text = randomText(3_000);
+      const textBytes = new TextEncoder().encode(text);
+      const split = Math.floor(textBytes.length / 2);
+
+      manager.handleDataStreamPacket(
+        headerPacket(streamId, 'textHeader', {
+          totalLength: BigInt(textBytes.length),
+          compression: DataStream_CompressionType.NONE,
+        }),
+        Encryption_Type.NONE,
+      );
+      const chunk0 = chunkPacket(streamId, 0, textBytes.slice(0, split));
+      manager.handleDataStreamPacket(chunk0, Encryption_Type.NONE);
+      // A replayed chunk (e.g. reconnect logic) must be dropped with a warning, not appended a
+      // second time — otherwise the payload is corrupted and exceeds `totalLength`.
+      manager.handleDataStreamPacket(chunk0, Encryption_Type.NONE);
+      manager.handleDataStreamPacket(
+        chunkPacket(streamId, 1, textBytes.slice(split)),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE);
+
+      const reader = await readerPromise;
+      expect(await reader.readAll()).toStrictEqual(text);
+    });
+
+    it('should drop a duplicate chunk index on an uncompressed byte stream and still decode', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+
+      const readerPromise = new Promise<ByteStreamReader>((resolve) => {
+        manager.registerByteStreamHandler('my-topic', (reader) => resolve(reader));
+      });
+
+      const streamId = crypto.randomUUID();
+      const bytes = randomBytes(3_000);
+      const split = Math.floor(bytes.length / 2);
+
+      manager.handleDataStreamPacket(
+        headerPacket(streamId, 'byteHeader', {
+          totalLength: BigInt(bytes.length),
+          compression: DataStream_CompressionType.NONE,
+        }),
+        Encryption_Type.NONE,
+      );
+      const chunk0 = chunkPacket(streamId, 0, bytes.slice(0, split));
+      manager.handleDataStreamPacket(chunk0, Encryption_Type.NONE);
+      manager.handleDataStreamPacket(chunk0, Encryption_Type.NONE);
+      manager.handleDataStreamPacket(
+        chunkPacket(streamId, 1, bytes.slice(split)),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE);
+
+      const reader = await readerPromise;
+      expect(concatChunks(await reader.readAll())).toStrictEqual(bytes);
+    });
+
+    it('should drop a higher-version chunk resent at an already-received index', async () => {
+      const manager = new IncomingDataStreamManager();
+      manager.setConnected(true);
+
+      const readerPromise = new Promise<TextStreamReader>((resolve) => {
+        manager.registerTextStreamHandler('my-topic', (reader) => resolve(reader));
+      });
+
+      const streamId = crypto.randomUUID();
+      const text = 'hello world';
+      const textBytes = new TextEncoder().encode(text);
+
+      manager.handleDataStreamPacket(
+        headerPacket(streamId, 'textHeader', { totalLength: BigInt(textBytes.length) }),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(chunkPacket(streamId, 0, textBytes, 0), Encryption_Type.NONE);
+      // Chunk-level `version` retcon is not supported: a reader that has already yielded chunk 0 to
+      // its consumer cannot retract it, so a resend at the same index is dropped like any other
+      // duplicate rather than superseding the original. See the note on
+      // `TextStreamReader.handleChunkReceived`.
+      manager.handleDataStreamPacket(
+        chunkPacket(streamId, 0, new TextEncoder().encode('goodbye world'), 1),
+        Encryption_Type.NONE,
+      );
+      manager.handleDataStreamPacket(trailerPacket(streamId), Encryption_Type.NONE);
+
+      const reader = await readerPromise;
+      expect(await reader.readAll()).toStrictEqual(text);
+    });
+
     it('should reframe multibyte UTF-8 on chunk boundaries when decompressing a text stream', async () => {
       const manager = new IncomingDataStreamManager();
       manager.setConnected(true);

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

@@ -241,7 +241,7 @@ export default class IncomingDataStreamManager {
             info,
             compressed
               ? inflateRawByteChunkStream(stream, streamHeader.streamId, this.maxPayloadByteLength)
-              : stream,
+              : stream.pipeThrough(ensureOrderedChunks(streamHeader.streamId)),
             // `totalLength` is the pre-compression size, and the reader counts decompressed bytes,
             // so it applies to both paths (mirrors text).
             bigIntToNumber(streamHeader.totalLength),
@@ -344,7 +344,7 @@ export default class IncomingDataStreamManager {
             info,
             compressed
               ? inflateRawChunkStream(stream, streamHeader.streamId, this.maxPayloadByteLength)
-              : stream,
+              : stream.pipeThrough(ensureOrderedChunks(streamHeader.streamId)),
             // `totalLength` is the pre-compression size, and the reader sees decompressed bytes, so
             // it applies to both paths.
             bigIntToNumber(streamHeader.totalLength),
@@ -367,7 +367,7 @@ export default class IncomingDataStreamManager {
           ),
         );
         this.byteStreamControllers.delete(chunk.streamId);
-      } else if (chunk.content.length > 0) {
+      } else {
         fileBuffer.controller.enqueue(chunk);
       }
     }
@@ -381,7 +381,7 @@ export default class IncomingDataStreamManager {
           ),
         );
         this.textStreamControllers.delete(chunk.streamId);
-      } else if (chunk.content.length > 0) {
+      } else {
         textBuffer.controller.enqueue(chunk);
       }
     }
@@ -455,25 +455,21 @@ function createInlineStream(
 ): ReadableStream<DataStream_Chunk> {
   return new ReadableStream<DataStream_Chunk>({
     start: async (controller) => {
-      try {
-        const bytes = await content;
-        controller.enqueue(
-          new DataStream_Chunk({ streamId, chunkIndex: BigInt(0), content: bytes }),
-        );
-        controller.close();
-      } catch (err) {
-        controller.error(err);
-      }
+      const bytes = await content;
+      controller.enqueue(new DataStream_Chunk({ streamId, chunkIndex: BigInt(0), content: bytes }));
+      controller.close();
     },
   });
 }
 
 /**
  * Validates that chunks are received in order, dropping duplicates and throwing if gaps are found.
  *
- * A stateful decompressor silently corrupts on duplicated or out-of-order input, so duplicates are
- * dropped (with a warning - in-order delivery is expected on the reliable channel, but reconnect
- * handling may replay) and a gap is a hard error. Shared by the text and byte deflate-raw decoders.
+ * Reassembly (and, for compressed streams, a stateful decompressor) silently corrupts on duplicated
+ * or out-of-order input, so duplicates are dropped (with a warning - in-order delivery is expected
+ * on the reliable channel, but reconnect handling may replay) and a gap is a hard error. Empty
+ * chunks consume their index and are then dropped, so they never reach the reader. Applied to every
+ * chunked stream, compressed or not.
  */
 function ensureOrderedChunks(
   streamId: string,
@@ -484,17 +480,20 @@ function ensureOrderedChunks(
       const index = bigIntToNumber(value.chunkIndex);
       if (index <= lastChunkIndex) {
         log.warn(
-          `ignoring duplicate chunk ${index} for compressed data stream ${streamId} (last processed: ${lastChunkIndex})`,
+          `ignoring duplicate chunk ${index} ${value.version > 0 ? `(version ${value.version})` : ''} for data stream ${streamId} (last processed: ${lastChunkIndex})`,
         );
         return;
       }
       if (index > lastChunkIndex + 1) {
         throw new DataStreamError(
-          `Missing chunk(s) ${lastChunkIndex + 1}..${index - 1} for compressed data stream ${streamId} - cannot continue decompressing`,
+          `Missing chunk(s) ${lastChunkIndex + 1}..${index - 1} for data stream ${streamId} - cannot reassemble payload`,
           DataStreamErrorReason.Incomplete,
         );
       }
       lastChunkIndex = index;
+      if (value.content.length === 0) {
+        return;
+      }
       controller.enqueue(value);
     },
   });

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

@@ -1,7 +1,6 @@
 import type { DataStream_Chunk } from '@livekit/protocol';
 import { DataStreamError, DataStreamErrorReason } from '../../errors';
 import type { BaseStreamInfo, ByteStreamInfo, TextStreamInfo } from '../../types';
-import { bigIntToNumber } from '../../utils';
 
 export type BaseStreamReaderReadAllOpts = {
   /** An AbortSignal can be used to terminate reads early. */
@@ -47,14 +46,11 @@ abstract class BaseStreamReader<T extends BaseStreamInfo> {
     this.bytesReceived = 0;
   }
 
-  protected abstract handleChunkReceived(chunk: DataStream_Chunk): void;
-
-  onProgress?: (progress: number | undefined) => void;
-
-  abstract readAll(opts?: BaseStreamReaderReadAllOpts): Promise<string | Array<Uint8Array>>;
-}
-
-export class ByteStreamReader extends BaseStreamReader<ByteStreamInfo> {
+  /**
+   * Counts a chunk's bytes against `totalByteSize` and reports progress. Chunk ordering and
+   * de-duplication happen upstream in the manager's `ensureOrderedChunks`, so every chunk reaching
+   * here is new and in order.
+   */
   protected handleChunkReceived(chunk: DataStream_Chunk) {
     this.bytesReceived += chunk.content.byteLength;
     this.validateBytesReceived();
@@ -65,8 +61,15 @@ export class ByteStreamReader extends BaseStreamReader<ByteStreamInfo> {
     this.onProgress?.(currentProgress);
   }
 
+  /**
+   * @param progress - progress of the stream between 0 and 1. Undefined for streams of unknown size
+   */
   onProgress?: (progress: number | undefined) => void;
 
+  abstract readAll(opts?: BaseStreamReaderReadAllOpts): Promise<string | Array<Uint8Array>>;
+}
+
+export class ByteStreamReader extends BaseStreamReader<ByteStreamInfo> {
   signal?: AbortSignal;
 
   [Symbol.asyncIterator]() {
@@ -151,53 +154,20 @@ export class ByteStreamReader extends BaseStreamReader<ByteStreamInfo> {
 }
 
 /**
- * A class to read chunks from a ReadableStream and provide them in a structured format.
+ * A class to read chunks from a ReadableStream and decode them as UTF-8 text.
+ *
+ * NOTE: chunk-level `version` (resending a chunk at an already-received `chunkIndex` to supersede
+ * it) is not supported. The reader used to rebuild the whole string from a per-index chunk map and
+ * yield it as `TextStreamChunk.collected`, which made superseding work; 5d4a6346 (#1410, text auto
+ * chunking) changed the iterator to yield each chunk's text as it arrives, and a streaming reader
+ * cannot retract text it has already handed to the consumer. No sender emits a versioned chunk.
  */
 export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
-  private receivedChunks: Map<number /* chunk index */, DataStream_Chunk>;
-
   signal?: AbortSignal;
 
-  /**
-   * A TextStreamReader instance can be used as an AsyncIterator that returns the entire string
-   * that has been received up to the current point in time.
-   */
-  constructor(
-    info: TextStreamInfo,
-    stream: ReadableStream<DataStream_Chunk>,
-    totalChunkCount?: number,
-  ) {
-    super(info, stream, totalChunkCount);
-    this.receivedChunks = new Map();
-  }
-
-  protected handleChunkReceived(chunk: DataStream_Chunk) {
-    const index = bigIntToNumber(chunk.chunkIndex);
-    const previousChunkAtIndex = this.receivedChunks.get(index);
-    if (previousChunkAtIndex && previousChunkAtIndex.version > chunk.version) {
-      // we have a newer version already, dropping the old one
-      return;
-    }
-    this.receivedChunks.set(index, chunk);
-
-    this.bytesReceived += chunk.content.byteLength;
-    this.validateBytesReceived();
-
-    const currentProgress = this.totalByteSize
-      ? this.bytesReceived / this.totalByteSize
-      : undefined;
-    this.onProgress?.(currentProgress);
-  }
-
-  /**
-   * @param progress - progress of the stream between 0 and 1. Undefined for streams of unknown size
-   */
-  onProgress?: (progress: number | undefined) => void;
-
   /**
    * Async iterator implementation to allow usage of `for await...of` syntax.
-   * Yields structured chunks from the stream.
-   *
+   * Yields each chunk's decoded text as it arrives - a delta, not the string accumulated so far.
    */
   [Symbol.asyncIterator]() {
     const reader = this.reader.getReader();