From 72a7177ae188d70d0d61ced1c1b9e6162a15a9b6 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 22 Jul 2026 13:46:51 -0400 Subject: [PATCH 01/15] feat: port data streams over to rust livekit-ffi version --- packages/livekit-rtc/src/participant.ts | 399 +++++++++++------------- packages/livekit-rtc/src/room.ts | 280 +++++++++-------- packages/livekit-rtc/src/utils.ts | 30 ++ 3 files changed, 359 insertions(+), 350 deletions(-) diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index ca761f26..b8f23fa7 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: 2024 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import { Mutex } from '@livekit/mutex'; +import { type Mutex } from '@livekit/mutex'; import { DisconnectReason, type OwnedParticipant, @@ -10,13 +10,16 @@ import { type ParticipantKindDetail, } from '@livekit/rtc-ffi-bindings'; import { + type ByteStreamOpenCallback, + ByteStreamOpenRequest, + type ByteStreamOpenResponse, + type ByteStreamWriterCloseCallback, + ByteStreamWriterCloseRequest, + type ByteStreamWriterCloseResponse, + type ByteStreamWriterWriteCallback, + ByteStreamWriterWriteRequest, + type ByteStreamWriterWriteResponse, ChatMessage as ChatMessageModel, - DataStream_ByteHeader, - DataStream_Chunk, - DataStream_Header, - DataStream_OperationType, - DataStream_TextHeader, - DataStream_Trailer, EditChatMessageRequest, TranscriptionSegment as ProtoTranscriptionSegment, type PublishDataCallback, @@ -34,15 +37,6 @@ import { type SendChatMessageCallback, SendChatMessageRequest, type SendChatMessageResponse, - type SendStreamChunkCallback, - SendStreamChunkRequest, - type SendStreamChunkResponse, - type SendStreamHeaderCallback, - SendStreamHeaderRequest, - type SendStreamHeaderResponse, - type SendStreamTrailerCallback, - SendStreamTrailerRequest, - type SendStreamTrailerResponse, type SetLocalAttributesCallback, SetLocalAttributesRequest, type SetLocalAttributesResponse, @@ -52,6 +46,23 @@ import { type SetLocalNameCallback, SetLocalNameRequest, type SetLocalNameResponse, + StreamByteOptions, + type StreamSendFileCallback, + StreamSendFileRequest, + type StreamSendFileResponse, + type StreamSendTextCallback, + StreamSendTextRequest, + type StreamSendTextResponse, + StreamTextOptions, + type TextStreamOpenCallback, + TextStreamOpenRequest, + type TextStreamOpenResponse, + type TextStreamWriterCloseCallback, + TextStreamWriterCloseRequest, + type TextStreamWriterCloseResponse, + type TextStreamWriterWriteCallback, + TextStreamWriterWriteRequest, + type TextStreamWriterWriteResponse, type TrackPublishOptions, type UnpublishTrackCallback, UnpublishTrackRequest, @@ -71,12 +82,10 @@ import { UnregisterRpcMethodRequest, } from '@livekit/rtc-ffi-bindings'; import type { PathLike } from 'node:fs'; -import { open, stat } from 'node:fs/promises'; +import { fileURLToPath } from 'node:url'; import { - type ByteStreamInfo, type ByteStreamOptions, ByteStreamWriter, - type TextStreamInfo, TextStreamWriter, } from './data_streams/index.js'; import { FfiClient, FfiHandle } from './ffi_client.js'; @@ -87,9 +96,8 @@ import type { RemoteTrackPublication, TrackPublication } from './track_publicati import { LocalTrackPublication } from './track_publication.js'; import type { Transcription } from './transcription.js'; import type { ChatMessage } from './types.js'; -import { numberToBigInt, splitUtf8 } from './utils.js'; - -const STREAM_CHUNK_SIZE = 15_000; +import { byteStreamInfoFromProto, textStreamInfoFromProto } from './utils.js'; +import { numberToBigInt } from './utils.js'; export abstract class Participant { /** @internal */ @@ -279,94 +287,55 @@ export class LocalParticipant extends Participant { senderIdentity?: string; totalSize?: number; }): Promise { - const senderIdentity = options?.senderIdentity ?? this.identity; - const streamId = options?.streamId ?? crypto.randomUUID(); - const destinationIdentities = options?.destinationIdentities; - - const info: TextStreamInfo = { - streamId: streamId, - mimeType: 'text/plain', - topic: options?.topic ?? '', - timestamp: Date.now(), - totalSize: options?.totalSize, - }; - - const headerReq = new SendStreamHeaderRequest({ - senderIdentity, - destinationIdentities, + const req = new TextStreamOpenRequest({ localParticipantHandle: this.ffi_handle.handle, - header: new DataStream_Header({ - streamId, - mimeType: info.mimeType, - topic: info.topic, - timestamp: numberToBigInt(info.timestamp), - totalLength: numberToBigInt(info.totalSize), + options: new StreamTextOptions({ + topic: options?.topic ?? '', attributes: options?.attributes, - contentHeader: { - case: 'textHeader', - value: new DataStream_TextHeader({ - operationType: DataStream_OperationType.CREATE, - version: 0, - replyToStreamId: '', - generated: false, - }), - }, + destinationIdentities: options?.destinationIdentities, + id: options?.streamId, + senderIdentity: options?.senderIdentity ?? this.identity, + totalSize: options?.totalSize, }), }); - await this.sendStreamHeader(headerReq); + const res = FfiClient.instance.request({ + message: { case: 'textStreamOpen', value: req }, + }); + + const cb = await FfiClient.instance.waitFor( + (ev) => ev.message.case == 'textStreamOpen' && ev.message.value.asyncId == res.asyncId, + { signal: this.disconnectSignal }, + ); + + if (cb.result.case !== 'writer') { + throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + } - let nextChunkId = 0; - const localHandle = this.ffi_handle.handle; - const sendTrailer = this.sendStreamTrailer; - const sendChunk = this.sendStreamChunk; + const writerHandle = cb.result.value.handle!.id!; + const info = textStreamInfoFromProto(cb.result.value.info!); + const writeText = this.textStreamWriterWrite; + const closeWriter = this.textStreamWriterClose; const writableStream = new WritableStream({ // Implement the sink async write(text) { - for (const textByteChunk of splitUtf8(text, STREAM_CHUNK_SIZE)) { - const chunkRequest = new SendStreamChunkRequest({ - senderIdentity, - localParticipantHandle: localHandle, - destinationIdentities, - chunk: new DataStream_Chunk({ - content: textByteChunk, - streamId, - chunkIndex: numberToBigInt(nextChunkId), - }), - }); - - await sendChunk(chunkRequest); - nextChunkId += 1; - } + await writeText(new TextStreamWriterWriteRequest({ writerHandle, text })); }, async close() { - const trailerReq = new SendStreamTrailerRequest({ - senderIdentity, - localParticipantHandle: localHandle, - destinationIdentities, - trailer: new DataStream_Trailer({ - streamId, - reason: '', - }), - }); - await sendTrailer(trailerReq); + await closeWriter(new TextStreamWriterCloseRequest({ writerHandle })); }, - // Send a trailer with the error reason so the remote side's stream + // Close the stream with the error reason so the remote side's stream // controller is closed instead of waiting for data that won't arrive. async abort(err) { log.error(err, 'Sink Error'); try { - const trailerReq = new SendStreamTrailerRequest({ - senderIdentity, - localParticipantHandle: localHandle, - destinationIdentities, - trailer: new DataStream_Trailer({ - streamId, + await closeWriter( + new TextStreamWriterCloseRequest({ + writerHandle, reason: err instanceof Error ? err.message : String(err ?? ''), }), - }); - await sendTrailer(trailerReq); + ); } catch { // Best-effort: the connection may already be gone. } @@ -387,13 +356,32 @@ export class LocalParticipant extends Participant { streamId?: string; }, ) { - const writer = await this.streamText({ - ...options, - totalSize: new TextEncoder().encode(text).byteLength, + const req = new StreamSendTextRequest({ + localParticipantHandle: this.ffi_handle.handle, + options: new StreamTextOptions({ + topic: options?.topic ?? '', + attributes: options?.attributes, + destinationIdentities: options?.destinationIdentities, + id: options?.streamId, + senderIdentity: this.identity, + }), + text, }); - await writer.write(text); - await writer.close(); - return writer.info; + + const res = FfiClient.instance.request({ + message: { case: 'sendText', value: req }, + }); + + const cb = await FfiClient.instance.waitFor( + (ev) => ev.message.case == 'sendText' && ev.message.value.asyncId == res.asyncId, + { signal: this.disconnectSignal }, + ); + + if (cb.result.case !== 'info') { + throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + } + + return textStreamInfoFromProto(cb.result.value); } async streamBytes(options?: { @@ -405,102 +393,56 @@ export class LocalParticipant extends Participant { mimeType?: string; totalSize?: number; }) { - const senderIdentity = this.identity; - const streamId = options?.streamId ?? crypto.randomUUID(); - const destinationIdentities = options?.destinationIdentities; - - const info: ByteStreamInfo = { - streamId: streamId, - mimeType: options?.mimeType ?? 'application/octet-stream', - topic: options?.topic ?? '', - timestamp: Date.now(), - attributes: options?.attributes, - totalSize: options?.totalSize, - name: options?.name ?? 'unknown', - }; - - const headerReq = new SendStreamHeaderRequest({ - senderIdentity, - destinationIdentities, + const req = new ByteStreamOpenRequest({ localParticipantHandle: this.ffi_handle.handle, - header: new DataStream_Header({ - streamId, - mimeType: info.mimeType, - topic: info.topic, - timestamp: numberToBigInt(info.timestamp), - attributes: info.attributes, - totalLength: numberToBigInt(info.totalSize), - contentHeader: { - case: 'byteHeader', - value: new DataStream_ByteHeader({ - name: info.name, - }), - }, + options: new StreamByteOptions({ + topic: options?.topic ?? '', + attributes: options?.attributes, + destinationIdentities: options?.destinationIdentities, + id: options?.streamId, + name: options?.name ?? 'unknown', + mimeType: options?.mimeType ?? 'application/octet-stream', + totalLength: numberToBigInt(options?.totalSize), + senderIdentity: this.identity, }), }); - await this.sendStreamHeader(headerReq); + const res = FfiClient.instance.request({ + message: { case: 'byteStreamOpen', value: req }, + }); + + const cb = await FfiClient.instance.waitFor( + (ev) => ev.message.case == 'byteStreamOpen' && ev.message.value.asyncId == res.asyncId, + { signal: this.disconnectSignal }, + ); - let chunkId = 0; - const localHandle = this.ffi_handle.handle; - const sendTrailer = this.sendStreamTrailer; - const sendChunk = this.sendStreamChunk; - const writeMutex = new Mutex(); + if (cb.result.case !== 'writer') { + throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + } + + const writerHandle = cb.result.value.handle!.id!; + const info = byteStreamInfoFromProto(cb.result.value.info!); + const writeBytes = this.byteStreamWriterWrite; + const closeWriter = this.byteStreamWriterClose; const writableStream = new WritableStream({ async write(chunk) { - const unlock = await writeMutex.lock(); - - let byteOffset = 0; - try { - while (byteOffset < chunk.byteLength) { - const subChunk = chunk.slice(byteOffset, byteOffset + STREAM_CHUNK_SIZE); - const chunkRequest = new SendStreamChunkRequest({ - senderIdentity, - localParticipantHandle: localHandle, - destinationIdentities, - chunk: new DataStream_Chunk({ - content: subChunk, - streamId, - chunkIndex: numberToBigInt(chunkId), - }), - }); - - await sendChunk(chunkRequest); - chunkId += 1; - byteOffset += subChunk.byteLength; - } - } finally { - unlock(); - } + await writeBytes(new ByteStreamWriterWriteRequest({ writerHandle, bytes: chunk })); }, async close() { - const trailerReq = new SendStreamTrailerRequest({ - senderIdentity, - localParticipantHandle: localHandle, - destinationIdentities, - trailer: new DataStream_Trailer({ - streamId, - reason: '', - }), - }); - await sendTrailer(trailerReq); + await closeWriter(new ByteStreamWriterCloseRequest({ writerHandle })); }, - // Send a trailer with the error reason so the remote side's stream + // Close the stream with the error reason so the remote side's stream // controller is closed instead of waiting for data that won't arrive. async abort(err) { log.error(err, 'Sink error'); try { - const trailerReq = new SendStreamTrailerRequest({ - senderIdentity, - localParticipantHandle: localHandle, - destinationIdentities, - trailer: new DataStream_Trailer({ - streamId, + await closeWriter( + new ByteStreamWriterCloseRequest({ + writerHandle, reason: err instanceof Error ? err.message : String(err ?? ''), }), - }); - await sendTrailer(trailerReq); + ); } catch { // Best-effort: the connection may already be gone. } @@ -514,77 +456,96 @@ export class LocalParticipant extends Participant { /** Sends a file provided as PathLike to specified recipients */ async sendFile(path: PathLike, options?: ByteStreamOptions) { - const fileStats = await stat(path); - const file = await open(path); - try { - const stream: ReadableStream = file.readableWebStream(); - const streamId = crypto.randomUUID(); - const destinationIdentities = options?.destinationIdentities; - - const writer = await this.streamBytes({ - streamId: streamId, - name: options?.name, - totalSize: fileStats.size, - destinationIdentities, - topic: options?.topic, - mimeType: options?.mimeType, + const filePath = path instanceof URL ? fileURLToPath(path) : path.toString(); + + const req = new StreamSendFileRequest({ + localParticipantHandle: this.ffi_handle.handle, + options: new StreamByteOptions({ + topic: options?.topic ?? '', attributes: options?.attributes, - }); + destinationIdentities: options?.destinationIdentities, + name: options?.name ?? 'unknown', + mimeType: options?.mimeType ?? 'application/octet-stream', + senderIdentity: this.identity, + }), + filePath, + }); - for await (const chunk of stream) { - await writer.write(chunk); - } - await writer.close(); - } finally { - await file.close(); + const res = FfiClient.instance.request({ + message: { case: 'sendFile', value: req }, + }); + + const cb = await FfiClient.instance.waitFor( + (ev) => ev.message.case == 'sendFile' && ev.message.value.asyncId == res.asyncId, + { signal: this.disconnectSignal }, + ); + + if (cb.result.case === 'error') { + throw new Error(cb.result.value.description); } } - private async sendStreamHeader(req: SendStreamHeaderRequest) { - const type = 'sendStreamHeader'; - const res = FfiClient.instance.request({ - message: { case: type, value: req }, + private textStreamWriterWrite = async (req: TextStreamWriterWriteRequest) => { + const res = FfiClient.instance.request({ + message: { case: 'textStreamWrite', value: req }, }); - const cb = await FfiClient.instance.waitFor( - (ev) => ev.message.case == type && ev.message.value.asyncId == res.asyncId, + const cb = await FfiClient.instance.waitFor( + (ev) => + ev.message.case == 'textStreamWriterWrite' && ev.message.value.asyncId == res.asyncId, { signal: this.disconnectSignal }, ); if (cb.error) { - throw new Error(cb.error); + throw new Error(cb.error.description); } - } + }; - private sendStreamChunk = async (req: SendStreamChunkRequest) => { - const type = 'sendStreamChunk'; - const res = FfiClient.instance.request({ - message: { case: type, value: req }, + private textStreamWriterClose = async (req: TextStreamWriterCloseRequest) => { + const res = FfiClient.instance.request({ + message: { case: 'textStreamClose', value: req }, }); - const cb = await FfiClient.instance.waitFor( - (ev) => ev.message.case == type && ev.message.value.asyncId == res.asyncId, + const cb = await FfiClient.instance.waitFor( + (ev) => + ev.message.case == 'textStreamWriterClose' && ev.message.value.asyncId == res.asyncId, { signal: this.disconnectSignal }, ); if (cb.error) { - throw new Error(cb.error); + throw new Error(cb.error.description); } }; - private sendStreamTrailer = async (req: SendStreamTrailerRequest) => { - const type = 'sendStreamTrailer'; - const res = FfiClient.instance.request({ - message: { case: type, value: req }, + private byteStreamWriterWrite = async (req: ByteStreamWriterWriteRequest) => { + const res = FfiClient.instance.request({ + message: { case: 'byteStreamWrite', value: req }, }); - const cb = await FfiClient.instance.waitFor( - (ev) => ev.message.case == type && ev.message.value.asyncId == res.asyncId, + const cb = await FfiClient.instance.waitFor( + (ev) => + ev.message.case == 'byteStreamWriterWrite' && ev.message.value.asyncId == res.asyncId, { signal: this.disconnectSignal }, ); if (cb.error) { - throw new Error(cb.error); + throw new Error(cb.error.description); + } + }; + + private byteStreamWriterClose = async (req: ByteStreamWriterCloseRequest) => { + const res = FfiClient.instance.request({ + message: { case: 'byteStreamClose', value: req }, + }); + + const cb = await FfiClient.instance.waitFor( + (ev) => + ev.message.case == 'byteStreamWriterClose' && ev.message.value.asyncId == res.asyncId, + { signal: this.disconnectSignal }, + ); + + if (cb.error) { + throw new Error(cb.error.description); } }; diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index 85c001c7..23f73836 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -9,12 +9,10 @@ import type { GetSessionStatsResponse, } from '@livekit/rtc-ffi-bindings'; import { DisconnectReason, type OwnedParticipant } from '@livekit/rtc-ffi-bindings'; -import type { - DataStream_Trailer, - DisconnectCallback, - TrackPublicationInfo, -} from '@livekit/rtc-ffi-bindings'; +import { type DisconnectCallback, type TrackPublicationInfo } from '@livekit/rtc-ffi-bindings'; import { + ByteStreamReaderReadIncrementalRequest, + type ByteStreamReaderReadIncrementalResponse, type ConnectCallback, ConnectRequest, type ConnectResponse, @@ -22,30 +20,27 @@ import { ConnectionState, ContinualGatheringPolicy, type DataPacketKind, - type DataStream_Chunk, - type DataStream_Header, + DataStream_Chunk, type DisconnectResponse, RoomOptions as FfiRoomOptions, type IceServer, IceTransportType, + type OwnedByteStreamReader, + type OwnedTextStreamReader, type ReadyForRoomEventResponse, RoomDataStreamOptions, type RoomInfo, type SimulateScenarioCallback, type SimulateScenarioKind, type SimulateScenarioResponse, + TextStreamReaderReadIncrementalRequest, + type TextStreamReaderReadIncrementalResponse, } from '@livekit/rtc-ffi-bindings'; import { TrackKind } from '@livekit/rtc-ffi-bindings'; import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter'; import EventEmitter from 'events'; import { ByteStreamReader, TextStreamReader } from './data_streams/stream_reader.js'; -import type { - ByteStreamHandler, - ByteStreamInfo, - StreamController, - TextStreamHandler, - TextStreamInfo, -} from './data_streams/types.js'; +import { type ByteStreamHandler, type TextStreamHandler } from './data_streams/types.js'; import type { E2EEOptions } from './e2ee.js'; import { E2EEManager, defaultE2EEOptions } from './e2ee.js'; import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js'; @@ -57,7 +52,12 @@ import { RemoteAudioTrack, RemoteVideoTrack } from './track.js'; import type { LocalTrackPublication, TrackPublication } from './track_publication.js'; import { RemoteTrackPublication } from './track_publication.js'; import type { ChatMessage } from './types.js'; -import { bigIntToNumber } from './utils.js'; +import { + bigIntToNumber, + byteStreamInfoFromProto, + numberToBigInt, + textStreamInfoFromProto, +} from './utils.js'; export { RoomDataStreamOptions }; @@ -104,8 +104,15 @@ export class Room extends (EventEmitter as new () => TypedEmitter */ private ffiEventLock = new Mutex(); - private byteStreamControllers = new Map>(); - private textStreamControllers = new Map>(); + // Active incoming stream readers keyed by their FFI reader handle id. Each entry + // owns the FfiClient listener feeding reader events into the ReadableStream. + private streamReaders = new Map< + bigint, + { + controller: ReadableStreamDefaultController; + listener: (ev: FfiEvent) => void; + } + >(); private byteStreamHandlers = new Map(); private textStreamHandlers = new Map(); @@ -431,28 +438,20 @@ export class Room extends (EventEmitter as new () => TypedEmitter if (this.hasCleanedUp) return; this.hasCleanedUp = true; - // Error all in-progress stream controllers to prevent FD leaks. - // Streams that were receiving data but never got a trailer (e.g. the sender - // disconnected mid-transfer) would otherwise keep their ReadableStream open - // indefinitely, leaking the underlying controller and any buffered chunks. + // Error all in-progress stream readers to prevent FD leaks. + // Streams that were receiving data but never reached end-of-stream (e.g. the + // sender disconnected mid-transfer) would otherwise keep their ReadableStream + // open indefinitely, leaking the underlying controller and any buffered chunks. // Using error() instead of close() signals an abnormal termination to consumers. - for (const [, streamController] of this.byteStreamControllers) { - try { - streamController.controller.error(new Error('Disconnected while receiving')); - } catch { - // controller may already be closed or errored - } - } - this.byteStreamControllers.clear(); - - for (const [, streamController] of this.textStreamControllers) { + for (const [, { controller, listener }] of this.streamReaders) { + FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); try { - streamController.controller.error(new Error('Disconnected while receiving')); + controller.error(new Error('Disconnected while receiving')); } catch { // controller may already be closed or errored } } - this.textStreamControllers.clear(); + this.streamReaders.clear(); // Detach every track from this room so attached FrameProcessors receive // onStreamInfoCleared/onCredentialsCleared and the tokenRefreshed listener @@ -850,12 +849,10 @@ export class Room extends (EventEmitter as new () => TypedEmitter this.emit(RoomEvent.Reconnected); } else if (ev.case == 'roomSidChanged') { this.emit(RoomEvent.RoomSidChanged, ev.value.sid!); - } else if (ev.case === 'streamHeaderReceived' && ev.value.header) { - this.handleStreamHeader(ev.value.header, ev.value.participantIdentity!); - } else if (ev.case === 'streamChunkReceived' && ev.value.chunk) { - this.handleStreamChunk(ev.value.chunk); - } else if (ev.case === 'streamTrailerReceived' && ev.value.trailer) { - this.handleStreamTrailer(ev.value.trailer); + } else if (ev.case === 'byteStreamOpened' && ev.value.reader) { + this.handleByteStreamOpened(ev.value.reader, ev.value.participantIdentity!); + } else if (ev.case === 'textStreamOpened' && ev.value.reader) { + this.handleTextStreamOpened(ev.value.reader, ev.value.participantIdentity!); } else if (ev.case === 'roomUpdated') { this.info = ev.value; this.emit(RoomEvent.RoomUpdated); @@ -940,104 +937,125 @@ export class Room extends (EventEmitter as new () => TypedEmitter return participant; } - private handleStreamHeader(streamHeader: DataStream_Header, participantIdentity: string) { - if (streamHeader.contentHeader.case === 'byteHeader') { - const streamHandlerCallback = this.byteStreamHandlers.get(streamHeader.topic ?? ''); + private handleByteStreamOpened(ownedReader: OwnedByteStreamReader, participantIdentity: string) { + const info = byteStreamInfoFromProto(ownedReader.info!); + const readerHandle = ownedReader.handle!.id!; - if (!streamHandlerCallback) { - log.debug( - 'ignoring incoming byte stream due to no handler for topic: %s', - streamHeader.topic ?? 'undefined', - ); - return; - } - let streamController: ReadableStreamDefaultController; - const stream = new ReadableStream({ - start: (controller) => { - streamController = controller; - this.byteStreamControllers.set(streamHeader.streamId!, { - header: streamHeader, - controller: streamController, - startTime: Date.now(), - }); - }, - }); - const info: ByteStreamInfo = { - streamId: streamHeader.streamId!, - name: streamHeader.contentHeader.value.name ?? 'unknown', - mimeType: streamHeader.mimeType!, - totalSize: streamHeader.totalLength ? Number(streamHeader.totalLength) : undefined, - topic: streamHeader.topic!, - timestamp: bigIntToNumber(streamHeader.timestamp!), - attributes: streamHeader.attributes, - }; - streamHandlerCallback( - new ByteStreamReader(info, stream, bigIntToNumber(streamHeader.totalLength)), - { identity: participantIdentity }, - ); - } else if (streamHeader.contentHeader.case === 'textHeader') { - const streamHandlerCallback = this.textStreamHandlers.get(streamHeader.topic ?? ''); - - if (!streamHandlerCallback) { - log.debug( - 'ignoring incoming text stream due to no handler for topic: %s', - streamHeader.topic ?? 'undefined', - ); - return; - } - let streamController: ReadableStreamDefaultController; - const stream = new ReadableStream({ - start: (controller) => { - streamController = controller; - this.textStreamControllers.set(streamHeader.streamId!, { - header: streamHeader, - controller: streamController, - startTime: Date.now(), - }); - }, - }); - const info: TextStreamInfo = { - streamId: streamHeader.streamId!, - mimeType: streamHeader.mimeType!, - totalSize: streamHeader.totalLength ? Number(streamHeader.totalLength) : undefined, - topic: streamHeader.topic!, - timestamp: Number(streamHeader.timestamp), - attributes: streamHeader.attributes, - }; - streamHandlerCallback( - new TextStreamReader(info, stream, bigIntToNumber(streamHeader.totalLength)), - { identity: participantIdentity }, - ); + const streamHandlerCallback = this.byteStreamHandlers.get(info.topic); + if (!streamHandlerCallback) { + log.debug('ignoring incoming byte stream due to no handler for topic: %s', info.topic); + // The native reader was never consumed — dispose it to free the handle + // and any chunks it has buffered. + new FfiHandle(readerHandle).dispose(); + return; } - } - private handleStreamChunk(chunk: DataStream_Chunk) { - const fileBuffer = this.byteStreamControllers.get(chunk.streamId!); - if (fileBuffer) { - if (chunk.content!.length > 0) { - fileBuffer.controller.enqueue(chunk); - } - } - const textBuffer = this.textStreamControllers.get(chunk.streamId!); - if (textBuffer) { - if (chunk.content!.length > 0) { - textBuffer.controller.enqueue(chunk); - } - } + const stream = new ReadableStream({ + start: (controller) => { + let nextChunkIndex = 0; + const listener = (ev: FfiEvent) => { + if ( + ev.message.case !== 'byteStreamReaderEvent' || + ev.message.value.readerHandle !== readerHandle + ) { + return; + } + const detail = ev.message.value.detail; + if (detail.case === 'chunkReceived') { + const content = detail.value.content!; + if (content.length > 0) { + controller.enqueue( + new DataStream_Chunk({ content, chunkIndex: numberToBigInt(nextChunkIndex++) }), + ); + } + } else if (detail.case === 'eos') { + // End-of-stream (including a remote abort carrying an error) closes the + // stream normally so consumers see EOF, matching the trailer behavior of + // the previous implementation. + FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); + this.streamReaders.delete(readerHandle); + controller.close(); + } + }; + FfiClient.instance.on(FfiClientEvent.FfiEvent, listener); + this.streamReaders.set(readerHandle, { controller, listener }); + + // Start the incremental read; the native reader buffers chunks received + // before this request, so nothing is lost. This consumes the FFI handle. + FfiClient.instance.request({ + message: { + case: 'byteReadIncremental', + value: new ByteStreamReaderReadIncrementalRequest({ readerHandle }), + }, + }); + }, + }); + + streamHandlerCallback( + new ByteStreamReader(info, stream, bigIntToNumber(ownedReader.info!.totalLength)), + { identity: participantIdentity }, + ); } - private handleStreamTrailer(trailer: DataStream_Trailer) { - const streamId = trailer.streamId!; - const fileBuffer = this.byteStreamControllers.get(streamId); - if (fileBuffer) { - fileBuffer.controller.close(); - this.byteStreamControllers.delete(streamId); - } - const textBuffer = this.textStreamControllers.get(streamId); - if (textBuffer) { - textBuffer.controller.close(); - this.textStreamControllers.delete(streamId); + private handleTextStreamOpened(ownedReader: OwnedTextStreamReader, participantIdentity: string) { + const info = textStreamInfoFromProto(ownedReader.info!); + const readerHandle = ownedReader.handle!.id!; + + const streamHandlerCallback = this.textStreamHandlers.get(info.topic); + if (!streamHandlerCallback) { + log.debug('ignoring incoming text stream due to no handler for topic: %s', info.topic); + // The native reader was never consumed — dispose it to free the handle + // and any chunks it has buffered. + new FfiHandle(readerHandle).dispose(); + return; } + + const textEncoder = new TextEncoder(); + const stream = new ReadableStream({ + start: (controller) => { + let nextChunkIndex = 0; + const listener = (ev: FfiEvent) => { + if ( + ev.message.case !== 'textStreamReaderEvent' || + ev.message.value.readerHandle !== readerHandle + ) { + return; + } + const detail = ev.message.value.detail; + if (detail.case === 'chunkReceived') { + const content = textEncoder.encode(detail.value.content!); + if (content.length > 0) { + controller.enqueue( + new DataStream_Chunk({ content, chunkIndex: numberToBigInt(nextChunkIndex++) }), + ); + } + } else if (detail.case === 'eos') { + // End-of-stream (including a remote abort carrying an error) closes the + // stream normally so consumers see EOF, matching the trailer behavior of + // the previous implementation. + FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); + this.streamReaders.delete(readerHandle); + controller.close(); + } + }; + FfiClient.instance.on(FfiClientEvent.FfiEvent, listener); + this.streamReaders.set(readerHandle, { controller, listener }); + + // Start the incremental read; the native reader buffers chunks received + // before this request, so nothing is lost. This consumes the FFI handle. + FfiClient.instance.request({ + message: { + case: 'textReadIncremental', + value: new TextStreamReaderReadIncrementalRequest({ readerHandle }), + }, + }); + }, + }); + + streamHandlerCallback( + new TextStreamReader(info, stream, bigIntToNumber(ownedReader.info!.totalLength)), + { identity: participantIdentity }, + ); } } diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts index 7bc5a650..090fabc0 100644 --- a/packages/livekit-rtc/src/utils.ts +++ b/packages/livekit-rtc/src/utils.ts @@ -1,6 +1,11 @@ // SPDX-FileCopyrightText: 2024 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 +import type { + ByteStreamInfo as ProtoByteStreamInfo, + TextStreamInfo as ProtoTextStreamInfo, +} from '@livekit/rtc-ffi-bindings'; +import type { ByteStreamInfo, TextStreamInfo } from './data_streams/types.js'; /** convert bigints to numbers preserving undefined values */ export function bigIntToNumber( @@ -40,3 +45,28 @@ export function splitUtf8(s: string, n: number): Uint8Array[] { } return result; } + +/** @internal */ +export function textStreamInfoFromProto(info: ProtoTextStreamInfo): TextStreamInfo { + return { + streamId: info.streamId!, + mimeType: info.mimeType!, + topic: info.topic!, + timestamp: bigIntToNumber(info.timestamp!), + totalSize: info.totalLength !== undefined ? bigIntToNumber(info.totalLength) : undefined, + attributes: info.attributes, + }; +} + +/** @internal */ +export function byteStreamInfoFromProto(info: ProtoByteStreamInfo): ByteStreamInfo { + return { + streamId: info.streamId!, + name: info.name ?? 'unknown', + mimeType: info.mimeType!, + topic: info.topic!, + timestamp: bigIntToNumber(info.timestamp!), + totalSize: info.totalLength !== undefined ? bigIntToNumber(info.totalLength) : undefined, + attributes: info.attributes, + }; +} From f5dfc6639db3ec9c0f3338cb5b2c8ea1e6ba0b5d Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 22 Jul 2026 13:54:16 -0400 Subject: [PATCH 02/15] fix: add missing changeset --- .changeset/quiet-ducks-yell.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/quiet-ducks-yell.md diff --git a/.changeset/quiet-ducks-yell.md b/.changeset/quiet-ducks-yell.md new file mode 100644 index 00000000..424930c7 --- /dev/null +++ b/.changeset/quiet-ducks-yell.md @@ -0,0 +1,5 @@ +--- +'@livekit/rtc-node': patch +--- + +Convert data streams to use livekit-ffi exposed data streams interface From 5f1c007c02b562f3b112f672ae8f51dba4651d12 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 22 Jul 2026 15:11:04 -0400 Subject: [PATCH 03/15] feat: add compress and data stream options --- .../livekit-rtc/src/data_streams/types.ts | 6 ++++ packages/livekit-rtc/src/participant.ts | 10 ++++-- packages/livekit-rtc/src/room.ts | 36 ++++++++++++------- 3 files changed, 38 insertions(+), 14 deletions(-) diff --git a/packages/livekit-rtc/src/data_streams/types.ts b/packages/livekit-rtc/src/data_streams/types.ts index 5f7c9ff4..24f5c3be 100644 --- a/packages/livekit-rtc/src/data_streams/types.ts +++ b/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 = ( diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index b8f23fa7..4b55e4ef 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -285,7 +285,6 @@ export class LocalParticipant extends Participant { destinationIdentities?: Array; streamId?: string; senderIdentity?: string; - totalSize?: number; }): Promise { const req = new TextStreamOpenRequest({ localParticipantHandle: this.ffi_handle.handle, @@ -295,7 +294,6 @@ export class LocalParticipant extends Participant { destinationIdentities: options?.destinationIdentities, id: options?.streamId, senderIdentity: options?.senderIdentity ?? this.identity, - totalSize: options?.totalSize, }), }); @@ -354,6 +352,12 @@ export class LocalParticipant extends Participant { attributes?: Record; destinationIdentities?: Array; streamId?: string; + /** + * 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; }, ) { const req = new StreamSendTextRequest({ @@ -364,6 +368,7 @@ export class LocalParticipant extends Participant { destinationIdentities: options?.destinationIdentities, id: options?.streamId, senderIdentity: this.identity, + compress: options?.compress, }), text, }); @@ -467,6 +472,7 @@ export class LocalParticipant extends Participant { name: options?.name ?? 'unknown', mimeType: options?.mimeType ?? 'application/octet-stream', senderIdentity: this.identity, + compress: options?.compress, }), filePath, }); diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index 23f73836..311b4868 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -22,13 +22,13 @@ import { type DataPacketKind, DataStream_Chunk, type DisconnectResponse, + RoomDataStreamOptions as FfiRoomDataStreamOptions, RoomOptions as FfiRoomOptions, type IceServer, IceTransportType, type OwnedByteStreamReader, type OwnedTextStreamReader, type ReadyForRoomEventResponse, - RoomDataStreamOptions, type RoomInfo, type SimulateScenarioCallback, type SimulateScenarioKind, @@ -59,8 +59,6 @@ import { textStreamInfoFromProto, } from './utils.js'; -export { RoomDataStreamOptions }; - export interface RtcConfiguration { iceTransportType: IceTransportType; continualGatheringPolicy: ContinualGatheringPolicy; @@ -73,6 +71,15 @@ export const defaultRtcConfiguration: RtcConfiguration = { iceServers: [], }; +export interface RoomDataStreamOptions { + /** + * Maximum decompressed payload size in bytes accepted for a single incoming + * data stream. Incoming streams exceeding this limit terminate with an error + * on the receiving side. Unset falls back to the SDK default. + */ + maxPayloadByteLength?: number; +} + export interface RoomOptions { autoSubscribe: boolean; dynacast: boolean; @@ -82,6 +89,8 @@ export interface RoomOptions { e2ee?: E2EEOptions; encryption?: E2EEOptions; rtcConfig?: RtcConfiguration; + /** Options controlling data stream behavior for this room. */ + dataStream?: RoomDataStreamOptions; } export const defaultRoomOptions = new FfiRoomOptions({ @@ -282,14 +291,10 @@ export class Room extends (EventEmitter as new () => TypedEmitter * @throws ConnectError - if connection fails */ async connect(url: string, token: string, opts?: RoomOptions) { - const options = { ...defaultRoomOptions, ...opts }; - // The Node SDK still implements data streams in TypeScript on top of raw FFI - // packets, so always advertise only legacy (v1) data stream support to other - // clients, preserving any other data stream options provided by the user. - options.dataStream = new RoomDataStreamOptions({ - ...options.dataStream, - useLegacyClientImplementation: true, - }); + // dataStream is a plain user-facing object; convert it to its proto + // message separately instead of spreading it into the FFI options. + const { dataStream, ...restOpts } = opts ?? {}; + const options = { ...defaultRoomOptions, ...restOpts }; const e2eeEnabled = options.encryption || options.e2ee; const e2eeOptions = options.encryption ? { ...defaultE2EEOptions, ...options.encryption } @@ -298,7 +303,14 @@ export class Room extends (EventEmitter as new () => TypedEmitter const req = new ConnectRequest({ url: url, token: token, - options, + options: { + ...options, + dataStream: dataStream + ? new FfiRoomDataStreamOptions({ + maxPayloadByteLength: numberToBigInt(dataStream.maxPayloadByteLength), + }) + : undefined, + }, }); FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onFfiEvent); From d5d7ede3d99873ec3a1c763a35f8bfe36ceeaa79 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 22 Jul 2026 15:11:31 -0400 Subject: [PATCH 04/15] fix: ensure that end of stream errors are exposed into the read stream and not dropped Previously they were just being ignored, which meant that a "data stream too big" error (probably along with others too) would not get surfaced. --- .../src/data_streams/stream_reader.ts | 10 +++++-- packages/livekit-rtc/src/room.ts | 26 +++++++++++++------ 2 files changed, 26 insertions(+), 10 deletions(-) diff --git a/packages/livekit-rtc/src/data_streams/stream_reader.ts b/packages/livekit-rtc/src/data_streams/stream_reader.ts index 7067d69d..e6ffd63f 100644 --- a/packages/livekit-rtc/src/data_streams/stream_reader.ts +++ b/packages/livekit-rtc/src/data_streams/stream_reader.ts @@ -66,7 +66,10 @@ export class ByteStreamReader extends BaseStreamReader { // 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; } }, @@ -153,7 +156,10 @@ export class TextStreamReader extends BaseStreamReader { 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; } }, diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index 311b4868..f2ff6674 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -981,12 +981,17 @@ export class Room extends (EventEmitter as new () => TypedEmitter ); } } else if (detail.case === 'eos') { - // End-of-stream (including a remote abort carrying an error) closes the - // stream normally so consumers see EOF, matching the trailer behavior of - // the previous implementation. FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); this.streamReaders.delete(readerHandle); - controller.close(); + const error = detail.value.error; + if (error) { + // Abnormal termination (e.g. remote abort, payload over the + // receiver's size limit): surface the error to the consumer + // instead of presenting a truncated payload as a clean EOF. + controller.error(new Error(error.description ?? 'stream terminated')); + } else { + controller.close(); + } } }; FfiClient.instance.on(FfiClientEvent.FfiEvent, listener); @@ -1042,12 +1047,17 @@ export class Room extends (EventEmitter as new () => TypedEmitter ); } } else if (detail.case === 'eos') { - // End-of-stream (including a remote abort carrying an error) closes the - // stream normally so consumers see EOF, matching the trailer behavior of - // the previous implementation. FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); this.streamReaders.delete(readerHandle); - controller.close(); + const error = detail.value.error; + if (error) { + // Abnormal termination (e.g. remote abort, payload over the + // receiver's size limit): surface the error to the consumer + // instead of presenting a truncated payload as a clean EOF. + controller.error(new Error(error.description ?? 'stream terminated')); + } else { + controller.close(); + } } }; FfiClient.instance.on(FfiClientEvent.FfiEvent, listener); From 1fa5e968b8caee92f5ebbd0ae178b0d7ac11bbcb Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 29 Jul 2026 16:56:18 -0400 Subject: [PATCH 05/15] fix: satisfy prettier in participant.ts The data streams port left four waitFor predicates wrapped in a way prettier reformats, which failed the CI format:check gate. Co-Authored-By: Claude Fable 5 --- packages/livekit-rtc/src/participant.ts | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index 4b55e4ef..b2755237 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -497,8 +497,7 @@ export class LocalParticipant extends Participant { }); const cb = await FfiClient.instance.waitFor( - (ev) => - ev.message.case == 'textStreamWriterWrite' && ev.message.value.asyncId == res.asyncId, + (ev) => ev.message.case == 'textStreamWriterWrite' && ev.message.value.asyncId == res.asyncId, { signal: this.disconnectSignal }, ); @@ -513,8 +512,7 @@ export class LocalParticipant extends Participant { }); const cb = await FfiClient.instance.waitFor( - (ev) => - ev.message.case == 'textStreamWriterClose' && ev.message.value.asyncId == res.asyncId, + (ev) => ev.message.case == 'textStreamWriterClose' && ev.message.value.asyncId == res.asyncId, { signal: this.disconnectSignal }, ); @@ -529,8 +527,7 @@ export class LocalParticipant extends Participant { }); const cb = await FfiClient.instance.waitFor( - (ev) => - ev.message.case == 'byteStreamWriterWrite' && ev.message.value.asyncId == res.asyncId, + (ev) => ev.message.case == 'byteStreamWriterWrite' && ev.message.value.asyncId == res.asyncId, { signal: this.disconnectSignal }, ); @@ -545,8 +542,7 @@ export class LocalParticipant extends Participant { }); const cb = await FfiClient.instance.waitFor( - (ev) => - ev.message.case == 'byteStreamWriterClose' && ev.message.value.asyncId == res.asyncId, + (ev) => ev.message.case == 'byteStreamWriterClose' && ev.message.value.asyncId == res.asyncId, { signal: this.disconnectSignal }, ); From 7899ac972b420eb24313dc2fa0128364df9c4657 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 10:33:17 -0400 Subject: [PATCH 06/15] fix: add sdk default into doc comment --- packages/livekit-rtc/src/room.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index f2ff6674..e76bcd24 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -75,7 +75,7 @@ export interface RoomDataStreamOptions { /** * Maximum decompressed payload size in bytes accepted for a single incoming * data stream. Incoming streams exceeding this limit terminate with an error - * on the receiving side. Unset falls back to the SDK default. + * on the receiving side. Unset falls back to the SDK default of 5gb. */ maxPayloadByteLength?: number; } From d93b835f3e507ab9ac60cc5961e3c1af098475ac Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 11:02:08 -0400 Subject: [PATCH 07/15] feat: add initial e2e tests --- packages/livekit-rtc/scripts/run-e2e.mjs | 3 +- packages/livekit-rtc/src/tests/e2e_common.ts | 149 ++++++++ .../src/tests/e2e_data_streams.test.ts | 335 ++++++++++++++++++ 3 files changed, 486 insertions(+), 1 deletion(-) create mode 100644 packages/livekit-rtc/src/tests/e2e_common.ts create mode 100644 packages/livekit-rtc/src/tests/e2e_data_streams.test.ts diff --git a/packages/livekit-rtc/scripts/run-e2e.mjs b/packages/livekit-rtc/scripts/run-e2e.mjs index c92f41b4..e6821010 100644 --- a/packages/livekit-rtc/scripts/run-e2e.mjs +++ b/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, diff --git a/packages/livekit-rtc/src/tests/e2e_common.ts b/packages/livekit-rtc/src/tests/e2e_common.ts new file mode 100644 index 00000000..040a1ded --- /dev/null +++ b/packages/livekit-rtc/src/tests/e2e_common.ts @@ -0,0 +1,149 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { AccessToken } from 'livekit-server-sdk'; +import { randomUUID } from 'node:crypto'; +import { setTimeout as delay } from 'node:timers/promises'; +import { describe } from 'vitest'; +import { Room, type RoomEvent } from '../index.js'; + +export const hasE2EEnv = + !!process.env.LIVEKIT_URL && !!process.env.LIVEKIT_API_KEY && !!process.env.LIVEKIT_API_SECRET; +// Explicitly typed so declaration emit doesn't reference vitest's +// non-exported internal suite types (TS4023). +export const describeE2E: typeof describe = hasE2EEnv + ? describe + : (describe.skip as typeof describe); +export const testTimeoutMs = 10_000; + +export type TestEnv = { + url: string; + apiKey: string; + apiSecret: string; +}; + +function normalizeLiveKitUrl(url: string): string { + if (url.startsWith('http://')) return `ws://${url.slice('http://'.length)}`; + if (url.startsWith('https://')) return `wss://${url.slice('https://'.length)}`; + return url; +} + +export function getTestEnv(): TestEnv { + if (!hasE2EEnv) { + throw new Error( + 'Missing required env vars for e2e: LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET', + ); + } + return { + url: normalizeLiveKitUrl(process.env.LIVEKIT_URL!), + apiKey: process.env.LIVEKIT_API_KEY!, + apiSecret: process.env.LIVEKIT_API_SECRET!, + }; +} + +export async function withTimeout( + promise: Promise, + timeoutMs: number, + message: string, +): Promise { + return await Promise.race([ + promise, + (async () => { + await delay(timeoutMs); + throw new Error(message); + })(), + ]); +} + +export async function waitFor( + condition: () => boolean, + opts: { timeoutMs: number; intervalMs?: number; debugName?: string }, +): Promise { + const intervalMs = opts.intervalMs ?? 50; + const deadline = Date.now() + opts.timeoutMs; + while (Date.now() < deadline) { + if (condition()) return; + await delay(intervalMs); + } + throw new Error(`Timed out waiting for condition${opts.debugName ? ` (${opts.debugName})` : ''}`); +} + +export async function createJoinToken(params: { + env: TestEnv; + roomName: string; + identity: string; + name: string; +}): Promise { + const token = new AccessToken(params.env.apiKey, params.env.apiSecret, { + identity: params.identity, + name: params.name, + ttl: '30m', + }); + token.addGrant({ + room: params.roomName, + roomJoin: true, + roomCreate: true, + canPublish: true, + canSubscribe: true, + }); + return await token.toJwt(); +} + +export async function connectTestRooms( + count: number, +): Promise<{ roomName: string; rooms: Room[] }> { + const env = getTestEnv(); + const roomName = `test_room_${randomUUID()}`; + const rooms = await Promise.all( + Array.from({ length: count }, async (_, i) => { + const token = await createJoinToken({ + env, + roomName, + identity: `p${i}`, + name: `Participant ${i}`, + }); + const room = new Room(); + await room.connect(env.url, token, { autoSubscribe: true, dynacast: false }); + return room; + }), + ); + + const start = Date.now(); + await waitFor(() => rooms.every((r) => r.remoteParticipants.size === count - 1), { + timeoutMs: 5000, + debugName: `participant visibility (${Date.now() - start}ms)`, + }); + + return { roomName, rooms }; +} + +export function waitForRoomEvent( + room: Room, + event: RoomEvent, + timeoutMs: number, + take: (...args: any[]) => R, +): Promise { + return withTimeout( + new Promise((resolve) => { + const handler = (...args: any[]) => { + // typed-emitter doesn't expose `.once` in the type surface, so do manual once+cleanup. + room.off(event as any, handler as any); + resolve(take(...args)); + }; + room.on(event as any, handler as any); + }), + timeoutMs, + `Timed out waiting for ${event}`, + ); +} + +export function concatUint8(chunks: Uint8Array[]): Uint8Array { + const len = chunks.reduce((acc, c) => acc + c.byteLength, 0); + const out = new Uint8Array(len); + let offset = 0; + for (const c of chunks) { + out.set(c, offset); + offset += c.byteLength; + } + return out; +} diff --git a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts new file mode 100644 index 00000000..6da8dad9 --- /dev/null +++ b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts @@ -0,0 +1,335 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +// +// End-to-end tests for data streams (text/byte streams over the FFI-backed +// rust implementation). Ported from rust-sdks `livekit/tests/data_stream_test.rs` +// where the node API surface allows; see comments on individual tests for +// intentional deviations. +import { afterAll, expect, it as itRaw } from 'vitest'; +import { dispose } from '../index.js'; +import { + concatUint8, + connectTestRooms, + describeE2E, + testTimeoutMs, + withTimeout, +} from './e2e_common.js'; + +// use concurrent testing if available on the runner (currently not supported by bun's api) +const it = typeof itRaw.concurrent === 'function' ? itRaw.concurrent : itRaw; + +/** Deterministic pseudo-random lowercase text (mulberry32 PRNG). + * + * Counterpart of rust's `pseudo_random_text`: random lowercase carries + * ~4.7 bits of entropy per 8-bit byte, so deflate compresses it well under + * its raw size — exercising the chunked-compressed wire path. */ +function pseudoRandomText(length: number, seed = 0x5eed): string { + let a = seed >>> 0; + const next = () => { + a = (a + 0x6d2b79f5) >>> 0; + let t = a; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; + const chars = new Array(length); + for (let i = 0; i < length; i++) { + chars[i] = String.fromCharCode(97 + Math.floor(next() * 26)); + } + return chars.join(''); +} + +/** Deterministic pseudo-random bytes (mulberry32 PRNG). + * + * Uniform random bytes are genuinely incompressible: deflate cannot shrink + * them, so the send path must fall back to an uncompressed wire format. + * Seeded for a deterministic, reproducible test. */ +function pseudoRandomBytes(length: number, seed = 0xc0ffee): Uint8Array { + let a = seed >>> 0; + const out = new Uint8Array(length); + for (let i = 0; i < length; i++) { + a = (a + 0x6d2b79f5) >>> 0; + let t = a; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + out[i] = (t ^ (t >>> 14)) & 0xff; + } + return out; +} + +describeE2E('livekit-rtc data streams e2e', () => { + afterAll(async () => { + await dispose(); + }); + + // Port of rust `test_send_text`. The `is_compressed` / `is_inline` info + // assertions are not portable: the FFI stream-info protos don't expose them. + it( + 'sends and receives a small text stream', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const senderIdentity = sendingRoom!.localParticipant!.identity; + + const topic = 'some-topic'; + const textToSend = 'some-text'; + + const receivedText = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader, sender) => { + expect(sender.identity).toBe(senderIdentity); + resolve(await reader.readAll()); + }); + }), + testTimeoutMs, + 'Timed out waiting for text stream', + ); + + const textInfo = await sendingRoom!.localParticipant!.sendText(textToSend, { topic }); + expect(textInfo.streamId).toBeTruthy(); + expect(Math.abs(textInfo.timestamp - Date.now())).toBeLessThanOrEqual(1_000); + expect(textInfo.totalSize).toBe(textToSend.length); + expect(textInfo.mimeType).toBe('text/plain'); + expect(textInfo.topic).toBe(topic); + + expect(await receivedText).toBe(textToSend); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + // Port of rust `test_send_bytes`. Node has no `sendBytes` (in-memory one-shot) + // wrapper yet, so this uses the incremental `streamBytes` writer — which by + // design is never inlined or compressed on the wire. + it( + 'sends and receives a small byte stream', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const senderIdentity = sendingRoom!.localParticipant!.identity; + + const topic = 'some-topic'; + const bytesToSend = new Uint8Array(16).fill(0xfa); + + const receivedBytes = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerByteStreamHandler(topic, async (reader, sender) => { + expect(sender.identity).toBe(senderIdentity); + const chunks = await reader.readAll(); + resolve(concatUint8(chunks)); + }); + }), + testTimeoutMs, + 'Timed out waiting for byte stream', + ); + + const writer = await sendingRoom!.localParticipant!.streamBytes({ + topic, + totalSize: bytesToSend.byteLength, + }); + await writer.write(bytesToSend); + await writer.close(); + + const byteInfo = writer.info; + expect(byteInfo.streamId).toBeTruthy(); + expect(Math.abs(byteInfo.timestamp - Date.now())).toBeLessThanOrEqual(1_000); + expect(byteInfo.mimeType).toBe('application/octet-stream'); + expect(byteInfo.topic).toBe(topic); + + expect(await receivedBytes).toEqual(bytesToSend); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + // Port of rust `test_send_large_compressible_text`: ~50 KB of deterministic + // pseudo-random lowercase is too big to inline and compresses well under its + // raw size, so the sender emits chunked deflate-raw and the receiver + // decompresses — validating the v2 compression path on the real wire. + it( + 'round-trips a large compressible text (chunked + compressed wire path)', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + + const topic = 'large-compressible-text'; + const text = pseudoRandomText(50_000); + + const receivedText = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader) => { + resolve(await reader.readAll()); + }); + }), + testTimeoutMs, + 'Timed out waiting for large text stream', + ); + + const info = await sendingRoom!.localParticipant!.sendText(text, { topic }); + expect(info.totalSize).toBe(text.length); + + expect(await receivedText).toBe(text); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + // Port of rust `test_send_large_incompressible_random_bytes`, adapted to the + // incremental `streamBytes` writer (no `sendBytes` wrapper in node yet): + // uniform random bytes exercise the chunked, uncompressed wire path with a + // payload spanning many chunks. + it( + 'round-trips large incompressible random bytes', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + + const topic = 'large-random-bytes'; + const payload = pseudoRandomBytes(1_800_000); + + const receivedBytes = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerByteStreamHandler(topic, async (reader) => { + resolve(concatUint8(await reader.readAll())); + }); + }), + testTimeoutMs * 2, + 'Timed out waiting for large byte stream', + ); + + const writer = await sendingRoom!.localParticipant!.streamBytes({ + topic, + totalSize: payload.byteLength, + }); + const writeChunkSize = 64 * 1024; + for (let offset = 0; offset < payload.byteLength; offset += writeChunkSize) { + await writer.write(payload.subarray(offset, offset + writeChunkSize)); + } + await writer.close(); + + const received = await receivedBytes; + expect(received.byteLength).toBe(payload.byteLength); + expect(received).toEqual(payload); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs * 3, + ); + + // Port of rust `test_send_large_bytes`: a 50 KB patterned (compressible) + // payload via the byte-stream path. + it( + 'round-trips large patterned bytes', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + + const topic = 'large-patterned-bytes'; + const payload = new Uint8Array(50_000); + for (let i = 0; i < payload.length; i++) payload[i] = i % 251; + + const receivedBytes = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerByteStreamHandler(topic, async (reader) => { + resolve(concatUint8(await reader.readAll())); + }); + }), + testTimeoutMs, + 'Timed out waiting for patterned byte stream', + ); + + const writer = await sendingRoom!.localParticipant!.streamBytes({ + topic, + totalSize: payload.byteLength, + }); + await writer.write(payload); + await writer.close(); + + expect(await receivedBytes).toEqual(payload); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + // Incremental writer path for text (multi-write, unicode content): ensures + // UTF-8-aware chunking on the sender doesn't split multi-byte characters. + it( + 'round-trips an incremental text stream with multi-byte characters', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + + const topic = 'incremental-unicode-text'; + const pieces = ['héllo wörld — ', '日本語のテキスト、', '🌍🚀 emoji tail']; + const expected = pieces.join(''); + + const receivedText = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader) => { + resolve(await reader.readAll()); + }); + }), + testTimeoutMs, + 'Timed out waiting for incremental text stream', + ); + + const writer = await sendingRoom!.localParticipant!.streamText({ topic }); + for (const piece of pieces) { + await writer.write(piece); + } + await writer.close(); + + expect(await receivedText).toBe(expected); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + // Moved from e2e.test.ts (kept alongside the other data-stream coverage). + it( + 'cleans up stream controllers when disconnecting during an active stream', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const topic = 'cleanup-stream-topic'; + + // Register a handler on the receiving side that will intentionally + // NOT fully consume the stream — simulating an abandoned transfer. + let readerReceived = false; + // eslint-disable-next-line @typescript-eslint/no-unused-vars + receivingRoom!.registerTextStreamHandler(topic, async (_reader, _sender) => { + readerReceived = true; + // Deliberately do not call reader.readAll() so the stream stays open + }); + + // Start sending a text stream but don't close it + const writer = await sendingRoom!.localParticipant!.streamText({ topic }); + await writer.write('partial data'); + + // Wait for the receiving side to get the stream header + const deadline = Date.now() + 5000; + while (!readerReceived && Date.now() < deadline) { + await new Promise((r) => setTimeout(r, 50)); + } + expect(readerReceived).toBe(true); + + // Disconnect the receiving room while the stream is still open. + // This should close the stream controller without throwing. + await receivingRoom!.disconnect(); + + // Also close the writer and disconnect the sender + await writer.close(); + await sendingRoom!.disconnect(); + + // If we got here without hanging or throwing, the stream controller + // was properly cleaned up on disconnect. + }, + testTimeoutMs, + ); +}); From bcf014acd5c9850cd231ae288e8e74c0b8d0eb32 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 11:21:25 -0400 Subject: [PATCH 08/15] fix: switch e2e test prng mechanism to something more conventional --- .../src/tests/e2e_data_streams.test.ts | 28 +++++-------------- 1 file changed, 7 insertions(+), 21 deletions(-) diff --git a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts index 6da8dad9..a2ed3c6f 100644 --- a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts +++ b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts @@ -19,41 +19,27 @@ import { // use concurrent testing if available on the runner (currently not supported by bun's api) const it = typeof itRaw.concurrent === 'function' ? itRaw.concurrent : itRaw; -/** Deterministic pseudo-random lowercase text (mulberry32 PRNG). +/** Pseudo-random lowercase text. * * Counterpart of rust's `pseudo_random_text`: random lowercase carries * ~4.7 bits of entropy per 8-bit byte, so deflate compresses it well under * its raw size — exercising the chunked-compressed wire path. */ -function pseudoRandomText(length: number, seed = 0x5eed): string { - let a = seed >>> 0; - const next = () => { - a = (a + 0x6d2b79f5) >>> 0; - let t = a; - t = Math.imul(t ^ (t >>> 15), t | 1); - t ^= t + Math.imul(t ^ (t >>> 7), t | 61); - return ((t ^ (t >>> 14)) >>> 0) / 4294967296; - }; +function pseudoRandomText(length: number): string { const chars = new Array(length); for (let i = 0; i < length; i++) { - chars[i] = String.fromCharCode(97 + Math.floor(next() * 26)); + chars[i] = String.fromCharCode(97 + Math.floor(Math.random() * 26)); } return chars.join(''); } -/** Deterministic pseudo-random bytes (mulberry32 PRNG). +/** Pseudo-random bytes. * * Uniform random bytes are genuinely incompressible: deflate cannot shrink - * them, so the send path must fall back to an uncompressed wire format. - * Seeded for a deterministic, reproducible test. */ -function pseudoRandomBytes(length: number, seed = 0xc0ffee): Uint8Array { - let a = seed >>> 0; + * them, so the send path must fall back to an uncompressed wire format. */ +function pseudoRandomBytes(length: number): Uint8Array { const out = new Uint8Array(length); for (let i = 0; i < length; i++) { - a = (a + 0x6d2b79f5) >>> 0; - let t = a; - t = Math.imul(t ^ (t >>> 15), t | 1); - t ^= t + Math.imul(t ^ (t >>> 7), t | 61); - out[i] = (t ^ (t >>> 14)) & 0xff; + out[i] = Math.floor(Math.random() * 256); } return out; } From f496f5d8af56dfbeee1cb1e9e01a1833fcb6ca29 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 11:22:25 -0400 Subject: [PATCH 09/15] fix: drop Date.now() dependencies in e2e tests --- packages/livekit-rtc/src/tests/e2e_data_streams.test.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts index a2ed3c6f..f23e108b 100644 --- a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts +++ b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts @@ -74,7 +74,6 @@ describeE2E('livekit-rtc data streams e2e', () => { const textInfo = await sendingRoom!.localParticipant!.sendText(textToSend, { topic }); expect(textInfo.streamId).toBeTruthy(); - expect(Math.abs(textInfo.timestamp - Date.now())).toBeLessThanOrEqual(1_000); expect(textInfo.totalSize).toBe(textToSend.length); expect(textInfo.mimeType).toBe('text/plain'); expect(textInfo.topic).toBe(topic); @@ -120,7 +119,6 @@ describeE2E('livekit-rtc data streams e2e', () => { const byteInfo = writer.info; expect(byteInfo.streamId).toBeTruthy(); - expect(Math.abs(byteInfo.timestamp - Date.now())).toBeLessThanOrEqual(1_000); expect(byteInfo.mimeType).toBe('application/octet-stream'); expect(byteInfo.topic).toBe(topic); From 699b94baef33cf930cbd46255566caacb99af22d Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 11:33:55 -0400 Subject: [PATCH 10/15] fix: ensure that progress measurements are tracked in bytes always --- .../src/data_streams/stream_reader.ts | 28 +++++++++---------- 1 file changed, 13 insertions(+), 15 deletions(-) diff --git a/packages/livekit-rtc/src/data_streams/stream_reader.ts b/packages/livekit-rtc/src/data_streams/stream_reader.ts index e6ffd63f..99eb80ae 100644 --- a/packages/livekit-rtc/src/data_streams/stream_reader.ts +++ b/packages/livekit-rtc/src/data_streams/stream_reader.ts @@ -26,7 +26,13 @@ abstract class BaseStreamReader { 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 { * A class to read chunks from a ReadableStream and provide them in a structured format. */ export class ByteStreamReader extends BaseStreamReader { - 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(); @@ -102,9 +100,9 @@ export class TextStreamReader extends BaseStreamReader { constructor( info: TextStreamInfo, stream: ReadableStream, - totalChunkCount?: number, + totalByteSize?: number, ) { - super(info, stream, totalChunkCount); + super(info, stream, totalByteSize); this.receivedChunks = new Map(); } @@ -116,10 +114,10 @@ export class TextStreamReader extends BaseStreamReader { 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); } /** From a6f8a35c15c39c2b5d3514662a804508350311e7 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 11:36:18 -0400 Subject: [PATCH 11/15] fix: ensure that abandoned data stream writers are closed on room disconnect --- packages/livekit-rtc/src/participant.ts | 116 ++++++++++++++++++------ packages/livekit-rtc/src/room.ts | 5 + 2 files changed, 93 insertions(+), 28 deletions(-) diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index b2755237..af60164a 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -174,6 +174,14 @@ export class LocalParticipant extends Participant { // pending FfiClient.waitFor() listeners so they don't leak. private disconnectSignal: AbortSignal; + // Handle ids of data stream writers that have been opened but not yet + // closed. The FFI close request consumes the handle (take_handle), so an + // entry is removed as soon as a close is sent for it; whatever is left when + // the room disconnects is dropped by disposeOpenStreamWriters(). Only the + // ids are kept — wrapping them in FfiHandle would make GC drop a handle the + // close request already consumed. + private openStreamWriters = new Set(); + trackPublications: Map = new Map(); constructor(info: OwnedParticipant, ffiEventLock: Mutex, disconnectSignal: AbortSignal) { @@ -312,28 +320,44 @@ export class LocalParticipant extends Participant { const writerHandle = cb.result.value.handle!.id!; const info = textStreamInfoFromProto(cb.result.value.info!); - const writeText = this.textStreamWriterWrite; - const closeWriter = this.textStreamWriterClose; + this.openStreamWriters.add(writerHandle); + + // Sends the close request at most once: it consumes the writer handle on + // the native side, so a second one would fail and, worse, the handle must + // no longer be dropped by disconnect cleanup. + const closeWriter = async (reason?: string) => { + if (!this.openStreamWriters.delete(writerHandle)) { + return; + } + await this.textStreamWriterClose(new TextStreamWriterCloseRequest({ writerHandle, reason })); + }; const writableStream = new WritableStream({ // Implement the sink - async write(text) { - await writeText(new TextStreamWriterWriteRequest({ writerHandle, text })); - }, - async close() { - await closeWriter(new TextStreamWriterCloseRequest({ writerHandle })); + write: async (text) => { + try { + await this.textStreamWriterWrite( + new TextStreamWriterWriteRequest({ writerHandle, text }), + ); + } catch (err) { + // A rejected write leaves the writable stream errored, and an errored + // stream never runs the sink's close()/abort() — a later close() just + // rejects with this error. Close the native writer here so its handle + // isn't held for the lifetime of the process and peers get an + // end-of-stream instead of a stream that never terminates. + await closeWriter(err instanceof Error ? err.message : String(err ?? '')).catch(() => { + // Best-effort: the connection may already be gone. + }); + throw err; + } }, + close: () => closeWriter(), // Close the stream with the error reason so the remote side's stream // controller is closed instead of waiting for data that won't arrive. - async abort(err) { + abort: async (err) => { log.error(err, 'Sink Error'); try { - await closeWriter( - new TextStreamWriterCloseRequest({ - writerHandle, - reason: err instanceof Error ? err.message : String(err ?? ''), - }), - ); + await closeWriter(err instanceof Error ? err.message : String(err ?? '')); } catch { // Best-effort: the connection may already be gone. } @@ -427,27 +451,43 @@ export class LocalParticipant extends Participant { const writerHandle = cb.result.value.handle!.id!; const info = byteStreamInfoFromProto(cb.result.value.info!); - const writeBytes = this.byteStreamWriterWrite; - const closeWriter = this.byteStreamWriterClose; + this.openStreamWriters.add(writerHandle); + + // Sends the close request at most once: it consumes the writer handle on + // the native side, so a second one would fail and, worse, the handle must + // no longer be dropped by disconnect cleanup. + const closeWriter = async (reason?: string) => { + if (!this.openStreamWriters.delete(writerHandle)) { + return; + } + await this.byteStreamWriterClose(new ByteStreamWriterCloseRequest({ writerHandle, reason })); + }; const writableStream = new WritableStream({ - async write(chunk) { - await writeBytes(new ByteStreamWriterWriteRequest({ writerHandle, bytes: chunk })); - }, - async close() { - await closeWriter(new ByteStreamWriterCloseRequest({ writerHandle })); + write: async (chunk) => { + try { + await this.byteStreamWriterWrite( + new ByteStreamWriterWriteRequest({ writerHandle, bytes: chunk }), + ); + } catch (err) { + // A rejected write leaves the writable stream errored, and an errored + // stream never runs the sink's close()/abort() — a later close() just + // rejects with this error. Close the native writer here so its handle + // isn't held for the lifetime of the process and peers get an + // end-of-stream instead of a stream that never terminates. + await closeWriter(err instanceof Error ? err.message : String(err ?? '')).catch(() => { + // Best-effort: the connection may already be gone. + }); + throw err; + } }, + close: () => closeWriter(), // Close the stream with the error reason so the remote side's stream // controller is closed instead of waiting for data that won't arrive. - async abort(err) { + abort: async (err) => { log.error(err, 'Sink error'); try { - await closeWriter( - new ByteStreamWriterCloseRequest({ - writerHandle, - reason: err instanceof Error ? err.message : String(err ?? ''), - }), - ); + await closeWriter(err instanceof Error ? err.message : String(err ?? '')); } catch { // Best-effort: the connection may already be gone. } @@ -551,6 +591,26 @@ export class LocalParticipant extends Participant { } }; + /** + * Drops the native writers of any data streams that were opened but never + * closed — a writer the caller abandoned, or one whose write failed. Their + * handles would otherwise be held by the FFI server for the lifetime of the + * process. The room is already gone at this point, so the handles are dropped + * directly rather than closed: there is no end-of-stream left to deliver. + * + * @internal + */ + disposeOpenStreamWriters() { + for (const writerHandle of this.openStreamWriters) { + try { + new FfiHandle(writerHandle).dispose(); + } catch (err) { + log.warn('failed to dispose data stream writer handle: %s', err); + } + } + this.openStreamWriters.clear(); + } + /** * Sends a chat message to participants in the room * diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index e76bcd24..c7df850a 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -472,6 +472,11 @@ export class Room extends (EventEmitter as new () => TypedEmitter // listener until GC. setRoom(null) is idempotent, so this is safe even if a // track was already detached (e.g. via unsubscribe/unpublish). if (this.localParticipant) { + // Data stream writers that were never closed (abandoned by the caller, or + // left errored by a failed write) keep their native writer alive until + // their handle is dropped. + this.localParticipant.disposeOpenStreamWriters(); + for (const pub of this.localParticipant.trackPublications.values()) { pub.track?.setRoom(null); } From 592699eab5a5800a48eda9b9bf7941670fc74dec Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 11:37:18 -0400 Subject: [PATCH 12/15] Update packages/livekit-rtc/src/utils.ts Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- packages/livekit-rtc/src/utils.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts index 090fabc0..1c274d59 100644 --- a/packages/livekit-rtc/src/utils.ts +++ b/packages/livekit-rtc/src/utils.ts @@ -53,7 +53,7 @@ export function textStreamInfoFromProto(info: ProtoTextStreamInfo): TextStreamIn mimeType: info.mimeType!, topic: info.topic!, timestamp: bigIntToNumber(info.timestamp!), - totalSize: info.totalLength !== undefined ? bigIntToNumber(info.totalLength) : undefined, + totalSize: info.totalLength ? bigIntToNumber(info.totalLength) : undefined, attributes: info.attributes, }; } From a32e2e6a0a8ae0c752c4e5bb9adb57ce3f328a15 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 11:41:04 -0400 Subject: [PATCH 13/15] fix: update structure of makeLocalParticipant mock to take into account a6f8a35 --- packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts b/packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts index 0ba4d163..fb680874 100644 --- a/packages/livekit-rtc/src/audio_stream_room_lifecycle.test.ts +++ b/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; } From bf7b381186451879cf24f84759877a00e85430b3 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 15:11:40 -0400 Subject: [PATCH 14/15] fix: drop receivedChunks map and add .close() method for manual early close --- .../src/data_streams/stream_reader.ts | 113 +++++++++++------- packages/livekit-rtc/src/room.ts | 59 ++++++--- 2 files changed, 112 insertions(+), 60 deletions(-) diff --git a/packages/livekit-rtc/src/data_streams/stream_reader.ts b/packages/livekit-rtc/src/data_streams/stream_reader.ts index 99eb80ae..d06c14c7 100644 --- a/packages/livekit-rtc/src/data_streams/stream_reader.ts +++ b/packages/livekit-rtc/src/data_streams/stream_reader.ts @@ -3,7 +3,6 @@ // SPDX-License-Identifier: Apache-2.0 import type { DataStream_Chunk } from '@livekit/rtc-ffi-bindings'; import { log } from '../log.js'; -import { bigIntToNumber } from '../utils.js'; import type { BaseStreamInfo, ByteStreamInfo, TextStreamInfo } from './types.js'; abstract class BaseStreamReader { @@ -15,6 +14,12 @@ abstract class BaseStreamReader { protected bytesReceived: number; + private closed = false; + + // The reader held by an in-progress iteration. Cancelling has to go through + // it, since it holds the stream's lock. + private activeReader: ReadableStreamDefaultReader | null = null; + get info() { return this._info; } @@ -34,6 +39,57 @@ abstract class BaseStreamReader { this.onProgress?.(currentProgress); } + /** Takes the stream's reader for an iteration, remembering it so that + * close() can cancel a read that is still in flight. */ + protected acquireReader(): ReadableStreamDefaultReader { + const reader = this.reader.getReader(); + this.activeReader = reader; + return reader; + } + + /** Releases the stream lock after an iteration ended on its own — at + * end-of-stream or on a stream error. The FFI subscription behind the stream + * was already dropped by that terminal event, so there is nothing to cancel. */ + protected finishIteration(reader: ReadableStreamDefaultReader) { + this.closed = true; + if (this.activeReader === reader) { + this.activeReader = null; + } + reader.releaseLock(); + } + + /** + * Stops receiving this stream and releases the resources behind it. + * + * A reader is subscribed to FFI events from the moment it is handed to a + * stream handler, and only unsubscribes once a read consumes the stream's + * end-of-stream event. So a reader that is abandoned part-way through, or + * that a handler never reads at all, has to be closed here or its + * subscription lives for the rest of the process. Reading to completion — or + * breaking out of a `for await` — closes the reader for you, and closing an + * already-closed reader is a no-op. + */ + async close(): Promise { + if (this.closed) { + return; + } + this.closed = true; + const active = this.activeReader; + this.activeReader = null; + try { + if (active) { + await active.cancel(); + active.releaseLock(); + } else { + await this.reader.cancel(); + } + } catch (error: unknown) { + // The stream was already closed or errored (e.g. the room disconnected + // mid-stream), in which case its subscription is already gone. + log.debug('error closing stream reader: %s', error); + } + } + onProgress?: (progress: number | undefined) => void; abstract readAll(): Promise>; @@ -44,7 +100,7 @@ abstract class BaseStreamReader { */ export class ByteStreamReader extends BaseStreamReader { [Symbol.asyncIterator]() { - const reader = this.reader.getReader(); + const reader = this.acquireReader(); return { next: async (): Promise> => { @@ -53,7 +109,7 @@ export class ByteStreamReader extends BaseStreamReader { if (done) { // Release the lock when the stream is exhausted so the // underlying ReadableStream can be garbage-collected. - reader.releaseLock(); + this.finishIteration(reader); return { done: true, value: undefined as unknown }; } else { this.handleChunkReceived(value); @@ -62,7 +118,7 @@ export class ByteStreamReader extends BaseStreamReader { } catch (error: unknown) { // Release the lock on error so it doesn't stay held when the // consumer never calls return() (e.g. breaking out of for-await). - reader.releaseLock(); + this.finishIteration(reader); log.error('error processing stream update: %s', error); // Propagate abnormal termination (e.g. remote abort, payload over // the receiver's size limit) instead of presenting the truncated @@ -71,8 +127,8 @@ export class ByteStreamReader extends BaseStreamReader { } }, - return(): IteratorResult { - reader.releaseLock(); + return: async (): Promise> => { + await this.close(); return { done: true, value: undefined }; }, }; @@ -91,44 +147,14 @@ export class ByteStreamReader extends BaseStreamReader { * A class to read chunks from a ReadableStream and provide them in a structured format. */ export class TextStreamReader extends BaseStreamReader { - private receivedChunks: Map; - - /** - * 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, - totalByteSize?: number, - ) { - super(info, stream, totalByteSize); - 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); - // 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); - } - /** * Async iterator implementation to allow usage of `for await...of` syntax. * Yields structured chunks from the stream. * */ [Symbol.asyncIterator]() { - const reader = this.reader.getReader(); + const reader = this.acquireReader(); const decoder = new TextDecoder(); - const receivedChunks = this.receivedChunks; return { next: async (): Promise> => { @@ -137,9 +163,7 @@ export class TextStreamReader extends BaseStreamReader { if (done) { // Release the lock when the stream is exhausted so the // underlying ReadableStream can be garbage-collected. - reader.releaseLock(); - // Clear received chunks so the buffered data can be GC'd. - receivedChunks.clear(); + this.finishIteration(reader); return { done: true, value: undefined }; } else { this.handleChunkReceived(value); @@ -151,8 +175,7 @@ export class TextStreamReader extends BaseStreamReader { } catch (error: unknown) { // Release the lock on error so it doesn't stay held when the // consumer never calls return() (e.g. breaking out of for-await). - reader.releaseLock(); - receivedChunks.clear(); + this.finishIteration(reader); log.error('error processing stream update: %s', error); // Propagate abnormal termination (e.g. remote abort, payload over // the receiver's size limit) instead of presenting the truncated @@ -161,10 +184,8 @@ export class TextStreamReader extends BaseStreamReader { } }, - return(): IteratorResult { - reader.releaseLock(); - // Clear received chunks so the buffered data can be GC'd. - receivedChunks.clear(); + return: async (): Promise> => { + await this.close(); return { done: true, value: undefined }; }, }; diff --git a/packages/livekit-rtc/src/room.ts b/packages/livekit-rtc/src/room.ts index c7df850a..32dff372 100644 --- a/packages/livekit-rtc/src/room.ts +++ b/packages/livekit-rtc/src/room.ts @@ -967,10 +967,21 @@ export class Room extends (EventEmitter as new () => TypedEmitter return; } + // Assigned by start(); unsubscribe() drops it once the stream is done with, + // whether that's end-of-stream or the consumer walking away early. + let listener: ((ev: FfiEvent) => void) | null = null; + const unsubscribe = () => { + if (!listener) { + return; + } + FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); + listener = null; + this.streamReaders.delete(readerHandle); + }; + const stream = new ReadableStream({ start: (controller) => { - let nextChunkIndex = 0; - const listener = (ev: FfiEvent) => { + listener = (ev: FfiEvent) => { if ( ev.message.case !== 'byteStreamReaderEvent' || ev.message.value.readerHandle !== readerHandle @@ -981,13 +992,10 @@ export class Room extends (EventEmitter as new () => TypedEmitter if (detail.case === 'chunkReceived') { const content = detail.value.content!; if (content.length > 0) { - controller.enqueue( - new DataStream_Chunk({ content, chunkIndex: numberToBigInt(nextChunkIndex++) }), - ); + controller.enqueue(new DataStream_Chunk({ content })); } } else if (detail.case === 'eos') { - FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); - this.streamReaders.delete(readerHandle); + unsubscribe(); const error = detail.value.error; if (error) { // Abnormal termination (e.g. remote abort, payload over the @@ -1011,6 +1019,13 @@ export class Room extends (EventEmitter as new () => TypedEmitter }, }); }, + // The consumer stopped reading before end-of-stream — ByteStreamReader + // .close(), or breaking out of a for-await. The eos event that would have + // unsubscribed is never consumed, so the listener would otherwise be run + // against every FFI event for the rest of the process. Only the + // subscription is released: the reader handle was consumed by the + // readIncremental request above, so there is nothing left to dispose. + cancel: () => unsubscribe(), }); streamHandlerCallback( @@ -1033,10 +1048,22 @@ export class Room extends (EventEmitter as new () => TypedEmitter } const textEncoder = new TextEncoder(); + + // Assigned by start(); unsubscribe() drops it once the stream is done with, + // whether that's end-of-stream or the consumer walking away early. + let listener: ((ev: FfiEvent) => void) | null = null; + const unsubscribe = () => { + if (!listener) { + return; + } + FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); + listener = null; + this.streamReaders.delete(readerHandle); + }; + const stream = new ReadableStream({ start: (controller) => { - let nextChunkIndex = 0; - const listener = (ev: FfiEvent) => { + listener = (ev: FfiEvent) => { if ( ev.message.case !== 'textStreamReaderEvent' || ev.message.value.readerHandle !== readerHandle @@ -1047,13 +1074,10 @@ export class Room extends (EventEmitter as new () => TypedEmitter if (detail.case === 'chunkReceived') { const content = textEncoder.encode(detail.value.content!); if (content.length > 0) { - controller.enqueue( - new DataStream_Chunk({ content, chunkIndex: numberToBigInt(nextChunkIndex++) }), - ); + controller.enqueue(new DataStream_Chunk({ content })); } } else if (detail.case === 'eos') { - FfiClient.instance.off(FfiClientEvent.FfiEvent, listener); - this.streamReaders.delete(readerHandle); + unsubscribe(); const error = detail.value.error; if (error) { // Abnormal termination (e.g. remote abort, payload over the @@ -1077,6 +1101,13 @@ export class Room extends (EventEmitter as new () => TypedEmitter }, }); }, + // The consumer stopped reading before end-of-stream — TextStreamReader + // .close(), or breaking out of a for-await. The eos event that would have + // unsubscribed is never consumed, so the listener would otherwise be run + // against every FFI event for the rest of the process. Only the + // subscription is released: the reader handle was consumed by the + // readIncremental request above, so there is nothing left to dispose. + cancel: () => unsubscribe(), }); streamHandlerCallback( From cd8e96e27297cca0b85dc904e1887ada75a1d787 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Fri, 31 Jul 2026 15:13:44 -0400 Subject: [PATCH 15/15] feat: add more e2e test cases matching bugs fixed in the python implementation --- .../src/tests/e2e_data_streams.test.ts | 216 ++++++++++++++++++ 1 file changed, 216 insertions(+) diff --git a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts index f23e108b..3a2d2bb2 100644 --- a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts +++ b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts @@ -7,18 +7,39 @@ // where the node API surface allows; see comments on individual tests for // intentional deviations. import { afterAll, expect, it as itRaw } from 'vitest'; +import type { Room, TextStreamReader } from '../index.js'; import { dispose } from '../index.js'; import { concatUint8, connectTestRooms, describeE2E, testTimeoutMs, + waitFor, withTimeout, } from './e2e_common.js'; // use concurrent testing if available on the runner (currently not supported by bun's api) const it = typeof itRaw.concurrent === 'function' ? itRaw.concurrent : itRaw; +/** How many of this room's incoming stream readers are still subscribed to FFI + * events. + * + * Reads the room's private reader registry rather than counting FfiClient + * listeners: an entry is added when a reader subscribes and removed when it + * unsubscribes, so the count is specific to this room and unaffected by the + * other tests running concurrently. */ +function subscribedReaderCount(room: Room): number { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return ((room as any).streamReaders as Map).size; +} + +async function waitForNoSubscribedReaders(room: Room): Promise { + await waitFor(() => subscribedReaderCount(room) === 0, { + timeoutMs: 5000, + debugName: 'stream reader to unsubscribe from FFI events', + }); +} + /** Pseudo-random lowercase text. * * Counterpart of rust's `pseudo_random_text`: random lowercase carries @@ -316,4 +337,199 @@ describeE2E('livekit-rtc data streams e2e', () => { }, testTimeoutMs, ); + + it( + 'releases the FFI subscription when a reader is abandoned mid-iteration', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const topic = 'abandon-topic'; + + // Unlike python, walking out of the loop is enough: `break` runs the + // async iterator's return(), which closes the reader for us. + const abandoned = withTimeout( + new Promise<{ reader: TextStreamReader; firstChunk: string }>((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader) => { + for await (const chunk of reader) { + resolve({ reader, firstChunk: chunk }); + break; // walk away without draining to end-of-stream + } + }); + }), + testTimeoutMs, + 'Timed out waiting for the reader to be abandoned', + ); + + await sendingRoom!.localParticipant!.sendText(pseudoRandomText(50_000), { topic }); + const { reader, firstChunk } = await abandoned; + expect(firstChunk.length).toBeGreaterThan(0); + + await waitForNoSubscribedReaders(receivingRoom!); + + // Closing an already-closed reader is a no-op. + await reader.close(); + expect(subscribedReaderCount(receivingRoom!)).toBe(0); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + it( + 'releases the FFI subscription when a reader is never read', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const topic = 'noread-topic'; + + const handed = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader) => { + resolve(reader); // never read + }); + }), + testTimeoutMs, + 'Timed out waiting for the stream handler to be called', + ); + + const writer = await sendingRoom!.localParticipant!.streamText({ topic }); + await writer.write('some data'); + + const reader = await handed; + expect(subscribedReaderCount(receivingRoom!)).toBe(1); + + await reader.close(); + expect(subscribedReaderCount(receivingRoom!)).toBe(0); + await reader.close(); // idempotent + + await writer.close(); + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + it( + 'ends an in-flight read when the reader is closed mid-stream', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const topic = 'close-during-read-topic'; + + const reading = withTimeout( + new Promise<{ reader: TextStreamReader; read: Promise }>((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader) => { + // readAll() blocks on a stream that is deliberately left open. + resolve({ reader, read: reader.readAll() }); + }); + }), + testTimeoutMs, + 'Timed out waiting for the stream handler to be called', + ); + + const writer = await sendingRoom!.localParticipant!.streamText({ topic }); + await writer.write('first chunk'); + + const { reader, read } = await reading; + // Let the read consume the first chunk and block waiting for more. + await waitFor(() => subscribedReaderCount(receivingRoom!) === 1, { + timeoutMs: 5000, + debugName: 'reader to subscribe', + }); + + await reader.close(); + + // The blocked read ends rather than hanging, with what it had so far. + const text = await withTimeout(read, testTimeoutMs, 'Timed out on the in-flight read'); + expect('first chunk'.startsWith(text)).toBe(true); + expect(subscribedReaderCount(receivingRoom!)).toBe(0); + + await writer.close(); + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + + it( + 'drops the subscription of an unread reader when the room disconnects', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const topic = 'disconnect-unread-topic'; + + const handed = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader) => { + resolve(reader); // never read + }); + }), + testTimeoutMs, + 'Timed out waiting for the stream handler to be called', + ); + + const writer = await sendingRoom!.localParticipant!.streamText({ topic }); + await writer.write('buffered '); + await writer.write('data'); + + const reader = await handed; + expect(subscribedReaderCount(receivingRoom!)).toBe(1); + + await receivingRoom!.disconnect(); + expect(subscribedReaderCount(receivingRoom!)).toBe(0); + + // The read settles rather than hanging. Which way it settles is a race + // the SDK doesn't control: the native side emits a clean end-of-stream + // when the room drops the stream (livekit-ffi `read_incremental` sends + // `eos { error: None }` once the channel closes), so either that lands + // first and the read returns the chunks buffered so far, or disconnect + // cleanup errors the stream first and the read reports that. Python is + // deterministic here because it injects a synthetic error end-of-stream. + const readResult = await withTimeout( + reader.readAll().then( + (text) => ({ text }), + (err: unknown) => ({ err }), + ), + testTimeoutMs, + 'Timed out reading a reader whose room disconnected', + ); + if ('err' in readResult) { + expect(String(readResult.err)).toMatch(/Disconnected while receiving/); + } else { + expect('buffered data'.startsWith(readResult.text)).toBe(true); + } + + await writer.close(); + await sendingRoom!.disconnect(); + }, + testTimeoutMs, + ); + + it( + 'needs no close once a reader has been read to completion', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const topic = 'drain-topic'; + + const drained = withTimeout( + new Promise<{ reader: TextStreamReader; text: string }>((resolve) => { + receivingRoom!.registerTextStreamHandler(topic, async (reader) => { + resolve({ reader, text: await reader.readAll() }); + }); + }), + testTimeoutMs, + 'Timed out waiting for the text stream', + ); + + await sendingRoom!.localParticipant!.sendText('hello', { topic }); + const { reader, text } = await drained; + expect(text).toBe('hello'); + + await waitForNoSubscribedReaders(receivingRoom!); + await reader.close(); // no-op + expect(subscribedReaderCount(receivingRoom!)).toBe(0); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); });