Port data streams over to rust livekit-ffi version

.changeset/quiet-ducks-yell.md

@@ -0,0 +1,5 @@
+---
+'@livekit/rtc-node': patch
+---
+
+Convert data streams to use livekit-ffi exposed data streams interface

packages/livekit-rtc/scripts/run-e2e.mjs

@@ -16,6 +16,7 @@ const PORT = 7880;
 const URL = `ws://${HOST}:${PORT}`;
 const API_KEY = 'devkey';
 const API_SECRET = 'secret';
+const TEST_FILES = ['src/tests/e2e.test.ts', 'src/tests/e2e_data_streams.test.ts'];
 
 async function tcpReady(host, port, timeoutMs) {
   const deadline = Date.now() + timeoutMs;
@@ -77,7 +78,7 @@ process.on('SIGTERM', () => onSignal('SIGTERM'));
 try {
   await tcpReady(HOST, PORT, 15_000);
 
-  const args = ['exec', 'vitest', 'run', 'src/tests/e2e.test.ts', ...process.argv.slice(2)];
+  const args = ['exec', 'vitest', 'run', ...TEST_FILES, ...process.argv.slice(2)];
   testProc = spawn('pnpm', args, {
     env: {
       ...process.env,

packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts

@@ -146,12 +146,15 @@ function makeEndTrackingStream(): { stream: AudioStreamSource; endCount: () => n
 
 function makeLocalParticipant(identity: string): LocalParticipant {
   // Bypass the FFI-touching constructor; set only the fields the lifecycle
-  // paths read (identity getter + trackPublications map).
+  // paths read (identity getter, trackPublications map, and the open data
+  // stream writers that disconnect cleanup disposes).
   const p = Object.create(LocalParticipant.prototype) as LocalParticipant;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   (p as any).info = { identity };
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   (p as any).trackPublications = new Map();
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  (p as any).openStreamWriters = new Set();
   return p;
 }
 

packages/livekit-rtc/src/data_streams/stream_reader.ts

@@ -26,7 +26,13 @@ abstract class BaseStreamReader<T extends BaseStreamInfo> {
     this.bytesReceived = 0;
   }
 
-  protected abstract handleChunkReceived(chunk: DataStream_Chunk): void;
+  protected handleChunkReceived(chunk: DataStream_Chunk) {
+    this.bytesReceived += chunk.content!.byteLength;
+    const currentProgress = this.totalByteSize
+      ? this.bytesReceived / this.totalByteSize
+      : undefined;
+    this.onProgress?.(currentProgress);
+  }
 
   onProgress?: (progress: number | undefined) => void;
 
@@ -37,14 +43,6 @@ abstract class BaseStreamReader<T extends BaseStreamInfo> {
  * A class to read chunks from a ReadableStream and provide them in a structured format.
  */
 export class ByteStreamReader extends BaseStreamReader<ByteStreamInfo> {
-  protected handleChunkReceived(chunk: DataStream_Chunk) {
-    this.bytesReceived += chunk.content!.byteLength;
-    const currentProgress = this.totalByteSize
-      ? this.bytesReceived / this.totalByteSize
-      : undefined;
-    this.onProgress?.(currentProgress);
-  }
-
   [Symbol.asyncIterator]() {
     const reader = this.reader.getReader();
 
@@ -66,7 +64,10 @@ export class ByteStreamReader extends BaseStreamReader<ByteStreamInfo> {
           // consumer never calls return() (e.g. breaking out of for-await).
           reader.releaseLock();
           log.error('error processing stream update: %s', error);
-          return { done: true, value: undefined as unknown };
+          // Propagate abnormal termination (e.g. remote abort, payload over
+          // the receiver's size limit) instead of presenting the truncated
+          // payload as a clean EOF.
+          throw error;
         }
       },
 
@@ -99,9 +100,9 @@ export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
   constructor(
     info: TextStreamInfo,
     stream: ReadableStream<DataStream_Chunk>,
-    totalChunkCount?: number,
+    totalByteSize?: number,
   ) {
-    super(info, stream, totalChunkCount);
+    super(info, stream, totalByteSize);
     this.receivedChunks = new Map();
   }
 
@@ -113,10 +114,10 @@ export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
       return;
     }
     this.receivedChunks.set(index, chunk);
-    const currentProgress = this.totalByteSize
-      ? this.receivedChunks.size / this.totalByteSize
-      : undefined;
-    this.onProgress?.(currentProgress);
+    // Progress is reported against the payload's total byte length, so it has
+    // to be measured in received bytes too — a chunk count would report e.g.
+    // 1/50000 for a payload that arrived as a single chunk.
+    super.handleChunkReceived(chunk);
   }
 
   /**
@@ -153,7 +154,10 @@ export class TextStreamReader extends BaseStreamReader<TextStreamInfo> {
           reader.releaseLock();
           receivedChunks.clear();
           log.error('error processing stream update: %s', error);
-          return { done: true, value: undefined };
+          // Propagate abnormal termination (e.g. remote abort, payload over
+          // the receiver's size limit) instead of presenting the truncated
+          // payload as a clean EOF.
+          throw error;
         }
       },
 

packages/livekit-rtc/src/data_streams/types.ts

@@ -42,6 +42,12 @@ export interface TextStreamOptions extends DataStreamOptions {
 export interface ByteStreamOptions extends DataStreamOptions {
   name?: string;
   onProgress?: (progress: number) => void;
+  /**
+   * Whether the payload may be compressed on the wire. Defaults to true;
+   * compression is only applied when it actually reduces the payload size
+   * and every recipient supports it.
+   */
+  compress?: boolean;
 }
 
 export type ByteStreamHandler = (