feat: add drop/wait options to sendLossyBytes to control whether bytes are buffered when the dc is in low status

This was encountered when implementing the e2e tests - a large data
track frame could be split up into potentially n packets, and if one
packet is dropped the whole frame is considered dropped. So the behavior
that is really wanted here is to buffer packets and send them once the
data channel can accept more data, even though it is technically a lossy
data channel.

Author: 1egoman

src/room/RTCEngine.ts

@@ -1413,22 +1413,34 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
   async sendLossyBytes(
     bytes: Uint8Array,
     kind: Exclude<DataChannelKind, DataChannelKind.RELIABLE>,
+    bufferStatusLowBehavior: 'drop' | 'wait' = 'drop',
   ) {
     // make sure we do have a data connection
     await this.ensurePublisherConnected(kind);
 
     const dc = this.dataChannelForKind(kind);
     if (dc) {
-      // lossy channel, drop messages to reduce latency
       if (!this.isBufferStatusLow(kind)) {
-        this.lossyDataDropCount += 1;
-        if (this.lossyDataDropCount % 100 === 0) {
-          this.log.warn(
-            `dropping lossy data channel messages, total dropped: ${this.lossyDataDropCount}`,
-            this.logContext,
-          );
+        // 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) {
+          case 'wait':
+            this.log.warn(`waiting for data channel buffer status to go low, ${dc.bufferedAmount} > ${dc.bufferedAmountLowThreshold}`, bytes, this.logContext);
+            await this.waitForBufferStatusLow(kind);
+            this.log.warn(`data channel buffer status no longer low, ${dc.bufferedAmount} <= ${dc.bufferedAmountLowThreshold}`, this.logContext);
+            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}`,
+                this.logContext,
+              );
+            }
+            return;
         }
-        return;
       }
       this.lossyDataStatCurrentBytes += bytes.byteLength;
 

src/room/Room.ts

@@ -289,7 +289,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
         this.emit(RoomEvent.LocalDataTrackUnpublished, event.sid);
       })
       .on('packetAvailable', ({ bytes }) => {
-        this.engine.sendLossyBytes(bytes, DataChannelKind.DATA_TRACK_LOSSY);
+        this.engine.sendLossyBytes(bytes, DataChannelKind.DATA_TRACK_LOSSY, 'wait');
       });
 
     this.disconnectLock = new Mutex();