FCE-3224: Fix reconnect race condition

packages/webrtc-client/src/ConnectionManager.ts

@@ -28,26 +28,51 @@ export class ConnectionManager {
   };
 
   public addTransceiversIfNeeded = (serverTracks: MediaEvent_OfferData_TrackTypes) => {
-    const recvTransceivers = this.connection.getTransceivers().filter((elem) => elem.direction === 'recvonly');
+    const recvTransceivers = this.connection.getTransceivers().filter(ConnectionManager.isLiveRecvTransceiver);
 
     const videoTransceiversAmount = recvTransceivers.filter((elem) => elem.receiver.track.kind === 'video').length;
     const audioTransceiversAmount = recvTransceivers.filter((elem) => elem.receiver.track.kind === 'audio').length;
 
-    const videoNeededTypes = Array<string>(serverTracks.video - videoTransceiversAmount).fill('video');
-    const audioNeededTypes = Array<string>(serverTracks.audio - audioTransceiversAmount).fill('audio');
+    const videoDelta = serverTracks.video - videoTransceiversAmount;
+    const audioDelta = serverTracks.audio - audioTransceiversAmount;
+
+    this.stopExcessRecvTransceivers('video', -videoDelta);
+    this.stopExcessRecvTransceivers('audio', -audioDelta);
+
+    const videoNeededTypes = Array<string>(Math.max(0, videoDelta)).fill('video');
+    const audioNeededTypes = Array<string>(Math.max(0, audioDelta)).fill('audio');
 
     [...videoNeededTypes, ...audioNeededTypes].forEach((kind) =>
       this.connection.addTransceiver(kind, { direction: 'recvonly' }),
     );
   };
 
+  private static isLiveRecvTransceiver = (t: RTCRtpTransceiver) =>
+    t.direction === 'recvonly' && t.currentDirection !== 'stopped';
+
+  private stopExcessRecvTransceivers = (kind: 'audio' | 'video', excess: number) => {
+    if (excess <= 0) return;
+
+    const candidates = this.connection
+      .getTransceivers()
+      .filter((t) => ConnectionManager.isLiveRecvTransceiver(t) && t.receiver.track?.kind === kind)
+      .sort((a, b) => {
+        const aOrphan = a.mid === null ? 0 : 1;
+        const bOrphan = b.mid === null ? 0 : 1;
+        return aOrphan - bOrphan;
+      });
+
+    candidates.slice(0, excess).forEach((transceiver) => transceiver.stop());
+  };
+
   public addTransceiver = (track: MediaStreamTrack, transceiverConfig: RTCRtpTransceiverInit) => {
     this.connection.addTransceiver(track, transceiverConfig);
   };
 
   public setOnTrackReady = (onTrackReady: (event: RTCTrackEvent) => void) => {
     this.connection.ontrack = onTrackReady;
   };
+
   public setRemoteDescription = async (data: RTCSessionDescriptionInit) => {
     await this.connection.setRemoteDescription(data);
   };

packages/webrtc-client/src/webRTCEndpoint.ts

@@ -7,6 +7,7 @@ import {
 } from '@fishjam-cloud/protobufs/peer';
 import type {
   MediaEvent as ServerMediaEvent,
+  MediaEvent_Connected,
   MediaEvent_OfferData,
   MediaEvent_SdpAnswer,
   MediaEvent_Track_SimulcastConfig,
@@ -154,42 +155,35 @@ export class WebRTCEndpoint extends (EventEmitter as new () => TypedEmitter<Requ
    * webrtcChannel.on("mediaEvent", (event) => webrtc.receiveMediaEvent(event.data));
    * ```
    */
-  public receiveMediaEvent = async (mediaEvent: SerializedMediaEvent) => {
-    const deserializedMediaEvent = deserializeServerMediaEvent(mediaEvent);
-
-    if (deserializedMediaEvent.connected) {
-      const connectedEvent = deserializedMediaEvent.connected;
-
-      this.proposedIceServers = connectedEvent.iceServers;
+  private mediaEventQueue: Promise<void> = Promise.resolve();
 
-      this.local.setLocalEndpointId(connectedEvent.endpointId);
-
-      const localEndpointMetadataJson = connectedEvent.endpointIdToEndpoint[connectedEvent.endpointId]?.metadataJson;
-      if (localEndpointMetadataJson) {
-        this.local.setEndpointMetadata(JSON.parse(localEndpointMetadataJson));
-      }
+  public receiveMediaEvent = (mediaEvent: SerializedMediaEvent): Promise<void> => {
+    const deserializedMediaEvent = deserializeServerMediaEvent(mediaEvent);
 
-      const connectedEndpoint = connectedEvent.endpointIdToEndpoint[connectedEvent.endpointId];
+    const next = this.mediaEventQueue.then(() => this.handleMediaEvent(deserializedMediaEvent));
+    this.mediaEventQueue = next.catch(() => undefined);
+    return next;
+  };
 
-      if (connectedEndpoint?.metadataJson) {
-        const parsedMetadata = JSON.parse(connectedEndpoint?.metadataJson);
-        this.local.setEndpointMetadata(parsedMetadata);
-      }
+  private handleConnected = (connectedEvent: MediaEvent_Connected) => {
+    this.proposedIceServers = connectedEvent.iceServers;
 
-      Object.entries(connectedEvent.endpointIdToEndpoint)
-        .filter(([endpointId]) => endpointId !== this.local.getEndpoint().id)
-        .forEach(([endpointId, endpoint]) => {
-          this.remote.addRemoteEndpoint(endpointId, endpoint.metadataJson, endpoint.trackIdToTrack);
-        });
+    this.local.setLocalEndpointId(connectedEvent.endpointId);
 
-      const remoteEndpoints = Object.values(this.remote.getRemoteEndpoints());
+    const localEndpoint = connectedEvent.endpointIdToEndpoint[connectedEvent.endpointId];
+    if (localEndpoint?.metadataJson) {
+      this.local.setEndpointMetadata(JSON.parse(localEndpoint.metadataJson));
+    }
 
-      this.emit('connected', this.local.getEndpoint().id, remoteEndpoints);
+    Object.entries(connectedEvent.endpointIdToEndpoint)
+      .filter(([endpointId]) => endpointId !== this.local.getEndpoint().id)
+      .forEach(([endpointId, endpoint]) => {
+        this.remote.addRemoteEndpoint(endpointId, endpoint.metadataJson, endpoint.trackIdToTrack);
+      });
 
-      return;
-    }
+    const remoteEndpoints = Object.values(this.remote.getRemoteEndpoints());
 
-    if (this.getEndpointId()) await this.handleMediaEvent(deserializedMediaEvent);
+    this.emit('connected', this.local.getEndpoint().id, remoteEndpoints);
   };
 
   private getEndpointId = () => this.local.getEndpoint().id;
@@ -251,6 +245,13 @@ export class WebRTCEndpoint extends (EventEmitter as new () => TypedEmitter<Requ
   }
 
   private handleMediaEvent = async (event: ServerMediaEvent) => {
+    if (event.connected) {
+      this.handleConnected(event.connected);
+      return;
+    }
+
+    if (!this.getEndpointId()) return;
+
     if (event.offerData) {
       await this.onOfferData(event.offerData);
     } else if (event.tracksAdded) {

packages/webrtc-client/tests/addTransceiversIfNeeded.test.ts

@@ -0,0 +1,115 @@
+import { MediaEvent_OfferData } from '@fishjam-cloud/protobufs/server';
+import { expect, it, vi } from 'vitest';
+
+import { ConnectionManager } from '../src/ConnectionManager';
+
+const makeTransceiver = (
+  overrides: Partial<RTCRtpTransceiver> & { kind?: string; dir?: RTCRtpTransceiverDirection; mid?: string | null },
+): RTCRtpTransceiver & { stop: ReturnType<typeof vi.fn> } => {
+  const stop = vi.fn();
+  const mid = 'mid' in overrides ? overrides.mid : 'someMid';
+  const transceiver = {
+    mid,
+    direction: overrides.dir ?? 'recvonly',
+    currentDirection: overrides.dir ?? 'recvonly',
+    receiver: { track: { kind: overrides.kind ?? 'video' } } as RTCRtpReceiver,
+    sender: { track: null } as unknown as RTCRtpSender,
+    setCodecPreferences: () => {},
+    stop,
+  } as unknown as RTCRtpTransceiver & { stop: ReturnType<typeof vi.fn> };
+  return transceiver;
+};
+
+const mockPcWithTransceivers = (initial: Array<RTCRtpTransceiver & { stop: ReturnType<typeof vi.fn> }>) => {
+  const transceivers = [...initial];
+  (global as any).RTCPeerConnection = class {
+    getTransceivers() {
+      return transceivers;
+    }
+    addTransceiver(kind: string): RTCRtpTransceiver {
+      const t = makeTransceiver({ kind, dir: 'recvonly', mid: null });
+      transceivers.push(t);
+      return t;
+    }
+  };
+  return transceivers;
+};
+
+it('clamps to zero when serverTracks equals current recvonly count', () => {
+  const transceivers = mockPcWithTransceivers([
+    makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '0' }),
+    makeTransceiver({ kind: 'audio', dir: 'recvonly', mid: '1' }),
+  ]);
+
+  const cm = new ConnectionManager([]);
+  cm.addTransceiversIfNeeded(MediaEvent_OfferData.create({ tracksTypes: { video: 1, audio: 1 } }).tracksTypes!);
+
+  expect(transceivers.length).toBe(2);
+  expect((transceivers[0] as any).stop).not.toHaveBeenCalled();
+  expect((transceivers[1] as any).stop).not.toHaveBeenCalled();
+});
+
+it('adds recvonly transceivers when serverTracks exceeds current count', () => {
+  const transceivers = mockPcWithTransceivers([makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '0' })]);
+
+  const cm = new ConnectionManager([]);
+  cm.addTransceiversIfNeeded(MediaEvent_OfferData.create({ tracksTypes: { video: 2, audio: 1 } }).tracksTypes!);
+
+  expect(transceivers.length).toBe(3);
+  expect(transceivers[1]!.receiver.track.kind).toBe('video');
+  expect(transceivers[2]!.receiver.track.kind).toBe('audio');
+});
+
+it('stops excess recvonly transceivers when serverTracks is below current count', () => {
+  const orphan = makeTransceiver({ kind: 'audio', dir: 'recvonly', mid: null });
+  const negotiated = makeTransceiver({ kind: 'audio', dir: 'recvonly', mid: '3' });
+  mockPcWithTransceivers([makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '0' }), negotiated, orphan]);
+
+  const cm = new ConnectionManager([]);
+  cm.addTransceiversIfNeeded(MediaEvent_OfferData.create({ tracksTypes: { video: 1, audio: 1 } }).tracksTypes!);
+
+  expect(orphan.stop).toHaveBeenCalled();
+  expect(negotiated.stop).not.toHaveBeenCalled();
+});
+
+it('does not throw on negative delta (previously Array(-1) crash)', () => {
+  mockPcWithTransceivers([
+    makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '0' }),
+    makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '1' }),
+  ]);
+
+  const cm = new ConnectionManager([]);
+  expect(() =>
+    cm.addTransceiversIfNeeded(MediaEvent_OfferData.create({ tracksTypes: { video: 1, audio: 0 } }).tracksTypes!),
+  ).not.toThrow();
+});
+
+it('ignores stopped recvonly transceivers when counting so new ones are still added', () => {
+  const stopped = makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '0' });
+  (stopped as unknown as { currentDirection: RTCRtpTransceiverDirection }).currentDirection = 'stopped';
+
+  const transceivers = mockPcWithTransceivers([stopped]);
+
+  const cm = new ConnectionManager([]);
+  cm.addTransceiversIfNeeded(MediaEvent_OfferData.create({ tracksTypes: { video: 1, audio: 0 } }).tracksTypes!);
+
+  expect(transceivers.length).toBe(2);
+  expect(transceivers[1]!.receiver.track.kind).toBe('video');
+  expect(stopped.stop).not.toHaveBeenCalled();
+});
+
+it('does not select stopped transceivers as stop candidates for excess', () => {
+  const stopped = makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '0' });
+  (stopped as unknown as { currentDirection: RTCRtpTransceiverDirection }).currentDirection = 'stopped';
+  const live = makeTransceiver({ kind: 'video', dir: 'recvonly', mid: '1' });
+  const orphan = makeTransceiver({ kind: 'video', dir: 'recvonly', mid: null });
+
+  mockPcWithTransceivers([stopped, live, orphan]);
+
+  const cm = new ConnectionManager([]);
+  cm.addTransceiversIfNeeded(MediaEvent_OfferData.create({ tracksTypes: { video: 1, audio: 0 } }).tracksTypes!);
+
+  expect(stopped.stop).not.toHaveBeenCalled();
+  expect(orphan.stop).toHaveBeenCalled();
+  expect(live.stop).not.toHaveBeenCalled();
+});

packages/webrtc-client/tests/events/connectedEvent.test.ts

@@ -53,44 +53,39 @@ it('Connecting to room with one peer', () =>
     webRTCEndpoint.receiveMediaEvent(serializeServerMediaEvent({ connected }));
   }));
 
-it('Connecting to room with one peer with one track', () =>
-  new Promise((done) => {
-    // Given
-    const webRTCEndpoint = new WebRTCEndpoint();
-    const trackAddedCallback = vi.fn((_x) => null);
-    const connectedCallback = vi.fn((_peerId, _peersInRoom) => null);
+it('Connecting to room with one peer with one track', async () => {
+  const webRTCEndpoint = new WebRTCEndpoint();
+  const trackAddedCallback = vi.fn((_x) => null);
+  const connectedCallback = vi.fn((_peerId, _peersInRoom) => null);
 
-    const connected = createConnectedEvent();
+  const connected = createConnectedEvent();
 
-    const otherEndpoint = createEmptyEndpoint();
+  const otherEndpoint = createEmptyEndpoint();
 
-    otherEndpoint.trackIdToTrack = { [exampleTrackId]: createTrackWithSimulcast() };
+  otherEndpoint.trackIdToTrack = { [exampleTrackId]: createTrackWithSimulcast() };
 
-    connected.endpointIdToEndpoint = {
-      ...connected.endpointIdToEndpoint,
-      [faker.string.uuid()]: otherEndpoint,
-    };
+  connected.endpointIdToEndpoint = {
+    ...connected.endpointIdToEndpoint,
+    [faker.string.uuid()]: otherEndpoint,
+  };
 
-    webRTCEndpoint.on('connected', (peerId: string, remotePeersInRoom: Endpoint[]) => {
-      connectedCallback(peerId, remotePeersInRoom);
-      expect(peerId).toBe(connected.endpointId);
-      expect(remotePeersInRoom.length).toBe(1);
-    });
+  webRTCEndpoint.on('connected', (peerId: string, remotePeersInRoom: Endpoint[]) => {
+    connectedCallback(peerId, remotePeersInRoom);
+    expect(peerId).toBe(connected.endpointId);
+    expect(remotePeersInRoom.length).toBe(1);
+  });
 
-    webRTCEndpoint.on('trackAdded', (ctx) => {
-      trackAddedCallback(ctx);
-      expect(ctx.trackId).toBe(exampleTrackId);
-      expect(ctx.simulcastConfig?.enabled).toBe(otherEndpoint.trackIdToTrack[exampleTrackId]!.simulcastConfig?.enabled);
-      done('');
-    });
+  webRTCEndpoint.on('trackAdded', (ctx) => {
+    trackAddedCallback(ctx);
+    expect(ctx.trackId).toBe(exampleTrackId);
+    expect(ctx.simulcastConfig?.enabled).toBe(otherEndpoint.trackIdToTrack[exampleTrackId]!.simulcastConfig?.enabled);
+  });
 
-    // When
-    webRTCEndpoint.receiveMediaEvent(serializeServerMediaEvent({ connected }));
+  await webRTCEndpoint.receiveMediaEvent(serializeServerMediaEvent({ connected }));
 
-    // Then
-    const remoteTracks = webRTCEndpoint.getRemoteTracks();
-    expect(Object.values(remoteTracks).length).toBe(1);
+  const remoteTracks = webRTCEndpoint.getRemoteTracks();
+  expect(Object.values(remoteTracks).length).toBe(1);
 
-    expect(trackAddedCallback.mock.calls).toHaveLength(1);
-    expect(connectedCallback.mock.calls).toHaveLength(1);
-  }));
+  expect(trackAddedCallback.mock.calls).toHaveLength(1);
+  expect(connectedCallback.mock.calls).toHaveLength(1);
+});