Address data channel thundering herd due to concurrent sendFile(...) calls

.changeset/metal-lemons-warn.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Fix data channel concurrent data stream send race condition

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<DataChannelKind, Mutex>();
+
   async waitForBufferStatusLow(kind: DataChannelKind) {
-    return new TypedPromise<void, UnexpectedConnectionState>(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<void, UnexpectedConnectionState>((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);
+    }
   }
 
   /**