From 17f66bb61d7e983fda6d531de4bb3edcdbbcefa1 Mon Sep 17 00:00:00 2001
From: ANDROIDHASSAN <84630851+ANDROIDHASSAN@users.noreply.github.com>
Date: Wed, 14 Oct 2026 20:08:38 +0530
Subject: [PATCH 1/6] fix(rtc): keep an AudioMixer input alive after a read
 timeout

The mixer read each stream with asyncio.wait_for(stream.__anext__(), ...).
On timeout wait_for cancels that read, and cancelling __anext__ of an async
generator closes it, so one late frame ended the stream and the rest of its
audio was dropped. Keep the timed-out read and await it again in the next
round; remove_stream() and aclose() cancel it.
---
 livekit-rtc/livekit/rtc/audio_mixer.py | 34 ++++++++++++---
 tests/rtc/test_mixer.py                | 60 +++++++++++++++++++++++++-
 2 files changed, 88 insertions(+), 6 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_mixer.py b/livekit-rtc/livekit/rtc/audio_mixer.py
index e2f28c6b..72b409fb 100644
--- a/livekit-rtc/livekit/rtc/audio_mixer.py
+++ b/livekit-rtc/livekit/rtc/audio_mixer.py
@@ -52,6 +52,9 @@ def __init__(
         """
         self._streams: set[_Stream] = set()
         self._buffers: dict[_Stream, np.ndarray] = {}
+        # A read that timed out is kept for the next round: cancelling it would close an
+        # async generator stream for good.
+        self._pending_reads: dict[_Stream, asyncio.Future[AudioFrame]] = {}
         self._sample_rate: int = sample_rate
         self._num_channels: int = num_channels
         self._chunk_size: int = blocksize if blocksize > 0 else int(sample_rate // 10)
@@ -90,6 +93,9 @@ def remove_stream(self, stream: AsyncIterator[AudioFrame]) -> None:
         """
         self._streams.discard(stream)
         self._buffers.pop(stream, None)
+        pending = self._pending_reads.pop(stream, None)
+        if pending is not None:
+            _discard_read(pending)
 
     def __aiter__(self) -> "AudioMixer":
         return self
@@ -110,6 +116,9 @@ async def aclose(self) -> None:
         self._mixer_task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
             await self._mixer_task
+        for pending in self._pending_reads.values():
+            _discard_read(pending)
+        self._pending_reads.clear()
 
     def end_input(self) -> None:
         """
@@ -174,13 +183,20 @@ async def _get_contribution(
         had_data = buf.shape[0] > 0
         exhausted = False
         while buf.shape[0] < self._chunk_size and not exhausted:
-            try:
-                frame = await asyncio.wait_for(
-                    stream.__anext__(), timeout=self._stream_timeout_ms / 1000
-                )
-            except asyncio.TimeoutError:
+            # The read stays in _pending_reads until it completes, so remove_stream()
+            # and aclose() can always cancel it.
+            read = self._pending_reads.get(stream)
+            if read is None:
+                read = asyncio.ensure_future(stream.__anext__())
+                self._pending_reads[stream] = read
+            done, _ = await asyncio.wait({read}, timeout=self._stream_timeout_ms / 1000)
+            if not done:
                 logger.warning(f"AudioMixer: stream {stream} timeout, ignoring")
                 break
+            if self._pending_reads.get(stream) is read:
+                del self._pending_reads[stream]
+            try:
+                frame = read.result()
             except StopAsyncIteration:
                 exhausted = True
                 break
@@ -198,3 +214,11 @@ async def _get_contribution(
                 np.empty((0, self._num_channels), dtype=np.int16),
             )
         return _Contribution(stream, contrib, buf, had_data, exhausted)
+
+
+def _discard_read(read: "asyncio.Future[AudioFrame]") -> None:
+    """Cancel a read that is no longer wanted, or consume the outcome of a finished one."""
+    if not read.done():
+        read.cancel()
+    elif not read.cancelled():
+        read.exception()  # mark a failed read as handled
diff --git a/tests/rtc/test_mixer.py b/tests/rtc/test_mixer.py
index 1ab75f4b..d7ee8e6e 100644
--- a/tests/rtc/test_mixer.py
+++ b/tests/rtc/test_mixer.py
@@ -1,9 +1,11 @@
 # type: ignore
 
+import asyncio
+
 import numpy as np
 import pytest
 
-from livekit.rtc import AudioMixer
+from livekit.rtc import AudioFrame, AudioMixer
 from livekit.rtc.utils import sine_wave_generator
 
 SAMPLE_RATE = 48000
@@ -56,3 +58,59 @@ async def test_mixer_two_sine_waves():
     # Assert that the peaks include 440Hz and 880Hz (with a tolerance of ±5 Hz)
     assert any(np.isclose(peak_freqs, 440, atol=5)), f"Expected 440 Hz in peaks, got: {peak_freqs}"
     assert any(np.isclose(peak_freqs, 880, atol=5)), f"Expected 880 Hz in peaks, got: {peak_freqs}"
+
+
+@pytest.mark.asyncio
+async def test_mixer_keeps_async_generator_after_a_timeout():
+    """
+    A stream that misses the timeout once must not be closed: cancelling the
+    pending read of an async generator ends it, which dropped the rest of its audio.
+    """
+    frames, block = 10, 480
+
+    async def late_once():
+        for i in range(frames):
+            if i == 3:
+                await asyncio.sleep(0.25)
+            yield AudioFrame(np.full(block, 1000, dtype=np.int16).tobytes(), SAMPLE_RATE, 1, block)
+
+    mixer = AudioMixer(
+        sample_rate=SAMPLE_RATE, num_channels=1, blocksize=block, stream_timeout_ms=100
+    )
+    mixer.add_stream(late_once())
+    mixer.end_input()
+
+    with_audio = 0
+    async for frame in mixer:
+        if np.frombuffer(frame.data.tobytes(), dtype=np.int16).any():
+            with_audio += 1
+    await mixer.aclose()
+
+    assert with_audio == frames
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("how", ["remove_stream", "aclose"])
+async def test_mixer_cancels_a_pending_read_when_the_stream_goes_away(how):
+    closed = asyncio.Event()
+
+    async def never_yields():
+        try:
+            await asyncio.Event().wait()
+            yield  # pragma: no cover
+        finally:
+            closed.set()
+
+    mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=50)
+    stream = never_yields()
+    mixer.add_stream(stream)
+    await asyncio.sleep(0.12)  # at least one round has timed out on the stream
+
+    if how == "remove_stream":
+        mixer.remove_stream(stream)
+    else:
+        await mixer.aclose()
+
+    await asyncio.wait_for(closed.wait(), timeout=1)
+    assert not mixer._pending_reads
+    await mixer.aclose()

From 8642fa86a9d0d92805ab8398d822591ba3985718 Mon Sep 17 00:00:00 2001
From: ANDROIDHASSAN <84630851+ANDROIDHASSAN@users.noreply.github.com>
Date: Wed, 14 Oct 2026 20:59:10 +0530
Subject: [PATCH 2/6] fix(rtc): finish pending reads on aclose and skip removed
 streams

aclose() now waits for cancelled reads to run their cleanup, and a read
cancelled by remove_stream() has its outcome consumed whenever it ends.

A round scheduled before remove_stream() no longer starts a new read on
the removed stream, which left an entry in _pending_reads (seen on the
Windows CI run).

A stream that raises asyncio.TimeoutError itself is handled like the
mixer's own timeout again, so the samples it already produced are kept.
---
 livekit-rtc/livekit/rtc/audio_mixer.py | 25 +++++++----
 tests/rtc/test_mixer.py                | 57 ++++++++++++++++++++++++++
 2 files changed, 75 insertions(+), 7 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_mixer.py b/livekit-rtc/livekit/rtc/audio_mixer.py
index 72b409fb..589f6dd1 100644
--- a/livekit-rtc/livekit/rtc/audio_mixer.py
+++ b/livekit-rtc/livekit/rtc/audio_mixer.py
@@ -116,9 +116,12 @@ async def aclose(self) -> None:
         self._mixer_task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
             await self._mixer_task
-        for pending in self._pending_reads.values():
-            _discard_read(pending)
+        pending = list(self._pending_reads.values())
         self._pending_reads.clear()
+        for read in pending:
+            _discard_read(read)
+        # let the streams finish their cleanup before the mixer counts as closed
+        await asyncio.gather(*pending, return_exceptions=True)
 
     def end_input(self) -> None:
         """
@@ -182,7 +185,8 @@ async def _get_contribution(
     ) -> _Contribution:
         had_data = buf.shape[0] > 0
         exhausted = False
-        while buf.shape[0] < self._chunk_size and not exhausted:
+        # a round scheduled before remove_stream() must not start a new read on that stream
+        while buf.shape[0] < self._chunk_size and not exhausted and stream in self._streams:
             # The read stays in _pending_reads until it completes, so remove_stream()
             # and aclose() can always cancel it.
             read = self._pending_reads.get(stream)
@@ -200,6 +204,10 @@ async def _get_contribution(
             except StopAsyncIteration:
                 exhausted = True
                 break
+            except asyncio.TimeoutError:
+                # raised by the stream itself: handled like the mixer's own timeout, as before
+                logger.warning(f"AudioMixer: stream {stream} timeout, ignoring")
+                break
             new_data = np.frombuffer(frame.data.tobytes(), dtype=np.int16).reshape(
                 -1, self._num_channels
             )
@@ -217,8 +225,11 @@ async def _get_contribution(
 
 
 def _discard_read(read: "asyncio.Future[AudioFrame]") -> None:
-    """Cancel a read that is no longer wanted, or consume the outcome of a finished one."""
-    if not read.done():
-        read.cancel()
-    elif not read.cancelled():
+    """Cancel a read that is no longer wanted and consume its outcome whenever it ends."""
+    read.cancel()
+    read.add_done_callback(_consume_outcome)
+
+
+def _consume_outcome(read: "asyncio.Future[AudioFrame]") -> None:
+    if not read.cancelled():
         read.exception()  # mark a failed read as handled
diff --git a/tests/rtc/test_mixer.py b/tests/rtc/test_mixer.py
index d7ee8e6e..8acde0b6 100644
--- a/tests/rtc/test_mixer.py
+++ b/tests/rtc/test_mixer.py
@@ -114,3 +114,60 @@ async def never_yields():
     await asyncio.wait_for(closed.wait(), timeout=1)
     assert not mixer._pending_reads
     await mixer.aclose()
+
+
+@pytest.mark.asyncio
+async def test_mixer_starts_no_read_on_a_removed_stream():
+    # a round can be scheduled before remove_stream() and run after it
+    async def never_yields():
+        await asyncio.Event().wait()
+        yield  # pragma: no cover
+
+    mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=10)
+    stream = never_yields()
+    mixer.add_stream(stream)
+    mixer.remove_stream(stream)
+
+    await mixer._get_contribution(stream, np.empty((0, 1), dtype=np.int16))
+    assert not mixer._pending_reads
+    await mixer.aclose()
+
+
+@pytest.mark.asyncio
+async def test_mixer_keeps_buffered_audio_when_a_stream_raises_timeout_error():
+    async def partial_then_timeout():
+        yield AudioFrame(np.full(240, 1000, dtype=np.int16).tobytes(), SAMPLE_RATE, 1, 240)
+        raise asyncio.TimeoutError
+
+    mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480)
+    mixer.add_stream(partial_then_timeout())
+    mixer.end_input()
+
+    with_audio = 0
+    async for frame in mixer:
+        if np.frombuffer(frame.data.tobytes(), dtype=np.int16).any():
+            with_audio += 1
+    await mixer.aclose()
+
+    assert with_audio == 1
+
+
+@pytest.mark.asyncio
+async def test_mixer_aclose_waits_for_a_cancelled_read_to_clean_up():
+    cleaned_up = False
+
+    async def slow_cleanup():
+        nonlocal cleaned_up
+        try:
+            await asyncio.Event().wait()
+            yield  # pragma: no cover
+        finally:
+            await asyncio.sleep(0.1)
+            cleaned_up = True
+
+    mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=50)
+    mixer.add_stream(slow_cleanup())
+    await asyncio.sleep(0.12)
+
+    await mixer.aclose()
+    assert cleaned_up

From 2fe5a4e70cb493b69db6d28b07b13c75ffe540de Mon Sep 17 00:00:00 2001
From: ANDROIDHASSAN <84630851+ANDROIDHASSAN@users.noreply.github.com>
Date: Wed, 14 Oct 2026 21:08:36 +0530
Subject: [PATCH 3/6] fix(rtc): bound the wait for cancelled reads in
 AudioMixer.aclose

A stream that catches CancelledError and keeps waiting made aclose()
hang. aclose() now waits for cancelled reads for at most
stream_timeout_ms; their outcome is still consumed when they end, and
cancelled reads are referenced until then.
---
 livekit-rtc/livekit/rtc/audio_mixer.py | 14 ++++++++++++--
 tests/rtc/test_mixer.py                | 26 +++++++++++++++++++++++++-
 2 files changed, 37 insertions(+), 3 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_mixer.py b/livekit-rtc/livekit/rtc/audio_mixer.py
index 589f6dd1..0d56657b 100644
--- a/livekit-rtc/livekit/rtc/audio_mixer.py
+++ b/livekit-rtc/livekit/rtc/audio_mixer.py
@@ -111,6 +111,8 @@ async def aclose(self) -> None:
         Immediately stop mixing and close the mixer.
 
         This cancels the mixing task, and any unconsumed output in the queue may be dropped.
+        Pending stream reads are cancelled, and their cleanup is awaited for up to
+        stream_timeout_ms.
         """
         self._ending = True
         self._mixer_task.cancel()
@@ -120,8 +122,10 @@ async def aclose(self) -> None:
         self._pending_reads.clear()
         for read in pending:
             _discard_read(read)
-        # let the streams finish their cleanup before the mixer counts as closed
-        await asyncio.gather(*pending, return_exceptions=True)
+        if pending:
+            # give the streams time to clean up, bounded so one that ignores
+            # cancellation cannot hang shutdown
+            await asyncio.wait(pending, timeout=self._stream_timeout_ms / 1000)
 
     def end_input(self) -> None:
         """
@@ -224,12 +228,18 @@ async def _get_contribution(
         return _Contribution(stream, contrib, buf, had_data, exhausted)
 
 
+# cancelled reads that may outlive their mixer, kept until their outcome is consumed
+_cancelled_reads: "set[asyncio.Future[AudioFrame]]" = set()
+
+
 def _discard_read(read: "asyncio.Future[AudioFrame]") -> None:
     """Cancel a read that is no longer wanted and consume its outcome whenever it ends."""
     read.cancel()
+    _cancelled_reads.add(read)  # the event loop holds tasks weakly
     read.add_done_callback(_consume_outcome)
 
 
 def _consume_outcome(read: "asyncio.Future[AudioFrame]") -> None:
+    _cancelled_reads.discard(read)
     if not read.cancelled():
         read.exception()  # mark a failed read as handled
diff --git a/tests/rtc/test_mixer.py b/tests/rtc/test_mixer.py
index 8acde0b6..94a1319f 100644
--- a/tests/rtc/test_mixer.py
+++ b/tests/rtc/test_mixer.py
@@ -162,7 +162,7 @@ async def slow_cleanup():
             await asyncio.Event().wait()
             yield  # pragma: no cover
         finally:
-            await asyncio.sleep(0.1)
+            await asyncio.sleep(0.01)  # well within stream_timeout_ms
             cleaned_up = True
 
     mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=50)
@@ -171,3 +171,27 @@ async def slow_cleanup():
 
     await mixer.aclose()
     assert cleaned_up
+
+
+@pytest.mark.asyncio
+async def test_mixer_aclose_returns_when_a_stream_ignores_cancellation():
+    release = asyncio.Event()
+
+    async def ignores_cancel():
+        while True:
+            try:
+                await release.wait()
+                return
+            except asyncio.CancelledError:
+                continue
+        yield  # pragma: no cover
+
+    mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=50)
+    mixer.add_stream(ignores_cancel())
+    await asyncio.sleep(0.12)
+
+    closing = asyncio.ensure_future(mixer.aclose())
+    done, _ = await asyncio.wait({closing}, timeout=1)
+    release.set()
+    await closing
+    assert done

From 09b6002ae8b2dabddfd54164853dd885edce5dc5 Mon Sep 17 00:00:00 2001
From: ANDROIDHASSAN <84630851+ANDROIDHASSAN@users.noreply.github.com>
Date: Wed, 14 Oct 2026 21:20:22 +0530
Subject: [PATCH 4/6] Revert "fix(rtc): bound the wait for cancelled reads in
 AudioMixer.aclose"

This reverts commit 2fe5a4e. aclose() goes back to awaiting the cancelled
reads in full, which is what main does (wait_for awaits the cancelled
read inside the mixer task, and aclose() awaits that task). A stream
that ignores cancellation hangs shutdown on main as well; changing that
is a separate decision from this fix.
---
 livekit-rtc/livekit/rtc/audio_mixer.py | 14 ++------------
 tests/rtc/test_mixer.py                | 26 +-------------------------
 2 files changed, 3 insertions(+), 37 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_mixer.py b/livekit-rtc/livekit/rtc/audio_mixer.py
index 0d56657b..589f6dd1 100644
--- a/livekit-rtc/livekit/rtc/audio_mixer.py
+++ b/livekit-rtc/livekit/rtc/audio_mixer.py
@@ -111,8 +111,6 @@ async def aclose(self) -> None:
         Immediately stop mixing and close the mixer.
 
         This cancels the mixing task, and any unconsumed output in the queue may be dropped.
-        Pending stream reads are cancelled, and their cleanup is awaited for up to
-        stream_timeout_ms.
         """
         self._ending = True
         self._mixer_task.cancel()
@@ -122,10 +120,8 @@ async def aclose(self) -> None:
         self._pending_reads.clear()
         for read in pending:
             _discard_read(read)
-        if pending:
-            # give the streams time to clean up, bounded so one that ignores
-            # cancellation cannot hang shutdown
-            await asyncio.wait(pending, timeout=self._stream_timeout_ms / 1000)
+        # let the streams finish their cleanup before the mixer counts as closed
+        await asyncio.gather(*pending, return_exceptions=True)
 
     def end_input(self) -> None:
         """
@@ -228,18 +224,12 @@ async def _get_contribution(
         return _Contribution(stream, contrib, buf, had_data, exhausted)
 
 
-# cancelled reads that may outlive their mixer, kept until their outcome is consumed
-_cancelled_reads: "set[asyncio.Future[AudioFrame]]" = set()
-
-
 def _discard_read(read: "asyncio.Future[AudioFrame]") -> None:
     """Cancel a read that is no longer wanted and consume its outcome whenever it ends."""
     read.cancel()
-    _cancelled_reads.add(read)  # the event loop holds tasks weakly
     read.add_done_callback(_consume_outcome)
 
 
 def _consume_outcome(read: "asyncio.Future[AudioFrame]") -> None:
-    _cancelled_reads.discard(read)
     if not read.cancelled():
         read.exception()  # mark a failed read as handled
diff --git a/tests/rtc/test_mixer.py b/tests/rtc/test_mixer.py
index 94a1319f..8acde0b6 100644
--- a/tests/rtc/test_mixer.py
+++ b/tests/rtc/test_mixer.py
@@ -162,7 +162,7 @@ async def slow_cleanup():
             await asyncio.Event().wait()
             yield  # pragma: no cover
         finally:
-            await asyncio.sleep(0.01)  # well within stream_timeout_ms
+            await asyncio.sleep(0.1)
             cleaned_up = True
 
     mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=50)
@@ -171,27 +171,3 @@ async def slow_cleanup():
 
     await mixer.aclose()
     assert cleaned_up
-
-
-@pytest.mark.asyncio
-async def test_mixer_aclose_returns_when_a_stream_ignores_cancellation():
-    release = asyncio.Event()
-
-    async def ignores_cancel():
-        while True:
-            try:
-                await release.wait()
-                return
-            except asyncio.CancelledError:
-                continue
-        yield  # pragma: no cover
-
-    mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=50)
-    mixer.add_stream(ignores_cancel())
-    await asyncio.sleep(0.12)
-
-    closing = asyncio.ensure_future(mixer.aclose())
-    done, _ = await asyncio.wait({closing}, timeout=1)
-    release.set()
-    await closing
-    assert done

From e57d4fd14c2ffdd43c98a885b8b6850730fe0b19 Mon Sep 17 00:00:00 2001
From: ANDROIDHASSAN <84630851+ANDROIDHASSAN@users.noreply.github.com>
Date: Wed, 14 Oct 2026 21:55:02 +0530
Subject: [PATCH 5/6] docs(rtc): note that AudioMixer streams must propagate
 cancellation

---
 livekit-rtc/livekit/rtc/audio_mixer.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/livekit-rtc/livekit/rtc/audio_mixer.py b/livekit-rtc/livekit/rtc/audio_mixer.py
index 589f6dd1..918c4c57 100644
--- a/livekit-rtc/livekit/rtc/audio_mixer.py
+++ b/livekit-rtc/livekit/rtc/audio_mixer.py
@@ -111,6 +111,8 @@ async def aclose(self) -> None:
         Immediately stop mixing and close the mixer.
 
         This cancels the mixing task, and any unconsumed output in the queue may be dropped.
+        Pending stream reads are cancelled and awaited, so a stream must let
+        asyncio.CancelledError propagate once its cleanup is done.
         """
         self._ending = True
         self._mixer_task.cancel()

From 50746677f9baec5a98d328b8fc4321c6f1b9d719 Mon Sep 17 00:00:00 2001
From: ANDROIDHASSAN <84630851+ANDROIDHASSAN@users.noreply.github.com>
Date: Wed, 14 Oct 2026 22:02:35 +0530
Subject: [PATCH 6/6] fix(rtc): drop a stream that keeps stalling after
 end_input

On main a read timeout closed the stream, so a stalled stream was
dropped and the drain after end_input() ended. With the read now kept
across timeouts, a stream that never yields again kept the drain alive
forever. After end_input(), a stream is now dropped after 10
consecutive timeouts; a single late frame is still kept.
---
 livekit-rtc/livekit/rtc/audio_mixer.py | 10 ++++++++++
 tests/rtc/test_mixer.py                | 20 ++++++++++++++++++++
 2 files changed, 30 insertions(+)

diff --git a/livekit-rtc/livekit/rtc/audio_mixer.py b/livekit-rtc/livekit/rtc/audio_mixer.py
index 918c4c57..f293184a 100644
--- a/livekit-rtc/livekit/rtc/audio_mixer.py
+++ b/livekit-rtc/livekit/rtc/audio_mixer.py
@@ -8,6 +8,9 @@
 
 _Stream = AsyncIterator[AudioFrame]
 
+# consecutive timeouts after which a stream is dropped once end_input() was called
+_MAX_STALLED_ROUNDS = 10
+
 
 @dataclass
 class _Contribution:
@@ -55,6 +58,7 @@ def __init__(
         # A read that timed out is kept for the next round: cancelling it would close an
         # async generator stream for good.
         self._pending_reads: dict[_Stream, asyncio.Future[AudioFrame]] = {}
+        self._stalled_rounds: dict[_Stream, int] = {}
         self._sample_rate: int = sample_rate
         self._num_channels: int = num_channels
         self._chunk_size: int = blocksize if blocksize > 0 else int(sample_rate // 10)
@@ -93,6 +97,7 @@ def remove_stream(self, stream: AsyncIterator[AudioFrame]) -> None:
         """
         self._streams.discard(stream)
         self._buffers.pop(stream, None)
+        self._stalled_rounds.pop(stream, None)
         pending = self._pending_reads.pop(stream, None)
         if pending is not None:
             _discard_read(pending)
@@ -198,7 +203,12 @@ async def _get_contribution(
             done, _ = await asyncio.wait({read}, timeout=self._stream_timeout_ms / 1000)
             if not done:
                 logger.warning(f"AudioMixer: stream {stream} timeout, ignoring")
+                stalled = self._stalled_rounds.get(stream, 0) + 1
+                self._stalled_rounds[stream] = stalled
+                # after end_input() a stream that keeps stalling is dropped, so the drain ends
+                exhausted = self._ending and stalled >= _MAX_STALLED_ROUNDS
                 break
+            self._stalled_rounds.pop(stream, None)
             if self._pending_reads.get(stream) is read:
                 del self._pending_reads[stream]
             try:
diff --git a/tests/rtc/test_mixer.py b/tests/rtc/test_mixer.py
index 8acde0b6..ffb450cc 100644
--- a/tests/rtc/test_mixer.py
+++ b/tests/rtc/test_mixer.py
@@ -89,6 +89,26 @@ async def late_once():
     assert with_audio == frames
 
 
+@pytest.mark.asyncio
+async def test_mixer_drops_a_stalled_stream_after_end_input():
+    async def one_frame_then_stall():
+        yield AudioFrame(np.full(480, 1000, dtype=np.int16).tobytes(), SAMPLE_RATE, 1, 480)
+        await asyncio.Event().wait()
+
+    mixer = AudioMixer(sample_rate=SAMPLE_RATE, num_channels=1, blocksize=480, stream_timeout_ms=50)
+    mixer.add_stream(one_frame_then_stall())
+    mixer.end_input()
+
+    async def drain():
+        return [frame async for frame in mixer]
+
+    draining = asyncio.ensure_future(drain())
+    done, _ = await asyncio.wait({draining}, timeout=1)
+    await mixer.aclose()
+    assert done
+    assert len(draining.result()) == 1
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("how", ["remove_stream", "aclose"])
 async def test_mixer_cancels_a_pending_read_when_the_stream_goes_away(how):