diff --git a/.changeset/loud-pears-negotiate.md b/.changeset/loud-pears-negotiate.md new file mode 100644 index 0000000000..183e4303c9 --- /dev/null +++ b/.changeset/loud-pears-negotiate.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Avoid attaching a new Closing/Restarting event listener for each negotiate call diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index 1dc28a2cf6..e8be48da8f 100644 --- a/src/room/RTCEngine.test.ts +++ b/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((_resolve, reject) => { + abortController.signal.addEventListener( + 'abort', + () => reject(new Error('negotiation aborted')), + { once: true }, + ); + }); + }); + Object.assign(engine as unknown as Record, { + _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); + }); + }); }); diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index e8bd78f39e..6467377223 100644 --- a/src/room/RTCEngine.ts +++ b/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); } }); }