fix(rtc): end AudioMixer iteration when endInput() runs with no streams left

.changeset/audio-mixer-end-input-wakeup.md

@@ -0,0 +1,5 @@
+---
+'@livekit/rtc-node': patch
+---
+
+`AudioMixer.endInput()` now ends iteration when no streams are left. Before, a consumer already waiting for a frame stayed blocked forever if `endInput()` was called after the last stream was removed (or before any was added).

packages/livekit-rtc/src/audio_mixer.test.ts

@@ -111,6 +111,69 @@ describe('AudioMixer', () => {
     expect(frames.length).toBe(0);
   });
 
+  it('ends iteration when endInput() is called while a consumer is waiting', async () => {
+    const mixer = new AudioMixer(48000, 1);
+
+    // Start consuming before endInput(), so the consumer is already parked
+    // waiting for a frame when the mixer is told no more streams are coming.
+    const consumed = (async () => {
+      const frames: AudioFrame[] = [];
+      for await (const frame of mixer) {
+        frames.push(frame);
+      }
+      return frames;
+    })();
+
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    mixer.endInput();
+
+    const result = await Promise.race([
+      consumed,
+      new Promise<'hung'>((resolve) => setTimeout(() => resolve('hung'), 1000)),
+    ]);
+    expect(result).toEqual([]);
+
+    await mixer.aclose();
+  });
+
+  it('ends iteration when endInput() follows removal of the last stream', async () => {
+    const mixer = new AudioMixer(48000, 1, { streamTimeoutMs: 10 });
+
+    // A stream that never produces audio, like a muted participant.
+    const silentStream = async function* (): AsyncGenerator<AudioFrame> {
+      await new Promise(() => {});
+    };
+    const stream = silentStream();
+
+    const originalWarn = console.warn;
+    console.warn = () => {};
+    try {
+      mixer.addStream(stream);
+
+      const consumed = (async () => {
+        const frames: AudioFrame[] = [];
+        for await (const frame of mixer) {
+          frames.push(frame);
+        }
+        return frames;
+      })();
+
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      mixer.removeStream(stream);
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      mixer.endInput();
+
+      const result = await Promise.race([
+        consumed,
+        new Promise<'hung'>((resolve) => setTimeout(() => resolve('hung'), 1000)),
+      ]);
+      expect(result).toEqual([]);
+    } finally {
+      console.warn = originalWarn;
+      await mixer.aclose();
+    }
+  });
+
   it('clips audio values to int16 range', async () => {
     const sampleRate = 48000;
     const numChannels = 1;

packages/livekit-rtc/src/audio_mixer.ts

@@ -185,6 +185,10 @@ export class AudioMixer {
    */
   endInput(): void {
     this.ending = true;
+    // No more streams will be added, so close the signal queue. If the mixer is
+    // parked waiting for a stream (none left), this wakes it so it can see it is
+    // ending, exit, and close the output queue for any waiting consumer.
+    this.streamSignal.close();
   }
 
   private async getNextFrame(): Promise<AudioFrame | null> {