diff --git a/.changeset/brave-crews-flash.md b/.changeset/brave-crews-flash.md new file mode 100644 index 0000000000..fb90fd2e27 --- /dev/null +++ b/.changeset/brave-crews-flash.md @@ -0,0 +1,5 @@ +--- +"livekit-client": patch +--- + +chore: extract data channel handling into manager class diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index d690e0a3fd..bcc00f4323 100644 --- a/src/room/RTCEngine.test.ts +++ b/src/room/RTCEngine.test.ts @@ -1,6 +1,6 @@ import { DataPacket, DataPacket_Kind, UserPacket } from '@livekit/protocol'; import { afterEach, describe, expect, it, vi } from 'vitest'; -import type { DataPacketBuffer, DataPacketItem } from '../utils/dataPacketBuffer'; +import type { DataPacketBuffer } from '../utils/dataPacketBuffer'; import RTCEngine, { DataChannelKind } from './RTCEngine'; import { roomOptionDefaults } from './defaults'; import { PublishDataError, UnexpectedConnectionState } from './errors'; @@ -228,17 +228,16 @@ describe('RTCEngine', () => { engine: RTCEngine, maxDataPacketSize: number = MAX_DATA_PACKET_SIZE, ) { - const send = vi.fn(); + const dc = new FakeDataChannel(); Object.assign(engine as unknown as Record, { + _isClosed: false, ensurePublisherConnected: vi.fn().mockResolvedValue(undefined), - waitForBufferHeadroomWithLock: vi.fn().mockResolvedValue(undefined), - updateAndEmitDCBufferStatus: vi.fn(), - dataChannelForKind: vi.fn(() => ({ send })), pcManager: { getMaxPublisherMessageSize: vi.fn(() => maxDataPacketSize), }, }); - return send; + attachFakeChannel(engine, 'reliableChannel', dc); + return dc.send; } it('rejects packets larger than the max data packet size', async () => { @@ -309,6 +308,23 @@ describe('RTCEngine', () => { const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + /** The reliable channel's private replay buffer — reached through casts, as tests do for engine privates. */ + const reliableBuffer = (engine: RTCEngine) => + (engine as unknown as { reliableChannel: { messageBuffer: DataPacketBuffer } }).reliableChannel + .messageBuffer; + + type ChannelField = 'reliableChannel' | 'lossyChannel' | 'dataTrackChannel'; + const engineChannel = (engine: RTCEngine, field: ChannelField) => + ( + engine as unknown as Record< + ChannelField, + { attach(dc: RTCDataChannel): void; invalidateWaiters(reason: string): void } + > + )[field]; + /** Attach a fake handle to one of the engine's flow-control wrappers. */ + const attachFakeChannel = (engine: RTCEngine, field: ChannelField, dc: FakeDataChannel) => + engineChannel(engine, field).attach(dc as unknown as RTCDataChannel); + describe('resendReliableMessagesForResume', () => { it('does not let a concurrent reliable send interleave into the resume replay', async () => { const engine = new RTCEngine(roomOptionDefaults); @@ -316,18 +332,17 @@ describe('RTCEngine', () => { Object.assign(engine as unknown as Record, { _isClosed: false, ensurePublisherConnected: vi.fn().mockResolvedValue(undefined), - dataChannelForKind: vi.fn(() => dc), pcManager: { getMaxPublisherMessageSize: vi.fn(() => 64 * 1024 - 1), }, }); + attachFakeChannel(engine, 'reliableChannel', dc); - // Two messages queued for replay, and a full buffer so the XX its first send. + // Two messages queued for replay, and a full buffer so the replay parks on + // waitForBufferHeadroom before its first send. const replayed1 = new Uint8Array([1]); const replayed2 = new Uint8Array([2]); - const buffer = ( - engine as unknown as { reliableMessageBuffer: { push: (item: DataPacketItem) => void } } - ).reliableMessageBuffer; + const buffer = reliableBuffer(engine); buffer.push({ data: replayed1, sequence: 1, sent: true }); buffer.push({ data: replayed2, sequence: 2, sent: true }); dc.bufferedAmount = 2 * 1024 * 1024; // above the reliable high-water mark @@ -359,59 +374,6 @@ describe('RTCEngine', () => { expect(dc.send.mock.calls[2][0]).not.toBe(replayed1); expect(dc.send.mock.calls[2][0]).not.toBe(replayed2); }); - - it('transmits a packet deferred mid-replay instead of marking it sent without sending', async () => { - const engine = new RTCEngine(roomOptionDefaults); - const dc = new FakeDataChannel(); - Object.assign(engine as unknown as Record, { - _isClosed: false, - // Reconnect still active: a send arriving during replay takes the deferral path. - attemptingReconnect: true, - ensurePublisherConnected: vi.fn().mockResolvedValue(undefined), - dataChannelForKind: vi.fn(() => dc), - pcManager: { - getMaxPublisherMessageSize: vi.fn(() => 64 * 1024 - 1), - }, - }); - - const buffer = (engine as unknown as { reliableMessageBuffer: DataPacketBuffer }) - .reliableMessageBuffer; - const replayed = new Uint8Array([1]); - buffer.push({ data: replayed, sequence: 1, sent: true }); - // Full buffer so replay parks on waitForBufferHeadroomWithLock before its first send. - dc.bufferedAmount = 2 * 1024 * 1024; - - const replay = ( - engine as unknown as { resendReliableMessagesForResume: (seq: number) => Promise } - ).resendReliableMessagesForResume(0); - await tick(); - - // A reliable send arrives while replay is parked; attemptingReconnect defers it into the - // buffer as sent:false, after the replay's first drain pass already started. - const deferred = new Uint8Array([2]); - const deferredSend = engine.sendDataPacket( - new DataPacket({ - kind: DataPacket_Kind.RELIABLE, - value: { case: 'user', value: new UserPacket({ payload: deferred }) }, - }), - DataChannelKind.RELIABLE, - ); - await tick(); - - dc.bufferedAmount = 0; - dc.dispatchEvent(new Event('bufferedamountlow')); - await Promise.all([replay, deferredSend]); - - // Pre-fix, replay sends only its snapshot ([replayed]) and blanket-marks the deferred packet - // sent without ever transmitting it — one send. The drain loop must transmit both (the - // deferred packet is serialized through sendDataPacket, so it's distinct from `replayed`), - // leaving nothing unsent for a later align to strand. - const sent = dc.send.mock.calls.map(([d]) => d); - expect(sent).toHaveLength(2); - expect(sent[0]).toBe(replayed); - expect(sent[1]).not.toBe(replayed); - expect(buffer.getUnsent()).toHaveLength(0); - }); }); describe('reliable sends during teardown windows', () => { @@ -425,13 +387,12 @@ describe('RTCEngine', () => { Object.assign(engine as unknown as Record, { _isClosed: false, ensurePublisherConnected: vi.fn().mockResolvedValue(undefined), - dataChannelForKind: vi.fn(() => dc), pcManager: { getMaxPublisherMessageSize: vi.fn(() => 64 * 1024 - 1), }, }); - return (engine as unknown as { reliableMessageBuffer: DataPacketBuffer }) - .reliableMessageBuffer; + attachFakeChannel(engine, 'reliableChannel', dc); + return reliableBuffer(engine); } it('resolves and queues the packet for replay when the wait is torn down transiently', async () => { @@ -443,9 +404,7 @@ describe('RTCEngine', () => { dc.bufferedAmount = 2 * 1024 * 1024; const send = engine.sendDataPacket(makePacket(1), DataChannelKind.RELIABLE); await tick(); - ( - engine as unknown as { invalidateDataChannelWaiters: (reason: string) => void } - ).invalidateDataChannelWaiters('data channels recreated'); + engineChannel(engine, 'reliableChannel').invalidateWaiters('data channels recreated'); // The send must not surface the teardown — the packet is queued for the resume replay. await expect(send).resolves.toBeUndefined(); @@ -471,9 +430,7 @@ describe('RTCEngine', () => { const send = engine.sendDataPacket(makePacket(1), DataChannelKind.RELIABLE); await tick(); Object.assign(engine as unknown as Record, { _isClosed: true }); - ( - engine as unknown as { invalidateDataChannelWaiters: (reason: string) => void } - ).invalidateDataChannelWaiters('engine closed'); + engineChannel(engine, 'reliableChannel').invalidateWaiters('engine closed'); await expect(send).rejects.toBeInstanceOf(UnexpectedConnectionState); expect(dc.send).not.toHaveBeenCalled(); @@ -496,78 +453,80 @@ describe('RTCEngine', () => { }); }); - describe('sendLossyBytes', () => { + describe('sendDataTrackFrame', () => { it('ensures the publisher is connected before sending (direct data-track path)', async () => { const engine = new RTCEngine(roomOptionDefaults); const dc = new FakeDataChannel(); // The channel only becomes available once the publisher connection has been established — // mirroring the lazily negotiated publisher case that Room's packetAvailable path hits. - let connected = false; const ensurePublisherConnected = vi.fn(async () => { - connected = true; + attachFakeChannel(engine, 'dataTrackChannel', dc); }); Object.assign(engine as unknown as Record, { _isClosed: false, ensurePublisherConnected, - dataChannelForKind: vi.fn(() => (connected ? dc : undefined)), }); - await engine.sendLossyBytes(new Uint8Array([1]), DataChannelKind.DATA_TRACK_LOSSY, 'wait'); + await engine.sendDataTrackFrame(new Uint8Array([1])); expect(ensurePublisherConnected).toHaveBeenCalledWith(DataChannelKind.DATA_TRACK_LOSSY); expect(dc.send).toHaveBeenCalledTimes(1); }); - it('only counts LOSSY bytes into the byterate stat that tunes the lossy drop threshold', async () => { + it('keeps each channel’s byterate stat isolated from the other’s traffic', async () => { const engine = new RTCEngine(roomOptionDefaults); const dc = new FakeDataChannel(); Object.assign(engine as unknown as Record, { _isClosed: false, ensurePublisherConnected: vi.fn().mockResolvedValue(undefined), - dataChannelForKind: vi.fn(() => dc), }); - const stat = () => - (engine as unknown as { lossyDataStatCurrentBytes: number }).lossyDataStatCurrentBytes; - - // Data-track traffic must not move the stat — it would inflate the LOSSY channel's - // dynamically tuned drop threshold with traffic that channel never carries. - await engine.sendLossyBytes(new Uint8Array(1000), DataChannelKind.DATA_TRACK_LOSSY, 'wait'); - expect(stat()).toBe(0); - - await engine.sendLossyBytes(new Uint8Array(100), DataChannelKind.LOSSY, 'drop'); - expect(stat()).toBe(100); + attachFakeChannel(engine, 'lossyChannel', dc); + attachFakeChannel(engine, 'dataTrackChannel', dc); + const lossyStat = () => + (engine as unknown as { lossyChannel: { statCurrentBytes: number } }).lossyChannel + .statCurrentBytes; + + // Data-track traffic (sendDataTrackFrame → data-track channel) must not move the LOSSY channel's + // stat — it would inflate the lossy channel's dynamically tuned drop threshold with traffic + // that channel never carries. + await engine.sendDataTrackFrame(new Uint8Array(1000)); + expect(lossyStat()).toBe(0); + + // A plain lossy publishData packet goes through sendDataPacket → lossy channel. + const lossyPacket = new DataPacket({ + kind: DataPacket_Kind.LOSSY, + value: { case: 'user', value: new UserPacket({ payload: new Uint8Array(100) }) }, + }); + await engine.sendDataPacket(lossyPacket, DataChannelKind.LOSSY); + expect(lossyStat()).toBeGreaterThan(0); }); }); - describe('waitForBufferHeadroomWithLock', () => { + describe('waitForBufferHeadroom', () => { it('rejects parked waiters and releases the lock when the data channels are invalidated', async () => { const engine = new RTCEngine(roomOptionDefaults); const dc = new FakeDataChannel(); Object.assign(engine as unknown as Record, { _isClosed: false, - dataChannelForKind: vi.fn(() => dc), }); + attachFakeChannel(engine, 'reliableChannel', dc); // Park a waiter: buffer above the reliable high-water mark, holding the headroom lock. dc.bufferedAmount = 2 * 1024 * 1024; - const parked = engine.waitForBufferHeadroomWithLock(DataChannelKind.RELIABLE); + const parked = engine.waitForBufferHeadroom(DataChannelKind.RELIABLE); // Swallow the expected rejection so it can't surface as unhandled before we assert on it. parked.catch(() => {}); await tick(); // The channel object gets abandoned (e.g. createDataChannels on the Safari resume path). - ( - engine as unknown as { invalidateDataChannelWaiters: (reason: string) => void } - ).invalidateDataChannelWaiters('data channels recreated'); + engineChannel(engine, 'reliableChannel').invalidateWaiters('data channels recreated'); await expect(parked).rejects.toBeInstanceOf(UnexpectedConnectionState); // The lock must be free again: a wait against the fresh, drained channel resolves instead // of queueing forever behind the stranded waiter. dc.bufferedAmount = 0; - await expect( - engine.waitForBufferHeadroomWithLock(DataChannelKind.RELIABLE), - ).resolves.toBeUndefined(); + await expect(engine.waitForBufferHeadroom(DataChannelKind.RELIABLE)).resolves.toBeUndefined(); }); }); diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index abacb738fc..ab0a7bb4ae 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -44,7 +44,6 @@ import { type UserPacket, } from '@livekit/protocol'; import { EventEmitter } from 'events'; -import type { Throws } from '@livekit/throws-transformer/throws'; import type { MediaAttributes } from 'sdp-transform'; import type TypedEventEmitter from 'typed-emitter'; import type { SignalOptions } from '../api/SignalClient'; @@ -64,11 +63,15 @@ import log, { LoggerNames, getLogger } from '../logger'; import type { InternalRoomOptions } from '../options'; import type { NonSharedUint8Array } from '../type-polyfills/non-shared-typed-arrays'; import TypedPromise from '../utils/TypedPromise'; -import { DataPacketBuffer } from '../utils/dataPacketBuffer'; import { TTLMap } from '../utils/ttlmap'; import PCTransport, { PCEvents } from './PCTransport'; import { PCTransportManager, PCTransportState } from './PCTransportManager'; import type { ReconnectContext, ReconnectPolicy } from './ReconnectPolicy'; +import { DataChannelManager } from './data-channel/DataChannelManager'; +import type { FlowControlledDataChannel } from './data-channel/FlowControlledDataChannel'; +import type { LossyDataChannel } from './data-channel/LossyDataChannel'; +import type { ReliableDataChannel } from './data-channel/ReliableDataChannel'; +import { DataChannelKind } from './data-channel/types'; import { DataTrackInfo } from './data-track/types'; import { roomConnectOptionDefaults } from './defaults'; import { @@ -103,24 +106,10 @@ import { toHttpUrl, } from './utils'; -const lossyDataChannel = '_lossy'; -const reliableDataChannel = '_reliable'; -const dataTrackDataChannel = '_data_track'; const minReconnectWait = 2 * 1000; const leaveReconnect = 'leave-reconnect'; const reliabeReceiveStateTTL = 30_000; -// Two-watermark flow control for the reliable and data-track channels. Senders fill the buffer -// freely up to the high-water mark; once it's exceeded they block until the browser's -// `bufferedamountlow` event (which we arm at the low-water mark) signals the buffer has drained. -// The gap between the marks keeps the SCTP send buffer saturated while we refill, so throughput -// isn't starved, while the high-water mark bounds the buffer well below the level that would abort -// the channel (see livekit/client-sdk-js#1995). -const reliableDataChannelWaterMarkLow = 64 * 1024; -const reliableDataChannelWaterMarkHigh = 1024 * 1024; -const lossyDataChannelWaterMarkLow = 8 * 1024; -const lossyDataChannelWaterMarkHigh = 256 * 1024; - const initialMediaSectionsAudio = 3; const initialMediaSectionsVideo = 3; @@ -132,11 +121,7 @@ enum PCState { Closed, } -export enum DataChannelKind { - RELIABLE = DataPacket_Kind.RELIABLE, - LOSSY = DataPacket_Kind.LOSSY, - DATA_TRACK_LOSSY = 2, -} +export { DataChannelKind }; // Default data-channel max message size (bytes), used when the remote SDP // answer does not advertise an `a=max-message-size` attribute (RFC 8841). @@ -180,22 +165,24 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit return !!this.reconnectTimeout; } - private lossyDC?: RTCDataChannel; - - // @ts-ignore noUnusedLocals - private lossyDCSub?: RTCDataChannel; - - private reliableDC?: RTCDataChannel; - - // @ts-ignore noUnusedLocals - private reliableDCSub?: RTCDataChannel; + /** + * Owns the data channels: the three flow-controlled publisher wrappers (engine-lifetime; the + * RTCDataChannel handles underneath are attached/detached as peer connections come and go, with + * waiter invalidation built into the turnover) plus the subscriber-side receive handles. + */ + private dataChannels: DataChannelManager; - private dataTrackDC?: RTCDataChannel; + private get reliableChannel(): ReliableDataChannel { + return this.dataChannels.reliable; + } - // @ts-ignore noUnusedLocals - private dataTrackDCSub?: RTCDataChannel; + private get lossyChannel(): LossyDataChannel { + return this.dataChannels.lossy; + } - private dcBufferStatus: Map; + private get dataTrackChannel(): LossyDataChannel { + return this.dataChannels.dataTrack; + } private subscriberPrimary: boolean = false; @@ -252,20 +239,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit private publisherConnectionPromise: Promise | undefined; - private reliableDataSequence: number = 1; - - private reliableMessageBuffer = new DataPacketBuffer(); - private reliableReceivedState: TTLMap = new TTLMap(reliabeReceiveStateTTL); - private lossyDataStatCurrentBytes: number = 0; - - private lossyDataStatByterate: number = 0; - - private lossyDataStatInterval: ReturnType | undefined; - - private lossyDataDropCount: number = 0; - private midToTrackId: { [key: string]: string } = {}; /** used to indicate whether the browser is currently waiting to reconnect */ @@ -283,11 +258,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.reconnectPolicy = this.options.reconnectPolicy; this.closingLock = new Mutex(); this.dataProcessLock = new Mutex(); - this.dcBufferStatus = new Map([ - [DataChannelKind.RELIABLE, true], - [DataChannelKind.LOSSY, true], - [DataChannelKind.DATA_TRACK_LOSSY, true], - ]); + this.dataChannels = new DataChannelManager({ + isEngineClosed: () => this.isClosed, + isReconnecting: () => this.attemptingReconnect, + onDataMessage: (message) => this.handleDataMessage(message), + onDataTrackMessage: (message) => this.handleDataTrackMessage(message), + onDataError: (event) => this.handleDataError(event), + onChannelClose: (kind) => this.handleDataChannelClose(kind)(), + onBufferStatusChanged: (kind, isLow) => + this.emit(EngineEvent.DCBufferStatusChanged, isLow, kind), + }); this.client.onParticipantUpdate = (updates) => this.emit(EngineEvent.ParticipantUpdate, updates); @@ -458,60 +438,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } async cleanupPeerConnections() { - // Reject parked headroom waiters up front: closing the peer connection is allowed (per spec) - // to transition data channels to 'closed' without firing events, which would otherwise strand - // a waiter holding the headroom lock. - this.invalidateDataChannelWaiters('peer connections cleaned up'); - - const dcCleanup = (dc: RTCDataChannel | undefined) => { - if (!dc) { - return; - } - - // Detach the data channel handlers before closing anything. Closing a peer connection tears - // down the SCTP transport, which can dispatch `error`/`close` events on the still-open data - // channels; if our handlers are still attached at that point, handleDataError logs a spurious - // "Unknown DataChannel error" during an otherwise graceful disconnect. Removing the handlers - // before dc.close()/pcManager.close() makes this deterministic regardless of how/when the - // browser dispatches those teardown events. See livekit/client-sdk-js#1953. - dc.onbufferedamountlow = null; - dc.onclose = null; - dc.onclosing = null; - dc.onerror = null; - dc.onmessage = null; - dc.onopen = null; - - dc.close(); - }; - dcCleanup(this.lossyDC); - dcCleanup(this.lossyDCSub); - dcCleanup(this.reliableDC); - dcCleanup(this.reliableDCSub); - dcCleanup(this.dataTrackDC); - dcCleanup(this.dataTrackDCSub); + this.dataChannels.teardown(); await this.pcManager?.close(); this.pcManager = undefined; - this.lossyDC = undefined; - this.lossyDCSub = undefined; - this.reliableDC = undefined; - this.reliableDCSub = undefined; - this.dataTrackDC = undefined; - this.dataTrackDCSub = undefined; - this.reliableMessageBuffer = new DataPacketBuffer(); - this.reliableDataSequence = 1; this.reliableReceivedState.clear(); } cleanupLossyDataStats() { - this.lossyDataStatByterate = 0; - this.lossyDataStatCurrentBytes = 0; - if (this.lossyDataStatInterval) { - clearInterval(this.lossyDataStatInterval); - this.lossyDataStatInterval = undefined; - } - this.lossyDataDropCount = 0; + this.lossyChannel.stopThresholdTuning(); } async cleanupClient() { @@ -580,7 +516,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } get dataSubscriberReadyState(): string | undefined { - return this.reliableDCSub?.readyState; + return this.dataChannelForKind(DataChannelKind.RELIABLE, true)?.readyState; } async getConnectedServerAddress(): Promise { @@ -890,121 +826,39 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit return; } - // Waiters parked on the old channel objects would never see another event from them once they - // are replaced below — reject them so the headroom lock is released and queued senders - // re-check against the new channels. - this.invalidateDataChannelWaiters('data channels recreated'); - - // clear old data channel callbacks if recreate - if (this.lossyDC) { - this.lossyDC.onmessage = null; - this.lossyDC.onerror = null; - this.lossyDC.onclose = null; - } - if (this.reliableDC) { - this.reliableDC.onmessage = null; - this.reliableDC.onerror = null; - this.reliableDC.onclose = null; - } - if (this.dataTrackDC) { - this.dataTrackDC.onmessage = null; - this.dataTrackDC.onerror = null; - this.dataTrackDC.onclose = null; - } - - // create data channels - this.lossyDC = this.pcManager.createPublisherDataChannel(lossyDataChannel, { - ordered: false, - maxRetransmits: 0, - }); - this.reliableDC = this.pcManager.createPublisherDataChannel(reliableDataChannel, { - ordered: true, - }); - this.dataTrackDC = this.pcManager.createPublisherDataChannel(dataTrackDataChannel, { - ordered: false, - maxRetransmits: 0, - }); - - // also handle messages over the pub channel, for backwards compatibility - this.lossyDC.onmessage = this.handleDataMessage; - this.reliableDC.onmessage = this.handleDataMessage; - this.dataTrackDC.onmessage = this.handleDataTrackMessage; - - // handle datachannel errors - this.lossyDC.onerror = this.handleDataError; - this.reliableDC.onerror = this.handleDataError; - this.dataTrackDC.onerror = this.handleDataError; - - // detect unexpected publisher data channel closes - this.lossyDC.onclose = this.handleDataChannelClose(DataChannelKind.LOSSY); - this.reliableDC.onclose = this.handleDataChannelClose(DataChannelKind.RELIABLE); - this.dataTrackDC.onclose = this.handleDataChannelClose(DataChannelKind.DATA_TRACK_LOSSY); - - // set up dc buffer threshold - if this is not set, it will default to 0 - this.lossyDC.bufferedAmountLowThreshold = lossyDataChannelWaterMarkLow; - this.reliableDC.bufferedAmountLowThreshold = reliableDataChannelWaterMarkLow; - this.dataTrackDC.bufferedAmountLowThreshold = lossyDataChannelWaterMarkLow; - - // handle buffer amount low events - this.lossyDC.onbufferedamountlow = () => this.handleBufferedAmountLow(DataChannelKind.LOSSY); - this.reliableDC.onbufferedamountlow = () => - this.handleBufferedAmountLow(DataChannelKind.RELIABLE); - this.dataTrackDC.onbufferedamountlow = () => - this.handleBufferedAmountLow(DataChannelKind.DATA_TRACK_LOSSY); - - this.cleanupLossyDataStats(); - this.lossyDataStatInterval = setInterval(() => { - this.lossyDataStatByterate = this.lossyDataStatCurrentBytes; - this.lossyDataStatCurrentBytes = 0; - - const dc = this.dataChannelForKind(DataChannelKind.LOSSY); - if (dc) { - // control buffered latency to ~100ms - const threshold = this.lossyDataStatByterate / 10; - dc.bufferedAmountLowThreshold = Math.min( - Math.max(threshold, lossyDataChannelWaterMarkLow), - lossyDataChannelWaterMarkHigh, - ); - } - }, 1000); + this.dataChannels.createPublisherChannels(this.pcManager); } private handleDataChannel = async ({ channel }: RTCDataChannelEvent) => { if (!channel) { return; } - let handler; - if (channel.label === reliableDataChannel) { - this.reliableDCSub = channel; - handler = this.handleDataMessage; - } else if (channel.label === lossyDataChannel) { - this.lossyDCSub = channel; - handler = this.handleDataMessage; - } else if (channel.label === dataTrackDataChannel) { - this.dataTrackDCSub = channel; - handler = this.handleDataTrackMessage; - } else { - return; + if (this.dataChannels.adoptSubscriberChannel(channel)) { + this.log.debug(`on data channel ${channel.id}, ${channel.label}`); } - this.log.debug(`on data channel ${channel.id}, ${channel.label}`); - channel.onmessage = handler; }; + /** Normalizes an incoming data-channel message into bytes, or logs and returns undefined. */ + private async decodeDataMessage(message: MessageEvent): Promise { + if (message.data instanceof ArrayBuffer) { + return new Uint8Array(message.data); + } + if (message.data instanceof Blob) { + return new Uint8Array(await message.data.arrayBuffer()); + } + this.log.error('unsupported data type', { data: message.data }); + return undefined; + } + private handleDataMessage = async (message: MessageEvent) => { // make sure to respect incoming data message order by processing message events one after the other const unlock = await this.dataProcessLock.lock(); try { - // decode - let buffer: ArrayBuffer | undefined; - if (message.data instanceof ArrayBuffer) { - buffer = message.data; - } else if (message.data instanceof Blob) { - buffer = await message.data.arrayBuffer(); - } else { - this.log.error('unsupported data type', { data: message.data }); + const bytes = await this.decodeDataMessage(message); + if (!bytes) { return; } - const dp = DataPacket.fromBinary(new Uint8Array(buffer)); + const dp = DataPacket.fromBinary(bytes); if (dp.sequence > 0 && dp.participantSid !== '') { const lastSeq = this.reliableReceivedState.get(dp.participantSid); @@ -1053,18 +907,11 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit }; private handleDataTrackMessage = async (message: MessageEvent) => { - // Decode / normalize into a common format - let buffer: ArrayBuffer | undefined; - if (message.data instanceof ArrayBuffer) { - buffer = message.data; - } else if (message.data instanceof Blob) { - buffer = await message.data.arrayBuffer(); - } else { - this.log.error('unsupported data type', { data: message.data }); + const bytes = await this.decodeDataMessage(message); + if (!bytes) { return; } - - this.emit('dataTrackPacketReceived', new Uint8Array(buffer)); + this.emit('dataTrackPacketReceived', bytes); }; private handleDataError = (event: Event) => { @@ -1103,10 +950,6 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } }; - private handleBufferedAmountLow = (channelKind: DataChannelKind) => { - this.updateAndEmitDCBufferStatus(channelKind); - }; - async createSender( track: LocalTrack, opts: TrackPublishOptions, @@ -1510,7 +1353,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // recreate publish datachannel if it's id is null // (for safari https://bugs.webkit.org/show_bug.cgi?id=184688) - if (this.reliableDC?.readyState === 'open' && this.reliableDC.id === null) { + const reliableDC = this.dataChannelForKind(DataChannelKind.RELIABLE); + if (reliableDC?.readyState === 'open' && reliableDC.id === null) { this.createDataChannels(); } @@ -1586,7 +1430,11 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } /* @internal */ - async sendDataPacket(packet: DataPacket, kind: DataChannelKind) { + async sendDataPacket( + packet: DataPacket, + /** Data-track frames don't come through here — they're sent pre-serialized via {@link sendDataTrackFrame }. */ + kind: Exclude, + ) { // make sure we do have a data connection await this.ensurePublisherConnected(kind); @@ -1608,8 +1456,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } if (kind === DataChannelKind.RELIABLE) { - packet.sequence = this.reliableDataSequence; - this.reliableDataSequence += 1; + packet.sequence = this.reliableChannel.nextSequence(); } const msg = packet.toBinary() as Uint8Array; @@ -1631,314 +1478,46 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit ); } - switch (kind) { - case DataChannelKind.LOSSY: - case DataChannelKind.DATA_TRACK_LOSSY: - return this.sendLossyBytes(msg, kind); - - case DataChannelKind.RELIABLE: { - if (this.attemptingReconnect) { - // A reconnect is already underway — queue for the resume replay instead of parking on a - // channel that is being torn down. The send resolves; delivery is deferred to the replay. - this.reliableMessageBuffer.push({ data: msg, sequence: packet.sequence, sent: false }); - return; - } - - const dc = this.dataChannelForKind(kind); - if (dc) { - try { - await this.waitForBufferHeadroomWithLock(kind); - } catch (error) { - if (this.isClosed) { - // No replay is coming after an engine close — surface the failure. - throw error; - } - // Transient teardown (the channel closed or was replaced while we waited): the - // reliable channel promises delivery across resume, so queue the packet for the - // replay instead of rejecting a send the app can't meaningfully retry. - this.reliableMessageBuffer.push({ data: msg, sequence: packet.sequence, sent: false }); - return; - } - - if (this.attemptingReconnect) { - // A reconnect began while we waited for headroom — same deal as above. - this.reliableMessageBuffer.push({ data: msg, sequence: packet.sequence, sent: false }); - return; - } - - this.reliableMessageBuffer.push({ data: msg, sequence: packet.sequence, sent: true }); - dc.send(msg); - } - - this.updateAndEmitDCBufferStatus(kind); - break; - } - } - } - - /* @internal */ - async sendLossyBytes( - bytes: NonSharedUint8Array, - kind: Exclude, - bufferStatusFullBehavior: 'drop' | 'wait' = 'drop', - ) { - // Make sure we do have a data connection. This matters most for the direct data-track path - // (Room's packetAvailable handler), which doesn't go through sendDataPacket: it both waits for - // the channel to be open and — on lazily negotiated publisher connections — is what kicks the - // negotiation off in the first place. The call is memoized, so the steady-state cost is one - // await on an already-resolved promise. - await this.ensurePublisherConnected(kind); - - const dc = this.dataChannelForKind(kind); - try { - if (dc) { - // Depending on the exact circumstance that data is being sent, either drop or wait for the - // buffer to drain below the high-water mark before continuing. - switch (bufferStatusFullBehavior) { - case 'wait': - if (!this.isBelowHighWaterMark(kind)) { - await this.waitForBufferHeadroomWithLock(kind); - } - break; - case 'drop': - // we check against the actual threshold on the DC here, as it is dynamic for the lossy DC - if (!this.isBelowLowWaterMark(kind)) { - // Drop messages to reduce latency - this.lossyDataDropCount += 1; - if (this.lossyDataDropCount % 100 === 0) { - this.log.warn( - `dropping lossy data channel messages, total dropped: ${this.lossyDataDropCount}`, - ); - } - return; - } - } - if (kind === DataChannelKind.LOSSY) { - // The byterate stat tunes the LOSSY channel's dynamic drop threshold; counting data-track - // bytes here would inflate it and let the lossy channel buffer far more latency than the - // ~100ms the tuning targets. - this.lossyDataStatCurrentBytes += bytes.byteLength; - } - - if (this.attemptingReconnect) { - return; - } - - dc.send(bytes); - } - - this.updateAndEmitDCBufferStatus(kind); - } catch (error: unknown) { - // ensure same surface behaviour as before with missing data channel being silently ignored, just log an error message for clarity - if (error instanceof TypeError) { - this.log.error(error); - } else { - throw error; - } - } - } - - private async resendReliableMessagesForResume(lastMessageSeq: number) { - await this.ensurePublisherConnected(DataChannelKind.RELIABLE); - const dc = this.dataChannelForKind(DataChannelKind.RELIABLE); - if (dc) { - this.reliableMessageBuffer.popToSequence(lastMessageSeq); - // Hold the headroom lock across the whole replay. Releasing it between messages would let a - // concurrent send — whose (newer) sequence was already assigned in sendDataPacket before it - // queued on the lock — hit the wire mid-replay, and receivers would then discard the - // remaining lower-sequence resent messages as duplicates. - const unlock = await this.getBufferHeadroomLock(DataChannelKind.RELIABLE).lock(); - try { - // Everything left after the ack cutoff must be re-handed to the current channel. - this.reliableMessageBuffer.markAllUnsent(); - // Drain in passes, re-scanning the live buffer each time: a send that arrives (deferred, - // sent:false) during our own awaits appends after this pass started, so we pick it up on - // the next one. We mark each packet only once we've actually handed it to the channel — a - // blanket "mark all sent" would flip such a late arrival to sent without transmitting it, - // and a later alignBufferedAmount would then drop it for good. If the loop throws - // mid-drain, unsent entries keep their flag and the next replay picks them up. - for ( - let batch = this.reliableMessageBuffer.getUnsent(); - batch.length > 0; - batch = this.reliableMessageBuffer.getUnsent() - ) { - for (const item of batch) { - // Respect flow control on resume too, so a large resend doesn't overflow the buffer. - // We already hold the lock across the whole replay, so use the lock-free wait. - await this.waitForBufferHeadroomWithoutLock(DataChannelKind.RELIABLE); - dc.send(item.data); - this.reliableMessageBuffer.markSent(item); - } - } - } finally { - unlock(); - } - } - this.updateAndEmitDCBufferStatus(DataChannelKind.RELIABLE); - } - - private updateAndEmitDCBufferStatus = (kind: DataChannelKind) => { + // The full-buffer policy (drop for lossy, wait/replay for reliable) is the channel's own, as + // is emitting the buffer-status change once the send settles. if (kind === DataChannelKind.RELIABLE) { - const dc = this.dataChannelForKind(kind); - if (dc) { - this.reliableMessageBuffer.alignBufferedAmount(dc.bufferedAmount); - } - } - try { - const status = this.isBelowLowWaterMark(kind); - if (typeof status !== 'undefined' && status !== this.dcBufferStatus.get(kind)) { - this.dcBufferStatus.set(kind, status); - this.emit(EngineEvent.DCBufferStatusChanged, status, kind); - } - } catch (e) { - this.log.warn('could not update buffer status', { error: e }); - } - }; - - /** - * Whether the send buffer has room to accept more data (the send gate). Senders proceed while - * this is true and block once it goes false. - */ - private isBelowHighWaterMark = (kind: DataChannelKind): Throws => { - const dc = this.dataChannelForKind(kind); - if (!dc) { - throw new TypeError(`Could not get data channel for kind ${kind}`); - } - // because RTCDatachannel has no high water mark built in we read the statically defined versions as constants - const highMark = - kind === DataChannelKind.RELIABLE - ? reliableDataChannelWaterMarkHigh - : lossyDataChannelWaterMarkHigh; - return dc.bufferedAmount <= highMark; - }; - - /** - * Whether the send buffer has drained to its low-water mark. Drives the public - * {@link EngineEvent.DCBufferStatusChanged} event. - */ - private isBelowLowWaterMark = (kind: DataChannelKind): Throws => { - const dc = this.dataChannelForKind(kind); - if (!dc) { - throw new TypeError(`Could not get data channel for kind ${kind}`); + await this.reliableChannel.send(msg, packet.sequence); + } else { + await this.lossyChannel.send(msg); } - // because RTCDatachannel has the threshold built in we can read it dynamically to account for changing thresholds over time - return dc.bufferedAmount <= dc.bufferedAmountLowThreshold; - }; - - /** Per-kind lock serializing senders that have to wait for the buffer to drain. */ - private waitForBufferHeadroomLocks = new Map(); + } /** - * Per-kind AbortController that cancels parked headroom waiters whenever a kind's channel object - * stops being current (replaced by createDataChannels or torn down by cleanupPeerConnections). A - * waiter is parked on the channel object it captured at wait entry; if that object is abandoned - * its events may never fire again, so the abort is what rejects the waiter and releases the - * headroom lock instead of stranding every future sender behind it. + * Sends pre-serialized bytes on the data-track channel. This is the one send path that doesn't + * go through {@link sendDataPacket} — Room's `packetAvailable` handler calls it directly with + * bytes the data-track pipeline already serialized. + * + * @internal */ - private waiterAbortControllers = new Map(); - - private waiterAbortSignal(kind: DataChannelKind): AbortSignal { - let controller = this.waiterAbortControllers.get(kind); - if (!controller) { - controller = new AbortController(); - this.waiterAbortControllers.set(kind, controller); - } - return controller.signal; + async sendDataTrackFrame(bytes: NonSharedUint8Array) { + // Make sure we do have a data connection: on lazily negotiated publisher connections this is + // what kicks negotiation off, and it waits for the channel to open. Memoized, so the + // steady-state cost is one await on an already-resolved promise. + await this.ensurePublisherConnected(DataChannelKind.DATA_TRACK_LOSSY); + await this.dataTrackChannel.send(bytes); } - /** Rejects all parked headroom waiters (all kinds); the next waiter gets a fresh controller. */ - private invalidateDataChannelWaiters(reason: string) { - for (const controller of this.waiterAbortControllers.values()) { - controller.abort(reason); - } - this.waiterAbortControllers.clear(); + private async resendReliableMessagesForResume(lastMessageSeq: number) { + await this.ensurePublisherConnected(DataChannelKind.RELIABLE); + await this.reliableChannel.replay(lastMessageSeq); } - /** Acquires the per-kind headroom lock, resolving with the unlock function. */ - private getBufferHeadroomLock(kind: DataChannelKind): Mutex { - let lock = this.waitForBufferHeadroomLocks.get(kind); - if (!lock) { - lock = new Mutex(); - this.waitForBufferHeadroomLocks.set(kind, lock); - } - return lock; + /** The flow-control gate for `kind` — see {@link FlowControlledDataChannel}. */ + private flowControlFor(kind: DataChannelKind): FlowControlledDataChannel { + return this.dataChannels.channelFor(kind); } /** - * Acquires the `kind` headroom lock, waits until the send buffer has room, then releases. The - * common single-send entry point. Callers that need to hold the lock across several sends (the - * resume replay) acquire it via {@link getBufferHeadroomLock} and call - * {@link waitForBufferHeadroomWithoutLock} per message instead. + * Resolves once the caller may send on the `kind` channel — see + * {@link FlowControlledDataChannel.waitForHeadroomWithLock}. */ - async waitForBufferHeadroomWithLock(kind: DataChannelKind) { - const unlock = await this.getBufferHeadroomLock(kind).lock(); - try { - await this.waitForBufferHeadroomWithoutLock(kind); - } finally { - unlock(); - } - } - - /** - * Waits until the send buffer for `kind` is at or below the high-water mark (draining, if - * needed, via the `bufferedamountlow` event). Does no locking of its own — the caller must - * already hold the kind's headroom lock (via {@link getBufferHeadroomLock}). The resume replay - * holds the lock across its whole batch and calls this per message so no other sender can - * interleave; the single-send path goes through {@link waitForBufferHeadroomWithLock}. - * - * Serializing through the per-kind lock means that, once the buffer drains, queued callers - * refill it one at a time (up to the high-water mark) rather than all at once and overflowing - * the SCTP send buffer (see livekit/client-sdk-js#1995). - */ - private async waitForBufferHeadroomWithoutLock( - kind: DataChannelKind, - ): Promise> { - if (this.isClosed) { - throw new UnexpectedConnectionState('engine closed'); - } - if (this.isBelowHighWaterMark(kind)) { - return; - } - const dc = this.dataChannelForKind(kind); - if (!dc) { - throw new UnexpectedConnectionState(`DataChannel not found, kind: ${kind}`); - } - const abortSignal = this.waiterAbortSignal(kind); - await new TypedPromise((resolve, reject) => { - const onBufferedAmountLow = () => { - cleanup(); - resolve(); - }; - const onDCClose = () => { - cleanup(); - reject( - new UnexpectedConnectionState(`DataChannel ${kind} closed while draining the buffer`), - ); - }; - const onAbort = () => { - cleanup(); - reject( - new UnexpectedConnectionState( - `DataChannel ${kind} was replaced or torn down while waiting for headroom`, - ), - ); - }; - const cleanup = () => { - dc.removeEventListener('bufferedamountlow', onBufferedAmountLow); - dc.removeEventListener('close', onDCClose); - abortSignal.removeEventListener('abort', onAbort); - }; - if (abortSignal.aborted) { - onAbort(); - return; - } - dc.addEventListener('bufferedamountlow', onBufferedAmountLow); - // Proxy along any error caused by the data channel closing while we wait. - dc.addEventListener('close', onDCClose); - // The channel object we're parked on can be abandoned without ever firing another event - // (e.g. createDataChannels replacing it); the abort is our way out. - abortSignal.addEventListener('abort', onAbort); - }); + async waitForBufferHeadroom(kind: DataChannelKind) { + return this.flowControlFor(kind).waitForHeadroomWithLock(); } /** @@ -2040,9 +1619,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // don't negotiate without any transceivers or data channel, it will generate sdp without ice frag then negotiate failed if ( this.pcManager.publisher.getTransceivers().length == 0 && - !this.lossyDC && - !this.reliableDC && - !this.dataTrackDC + !this.dataChannels.hasPublisherChannels ) { this.createDataChannels(); } @@ -2091,26 +1668,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } dataChannelForKind(kind: DataChannelKind, sub?: boolean): RTCDataChannel | undefined { - switch (kind) { - case DataChannelKind.RELIABLE: - if (!sub) { - return this.reliableDC; - } else { - return this.reliableDCSub; - } - case DataChannelKind.LOSSY: - if (!sub) { - return this.lossyDC; - } else { - return this.lossyDCSub; - } - case DataChannelKind.DATA_TRACK_LOSSY: - if (!sub) { - return this.dataTrackDC; - } else { - return this.dataTrackDCSub; - } - } + return this.dataChannels.getHandle(kind, sub); } /** @internal */ diff --git a/src/room/Room.ts b/src/room/Room.ts index b09f79bb8f..f8e2b2aaf0 100644 --- a/src/room/Room.ts +++ b/src/room/Room.ts @@ -308,7 +308,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) }) .on('packetAvailable', ({ handle, bytes }) => { this.engine - .sendLossyBytes(bytes, DataChannelKind.DATA_TRACK_LOSSY, 'wait') + .sendDataTrackFrame(bytes) .finally(() => this.outgoingDataTrackManager.handlePacketSendComplete(handle)); }); diff --git a/src/room/data-channel/DataChannelManager.test.ts b/src/room/data-channel/DataChannelManager.test.ts new file mode 100644 index 0000000000..c37861d8a2 --- /dev/null +++ b/src/room/data-channel/DataChannelManager.test.ts @@ -0,0 +1,152 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { PCTransportManager } from '../PCTransportManager'; +import { UnexpectedConnectionState } from '../errors'; +import { DataChannelManager, type DataChannelManagerOptions } from './DataChannelManager'; +import { DataChannelKind } from './types'; + +class FakeDataChannel extends EventTarget { + bufferedAmount = 0; + + bufferedAmountLowThreshold = 0; + + onmessage: ((message: MessageEvent) => void) | null = null; + + onerror: ((event: Event) => void) | null = null; + + onclose: (() => void) | null = null; + + onbufferedamountlow: (() => void) | null = null; + + send = vi.fn(); + + close = vi.fn(); + + constructor(public label: string) { + super(); + } +} + +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +function makeManager(overrides?: Partial) { + const opts = { + isEngineClosed: () => false, + isReconnecting: () => false, + onDataMessage: vi.fn(), + onDataTrackMessage: vi.fn(), + onDataError: vi.fn(), + onChannelClose: vi.fn(), + onBufferStatusChanged: vi.fn(), + ...overrides, + }; + const manager = new DataChannelManager(opts); + const created: Record = {}; + const pcManager = { + createPublisherDataChannel: vi.fn((label: string) => { + const dc = new FakeDataChannel(label); + created[label] = dc; + return dc as unknown as RTCDataChannel; + }), + } as unknown as PCTransportManager; + return { manager, opts, pcManager, created }; +} + +describe('DataChannelManager', () => { + it('creates and wires the three publisher channels', () => { + const { manager, opts, pcManager, created } = makeManager(); + + manager.createPublisherChannels(pcManager); + + expect(Object.keys(created).sort()).toEqual(['_data_track', '_lossy', '_reliable']); + expect(manager.getHandle(DataChannelKind.RELIABLE)).toBe(created._reliable); + expect(manager.hasPublisherChannels).toBe(true); + for (const dc of Object.values(created)) { + expect(dc.onmessage).toBeTruthy(); + expect(dc.onerror).toBeTruthy(); + expect(dc.onclose).toBeTruthy(); + expect(dc.onbufferedamountlow).toBeTruthy(); + expect(dc.bufferedAmountLowThreshold).toBeGreaterThan(0); + } + + // Close handler reports the right kind. + created._reliable.onclose!(); + expect(opts.onChannelClose).toHaveBeenCalledWith(DataChannelKind.RELIABLE); + + // A drain event on the data-track channel refreshes its status; since the buffer starts empty + // (below the low mark) and the wrapper's initial status is already "low", no change fires — + // fill it first so the drain is an observable not-low → low transition. + const dataTrackDc = created._data_track; + dataTrackDc.bufferedAmount = 10 * 1024 * 1024; + dataTrackDc.onbufferedamountlow!(); // refresh: now "not low" + dataTrackDc.bufferedAmount = 0; + dataTrackDc.onbufferedamountlow!(); // refresh: back to "low" + expect(opts.onBufferStatusChanged).toHaveBeenCalledWith( + DataChannelKind.DATA_TRACK_LOSSY, + false, + ); + expect(opts.onBufferStatusChanged).toHaveBeenCalledWith(DataChannelKind.DATA_TRACK_LOSSY, true); + + manager.lossy.stopThresholdTuning(); + }); + + it('recreating channels rejects waiters parked on the replaced handles', async () => { + const { manager, pcManager, created } = makeManager(); + manager.createPublisherChannels(pcManager); + + // Park a reliable sender on the first-generation channel. + created._reliable.bufferedAmount = 2 * 1024 * 1024; + const parked = manager.reliable.waitForHeadroomWithLock(); + parked.catch(() => {}); + await tick(); + + // Safari null-id path: channels recreated without closing the old ones. + manager.createPublisherChannels(pcManager); + + await expect(parked).rejects.toBeInstanceOf(UnexpectedConnectionState); + // The gate recovers against the fresh (empty) channel. + await expect(manager.reliable.waitForHeadroomWithLock()).resolves.toBeUndefined(); + + manager.lossy.stopThresholdTuning(); + }); + + it('adopts subscriber channels by label and rejects unknown labels', () => { + const { manager, opts } = makeManager(); + + const sub = new FakeDataChannel('_reliable'); + expect(manager.adoptSubscriberChannel(sub as unknown as RTCDataChannel)).toBe(true); + expect(manager.getHandle(DataChannelKind.RELIABLE, true)).toBe(sub); + expect(sub.onmessage).toBe(opts.onDataMessage); + + const dataTrackSub = new FakeDataChannel('_data_track'); + expect(manager.adoptSubscriberChannel(dataTrackSub as unknown as RTCDataChannel)).toBe(true); + expect(dataTrackSub.onmessage).toBe(opts.onDataTrackMessage); + + expect( + manager.adoptSubscriberChannel(new FakeDataChannel('_other') as unknown as RTCDataChannel), + ).toBe(false); + }); + + it('teardown rejects parked waiters, closes all handles, and detaches', async () => { + const { manager, pcManager, created } = makeManager(); + manager.createPublisherChannels(pcManager); + const sub = new FakeDataChannel('_lossy'); + manager.adoptSubscriberChannel(sub as unknown as RTCDataChannel); + + created._reliable.bufferedAmount = 2 * 1024 * 1024; + const parked = manager.reliable.waitForHeadroomWithLock(); + parked.catch(() => {}); + await tick(); + + manager.teardown(); + + await expect(parked).rejects.toBeInstanceOf(UnexpectedConnectionState); + for (const dc of [...Object.values(created), sub]) { + expect(dc.close).toHaveBeenCalled(); + expect(dc.onmessage).toBeNull(); + } + expect(manager.hasPublisherChannels).toBe(false); + expect(manager.getHandle(DataChannelKind.LOSSY, true)).toBeUndefined(); + + manager.lossy.stopThresholdTuning(); + }); +}); diff --git a/src/room/data-channel/DataChannelManager.ts b/src/room/data-channel/DataChannelManager.ts new file mode 100644 index 0000000000..99f1ab08aa --- /dev/null +++ b/src/room/data-channel/DataChannelManager.ts @@ -0,0 +1,237 @@ +import type { PCTransportManager } from '../PCTransportManager'; +import { FlowControlledDataChannel } from './FlowControlledDataChannel'; +import { LossyDataChannel } from './LossyDataChannel'; +import { ReliableDataChannel } from './ReliableDataChannel'; +import { DataChannelKind, dataChannelHighWaterMark, dataChannelLowWaterMark } from './types'; + +const lossyDataChannelLabel = '_lossy'; +const reliableDataChannelLabel = '_reliable'; +const dataTrackDataChannelLabel = '_data_track'; + +export interface DataChannelManagerOptions { + /** Whether the owning engine has been closed — a closed engine rejects headroom waiters. */ + isEngineClosed: () => boolean; + /** + * Whether a reconnect attempt is underway: reliable sends defer to the resume replay and lossy + * sends are skipped while this is true. + */ + isReconnecting: () => boolean; + onDataMessage: (message: MessageEvent) => void; + onDataTrackMessage: (message: MessageEvent) => void; + onDataError: (event: Event) => void; + onChannelClose: (kind: DataChannelKind) => void; + /** A channel's buffer crossed its low-water mark (debounced). Drives DCBufferStatusChanged. */ + onBufferStatusChanged: (kind: DataChannelKind, isLow: boolean) => void; +} + +/** + * Owns the engine's data channels: the three flow-controlled publisher channel wrappers (which + * live for the engine's lifetime and have RTCDataChannel handles attached/detached as peer + * connections come and go) plus the subscriber-side receive handles adopted by label. + * + * Handle turnover goes through {@link FlowControlledDataChannel.attach}/`detach`, which reject + * parked headroom waiters as a built-in — there is no separate invalidation step to forget. + */ +export class DataChannelManager { + readonly reliable: ReliableDataChannel; + + readonly lossy: LossyDataChannel; + + readonly dataTrack: LossyDataChannel; + + private reliableSub?: RTCDataChannel; + + private lossySub?: RTCDataChannel; + + private dataTrackSub?: RTCDataChannel; + + private opts: DataChannelManagerOptions; + + constructor(opts: DataChannelManagerOptions) { + this.opts = opts; + const flowControlOptions = (kind: DataChannelKind) => ({ + kind, + lowWaterMark: dataChannelLowWaterMark(kind), + highWaterMark: dataChannelHighWaterMark(kind), + isEngineClosed: opts.isEngineClosed, + onBufferStatusChanged: (isLow: boolean) => opts.onBufferStatusChanged(kind, isLow), + }); + this.reliable = new ReliableDataChannel({ + ...flowControlOptions(DataChannelKind.RELIABLE), + isDeferringSends: opts.isReconnecting, + }); + this.lossy = new LossyDataChannel({ + ...flowControlOptions(DataChannelKind.LOSSY), + // Classic lossy user data: a stale packet is worthless, so drop instead of queueing. + bufferFullBehavior: 'drop', + shouldSkipSends: opts.isReconnecting, + }); + this.dataTrack = new LossyDataChannel({ + ...flowControlOptions(DataChannelKind.DATA_TRACK_LOSSY), + // Data tracks backpressure the producer instead — it decides what to skip at frame + // granularity rather than the engine dropping arbitrary chunks out of frames. + bufferFullBehavior: 'wait', + shouldSkipSends: opts.isReconnecting, + }); + } + + /** The flow-control wrapper for `kind`. */ + channelFor(kind: DataChannelKind): FlowControlledDataChannel { + switch (kind) { + case DataChannelKind.RELIABLE: + return this.reliable; + case DataChannelKind.LOSSY: + return this.lossy; + case DataChannelKind.DATA_TRACK_LOSSY: + return this.dataTrack; + } + } + + /** The raw RTCDataChannel handle for `kind`, publisher side by default. */ + getHandle(kind: DataChannelKind, subscriber: boolean = false): RTCDataChannel | undefined { + if (!subscriber) { + return this.channelFor(kind).channelHandle; + } + switch (kind) { + case DataChannelKind.RELIABLE: + return this.reliableSub; + case DataChannelKind.LOSSY: + return this.lossySub; + case DataChannelKind.DATA_TRACK_LOSSY: + return this.dataTrackSub; + } + } + + get hasPublisherChannels(): boolean { + return Boolean( + this.reliable.channelHandle || this.lossy.channelHandle || this.dataTrack.channelHandle, + ); + } + + /** + * Creates the three publisher data channels on the given transport, wires their handlers, and + * attaches them to the wrappers — attaching rejects any waiters still parked on replaced + * channel objects. + */ + createPublisherChannels(pcManager: PCTransportManager) { + // clear old data channel callbacks if recreate + for (const channel of [this.lossy, this.reliable, this.dataTrack]) { + const old = channel.channelHandle; + if (old) { + old.onmessage = null; + old.onerror = null; + old.onclose = null; + } + } + + const wire = ( + channel: FlowControlledDataChannel, + dc: RTCDataChannel, + onMessage: (message: MessageEvent) => void, + ) => { + // also handle messages over the pub channel, for backwards compatibility + dc.onmessage = onMessage; + // handle datachannel errors + dc.onerror = this.opts.onDataError; + // detect unexpected publisher data channel closes + dc.onclose = () => this.opts.onChannelClose(channel.kind); + // set up dc buffer threshold - if this is not set, it will default to 0 + dc.bufferedAmountLowThreshold = channel.lowWaterMark; + // handle buffer amount low events + dc.onbufferedamountlow = () => channel.refreshBufferStatus(); + channel.attach(dc); + }; + + wire( + this.lossy, + pcManager.createPublisherDataChannel(lossyDataChannelLabel, { + ordered: false, + maxRetransmits: 0, + }), + this.opts.onDataMessage, + ); + wire( + this.reliable, + pcManager.createPublisherDataChannel(reliableDataChannelLabel, { + ordered: true, + }), + this.opts.onDataMessage, + ); + wire( + this.dataTrack, + pcManager.createPublisherDataChannel(dataTrackDataChannelLabel, { + ordered: false, + maxRetransmits: 0, + }), + this.opts.onDataTrackMessage, + ); + + this.lossy.startThresholdTuning(); + } + + /** + * Adopts a subscriber-side data channel by label, wiring the matching receive handler. + * Returns false for labels this manager doesn't own. + */ + adoptSubscriberChannel(channel: RTCDataChannel): boolean { + let handler: (message: MessageEvent) => void; + if (channel.label === reliableDataChannelLabel) { + this.reliableSub = channel; + handler = this.opts.onDataMessage; + } else if (channel.label === lossyDataChannelLabel) { + this.lossySub = channel; + handler = this.opts.onDataMessage; + } else if (channel.label === dataTrackDataChannelLabel) { + this.dataTrackSub = channel; + handler = this.opts.onDataTrackMessage; + } else { + return false; + } + channel.onmessage = handler; + return true; + } + + /** + * Tears down all channels for a peer-connection cleanup: rejects parked waiters (detach — the + * spec allows `pc.close()` to transition channels to 'closed' without firing events, so waiting + * for browser close events is not an option), strips handlers, closes the handles, and resets + * the reliable session state. + */ + teardown() { + const dcCleanup = (dc: RTCDataChannel | undefined) => { + if (!dc) { + return; + } + + // Detach the data channel handlers before closing anything. Closing a peer connection tears + // down the SCTP transport, which can dispatch `error`/`close` events on the still-open data + // channels; if our handlers are still attached at that point, handleDataError logs a spurious + // "Unknown DataChannel error" during an otherwise graceful disconnect. Removing the handlers + // before dc.close()/pcManager.close() makes this deterministic regardless of how/when the + // browser dispatches those teardown events. See livekit/client-sdk-js#1953. + dc.onbufferedamountlow = null; + dc.onclose = null; + dc.onclosing = null; + dc.onerror = null; + dc.onmessage = null; + dc.onopen = null; + + dc.close(); + }; + + for (const channel of [this.lossy, this.reliable, this.dataTrack]) { + const dc = channel.channelHandle; + channel.detach('peer connections cleaned up'); + dcCleanup(dc); + } + dcCleanup(this.lossySub); + dcCleanup(this.reliableSub); + dcCleanup(this.dataTrackSub); + this.lossySub = undefined; + this.reliableSub = undefined; + this.dataTrackSub = undefined; + + // Full teardown starts the session over: drop replay state and restart sequencing. + this.reliable.reset(); + } +} diff --git a/src/room/data-channel/FlowControlledDataChannel.test.ts b/src/room/data-channel/FlowControlledDataChannel.test.ts new file mode 100644 index 0000000000..c1e6323e1e --- /dev/null +++ b/src/room/data-channel/FlowControlledDataChannel.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it, vi } from 'vitest'; +import { UnexpectedConnectionState } from '../errors'; +import { FlowControlledDataChannel } from './FlowControlledDataChannel'; +import { DataChannelKind } from './types'; + +class FakeDataChannel extends EventTarget { + bufferedAmount = 0; + + bufferedAmountLowThreshold = 64; +} + +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +// Pass `dc: null` for a handle-less channel; omit it to get a fresh one. `null` (not `undefined`) +// is deliberate — a destructuring default fills in on `undefined`, so only `null` passes through. +function makeChannel({ + dc = new FakeDataChannel(), + engineClosed = false, +}: { dc?: FakeDataChannel | null; engineClosed?: boolean } = {}) { + const state = { engineClosed }; + const onBufferStatusChanged = vi.fn(); + const channel = new FlowControlledDataChannel({ + kind: DataChannelKind.RELIABLE, + lowWaterMark: 64, + highWaterMark: 1024, + isEngineClosed: () => state.engineClosed, + onBufferStatusChanged, + }); + if (dc) { + channel.attach(dc as unknown as RTCDataChannel); + } + return { channel, dc: dc as FakeDataChannel, state, onBufferStatusChanged }; +} + +describe('FlowControlledDataChannel', () => { + describe('refreshBufferStatus', () => { + it('notifies only on a change and in both directions', () => { + const { channel, dc, onBufferStatusChanged } = makeChannel(); + + // Starts "low" (empty buffer); a refresh that stays low emits nothing. + channel.refreshBufferStatus(); + expect(onBufferStatusChanged).not.toHaveBeenCalled(); + + // Cross above the low mark → one not-low notification. + dc.bufferedAmount = 512; + channel.refreshBufferStatus(); + channel.refreshBufferStatus(); // debounced: no second fire + expect(onBufferStatusChanged.mock.calls).toEqual([[false]]); + + // Drain back below → one low notification. + dc.bufferedAmount = 0; + channel.refreshBufferStatus(); + expect(onBufferStatusChanged.mock.calls).toEqual([[false], [true]]); + }); + + it('is a no-op without a handle (no throw, no notification)', () => { + const { channel, onBufferStatusChanged } = makeChannel({ dc: null }); + expect(() => channel.refreshBufferStatus()).not.toThrow(); + expect(onBufferStatusChanged).not.toHaveBeenCalled(); + }); + }); + + it('reports watermark status against the given channel', () => { + const { channel, dc } = makeChannel(); + const handle = dc as unknown as RTCDataChannel; + dc.bufferedAmount = 0; + expect(channel.isBelowHighWaterMark(handle)).toBe(true); + expect(channel.isBelowLowWaterMark(handle)).toBe(true); + + dc.bufferedAmount = 512; // between low (64) and high (1024) + expect(channel.isBelowHighWaterMark(handle)).toBe(true); + expect(channel.isBelowLowWaterMark(handle)).toBe(false); + + dc.bufferedAmount = 2048; + expect(channel.isBelowHighWaterMark(handle)).toBe(false); + }); + + it('waiting for headroom without a handle rejects with a connection error', async () => { + const { channel } = makeChannel({ dc: null }); + await expect(channel.waitForHeadroomWithLock()).rejects.toBeInstanceOf( + UnexpectedConnectionState, + ); + }); + + it('resolves immediately while below the high-water mark', async () => { + const { channel, dc } = makeChannel(); + dc.bufferedAmount = 1024; + await expect(channel.waitForHeadroomWithLock()).resolves.toBeUndefined(); + }); + + it('parks above the high-water mark and resumes on bufferedamountlow', async () => { + const { channel, dc } = makeChannel(); + dc.bufferedAmount = 2048; + const resolved = vi.fn(); + const wait = channel.waitForHeadroomWithLock().then(resolved); + await tick(); + expect(resolved).not.toHaveBeenCalled(); + + dc.bufferedAmount = 32; + dc.dispatchEvent(new Event('bufferedamountlow')); + await wait; + expect(resolved).toHaveBeenCalled(); + }); + + it('rejects a parked waiter when the channel closes', async () => { + const { channel, dc } = makeChannel(); + dc.bufferedAmount = 2048; + const wait = channel.waitForHeadroomWithLock(); + wait.catch(() => {}); + await tick(); + + dc.dispatchEvent(new Event('close')); + await expect(wait).rejects.toBeInstanceOf(UnexpectedConnectionState); + }); + + it('rejects a parked waiter on invalidateWaiters and recovers with a fresh controller', async () => { + const { channel, dc } = makeChannel(); + dc.bufferedAmount = 2048; + const wait = channel.waitForHeadroomWithLock(); + wait.catch(() => {}); + await tick(); + + channel.invalidateWaiters('channel replaced'); + await expect(wait).rejects.toBeInstanceOf(UnexpectedConnectionState); + + // Fresh controller: the gate is usable again and the lock was released. + dc.bufferedAmount = 0; + await expect(channel.waitForHeadroomWithLock()).resolves.toBeUndefined(); + }); + + it('rejects immediately when the engine is closed', async () => { + const { channel, state } = makeChannel(); + state.engineClosed = true; + await expect(channel.waitForHeadroomWithLock()).rejects.toBeInstanceOf( + UnexpectedConnectionState, + ); + }); + + it('serializes waiters FIFO through the headroom lock', async () => { + const { channel, dc } = makeChannel(); + dc.bufferedAmount = 2048; + const order: number[] = []; + const first = channel.waitForHeadroomWithLock().then(() => order.push(1)); + const second = channel.waitForHeadroomWithLock().then(() => order.push(2)); + await tick(); + + // One drain event wakes the head waiter; the second re-checks under the lock and, with the + // buffer now low, proceeds right after — strictly in arrival order. + dc.bufferedAmount = 0; + dc.dispatchEvent(new Event('bufferedamountlow')); + await Promise.all([first, second]); + expect(order).toEqual([1, 2]); + }); +}); diff --git a/src/room/data-channel/FlowControlledDataChannel.ts b/src/room/data-channel/FlowControlledDataChannel.ts new file mode 100644 index 0000000000..0d00408901 --- /dev/null +++ b/src/room/data-channel/FlowControlledDataChannel.ts @@ -0,0 +1,216 @@ +import { Mutex } from '@livekit/mutex'; +import TypedPromise from '../../utils/TypedPromise'; +import { UnexpectedConnectionState } from '../errors'; +import type { DataChannelKind } from './types'; + +export interface FlowControlledDataChannelOptions { + kind: DataChannelKind; + /** Buffer level (bytes) at which blocked senders resume; armed as `bufferedAmountLowThreshold`. */ + lowWaterMark: number; + /** Buffer level (bytes) above which senders block until the buffer drains to the low mark. */ + highWaterMark: number; + /** Whether the owning engine has been closed — a closed engine rejects waiters immediately. */ + isEngineClosed: () => boolean; + /** + * Notified when the buffer crosses the low-water mark in either direction (debounced: fires only + * on an actual change). Drives the engine's public DCBufferStatusChanged event. + */ + onBufferStatusChanged?: (isLow: boolean) => void; +} + +/** + * Two-watermark flow control for one data channel kind. + * + * Owns the per-kind headroom gate: senders proceed freely while the buffer is at or below the + * high-water mark and otherwise block — serialized FIFO through a mutex — until the browser's + * `bufferedamountlow` event (armed at the low-water mark) signals the buffer has drained. The + * serialization prevents woken senders from all refilling at once and overflowing the SCTP send + * buffer (see livekit/client-sdk-js#1995). + * + * Waiters are parked on the channel object captured at wait entry. If that object stops being + * current — replaced or torn down — its events may never fire again, so the owner must call + * {@link invalidateWaiters}, which aborts parked waiters (releasing the gate); the next waiter + * gets a fresh controller. + */ +export class FlowControlledDataChannel { + readonly kind: DataChannelKind; + + readonly lowWaterMark: number; + + readonly highWaterMark: number; + + protected isEngineClosed: () => boolean; + + private onBufferStatusChanged?: (isLow: boolean) => void; + + /** Last emitted low-water status; starts true (an empty buffer is below the mark). */ + private bufferStatusLow = true; + + private handle?: RTCDataChannel; + + private headroomLock = new Mutex(); + + /** Cancels parked headroom waiters when the handle is replaced or torn down. */ + private waiterAbortController = new AbortController(); + + constructor(opts: FlowControlledDataChannelOptions) { + this.kind = opts.kind; + this.lowWaterMark = opts.lowWaterMark; + this.highWaterMark = opts.highWaterMark; + this.isEngineClosed = opts.isEngineClosed; + this.onBufferStatusChanged = opts.onBufferStatusChanged; + } + + /** The currently attached RTCDataChannel handle, if any. */ + get channelHandle(): RTCDataChannel | undefined { + return this.handle; + } + + /** + * Attaches the channel handle this wrapper controls. Replacing an existing handle rejects + * parked waiters — their events would never fire again on the abandoned object — and installs a + * fresh controller, so queued senders re-check against the new channel. Wrappers outlive their + * handles: this is the one place handle turnover happens, which is what makes stranding a + * waiter structurally impossible. + */ + attach(dc: RTCDataChannel) { + if (this.handle && this.handle !== dc) { + this.invalidateWaiters('data channel replaced'); + } + this.handle = dc; + } + + /** Detaches the handle on teardown, rejecting parked waiters. */ + detach(reason: string = 'data channel torn down') { + if (this.handle) { + this.invalidateWaiters(reason); + } + this.handle = undefined; + } + + protected getChannel(): RTCDataChannel | undefined { + return this.handle; + } + + /** + * Whether the send buffer has room to accept more data (the send gate). Senders proceed while + * this is true and block once it goes false. Callers resolve the handle (and decide what an + * absent one means) before asking. + */ + isBelowHighWaterMark(dc: RTCDataChannel): boolean { + // RTCDataChannel has no built-in high-water mark, so we compare against our static mark. + return dc.bufferedAmount <= this.highWaterMark; + } + + /** + * Whether the send buffer has drained to its low-water mark. Drives the engine's public + * DCBufferStatusChanged event. + */ + isBelowLowWaterMark(dc: RTCDataChannel): boolean { + // Read the channel's own threshold: it is tuned dynamically for the lossy channel. + return dc.bufferedAmount <= dc.bufferedAmountLowThreshold; + } + + /** + * Acquires the headroom lock, resolving with the unlock function. Batch senders (the resume + * replay) hold it across all of their sends so no other sender can interleave, calling + * {@link waitForHeadroomWithoutLock} per message to respect flow control within the batch. + */ + lockHeadroom(): Promise<() => void> { + return this.headroomLock.lock(); + } + + /** + * Resolves once the caller may send on this channel: immediately while the send buffer is at or + * below its high-water mark, otherwise once the buffer has drained to the low-water mark (the + * `bufferedamountlow` event). Callers are serialized through the headroom lock so that, when + * the buffer drains, they refill it one at a time (up to the high-water mark) rather than all + * sending at once and overflowing the SCTP send buffer (see livekit/client-sdk-js#1995). The + * closed/buffer checks run inside the lock so queued callers proceed in FIFO order. + */ + async waitForHeadroomWithLock() { + const unlock = await this.lockHeadroom(); + try { + await this.waitForHeadroomWithoutLock(); + } finally { + unlock(); + } + } + + /** Core wait of {@link waitForHeadroomWithLock}. The caller must hold the headroom lock. */ + async waitForHeadroomWithoutLock() { + if (this.isEngineClosed()) { + throw new UnexpectedConnectionState('engine closed'); + } + const dc = this.getChannel(); + if (!dc) { + throw new UnexpectedConnectionState(`DataChannel not found, kind: ${this.kind}`); + } + if (this.isBelowHighWaterMark(dc)) { + return; + } + const abortSignal = this.waiterAbortController.signal; + await new TypedPromise((resolve, reject) => { + const onBufferedAmountLow = () => { + cleanup(); + resolve(); + }; + const onDCClose = () => { + cleanup(); + reject( + new UnexpectedConnectionState( + `DataChannel ${this.kind} closed while draining the buffer`, + ), + ); + }; + const onAbort = () => { + cleanup(); + reject( + new UnexpectedConnectionState( + `DataChannel ${this.kind} was replaced or torn down while waiting for headroom`, + ), + ); + }; + const cleanup = () => { + dc.removeEventListener('bufferedamountlow', onBufferedAmountLow); + dc.removeEventListener('close', onDCClose); + abortSignal.removeEventListener('abort', onAbort); + }; + if (abortSignal.aborted) { + onAbort(); + return; + } + dc.addEventListener('bufferedamountlow', onBufferedAmountLow); + // Proxy along any error caused by the data channel closing while we wait. + dc.addEventListener('close', onDCClose); + // The channel object we're parked on can be abandoned without ever firing another event + // (e.g. the engine recreating channels); the abort is our way out. + abortSignal.addEventListener('abort', onAbort); + }); + } + + /** Rejects all parked headroom waiters; the next waiter gets a fresh controller. */ + invalidateWaiters(reason: string) { + this.waiterAbortController.abort(reason); + this.waiterAbortController = new AbortController(); + } + + /** + * Recomputes whether the buffer has drained to the low-water mark and, if that changed since the + * last check, notifies the status listener. Two independent triggers land here: a send (which + * raises the buffer) and the `bufferedamountlow` drain event (which lowers it) — the latter has + * no send to hang the work off, which is why this is a shared entry point rather than a tail of + * `send`. + */ + refreshBufferStatus() { + const dc = this.getChannel(); + if (!dc) { + return; + } + const isLow = this.isBelowLowWaterMark(dc); + if (isLow !== this.bufferStatusLow) { + this.bufferStatusLow = isLow; + this.onBufferStatusChanged?.(isLow); + } + } +} diff --git a/src/room/data-channel/LossyDataChannel.test.ts b/src/room/data-channel/LossyDataChannel.test.ts new file mode 100644 index 0000000000..3bcd26b742 --- /dev/null +++ b/src/room/data-channel/LossyDataChannel.test.ts @@ -0,0 +1,118 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { LossyDataChannel } from './LossyDataChannel'; +import { DataChannelKind } from './types'; + +class FakeDataChannel extends EventTarget { + bufferedAmount = 0; + + bufferedAmountLowThreshold = 64; + + send = vi.fn(); +} + +// Pass `dc: null` for a handle-less channel; omit it to get a fresh one. `null` (not `undefined`) +// is deliberate — a destructuring default fills in on `undefined`, so only `null` passes through. +function makeChannel( + bufferFullBehavior: 'drop' | 'wait', + { dc = new FakeDataChannel() }: { dc?: FakeDataChannel | null } = {}, +) { + const state = { engineClosed: false, skipSends: false }; + const channel = new LossyDataChannel({ + kind: DataChannelKind.LOSSY, + lowWaterMark: 64, + highWaterMark: 1024, + isEngineClosed: () => state.engineClosed, + bufferFullBehavior, + shouldSkipSends: () => state.skipSends, + }); + if (dc) { + channel.attach(dc as unknown as RTCDataChannel); + } + const stats = channel as unknown as { statCurrentBytes: number; dropCount: number }; + return { channel, dc, state, stats }; +} + +describe('LossyDataChannel', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('drop policy: sends below the threshold and counts bytes', async () => { + const { channel, dc, stats } = makeChannel('drop'); + const msg = new Uint8Array(10); + + await channel.send(msg); + + expect(dc.send).toHaveBeenCalledWith(msg); + expect(stats.statCurrentBytes).toBe(10); + }); + + it('drop policy: discards while the buffer is above the dc threshold', async () => { + const { channel, dc, stats } = makeChannel('drop'); + dc.bufferedAmount = 128; // above the 64-byte dc threshold + + await channel.send(new Uint8Array(10)); + + expect(dc.send).not.toHaveBeenCalled(); + expect(stats.dropCount).toBe(1); + expect(stats.statCurrentBytes).toBe(0); + }); + + it('wait policy: parks above the high-water mark instead of dropping', async () => { + const { channel, dc } = makeChannel('wait'); + dc.bufferedAmount = 2048; + const resolved = vi.fn(); + const send = channel.send(new Uint8Array(10)).then(resolved); + await vi.advanceTimersByTimeAsync(0); + expect(resolved).not.toHaveBeenCalled(); + + dc.bufferedAmount = 0; + dc.dispatchEvent(new Event('bufferedamountlow')); + await send; + expect(dc.send).toHaveBeenCalledTimes(1); + }); + + it('skips sends while a reconnect is underway', async () => { + const { channel, dc, state } = makeChannel('drop'); + state.skipSends = true; + + await channel.send(new Uint8Array(10)); + + expect(dc.send).not.toHaveBeenCalled(); + }); + + it('tunes the dc threshold to ~100ms of observed byterate, clamped to the watermarks', async () => { + const { channel, dc } = makeChannel('drop'); + channel.startThresholdTuning(); + + // 10_000 bytes/second → threshold byterate/10 = 1000, clamped to the 1024 high mark? No — + // 1000 is within [64, 1024], so it applies as-is. + await channel.send(new Uint8Array(10_000)); + await vi.advanceTimersByTimeAsync(1000); + expect(dc.bufferedAmountLowThreshold).toBe(1000); + + // Idle second → byterate 0 → clamps up to the low-water mark. + await vi.advanceTimersByTimeAsync(1000); + expect(dc.bufferedAmountLowThreshold).toBe(64); + + channel.stopThresholdTuning(); + }); + + it('stopThresholdTuning halts adjustments and resets stats', async () => { + const { channel, dc, stats } = makeChannel('drop'); + channel.startThresholdTuning(); + await channel.send(new Uint8Array(500)); + + channel.stopThresholdTuning(); + expect(stats.statCurrentBytes).toBe(0); + expect(stats.dropCount).toBe(0); + + const before = dc.bufferedAmountLowThreshold; + await vi.advanceTimersByTimeAsync(2000); + expect(dc.bufferedAmountLowThreshold).toBe(before); + }); +}); diff --git a/src/room/data-channel/LossyDataChannel.ts b/src/room/data-channel/LossyDataChannel.ts new file mode 100644 index 0000000000..2fa22a8b2d --- /dev/null +++ b/src/room/data-channel/LossyDataChannel.ts @@ -0,0 +1,125 @@ +import log from '../../logger'; +import type { NonSharedUint8Array } from '../../type-polyfills/non-shared-typed-arrays'; +import CriticalTimers from '../timers'; +import { + FlowControlledDataChannel, + type FlowControlledDataChannelOptions, +} from './FlowControlledDataChannel'; + +export interface LossyDataChannelOptions extends FlowControlledDataChannelOptions { + /** + * What to do with a send while the buffer is full: `drop` discards it to keep latency bounded + * (the classic lossy channel), `wait` backpressures the producer until there is headroom (the + * data-track channel, whose producer decides what to skip at frame granularity). + */ + bufferFullBehavior: 'drop' | 'wait'; + /** Sends are silently discarded while this is true (a reconnect attempt is underway). */ + shouldSkipSends: () => boolean; +} + +/** + * A lossy channel: flow control plus a per-instance full-buffer policy. + * + * Each instance owns its own byterate stat, drop counter, and (when tuning is started) the + * dynamic `bufferedAmountLowThreshold` adjustment that keeps the drop gate at roughly 100ms of + * buffered latency. Keeping these per instance is what prevents one channel's traffic from + * steering another channel's policy. + */ +export class LossyDataChannel extends FlowControlledDataChannel { + private bufferFullBehavior: 'drop' | 'wait'; + + private shouldSkipSends: () => boolean; + + private statCurrentBytes = 0; + + private statByterate = 0; + + private statInterval: ReturnType | undefined; + + private dropCount = 0; + + constructor(opts: LossyDataChannelOptions) { + super(opts); + this.bufferFullBehavior = opts.bufferFullBehavior; + this.shouldSkipSends = opts.shouldSkipSends; + } + + /** Sends prepared bytes with this channel's full-buffer policy (drop or wait). */ + async send(msg: NonSharedUint8Array) { + const dc = this.getChannel(); + if (!dc) { + return; + } + // Depending on the channel's policy, either drop or wait for the buffer to drain below the + // high-water mark before continuing. + switch (this.bufferFullBehavior) { + case 'wait': + if (!this.isBelowHighWaterMark(dc)) { + await this.waitForHeadroomWithLock(); + } + break; + case 'drop': + // We check against the actual threshold on the DC here, as it is tuned dynamically. + if (!this.isBelowLowWaterMark(dc)) { + // Drop messages to reduce latency + this.dropCount += 1; + if (this.dropCount % 100 === 0) { + log.warn(`dropping lossy data channel messages, total dropped: ${this.dropCount}`); + } + return; + } + } + this.statCurrentBytes += msg.byteLength; + + if (this.shouldSkipSends()) { + return; + } + + try { + dc.send(msg); + this.refreshBufferStatus(); + } catch (error: unknown) { + // Preserve prior surface behaviour: a send that fails because the channel is closing is + // logged, not thrown, so lossy/data-track sends don't reject during teardown windows. + if (error instanceof TypeError) { + log.error(error); + } else { + throw error; + } + } + } + + /** + * Starts the once-per-second adjustment of the channel's `bufferedAmountLowThreshold` to the + * observed byterate, keeping the drop gate at roughly 100ms of buffered latency (clamped to + * the watermarks). Restarts cleanly if already running. + */ + startThresholdTuning() { + this.stopThresholdTuning(); + this.statInterval = CriticalTimers.setInterval(() => { + this.statByterate = this.statCurrentBytes; + this.statCurrentBytes = 0; + + const dc = this.getChannel(); + if (dc) { + // control buffered latency to ~100ms + const threshold = this.statByterate / 10; + dc.bufferedAmountLowThreshold = Math.min( + Math.max(threshold, this.lowWaterMark), + this.highWaterMark, + ); + } + }, 1000); + } + + /** Stops the threshold tuning and resets the stats and drop counter. */ + stopThresholdTuning() { + this.statByterate = 0; + this.statCurrentBytes = 0; + if (this.statInterval) { + CriticalTimers.clearInterval(this.statInterval); + this.statInterval = undefined; + } + this.dropCount = 0; + } +} diff --git a/src/room/data-channel/ReliableDataChannel.test.ts b/src/room/data-channel/ReliableDataChannel.test.ts new file mode 100644 index 0000000000..ac3a096e3a --- /dev/null +++ b/src/room/data-channel/ReliableDataChannel.test.ts @@ -0,0 +1,158 @@ +import { describe, expect, it, vi } from 'vitest'; +import { UnexpectedConnectionState } from '../errors'; +import { ReliableDataChannel } from './ReliableDataChannel'; +import { DataChannelKind } from './types'; + +class FakeDataChannel extends EventTarget { + bufferedAmount = 0; + + bufferedAmountLowThreshold = 64; + + send = vi.fn(); +} + +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +// Pass `dc: null` for a handle-less channel; omit it to get a fresh one. `null` (not `undefined`) +// is deliberate — a destructuring default fills in on `undefined`, so only `null` passes through. +function makeChannel({ dc = new FakeDataChannel() }: { dc?: FakeDataChannel | null } = {}) { + const state = { engineClosed: false, deferring: false }; + const channel = new ReliableDataChannel({ + kind: DataChannelKind.RELIABLE, + lowWaterMark: 64, + highWaterMark: 1024, + isEngineClosed: () => state.engineClosed, + isDeferringSends: () => state.deferring, + }); + if (dc) { + channel.attach(dc as unknown as RTCDataChannel); + } + const buffer = (channel as unknown as { messageBuffer: { getAll(): unknown[]; length: number } }) + .messageBuffer as unknown as { + getAll(): Array<{ data: Uint8Array; sequence: number; sent: boolean }>; + length: number; + }; + return { channel, dc, state, buffer }; +} + +describe('ReliableDataChannel', () => { + it('hands out monotonic sequences and resets them with the session', () => { + const { channel } = makeChannel(); + expect(channel.nextSequence()).toBe(1); + expect(channel.nextSequence()).toBe(2); + channel.reset(); + expect(channel.nextSequence()).toBe(1); + }); + + it('sends below the high-water mark and retains the packet for replay as sent', async () => { + const { channel, dc, buffer } = makeChannel(); + const msg = new Uint8Array([1]); + + await channel.send(msg, channel.nextSequence()); + + expect(dc.send).toHaveBeenCalledWith(msg); + expect(buffer.getAll()).toEqual([{ data: msg, sequence: 1, sent: true }]); + }); + + it('queues unsent while sends are deferred (reconnect window)', async () => { + const { channel, dc, state, buffer } = makeChannel(); + state.deferring = true; + + await channel.send(new Uint8Array([1]), channel.nextSequence()); + + expect(dc.send).not.toHaveBeenCalled(); + expect(buffer.getAll()[0].sent).toBe(false); + }); + + it('queues unsent and resolves when the headroom wait is torn down transiently', async () => { + const { channel, dc, buffer } = makeChannel(); + dc.bufferedAmount = 2048; // above high mark → send parks + const send = channel.send(new Uint8Array([1]), channel.nextSequence()); + await tick(); + + channel.invalidateWaiters('channel replaced'); + + await expect(send).resolves.toBeUndefined(); + expect(dc.send).not.toHaveBeenCalled(); + expect(buffer.getAll()[0].sent).toBe(false); + }); + + it('rejects when the engine is closed while waiting', async () => { + const { channel, dc, state } = makeChannel(); + dc.bufferedAmount = 2048; + const send = channel.send(new Uint8Array([1]), channel.nextSequence()); + send.catch(() => {}); + await tick(); + + state.engineClosed = true; + channel.invalidateWaiters('engine closed'); + + await expect(send).rejects.toBeInstanceOf(UnexpectedConnectionState); + }); + + it('replay drops acked packets, resends the rest in order, and marks them sent', async () => { + const { channel, dc, buffer } = makeChannel(); + const acked = new Uint8Array([1]); + const unacked = new Uint8Array([2]); + await channel.send(acked, channel.nextSequence()); + await channel.send(unacked, channel.nextSequence()); + // A packet queued unsent during the reconnect window joins the replay. + const queued = new Uint8Array([3]); + (channel as unknown as { isDeferringSends: () => boolean }).isDeferringSends = () => true; + await channel.send(queued, 3); + dc.send.mockClear(); + + await channel.replay(1); // server acked sequence 1 + + expect(dc.send.mock.calls.map(([data]) => data)).toEqual([unacked, queued]); + expect(buffer.getAll().every((item) => item.sent)).toBe(true); + }); + + it('transmits a packet deferred mid-replay instead of marking it sent without sending', async () => { + const { channel, dc, state, buffer } = makeChannel(); + // A packet sent before the disconnect, still buffered (unacked) at resume. + await channel.send(new Uint8Array([1]), channel.nextSequence()); + dc.send.mockClear(); + + // Replay parks on a full buffer, giving an await window mid-drain. + dc.bufferedAmount = 2048; + const replay = channel.replay(0); + await tick(); + + // A send arrives while replay is parked; the reconnect is still active, so it defers into the + // buffer as sent:false — after replay's first drain pass already started. + state.deferring = true; + await channel.send(new Uint8Array([2]), channel.nextSequence()); + state.deferring = false; + + dc.bufferedAmount = 0; + dc.dispatchEvent(new Event('bufferedamountlow')); + await replay; + + // The drain loop must transmit both; a blanket mark-all-sent would flip the late arrival to + // sent without sending it, and a later align would then strand it. + expect(dc.send.mock.calls.map(([d]) => d[0])).toEqual([1, 2]); + expect(buffer.getAll().filter((i) => !i.sent)).toHaveLength(0); + }); + + it('holds the headroom lock across the whole replay so new sends cannot interleave', async () => { + const { channel, dc } = makeChannel(); + (channel as unknown as { isDeferringSends: () => boolean }).isDeferringSends = () => true; + await channel.send(new Uint8Array([1]), 1); + await channel.send(new Uint8Array([2]), 2); + (channel as unknown as { isDeferringSends: () => boolean }).isDeferringSends = () => false; + + // Replay parks on a full buffer while a concurrent send races it. + dc.bufferedAmount = 2048; + const replay = channel.replay(0); + await tick(); + const concurrent = channel.send(new Uint8Array([9]), channel.nextSequence()); + await tick(); + + dc.bufferedAmount = 0; + dc.dispatchEvent(new Event('bufferedamountlow')); + await Promise.all([replay, concurrent]); + + expect(dc.send.mock.calls.map(([data]) => data[0])).toEqual([1, 2, 9]); + }); +}); diff --git a/src/room/data-channel/ReliableDataChannel.ts b/src/room/data-channel/ReliableDataChannel.ts new file mode 100644 index 0000000000..25f44f01a3 --- /dev/null +++ b/src/room/data-channel/ReliableDataChannel.ts @@ -0,0 +1,153 @@ +import type { NonSharedUint8Array } from '../../type-polyfills/non-shared-typed-arrays'; +import { DataPacketBuffer } from '../../utils/dataPacketBuffer'; +import { + FlowControlledDataChannel, + type FlowControlledDataChannelOptions, +} from './FlowControlledDataChannel'; + +export interface ReliableDataChannelOptions extends FlowControlledDataChannelOptions { + /** + * Whether sends should currently be deferred to the resume replay instead of hitting the wire + * (i.e. a reconnect attempt is underway). Read at send time so the reliable channel matches the + * engine's reconnect state without owning it. + */ + isDeferringSends: () => boolean; +} + +/** + * The reliable channel: flow control plus delivery-across-resume semantics. + * + * Every packet gets a monotonic sequence (stamped into the protobuf by the caller before + * serialization, via {@link nextSequence}) and is retained in a replay buffer until the channel's + * `bufferedAmount` confirms it has been handed to the transport. Sends that land in a reconnect + * window — or whose headroom wait is torn down transiently — are queued unsent and resolve; + * {@link replay} delivers them (plus any unacked packets) after a resume. Only an engine close + * rejects, because no replay is coming after that. + */ +export class ReliableDataChannel extends FlowControlledDataChannel { + private messageBuffer = new DataPacketBuffer(); + + private sequence = 1; + + private isDeferringSends: () => boolean; + + constructor(opts: ReliableDataChannelOptions) { + super(opts); + this.isDeferringSends = opts.isDeferringSends; + } + + /** + * Claims the next packet sequence. The caller stamps it into the packet before serialization, + * then passes it back to {@link send} so the replay buffer stays keyed by wire sequence. + */ + nextSequence(): number { + const sequence = this.sequence; + this.sequence += 1; + return sequence; + } + + /** + * Sends prepared bytes with reliable semantics. Resolves once the packet has either been handed + * to the channel or queued for the resume replay; throws only when the engine is closed. + */ + async send(msg: NonSharedUint8Array, sequence: number) { + if (this.isDeferringSends()) { + // A reconnect is already underway — queue for the resume replay instead of parking on a + // channel that is being torn down. The send resolves; delivery is deferred to the replay. + this.messageBuffer.push({ data: msg, sequence, sent: false }); + return; + } + + const dc = this.getChannel(); + if (!dc) { + return; + } + + try { + await this.waitForHeadroomWithLock(); + } catch (error) { + if (this.isEngineClosed()) { + // No replay is coming after an engine close — surface the failure. + throw error; + } + // Transient teardown (the channel closed or was replaced while we waited): the reliable + // channel promises delivery across resume, so queue the packet for the replay instead of + // rejecting a send the app can't meaningfully retry. + this.messageBuffer.push({ data: msg, sequence, sent: false }); + return; + } + + if (this.isDeferringSends()) { + // A reconnect began while we waited for headroom — same deal as above. + this.messageBuffer.push({ data: msg, sequence, sent: false }); + return; + } + + this.messageBuffer.push({ data: msg, sequence, sent: true }); + dc.send(msg); + this.refreshBufferStatus(); + } + + /** + * Replays the buffered backlog after a resume: drops everything the server acked + * (`lastMessageSeq`), then re-sends the rest in order. The headroom lock is held across the + * whole replay — releasing it between messages would let a concurrent send (whose newer + * sequence was already assigned before it queued on the lock) hit the wire mid-replay, and + * receivers would then discard the remaining lower-sequence resent messages as duplicates. + */ + async replay(lastMessageSeq: number) { + const dc = this.getChannel(); + if (!dc) { + return; + } + this.messageBuffer.popToSequence(lastMessageSeq); + const unlock = await this.lockHeadroom(); + try { + // Everything left after the ack cutoff must be re-handed to the current channel. + this.messageBuffer.markAllUnsent(); + // Drain in passes, re-scanning the live buffer each time: a send that arrives (deferred, + // sent:false) during our own awaits appends after this pass started, so we pick it up on + // the next one. Mark each packet only once we've actually handed it to the channel — a + // blanket "mark all sent" would flip such a late arrival to sent without transmitting it, + // and a later alignBufferedAmount would then drop it for good. If the loop throws + // mid-drain, unsent entries keep their flag and the next replay picks them up. + for ( + let batch = this.messageBuffer.getUnsent(); + batch.length > 0; + batch = this.messageBuffer.getUnsent() + ) { + for (const item of batch) { + // Respect flow control on resume too, so a large resend doesn't overflow the buffer. + await this.waitForHeadroomWithoutLock(); + dc.send(item.data); + this.messageBuffer.markSent(item); + } + } + } finally { + unlock(); + } + this.refreshBufferStatus(); + } + + /** + * Before recomputing status, trim packets the transport has now delivered — a send or a drain + * may have acked buffered packets, and the replay buffer is keyed off the channel's buffered + * bytes. + */ + override refreshBufferStatus() { + const dc = this.channelHandle; + if (dc) { + this.messageBuffer.alignBufferedAmount(dc.bufferedAmount); + } + super.refreshBufferStatus(); + } + + /** + * Drops all replay state and restarts sequencing. Only valid on a full reconnect, where the + * session (and the receivers' sequence tracking) starts over. + */ + reset() { + this.messageBuffer = new DataPacketBuffer(); + this.sequence = 1; + } +} diff --git a/src/room/data-channel/types.ts b/src/room/data-channel/types.ts new file mode 100644 index 0000000000..a2ab65fbd8 --- /dev/null +++ b/src/room/data-channel/types.ts @@ -0,0 +1,30 @@ +import { DataPacket_Kind } from '@livekit/protocol'; + +export enum DataChannelKind { + RELIABLE = DataPacket_Kind.RELIABLE, + LOSSY = DataPacket_Kind.LOSSY, + DATA_TRACK_LOSSY = 2, +} + +// Two-watermark flow control for the reliable and data-track channels. Senders fill the buffer +// freely up to the high-water mark; once it's exceeded they block until the browser's +// `bufferedamountlow` event (which we arm at the low-water mark) signals the buffer has drained. +// The gap between the marks keeps the SCTP send buffer saturated while we refill, so throughput +// isn't starved, while the high-water mark bounds the buffer well below the level that would abort +// the channel (see livekit/client-sdk-js#1995). +export const reliableDataChannelWaterMarkLow = 64 * 1024; +export const reliableDataChannelWaterMarkHigh = 1024 * 1024; +export const lossyDataChannelWaterMarkLow = 8 * 1024; +export const lossyDataChannelWaterMarkHigh = 256 * 1024; + +export function dataChannelLowWaterMark(kind: DataChannelKind): number { + return kind === DataChannelKind.RELIABLE + ? reliableDataChannelWaterMarkLow + : lossyDataChannelWaterMarkLow; +} + +export function dataChannelHighWaterMark(kind: DataChannelKind): number { + return kind === DataChannelKind.RELIABLE + ? reliableDataChannelWaterMarkHigh + : lossyDataChannelWaterMarkHigh; +}