diff --git a/.changeset/data-channel-buffer-range.md b/.changeset/data-channel-buffer-range.md new file mode 100644 index 0000000000..f76cf60a5c --- /dev/null +++ b/.changeset/data-channel-buffer-range.md @@ -0,0 +1,5 @@ +--- +"livekit-client": patch +--- + +Fix the reliable data channel dying under concurrent / multi-packet writes (#1995). Data-channel sends now use two-watermark flow control (fill up to a high-water mark, resume when drained to a low-water mark) and serialize contended senders through a per-kind lock, which prevents the SCTP send buffer from overflowing while keeping throughput saturated for data tracks. diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index d4c510a44e..d690e0a3fd 100644 --- a/src/room/RTCEngine.test.ts +++ b/src/room/RTCEngine.test.ts @@ -1,8 +1,9 @@ import { DataPacket, DataPacket_Kind, UserPacket } from '@livekit/protocol'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { DataPacketBuffer, DataPacketItem } from '../utils/dataPacketBuffer'; import RTCEngine, { DataChannelKind } from './RTCEngine'; import { roomOptionDefaults } from './defaults'; -import { PublishDataError } from './errors'; +import { PublishDataError, UnexpectedConnectionState } from './errors'; describe('RTCEngine', () => { const originalRTCRtpSender = window.RTCRtpSender; @@ -230,7 +231,7 @@ describe('RTCEngine', () => { const send = vi.fn(); Object.assign(engine as unknown as Record, { ensurePublisherConnected: vi.fn().mockResolvedValue(undefined), - waitForBufferStatusLow: vi.fn().mockResolvedValue(undefined), + waitForBufferHeadroomWithLock: vi.fn().mockResolvedValue(undefined), updateAndEmitDCBufferStatus: vi.fn(), dataChannelForKind: vi.fn(() => ({ send })), pcManager: { @@ -298,6 +299,278 @@ describe('RTCEngine', () => { }); }); + class FakeDataChannel extends EventTarget { + bufferedAmount = 0; + + bufferedAmountLowThreshold = 64 * 1024; + + send = vi.fn(); + } + + const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + + describe('resendReliableMessagesForResume', () => { + it('does not let a concurrent reliable send interleave into the resume replay', 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), + pcManager: { + getMaxPublisherMessageSize: vi.fn(() => 64 * 1024 - 1), + }, + }); + + // Two messages queued for replay, and a full buffer so the XX 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; + 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 + + const replay = ( + engine as unknown as { resendReliableMessagesForResume: (seq: number) => Promise } + ).resendReliableMessagesForResume(0); + await tick(); + + // A send racing the replay: its sequence is assigned immediately, but it must not hit the + // wire before the replayed (lower-sequence) messages, or receivers discard those as dupes. + const concurrentSend = engine.sendDataPacket( + new DataPacket({ + kind: DataPacket_Kind.RELIABLE, + value: { case: 'user', value: new UserPacket({ payload: new Uint8Array([3]) }) }, + }), + DataChannelKind.RELIABLE, + ); + await tick(); + + // Buffer drains: the replay must finish its whole batch before the concurrent send. + dc.bufferedAmount = 0; + dc.dispatchEvent(new Event('bufferedamountlow')); + await Promise.all([replay, concurrentSend]); + + expect(dc.send).toHaveBeenCalledTimes(3); + expect(dc.send.mock.calls[0][0]).toBe(replayed1); + expect(dc.send.mock.calls[1][0]).toBe(replayed2); + 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', () => { + const makePacket = (byte: number) => + new DataPacket({ + kind: DataPacket_Kind.RELIABLE, + value: { case: 'user', value: new UserPacket({ payload: new Uint8Array([byte]) }) }, + }); + + function stubEngine(engine: RTCEngine, dc: FakeDataChannel) { + 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; + } + + it('resolves and queues the packet for replay when the wait is torn down transiently', async () => { + const engine = new RTCEngine(roomOptionDefaults); + const dc = new FakeDataChannel(); + const buffer = stubEngine(engine, dc); + + // Park the send on a full buffer, then invalidate the channel (reconnect/replacement). + 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'); + + // The send must not surface the teardown — the packet is queued for the resume replay. + await expect(send).resolves.toBeUndefined(); + expect(dc.send).not.toHaveBeenCalled(); + expect(buffer.length).toBe(1); + expect(buffer.getAll()[0].sent).toBe(false); + + // The replay then delivers it. + dc.bufferedAmount = 0; + await ( + engine as unknown as { resendReliableMessagesForResume: (seq: number) => Promise } + ).resendReliableMessagesForResume(0); + expect(dc.send).toHaveBeenCalledTimes(1); + expect(buffer.getAll()[0].sent).toBe(true); + }); + + it('still rejects when the engine is closed while waiting', async () => { + const engine = new RTCEngine(roomOptionDefaults); + const dc = new FakeDataChannel(); + stubEngine(engine, dc); + + dc.bufferedAmount = 2 * 1024 * 1024; + 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'); + + await expect(send).rejects.toBeInstanceOf(UnexpectedConnectionState); + expect(dc.send).not.toHaveBeenCalled(); + }); + + it('queues without waiting while a reconnect attempt is in progress', async () => { + const engine = new RTCEngine(roomOptionDefaults); + const dc = new FakeDataChannel(); + const buffer = stubEngine(engine, dc); + Object.assign(engine as unknown as Record, { attemptingReconnect: true }); + + // Even with a full buffer, the send resolves immediately instead of parking. + dc.bufferedAmount = 2 * 1024 * 1024; + await expect( + engine.sendDataPacket(makePacket(1), DataChannelKind.RELIABLE), + ).resolves.toBeUndefined(); + expect(dc.send).not.toHaveBeenCalled(); + expect(buffer.length).toBe(1); + expect(buffer.getAll()[0].sent).toBe(false); + }); + }); + + describe('sendLossyBytes', () => { + 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; + }); + 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'); + + 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 () => { + 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); + }); + }); + + describe('waitForBufferHeadroomWithLock', () => { + 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), + }); + + // 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); + // 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'); + + 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(); + }); + }); + describe('handleDataChannelClose', () => { function stubCloseEnv( engine: RTCEngine, diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index f3f6028b92..abacb738fc 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -44,6 +44,7 @@ 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'; @@ -91,7 +92,6 @@ import type { TrackPublishOptions, VideoCodec } from './track/options'; import { getTrackPublicationInfo } from './track/utils'; import type { LoggerOptions } from './types'; import { - Future, isPublisherOfferWithJoinSupported, isReactNative, isVideoCodec, @@ -109,8 +109,17 @@ const dataTrackDataChannel = '_data_track'; const minReconnectWait = 2 * 1000; const leaveReconnect = 'leave-reconnect'; const reliabeReceiveStateTTL = 30_000; -const lossyDataChannelBufferThresholdMin = 8 * 1024; -const lossyDataChannelBufferThresholdMax = 256 * 1024; + +// 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; @@ -262,8 +271,6 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit /** used to indicate whether the browser is currently waiting to reconnect */ private isWaitingForNetworkReconnect: boolean = false; - private bufferStatusLowClosingFuture = new Future(); - constructor(private options: InternalRoomOptions) { super(); this.log = getLogger(options.loggerName ?? LoggerNames.Engine, () => this.logContext); @@ -297,13 +304,6 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.client.onParticipantUpdate = (updates) => this.emit(EngineEvent.ParticipantUpdate, updates); this.client.onJoined = (joinResponse) => this.emit(EngineEvent.Joined, joinResponse); - - this.on(EngineEvent.Closing, () => { - this.bufferStatusLowClosingFuture.reject?.(new UnexpectedConnectionState('engine closed')); - }); - // Swallow the rejection at the source so it doesn't surface as an unhandled promise rejection - // when no waitForBufferStatusLow callers are attached. - this.bufferStatusLowClosingFuture.promise.catch(() => {}); } /** @internal */ @@ -458,6 +458,11 @@ 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; @@ -885,6 +890,11 @@ 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; @@ -930,10 +940,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.reliableDC.onclose = this.handleDataChannelClose(DataChannelKind.RELIABLE); this.dataTrackDC.onclose = this.handleDataChannelClose(DataChannelKind.DATA_TRACK_LOSSY); - // set up dc buffer threshold, set to 64kB (otherwise 0 by default) - this.lossyDC.bufferedAmountLowThreshold = 65535; - this.reliableDC.bufferedAmountLowThreshold = 65535; - this.dataTrackDC.bufferedAmountLowThreshold = 65535; + // 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); @@ -952,8 +962,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // control buffered latency to ~100ms const threshold = this.lossyDataStatByterate / 10; dc.bufferedAmountLowThreshold = Math.min( - Math.max(threshold, lossyDataChannelBufferThresholdMin), - lossyDataChannelBufferThresholdMax, + Math.max(threshold, lossyDataChannelWaterMarkLow), + lossyDataChannelWaterMarkHigh, ); } }, 1000); @@ -1505,7 +1515,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } if (res?.lastMessageSeq) { - this.resendReliableMessagesForResume(res.lastMessageSeq); + this.resendReliableMessagesForResume(res.lastMessageSeq).catch((error) => { + this.log.warn('failed to resend reliable messages after resume', { + ...this.logContext, + error, + }); + }); } // resume success @@ -1621,21 +1636,43 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit case DataChannelKind.DATA_TRACK_LOSSY: return this.sendLossyBytes(msg, kind); - case DataChannelKind.RELIABLE: + 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) { - await this.waitForBufferStatusLow(kind); - this.reliableMessageBuffer.push({ data: msg, sequence: packet.sequence }); + 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; + } } } @@ -1643,42 +1680,62 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit async sendLossyBytes( bytes: NonSharedUint8Array, kind: Exclude, - bufferStatusLowBehavior: 'drop' | 'wait' = 'drop', + bufferStatusFullBehavior: 'drop' | 'wait' = 'drop', ) { - // make sure we do have a data connection + // 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); - if (dc) { - if (!this.isBufferStatusLow(kind)) { + try { + if (dc) { // Depending on the exact circumstance that data is being sent, either drop or wait for the - // buffer status to not be low before continuing. - switch (bufferStatusLowBehavior) { + // buffer to drain below the high-water mark before continuing. + switch (bufferStatusFullBehavior) { case 'wait': - await this.waitForBufferStatusLow(kind); + if (!this.isBelowHighWaterMark(kind)) { + await this.waitForBufferHeadroomWithLock(kind); + } break; case 'drop': - // this.log.warn(`dropping lossy data channel message`, this.logContext); - // 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}`, - ); + // 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; } - return; } - } - this.lossyDataStatCurrentBytes += bytes.byteLength; + 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; + if (this.attemptingReconnect) { + return; + } + + dc.send(bytes); } - 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; + } } - - this.updateAndEmitDCBufferStatus(kind); } private async resendReliableMessagesForResume(lastMessageSeq: number) { @@ -1686,9 +1743,36 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit const dc = this.dataChannelForKind(DataChannelKind.RELIABLE); if (dc) { this.reliableMessageBuffer.popToSequence(lastMessageSeq); - this.reliableMessageBuffer.getAll().forEach((msg) => { - dc.send(msg.data); - }); + // 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); } @@ -1700,39 +1784,160 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.reliableMessageBuffer.alignBufferedAmount(dc.bufferedAmount); } } - - const status = this.isBufferStatusLow(kind); - if (typeof status !== 'undefined' && status !== this.dcBufferStatus.get(kind)) { - this.dcBufferStatus.set(kind, status); - this.emit(EngineEvent.DCBufferStatusChanged, status, kind); + 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 }); } }; - private isBufferStatusLow = (kind: DataChannelKind): boolean | undefined => { + /** + * 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) { - return dc.bufferedAmount <= dc.bufferedAmountLowThreshold; + 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}`); } + // because RTCDatachannel has the threshold built in we can read it dynamically to account for changing thresholds over time + return dc.bufferedAmount <= dc.bufferedAmountLowThreshold; }; - async waitForBufferStatusLow(kind: DataChannelKind) { - return new TypedPromise(async (resolve, reject) => { - if (this.isClosed) { - reject(new UnexpectedConnectionState('engine closed')); - } - if (this.isBufferStatusLow(kind)) { + /** 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. + */ + 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; + } + + /** 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(); + } + + /** 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; + } + + /** + * 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. + */ + 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(); - } else { - const dc = this.dataChannelForKind(kind); - if (!dc) { - reject(new UnexpectedConnectionState(`DataChannel not found, kind: ${kind}`)); - return; - } - this.bufferStatusLowClosingFuture.promise.catch((e) => reject(e)); - dc.addEventListener('bufferedamountlow', () => resolve(), { - once: true, - }); + }; + 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); }); } diff --git a/src/utils/dataPacketBuffer.test.ts b/src/utils/dataPacketBuffer.test.ts new file mode 100644 index 0000000000..567b45a6a2 --- /dev/null +++ b/src/utils/dataPacketBuffer.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it } from 'vitest'; +import { DataPacketBuffer } from './dataPacketBuffer'; + +const item = (sequence: number, size: number, sent: boolean) => ({ + data: new Uint8Array(size), + sequence, + sent, +}); + +describe('DataPacketBuffer', () => { + it('trims sent packets down to the buffered amount', () => { + const buffer = new DataPacketBuffer(); + buffer.push(item(1, 100, true)); + buffer.push(item(2, 100, true)); + buffer.push(item(3, 100, true)); + + // 150 buffered bytes cover the last two packets; the first is fully delivered. + buffer.alignBufferedAmount(150); + + expect(buffer.getAll().map((i) => i.sequence)).toEqual([2, 3]); + }); + + it('never trims unsent packets, even when the buffered amount is zero', () => { + const buffer = new DataPacketBuffer(); + buffer.push(item(1, 100, false)); + buffer.push(item(2, 100, false)); + + buffer.alignBufferedAmount(0); + + expect(buffer.length).toBe(2); + }); + + it('does not let unsent tail bytes cause over-trimming of the sent prefix', () => { + const buffer = new DataPacketBuffer(); + buffer.push(item(1, 100, true)); + buffer.push(item(2, 100, true)); + // Queued during a reconnect window — not part of the channel's bufferedAmount. + buffer.push(item(3, 100, false)); + buffer.push(item(4, 100, false)); + + // Both sent packets are still in flight; nothing may be trimmed. With size accounting based + // on the total (400) instead of sent bytes (200), both sent packets would be popped here. + buffer.alignBufferedAmount(200); + + expect(buffer.getAll().map((i) => i.sequence)).toEqual([1, 2, 3, 4]); + }); + + it('markSent makes a single packet eligible for trimming and tracks sent size', () => { + const buffer = new DataPacketBuffer(); + const first = item(1, 100, false); + const second = item(2, 100, false); + buffer.push(first); + buffer.push(second); + + // Only the marked packet can be trimmed; the still-unsent one blocks the front. + buffer.markSent(first); + buffer.alignBufferedAmount(0); + expect(buffer.getAll().map((i) => i.sequence)).toEqual([1, 2]); + + buffer.markSent(second); + buffer.alignBufferedAmount(50); + expect(buffer.getAll().map((i) => i.sequence)).toEqual([2]); + }); + + it('getUnsent returns only unsent packets, in order', () => { + const buffer = new DataPacketBuffer(); + const a = item(1, 100, false); + const b = item(2, 100, false); + buffer.push(a); + buffer.push(b); + buffer.push(item(3, 100, false)); + buffer.markSent(b); + + expect(buffer.getUnsent().map((i) => i.sequence)).toEqual([1, 3]); + }); + + it('markAllUnsent flags every packet for re-send and resets sent size', () => { + const buffer = new DataPacketBuffer(); + buffer.push(item(1, 100, true)); + buffer.push(item(2, 100, true)); + + buffer.markAllUnsent(); + expect(buffer.getUnsent().map((i) => i.sequence)).toEqual([1, 2]); + + // With everything unsent, nothing can be trimmed even at a zero buffered amount. + buffer.alignBufferedAmount(0); + expect(buffer.getAll().map((i) => i.sequence)).toEqual([1, 2]); + }); + + it('popToSequence drops acked packets regardless of sent state', () => { + const buffer = new DataPacketBuffer(); + buffer.push(item(1, 100, true)); + buffer.push(item(2, 100, false)); + buffer.push(item(3, 100, false)); + + buffer.popToSequence(2); + + expect(buffer.getAll().map((i) => i.sequence)).toEqual([3]); + }); +}); diff --git a/src/utils/dataPacketBuffer.ts b/src/utils/dataPacketBuffer.ts index f91e4933ef..a287ebaf35 100644 --- a/src/utils/dataPacketBuffer.ts +++ b/src/utils/dataPacketBuffer.ts @@ -3,6 +3,13 @@ import type { NonSharedUint8Array } from '../type-polyfills/non-shared-typed-arr export interface DataPacketItem { data: NonSharedUint8Array; sequence: number; + /** + * Whether the packet has been handed to the data channel. Unsent packets are queued for the + * resume replay (e.g. sends that landed in a reconnect window) and must never be trimmed by + * {@link DataPacketBuffer.alignBufferedAmount}, which reasons about the channel's buffered + * bytes — those only ever contain sent packets. + */ + sent: boolean; } export class DataPacketBuffer { @@ -10,15 +17,23 @@ export class DataPacketBuffer { private _totalSize = 0; + private _sentSize = 0; + push(item: DataPacketItem) { this.buffer.push(item); this._totalSize += item.data.byteLength; + if (item.sent) { + this._sentSize += item.data.byteLength; + } } pop(): DataPacketItem | undefined { const item = this.buffer.shift(); if (item) { this._totalSize -= item.data.byteLength; + if (item.sent) { + this._sentSize -= item.data.byteLength; + } } return item; } @@ -27,6 +42,31 @@ export class DataPacketBuffer { return this.buffer.slice(); } + /** Every queued packet not yet handed to the channel, in sequence order. */ + getUnsent(): DataPacketItem[] { + return this.buffer.filter((item) => !item.sent); + } + + /** Marks a single queued packet as handed to the channel. */ + markSent(item: DataPacketItem) { + if (!item.sent) { + item.sent = true; + this._sentSize += item.data.byteLength; + } + } + + /** + * Marks every queued packet as not-yet-sent. Used at the start of a resume replay: whatever is + * still buffered was sent on the previous channel (or deferred) and must be re-handed to the + * current one, so none of it counts as sent until the replay actually transmits it. + */ + markAllUnsent() { + for (const item of this.buffer) { + item.sent = false; + } + this._sentSize = 0; + } + popToSequence(sequence: number) { while (this.buffer.length > 0) { const first = this.buffer[0]; @@ -41,7 +81,12 @@ export class DataPacketBuffer { alignBufferedAmount(bufferedAmount: number) { while (this.buffer.length > 0) { const first = this.buffer[0]; - if (this._totalSize - first.data.byteLength <= bufferedAmount) { + // Unsent packets aren't part of the channel's bufferedAmount and are still awaiting the + // resume replay — trimming them would silently lose them. + if (!first.sent) { + break; + } + if (this._sentSize - first.data.byteLength <= bufferedAmount) { break; } this.pop();