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 diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index f3f6028b92..865f92e536 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -1715,25 +1715,58 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } }; + /** 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; } - this.bufferStatusLowClosingFuture.promise.catch((e) => reject(e)); - dc.addEventListener('bufferedamountlow', () => resolve(), { - once: true, + 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); + } } /**