Fix RTCEngine.negotiate EngineEvent.Closing / EngineEvent.Restarting listeners to not subscribe to a new listener every call

.changeset/loud-pears-negotiate.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Avoid attaching a new Closing/Restarting event listener for each negotiate call

src/room/RTCEngine.test.ts

@@ -10,6 +10,7 @@ import { PCTransportState } from './PCTransportManager';
 import RTCEngine, { DataChannelKind } from './RTCEngine';
 import { roomOptionDefaults } from './defaults';
 import { PublishDataError, UnexpectedConnectionState } from './errors';
+import { EngineEvent } from './events';
 
 describe('RTCEngine', () => {
   const originalRTCRtpSender = window.RTCRtpSender;
@@ -919,4 +920,87 @@ describe('RTCEngine', () => {
       expect(handleDisconnect).not.toHaveBeenCalled();
     });
   });
+
+  describe('negotiate', () => {
+    /**
+     * An engine whose `pcManager` parks in `negotiate()` until its abort controller fires, so
+     * several negotiations can be held in flight at once. The controllers are captured so tests
+     * can check the abort actually reached each in-flight call.
+     */
+    function stubNegotiableEngine(engine: RTCEngine) {
+      const controllers: AbortController[] = [];
+      const negotiate = vi.fn((abortController: AbortController) => {
+        controllers.push(abortController);
+        return new Promise<void>((_resolve, reject) => {
+          abortController.signal.addEventListener(
+            'abort',
+            () => reject(new Error('negotiation aborted')),
+            { once: true },
+          );
+        });
+      });
+      Object.assign(engine as unknown as Record<string, unknown>, {
+        _isClosed: false,
+        pcManager: {
+          requirePublisher: vi.fn(),
+          // a non-empty transceiver list keeps negotiate() off the createDataChannels path
+          publisher: { getTransceivers: () => [{}], off: vi.fn(), once: vi.fn() },
+          negotiate,
+        },
+        handleDisconnect: vi.fn(),
+      });
+      return { negotiate, controllers };
+    }
+
+    const pendingAborts = (engine: RTCEngine) =>
+      (engine as unknown as { pendingNegotiationAborts: Set<() => void> }).pendingNegotiationAborts;
+
+    it('does not add a Closing/Restarting listener per in-flight negotiate call', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const { negotiate, controllers } = stubNegotiableEngine(engine);
+
+      // A burst of server-initiated renegotiations used to stack a listener pair per call, which
+      // trips the emitter's max-listener warning at 11.
+      const closingBaseline = engine.listenerCount(EngineEvent.Closing);
+      const restartingBaseline = engine.listenerCount(EngineEvent.Restarting);
+
+      const pending = Array.from({ length: 20 }, () => engine.negotiate());
+      await tick();
+
+      expect(negotiate).toHaveBeenCalledTimes(20);
+      expect(engine.listenerCount(EngineEvent.Closing)).toBe(closingBaseline);
+      expect(engine.listenerCount(EngineEvent.Restarting)).toBe(restartingBaseline);
+
+      // The single central listener still has to reach every one of them.
+      engine.emit(EngineEvent.Closing);
+      await expect(Promise.all(pending)).resolves.toHaveLength(20);
+      expect(controllers).toHaveLength(20);
+      expect(controllers.every((c) => c.signal.aborted)).toBe(true);
+      expect(pendingAborts(engine).size).toBe(0);
+    });
+
+    it('aborts in-flight negotiations on restart without disarming later ones', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const { controllers } = stubNegotiableEngine(engine);
+
+      const firstBatch = [engine.negotiate(), engine.negotiate()];
+      await tick();
+      engine.emit(EngineEvent.Restarting);
+      await expect(Promise.all(firstBatch)).resolves.toHaveLength(2);
+      expect(controllers.every((c) => c.signal.aborted)).toBe(true);
+
+      // Restarting fires on every reconnect attempt, so the fan-out point has to re-arm: a
+      // negotiation started after a restart must not be aborted on arrival.
+      const afterRestart = engine.negotiate();
+      await tick();
+      expect(controllers).toHaveLength(3);
+      expect(controllers[2].signal.aborted).toBe(false);
+      expect(pendingAborts(engine).size).toBe(1);
+
+      engine.emit(EngineEvent.Restarting);
+      await expect(afterRestart).resolves.toBeUndefined();
+      expect(controllers[2].signal.aborted).toBe(true);
+      expect(pendingAborts(engine).size).toBe(0);
+    });
+  });
 });

src/room/RTCEngine.ts

@@ -260,6 +260,14 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
   /** timestamp (ms) the primary transport entered `CONNECTING`, used to bound how long we tolerate it */
   private transportConnectingSince?: number;
 
+  /**
+   * Abort handlers for the in-flight `negotiate()` calls, which a single central
+   * Closing/Restarting listener pair fans out to. Registering a listener pair per call instead
+   * accumulates them on the engine emitter and trips its max-listener warning under
+   * renegotiation bursts.
+   */
+  private pendingNegotiationAborts = new Set<() => void>();
+
   constructor(private options: InternalRoomOptions) {
     super();
     this.log = getLogger(options.loggerName ?? LoggerNames.Engine, () => this.logContext);
@@ -300,6 +308,15 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     this.client.onParticipantUpdate = (updates) =>
       this.emit(EngineEvent.ParticipantUpdate, updates);
     this.client.onJoined = (joinResponse) => this.emit(EngineEvent.Joined, joinResponse);
+
+    const abortPendingNegotiations = () => {
+      // Iterate a copy: each handler removes itself from the set as its negotiation settles.
+      for (const abort of Array.from(this.pendingNegotiationAborts)) {
+        abort();
+      }
+    };
+    this.on(EngineEvent.Closing, abortPendingNegotiations);
+    this.on(EngineEvent.Restarting, abortPendingNegotiations);
   }
 
   /** @internal */
@@ -1786,8 +1803,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       if (this.isClosed) {
         reject(new NegotiationError('cannot negotiate on closed engine'));
       }
-      this.on(EngineEvent.Closing, handleClosed);
-      this.on(EngineEvent.Restarting, handleClosed);
+      this.pendingNegotiationAborts.add(handleClosed);
       this.pcManager.publisher.off(PCEvents.RTPVideoPayloadTypes, this.onRtpMapAvailable);
       this.pcManager.publisher.once(PCEvents.RTPVideoPayloadTypes, this.onRtpMapAvailable);
 
@@ -1811,8 +1827,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
           reject(new Error(String(e)));
         }
       } finally {
-        this.off(EngineEvent.Closing, handleClosed);
-        this.off(EngineEvent.Restarting, handleClosed);
+        this.pendingNegotiationAborts.delete(handleClosed);
       }
     });
   }