From 1f264d2fc7bc7c65529a521c772e4d56635fdc50 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Mon, 13 Jul 2026 16:46:03 -0400 Subject: [PATCH 1/4] fix: ensure waitForBufferStatusLow resolves all awaiters in order, not all at once --- src/room/RTCEngine.ts | 36 +++++++++++++++++++++++++++++++++--- 1 file changed, 33 insertions(+), 3 deletions(-) diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index f3f6028b92..434dfdaebe 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -1715,6 +1715,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } }; + /** Abort controller which when called will remove the `bufferedamountlow` event on the data channel. */ + private waitForBufferStatusLowAbortController: AbortController | null = null; + /** List of resolve functions which are waiting to be called by {@link waitForBufferStatusLow}. + * Note that these will be called in order, validating that the buffer status is still low + * between each call. */ + private waitForBufferedStatusLowResolves: Array<() => void> = []; async waitForBufferStatusLow(kind: DataChannelKind) { return new TypedPromise(async (resolve, reject) => { if (this.isClosed) { @@ -1728,10 +1734,34 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit reject(new UnexpectedConnectionState(`DataChannel not found, kind: ${kind}`)); return; } + + // Proxy along any errors due to the engine closing this.bufferStatusLowClosingFuture.promise.catch((e) => reject(e)); - dc.addEventListener('bufferedamountlow', () => resolve(), { - once: true, - }); + + // Store resolve so that when the bufferedamountlow event fires, all resolve calls can be + // fired in order until the data channel buffer fills up again. + this.waitForBufferedStatusLowResolves.push(resolve); + if (!this.waitForBufferStatusLowAbortController) { + this.waitForBufferStatusLowAbortController = new AbortController(); + dc.addEventListener('bufferedamountlow', () => { + while (true) { + const resolve = this.waitForBufferedStatusLowResolves[0]; + if (!resolve) { + this.waitForBufferedStatusLowResolves = []; + this.waitForBufferStatusLowAbortController?.abort(); + this.waitForBufferStatusLowAbortController = null; + break; + } + if (!this.isBufferStatusLow(kind)) { + // Buffer status no longer low, bail out and resume once the next bufferedamountlow + // event fires again. + break; + } + resolve(); + this.waitForBufferedStatusLowResolves.shift(); + } + }, { signal: this.waitForBufferStatusLowAbortController.signal }); + } } }); } From 5a4f6d2e6206a853ed865dd559b3ed7a90d93c0c Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Mon, 13 Jul 2026 16:49:15 -0400 Subject: [PATCH 2/4] fix: defer to event loop in waitForBufferedStatusLow --- src/room/RTCEngine.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 434dfdaebe..21119c5995 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -1743,7 +1743,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.waitForBufferedStatusLowResolves.push(resolve); if (!this.waitForBufferStatusLowAbortController) { this.waitForBufferStatusLowAbortController = new AbortController(); - dc.addEventListener('bufferedamountlow', () => { + dc.addEventListener('bufferedamountlow', async () => { while (true) { const resolve = this.waitForBufferedStatusLowResolves[0]; if (!resolve) { @@ -1758,6 +1758,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit break; } resolve(); + // Defer to the event loop so any `await dc.send(...)` calls can fire and fill back up + // the data channel. + await new Promise((r) => setTimeout(r, 0)); this.waitForBufferedStatusLowResolves.shift(); } }, { signal: this.waitForBufferStatusLowAbortController.signal }); From d4a824cab443850fdda474e803d665122ae87634 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Mon, 13 Jul 2026 17:15:16 -0400 Subject: [PATCH 3/4] refactor: convert from manual approach to mutex approach --- src/room/RTCEngine.ts | 92 +++++++++++++++++++++---------------------- 1 file changed, 46 insertions(+), 46 deletions(-) diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 21119c5995..865f92e536 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -1715,58 +1715,58 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } }; - /** Abort controller which when called will remove the `bufferedamountlow` event on the data channel. */ - private waitForBufferStatusLowAbortController: AbortController | null = null; - /** List of resolve functions which are waiting to be called by {@link waitForBufferStatusLow}. - * Note that these will be called in order, validating that the buffer status is still low - * between each call. */ - private waitForBufferedStatusLowResolves: Array<() => void> = []; + /** Per-kind lock serializing callers of {@link waitForBufferStatusLow} that have to wait. */ + private waitForBufferStatusLowLocks = new Map(); + async waitForBufferStatusLow(kind: DataChannelKind) { - return new TypedPromise(async (resolve, reject) => { - if (this.isClosed) { - reject(new UnexpectedConnectionState('engine closed')); - } - if (this.isBufferStatusLow(kind)) { - resolve(); - } else { + if (this.isClosed) { + throw new UnexpectedConnectionState('engine closed'); + } + if (this.isBufferStatusLow(kind)) { + return; + } + + // Slow path: the buffer is full, so serialize waiters. Only one proceeds per capacity window; + // without this, N concurrent senders all observe the bufferedamountlow signal in the same tick + // and call dc.send() together, blowing past the SCTP send buffer (see #1995). + let lock = this.waitForBufferStatusLowLocks.get(kind); + if (!lock) { + lock = new Mutex(); + this.waitForBufferStatusLowLocks.set(kind, lock); + } + const unlock = await lock.lock(); + try { + await new TypedPromise((resolve, reject) => { + // Re-check after acquiring the lock - the engine may have closed and the buffer may have + // drained while we were queued. + if (this.isClosed) { + reject(new UnexpectedConnectionState('engine closed')); + return; + } + if (this.isBufferStatusLow(kind)) { + resolve(); + return; + } const dc = this.dataChannelForKind(kind); if (!dc) { reject(new UnexpectedConnectionState(`DataChannel not found, kind: ${kind}`)); return; } - - // Proxy along any errors due to the engine closing - this.bufferStatusLowClosingFuture.promise.catch((e) => reject(e)); - - // Store resolve so that when the bufferedamountlow event fires, all resolve calls can be - // fired in order until the data channel buffer fills up again. - this.waitForBufferedStatusLowResolves.push(resolve); - if (!this.waitForBufferStatusLowAbortController) { - this.waitForBufferStatusLowAbortController = new AbortController(); - dc.addEventListener('bufferedamountlow', async () => { - while (true) { - const resolve = this.waitForBufferedStatusLowResolves[0]; - if (!resolve) { - this.waitForBufferedStatusLowResolves = []; - this.waitForBufferStatusLowAbortController?.abort(); - this.waitForBufferStatusLowAbortController = null; - break; - } - if (!this.isBufferStatusLow(kind)) { - // Buffer status no longer low, bail out and resume once the next bufferedamountlow - // event fires again. - break; - } - resolve(); - // Defer to the event loop so any `await dc.send(...)` calls can fire and fill back up - // the data channel. - await new Promise((r) => setTimeout(r, 0)); - this.waitForBufferedStatusLowResolves.shift(); - } - }, { signal: this.waitForBufferStatusLowAbortController.signal }); - } - } - }); + const onBufferedAmountLow = () => resolve(); + dc.addEventListener('bufferedamountlow', onBufferedAmountLow, { once: true }); + // Proxy along any errors due to the engine closing. + this.bufferStatusLowClosingFuture.promise.catch((e) => { + dc.removeEventListener('bufferedamountlow', onBufferedAmountLow); + reject(e); + }); + }); + } finally { + // Release only after the caller's synchronous dc.send() has run, so the next waiter reads the + // updated bufferedAmount rather than the stale (still-low) value. The caller's continuation is + // scheduled when this promise settles; deferring the unlock to a later task lets that send run + // first. + setTimeout(unlock, 0); + } } /** From 042e6d48f00548da2b1314f4bc9241a37adf388f Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Mon, 13 Jul 2026 17:10:58 -0400 Subject: [PATCH 4/4] fix: add missing changeset --- .changeset/metal-lemons-warn.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/metal-lemons-warn.md diff --git a/.changeset/metal-lemons-warn.md b/.changeset/metal-lemons-warn.md new file mode 100644 index 0000000000..6ca2ec111d --- /dev/null +++ b/.changeset/metal-lemons-warn.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Fix data channel concurrent data stream send race condition