feat(client): Client Call Event Reporting

packages/client/src/rtc/BasePeerConnection.ts

@@ -16,6 +16,8 @@ import { StatsTracer, Tracer, traceRTCPeerConnection } from '../stats';
 import {
   BasePeerConnectionOpts,
   OnIceConnected,
+  OnPeerConnectionAttemptStarted,
+  OnPeerConnectionConnected,
   OnReconnectionNeeded,
   ReconnectReason,
 } from './types';
@@ -37,8 +39,12 @@ export abstract class BasePeerConnection {
 
   private onReconnectionNeeded?: OnReconnectionNeeded;
   private onIceConnected?: OnIceConnected;
+  private onPeerConnectionConnected?: OnPeerConnectionConnected;
+  private onPeerConnectionAttemptStarted?: OnPeerConnectionAttemptStarted;
   private readonly iceRestartDelay: number;
   private iceHasEverConnected = false;
+  private peerConnectionHasEverConnected = false;
+  private peerConnectionAttemptStarted = false;
   private iceRestartTimeout?: NodeJS.Timeout;
   private preConnectStuckTimeout?: NodeJS.Timeout;
   protected isIceRestarting = false;
@@ -65,6 +71,8 @@ export abstract class BasePeerConnection {
       dispatcher,
       onReconnectionNeeded,
       onIceConnected,
+      onPeerConnectionConnected,
+      onPeerConnectionAttemptStarted,
       tag,
       enableTracing,
       clientPublishOptions,
@@ -80,6 +88,8 @@ export abstract class BasePeerConnection {
     this.tag = tag;
     this.onReconnectionNeeded = onReconnectionNeeded;
     this.onIceConnected = onIceConnected;
+    this.onPeerConnectionConnected = onPeerConnectionConnected;
+    this.onPeerConnectionAttemptStarted = onPeerConnectionAttemptStarted;
     this.logger = videoLoggerSystem.getLogger(
       peerType === PeerType.SUBSCRIBER ? 'Subscriber' : 'Publisher',
       { tags: [tag] },
@@ -122,6 +132,8 @@ export abstract class BasePeerConnection {
     this.preConnectStuckTimeout = undefined;
     this.onReconnectionNeeded = undefined;
     this.onIceConnected = undefined;
+    this.onPeerConnectionConnected = undefined;
+    this.onPeerConnectionAttemptStarted = undefined;
     this.isDisposed = true;
     this.detachEventHandlers();
     this.pc.close();
@@ -239,6 +251,12 @@ export abstract class BasePeerConnection {
     return this.trackIdToTrackType.get(trackId);
   };
 
+  protected beginPeerConnectionAttempt = () => {
+    if (this.peerConnectionAttemptStarted) return;
+    this.peerConnectionAttemptStarted = true;
+    this.onPeerConnectionAttemptStarted?.(this.peerType);
+  };
+
   /**
    * Checks if the `RTCPeerConnection` is healthy.
    * It checks the ICE connection state and the peer connection state.
@@ -318,6 +336,11 @@ export abstract class BasePeerConnection {
       }
     }
 
+    if (state === 'connected' && !this.peerConnectionHasEverConnected) {
+      this.peerConnectionHasEverConnected = true;
+      this.onPeerConnectionConnected?.(this.peerType);
+    }
+
     // we can't recover from a failed connection state (contrary to ICE)
     if (state === 'failed') {
       this.onReconnectionNeeded?.(

packages/client/src/rtc/Publisher.ts

@@ -425,6 +425,7 @@ export class Publisher extends BasePeerConnection {
 
       try {
         this.isIceRestarting = options?.iceRestart ?? false;
+        this.beginPeerConnectionAttempt();
         await this.pc.setLocalDescription(offer);
 
         const { sdp: baseSdp = '' } = offer;

packages/client/src/rtc/Subscriber.ts

@@ -191,6 +191,7 @@ export class Subscriber extends BasePeerConnection {
         );
       }
     }
+    this.beginPeerConnectionAttempt();
     await this.pc.setLocalDescription(answer);
 
     await this.sfuClient.sendAnswer({

packages/client/src/rtc/__tests__/Call.reconnect.test.ts

@@ -18,6 +18,8 @@ import { Subscriber } from '../Subscriber';
 import { Dispatcher } from '../Dispatcher';
 import { StreamSfuClient } from '../../StreamSfuClient';
 import { IceTrickleBuffer } from '../IceTrickleBuffer';
+import { settled } from '../../helpers/concurrency';
+import { promiseWithResolvers } from '../../helpers/promise';
 
 vi.mock('../../StreamSfuClient', () => ({
   StreamSfuClient: vi.fn(),
@@ -326,6 +328,51 @@ describe('Call reconnect stopping conditions', () => {
 
       expect(call['consecutiveNegotiationFailures']).toBe(0);
     });
+
+    it('skips a queued reconnect after a previous reconnect already recovered the call', async () => {
+      const networkAvailableTask = promiseWithResolvers<void>();
+      const recoveredSfuClient = {
+        close: vi.fn(),
+        isHealthy: true,
+        tag: 'recovered',
+      } as unknown as StreamSfuClient;
+
+      call.state.setCallingState(CallingState.OFFLINE);
+      call['networkAvailableTask'] = networkAvailableTask;
+
+      const rejoinSpy = vi
+        .spyOn(
+          call as unknown as { reconnectRejoin: () => Promise<void> },
+          'reconnectRejoin',
+        )
+        .mockImplementation(async () => {
+          call['sfuClient'] = recoveredSfuClient;
+          call.state.setCallingState(CallingState.JOINED);
+        });
+
+      const firstReconnect = call['reconnect'](
+        WebsocketReconnectStrategy.REJOIN,
+        'reason1',
+      );
+      const secondReconnect = call['reconnect'](
+        WebsocketReconnectStrategy.REJOIN,
+        'reason2',
+      );
+      const thirdReconnect = call['reconnect'](
+        WebsocketReconnectStrategy.REJOIN,
+        'reason3',
+      );
+
+      await new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+      networkAvailableTask.resolve();
+      call['networkAvailableTask'] = undefined;
+
+      await Promise.all([firstReconnect, secondReconnect, thirdReconnect]);
+      await settled(call['reconnectConcurrencyTag']);
+
+      expect(rejoinSpy).toHaveBeenCalledTimes(1);
+    });
   });
 
   describe('tunability setters', () => {

packages/client/src/rtc/types.ts

@@ -51,6 +51,8 @@ export type OnReconnectionNeeded = (
  * recovered, not merely when the SFU join handshake succeeded.
  */
 export type OnIceConnected = (peerType: PeerType) => void;
+export type OnPeerConnectionConnected = (peerType: PeerType) => void;
+export type OnPeerConnectionAttemptStarted = (peerType: PeerType) => void;
 
 export type BasePeerConnectionOpts = {
   sfuClient: StreamSfuClient;
@@ -59,6 +61,8 @@ export type BasePeerConnectionOpts = {
   dispatcher: Dispatcher;
   onReconnectionNeeded?: OnReconnectionNeeded;
   onIceConnected?: OnIceConnected;
+  onPeerConnectionConnected?: OnPeerConnectionConnected;
+  onPeerConnectionAttemptStarted?: OnPeerConnectionAttemptStarted;
   tag: string;
   enableTracing: boolean;
   iceRestartDelay?: number;