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)
diff --git a/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts b/src/room/data-stream/incoming/IncomingDataStreamManager.test.ts
index c973973ad5..b2d602b281 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 () => {
@@ -1795,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<TextStreamReader> = [];
+      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');
+    });
   });
 });
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(