e2e reliablility for data channel

.changeset/quick-turkeys-post.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Improve e2e reliablility of data channel

package.json

@@ -55,7 +55,7 @@
   },
   "dependencies": {
     "@livekit/mutex": "1.1.1",
-    "@livekit/protocol": "1.38.0",
+    "@livekit/protocol": "1.39.1",
     "events": "^3.3.0",
     "loglevel": "^1.9.2",
     "sdp-transform": "^2.15.0",

src/api/SignalClient.ts

@@ -252,7 +252,10 @@ export class SignalClient {
       sid,
       reconnectReason: reason,
     });
-    return res;
+    if (res instanceof ReconnectResponse) {
+      return res;
+    }
+    return;
   }
 
   private connect(

src/room/RTCEngine.ts

@@ -5,6 +5,7 @@ import {
   ClientConfiguration,
   type ConnectionQualityUpdate,
   DataChannelInfo,
+  DataChannelReceiveState,
   DataPacket,
   DataPacket_Kind,
   DisconnectReason,
@@ -44,6 +45,8 @@ import {
 } from '../api/SignalClient';
 import log, { LoggerNames, getLogger } from '../logger';
 import type { InternalRoomOptions } from '../options';
+import { DataPacketBuffer } from '../utils/dataPacketBuffer';
+import { TTLMap } from '../utils/ttlmap';
 import PCTransport, { PCEvents } from './PCTransport';
 import { PCTransportManager, PCTransportState } from './PCTransportManager';
 import type { ReconnectContext, ReconnectPolicy } from './ReconnectPolicy';
@@ -81,6 +84,7 @@ const lossyDataChannel = '_lossy';
 const reliableDataChannel = '_reliable';
 const minReconnectWait = 2 * 1000;
 const leaveReconnect = 'leave-reconnect';
+const reliabeReceiveStateTTL = 30_000;
 
 enum PCState {
   New,
@@ -178,6 +182,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
   private publisherConnectionPromise: Promise<void> | undefined;
 
+  private reliableDataSequence: number = 1;
+
+  private reliableMessageBuffer = new DataPacketBuffer();
+
+  private reliableReceivedState: TTLMap<string, number> = new TTLMap(reliabeReceiveStateTTL);
+
   constructor(private options: InternalRoomOptions) {
     super();
     this.log = getLogger(options.loggerName ?? LoggerNames.Engine);
@@ -310,6 +320,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     this.lossyDCSub = undefined;
     this.reliableDC = undefined;
     this.reliableDCSub = undefined;
+    this.reliableMessageBuffer = new DataPacketBuffer();
+    this.reliableDataSequence = 1;
+    this.reliableReceivedState.clear();
   }
 
   async cleanupClient() {
@@ -677,6 +690,15 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       }
       const dp = DataPacket.fromBinary(new Uint8Array(buffer));
 
+      if (dp.sequence > 0 && dp.participantSid !== '') {
+        const lastSeq = this.reliableReceivedState.get(dp.participantSid);
+        if (lastSeq && dp.sequence <= lastSeq) {
+          // ignore duplicate or out-of-order packets in reliable channel
+          return;
+        }
+        this.reliableReceivedState.set(dp.participantSid, dp.sequence);
+      }
+
       if (dp.value?.case === 'speaker') {
         // dispatch speaker updates
         this.emit(EngineEvent.ActiveSpeakersUpdate, dp.value.value.speakers);
@@ -1033,6 +1055,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     if (res) {
       const rtcConfig = this.makeRTCConfiguration(res);
       this.pcManager.updateConfiguration(rtcConfig);
+      if (this.latestJoinResponse) {
+        this.latestJoinResponse.serverInfo = res.serverInfo;
+      }
     } else {
       this.log.warn('Did not receive reconnect response', this.logContext);
     }
@@ -1059,6 +1084,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       this.createDataChannels();
     }
 
+    if (res?.lastMessageSeq) {
+      this.resendReliableMessagesForResume(res.lastMessageSeq);
+    }
+
     // resume success
     this.emit(EngineEvent.Resumed);
   }
@@ -1151,19 +1180,42 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
   /* @internal */
   async sendDataPacket(packet: DataPacket, kind: DataPacket_Kind) {
-    const msg = packet.toBinary();
-
     // make sure we do have a data connection
     await this.ensurePublisherConnected(kind);
 
+    if (kind === DataPacket_Kind.RELIABLE) {
+      packet.sequence = this.reliableDataSequence;
+      this.reliableDataSequence += 1;
+    }
+    const msg = packet.toBinary();
     const dc = this.dataChannelForKind(kind);
     if (dc) {
+      if (kind === DataPacket_Kind.RELIABLE) {
+        this.reliableMessageBuffer.push({ data: msg, sequence: packet.sequence });
+      }
+
+      if (this.attemptingReconnect) {
+        return;
+      }
+
       dc.send(msg);
     }
 
     this.updateAndEmitDCBufferStatus(kind);
   }
 
+  private async resendReliableMessagesForResume(lastMessageSeq: number) {
+    await this.ensurePublisherConnected(DataPacket_Kind.RELIABLE);
+    const dc = this.dataChannelForKind(DataPacket_Kind.RELIABLE);
+    if (dc) {
+      this.reliableMessageBuffer.popToSequence(lastMessageSeq);
+      this.reliableMessageBuffer.getAll().forEach((msg) => {
+        dc.send(msg.data);
+      });
+    }
+    this.updateAndEmitDCBufferStatus(DataPacket_Kind.RELIABLE);
+  }
+
   private updateAndEmitDCBufferStatus = (kind: DataPacket_Kind) => {
     const status = this.isBufferStatusLow(kind);
     if (typeof status !== 'undefined' && status !== this.dcBufferStatus.get(kind)) {
@@ -1175,6 +1227,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
   private isBufferStatusLow = (kind: DataPacket_Kind): boolean | undefined => {
     const dc = this.dataChannelForKind(kind);
     if (dc) {
+      if (kind === DataPacket_Kind.RELIABLE) {
+        this.reliableMessageBuffer.alignBufferedAmount(dc.bufferedAmount);
+      }
       return dc.bufferedAmount <= dc.bufferedAmountLowThreshold;
     }
   };
@@ -1409,6 +1464,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
         publishTracks: getTrackPublicationInfo(localTracks),
         dataChannels: this.dataChannelsInfo(),
         trackSidsDisabled,
+        datachannelReceiveStates: this.reliableReceivedState.map((seq, sid) => {
+          return new DataChannelReceiveState({
+            publisherSid: sid,
+            lastSeq: seq,
+          });
+        }),
       }),
     );
   }

src/utils/dataPacketBuffer.ts

@@ -0,0 +1,52 @@
+export interface DataPacketItem {
+  data: Uint8Array;
+  sequence: number;
+}
+
+export class DataPacketBuffer {
+  private buffer: DataPacketItem[] = [];
+
+  private _totalSize = 0;
+
+  push(item: DataPacketItem) {
+    this.buffer.push(item);
+    this._totalSize += item.data.byteLength;
+  }
+
+  pop(): DataPacketItem | undefined {
+    const item = this.buffer.shift();
+    if (item) {
+      this._totalSize -= item.data.byteLength;
+    }
+    return item;
+  }
+
+  getAll(): DataPacketItem[] {
+    return this.buffer.slice();
+  }
+
+  popToSequence(sequence: number) {
+    while (this.buffer.length > 0) {
+      const first = this.buffer[0];
+      if (first.sequence <= sequence) {
+        this.pop();
+      } else {
+        break;
+      }
+    }
+  }
+
+  alignBufferedAmount(bufferedAmount: number) {
+    while (this.buffer.length > 0) {
+      const first = this.buffer[0];
+      if (this._totalSize - first.data.byteLength <= bufferedAmount) {
+        break;
+      }
+      this.pop();
+    }
+  }
+
+  get length(): number {
+    return this.buffer.length;
+  }
+}