From 8ae57924719b13e8d0be69c7411c634c333576ef Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Sat, 17 Oct 2026 00:17:25 +0500
Subject: [PATCH 1/6] fix(rtc): offload FFI handle disposal to thread in aclose
 and add sync close (#832)

- In AudioSource, AudioStream, VideoSource, and VideoStream, add a synchronous close() method.
- In AudioSource, VideoSource, AudioStream, and VideoStream, run self._ffi_handle.dispose() via asyncio.to_thread in aclose() so the synchronous native drop call does not block the asyncio event loop.
- In AudioSource, cancel any pending _join_handle timer and resolve _join_fut when close() or aclose() is called so waiters on wait_for_playout are immediately unblocked.
- In VideoSource.capture_frame, safely ignore frames if the handle is already disposed.

Fixes #832
---
 livekit-rtc/livekit/rtc/audio_source.py       | 17 +++-
 livekit-rtc/livekit/rtc/audio_stream.py       | 13 ++-
 livekit-rtc/livekit/rtc/video_source.py       | 18 +++-
 livekit-rtc/livekit/rtc/video_stream.py       |  8 +-
 livekit-rtc/tests/test_source_stream_close.py | 95 +++++++++++++++++++
 5 files changed, 145 insertions(+), 6 deletions(-)
 create mode 100644 livekit-rtc/tests/test_source_stream_close.py

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index 680cc803..e6b171e8 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -173,6 +173,10 @@ async def wait_for_playout(self) -> None:
         await asyncio.shield(self._join_fut)
 
     def _release_waiter(self) -> None:
+        if self._join_handle is not None:
+            self._join_handle.cancel()
+            self._join_handle = None
+
         if self._join_fut is None:
             return  # could be None when clear_queue is called
 
@@ -183,9 +187,18 @@ def _release_waiter(self) -> None:
         self._q_size = 0.0
         self._join_fut = None
 
-    async def aclose(self) -> None:
-        """Close the audio source
+    def close(self) -> None:
+        """Close the audio source.
 
         This method cleans up resources associated with the audio source.
         """
+        self._release_waiter()
         self._ffi_handle.dispose()
+
+    async def aclose(self) -> None:
+        """Close the audio source without blocking the event loop.
+
+        This method cleans up resources associated with the audio source.
+        """
+        self._release_waiter()
+        await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/audio_stream.py b/livekit-rtc/livekit/rtc/audio_stream.py
index 802c0830..76976c21 100644
--- a/livekit-rtc/livekit/rtc/audio_stream.py
+++ b/livekit-rtc/livekit/rtc/audio_stream.py
@@ -324,15 +324,24 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
+    def close(self) -> None:
+        """Synchronously close the audio stream and release its resources."""
+        if self._track is not None:
+            self._track._unregister_audio_stream(self)
+        self._ffi_handle.dispose()
+        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
+        if self._processor is not None and self._processor_auto_close:
+            self._processor._close()
+
     async def aclose(self) -> None:
         """Asynchronously close the audio stream.
 
         This method cleans up resources associated with the audio stream and waits for
-        any pending operations to complete.
+        any pending operations to complete without blocking the event loop.
         """
         if self._track is not None:
             self._track._unregister_audio_stream(self)
-        self._ffi_handle.dispose()
+        await asyncio.to_thread(self._ffi_handle.dispose)
         await self._task
         if self._processor is not None and self._processor_auto_close:
             self._processor._close()
diff --git a/livekit-rtc/livekit/rtc/video_source.py b/livekit-rtc/livekit/rtc/video_source.py
index c8dd2a02..ad34fb2d 100644
--- a/livekit-rtc/livekit/rtc/video_source.py
+++ b/livekit-rtc/livekit/rtc/video_source.py
@@ -14,6 +14,8 @@
 
 from __future__ import annotations
 
+import asyncio
+
 from ._ffi_client import FfiHandle, FfiClient
 from ._proto import ffi_pb2 as proto_ffi
 from ._proto import video_frame_pb2 as proto_video
@@ -61,6 +63,9 @@ def capture_frame(
         metadata: proto_video.FrameMetadata | None = None,
     ) -> None:
         """Capture a frame, optionally attaching frame metadata."""
+        if self._ffi_handle.disposed:
+            return
+
         req = proto_ffi.FfiRequest()
         req.capture_video_frame.source_handle = self._ffi_handle.handle
         req.capture_video_frame.buffer.CopyFrom(frame._proto_info())
@@ -70,5 +75,16 @@ def capture_frame(
             req.capture_video_frame.metadata.CopyFrom(metadata)
         FfiClient.instance.request(req)
 
-    async def aclose(self) -> None:
+    def close(self) -> None:
+        """Close the video source.
+
+        This method cleans up resources associated with the video source.
+        """
         self._ffi_handle.dispose()
+
+    async def aclose(self) -> None:
+        """Close the video source without blocking the event loop.
+
+        This method cleans up resources associated with the video source.
+        """
+        await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py
index 07938fe5..ca0ef675 100644
--- a/livekit-rtc/livekit/rtc/video_stream.py
+++ b/livekit-rtc/livekit/rtc/video_stream.py
@@ -170,8 +170,14 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
-    async def aclose(self) -> None:
+    def close(self) -> None:
+        """Synchronously close the video stream and release its resources."""
         self._ffi_handle.dispose()
+        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
+
+    async def aclose(self) -> None:
+        """Asynchronously close the video stream without blocking the event loop."""
+        await asyncio.to_thread(self._ffi_handle.dispose)
         await self._task
 
     def _is_event(self, e: proto_ffi.FfiEvent) -> bool:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
new file mode 100644
index 00000000..cd84ac1d
--- /dev/null
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -0,0 +1,95 @@
+# Copyright 2026 LiveKit, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+
+import pytest
+
+from livekit import rtc
+
+
+@pytest.mark.asyncio
+async def test_audio_source_aclose_releases_waiter_and_cancels_timer() -> None:
+    source = rtc.AudioSource(48000, 1)
+    frame = rtc.AudioFrame.create(48000, 1, 48000)
+    await source.capture_frame(frame)
+
+    assert source._join_fut is not None
+    assert source._join_handle is not None
+
+    await source.aclose()
+
+    assert source._ffi_handle.disposed
+    assert source._join_fut is None
+    assert source._join_handle is None
+
+    # After aclose, wait_for_playout should return immediately without hanging
+    await source.wait_for_playout()
+
+    # Repeated calls are idempotent
+    await source.aclose()
+
+
+def test_audio_source_sync_close() -> None:
+    source = rtc.AudioSource(48000, 1)
+    assert not source._ffi_handle.disposed
+
+    source.close()
+    assert source._ffi_handle.disposed
+
+    # Repeated close is idempotent
+    source.close()
+
+
+@pytest.mark.asyncio
+async def test_audio_source_capture_after_close_is_noop() -> None:
+    source = rtc.AudioSource(48000, 1)
+    source.close()
+
+    frame = rtc.AudioFrame.create(48000, 1, 480)
+    # Should safely return without error
+    await source.capture_frame(frame)
+
+
+def test_video_source_sync_close() -> None:
+    source = rtc.VideoSource(640, 480)
+    assert not source._ffi_handle.disposed
+
+    source.close()
+    assert source._ffi_handle.disposed
+
+    # Repeated close is idempotent
+    source.close()
+
+
+@pytest.mark.asyncio
+async def test_video_source_aclose_and_capture_after_close() -> None:
+    source = rtc.VideoSource(640, 480)
+    assert not source._ffi_handle.disposed
+
+    await source.aclose()
+    assert source._ffi_handle.disposed
+
+    # Capture after close is safely ignored
+    frame = rtc.VideoFrame(
+        640,
+        480,
+        rtc.VideoBufferType.RGBA,
+        b"\x00" * (640 * 480 * 4),
+    )
+    source.capture_frame(frame)
+
+    # Repeated aclose is idempotent
+    await source.aclose()

From c97bbf9bd1a189e7158e33339991059b0498d035 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 9 Nov 2026 20:15:01 +0500
Subject: [PATCH 2/6] fix(rtc): handle cross-thread close, safe mock
 attributes, and isolate FFI lifecycle test

# Conflicts:
#	livekit-rtc/tests/test_ffi_handle_lifecycle.py
---
 livekit-rtc/livekit/rtc/audio_source.py       | 30 ++++++++++++++++---
 livekit-rtc/livekit/rtc/audio_stream.py       | 13 ++------
 livekit-rtc/livekit/rtc/video_source.py       |  5 ++--
 livekit-rtc/livekit/rtc/video_stream.py       |  8 +----
 livekit-rtc/tests/test_source_stream_close.py | 27 ++++++++++++-----
 5 files changed, 52 insertions(+), 31 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index e6b171e8..93574a5c 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -53,7 +53,17 @@ def __init__(
         """
         self._sample_rate = sample_rate
         self._num_channels = num_channels
-        self._loop = loop or asyncio.get_event_loop()
+
+        if loop is not None:
+            self._loop = loop
+        else:
+            try:
+                self._loop = asyncio.get_running_loop()
+            except RuntimeError:
+                try:
+                    self._loop = asyncio.get_event_loop()
+                except RuntimeError:
+                    self._loop = asyncio.new_event_loop()
 
         req = proto_ffi.FfiRequest()
         req.new_audio_source.type = proto_audio_frame.AudioSourceType.AUDIO_SOURCE_NATIVE
@@ -116,7 +126,7 @@ async def capture_frame(self, frame: AudioFrame) -> None:
             Exception: If there is an error during frame capture.
         """
 
-        if frame.samples_per_channel == 0 or self._ffi_handle.disposed:
+        if frame.samples_per_channel == 0 or getattr(self._ffi_handle, "disposed", False):
             return
 
         now = time.monotonic()
@@ -192,7 +202,18 @@ def close(self) -> None:
 
         This method cleans up resources associated with the audio source.
         """
-        self._release_waiter()
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            running_loop = None
+
+        if running_loop is self._loop:
+            self._release_waiter()
+        elif self._loop.is_running():
+            self._loop.call_soon_threadsafe(self._release_waiter)
+        else:
+            self._release_waiter()
+
         self._ffi_handle.dispose()
 
     async def aclose(self) -> None:
@@ -201,4 +222,5 @@ async def aclose(self) -> None:
         This method cleans up resources associated with the audio source.
         """
         self._release_waiter()
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        if not getattr(self._ffi_handle, "disposed", False):
+            await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/audio_stream.py b/livekit-rtc/livekit/rtc/audio_stream.py
index 76976c21..802c0830 100644
--- a/livekit-rtc/livekit/rtc/audio_stream.py
+++ b/livekit-rtc/livekit/rtc/audio_stream.py
@@ -324,24 +324,15 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
-    def close(self) -> None:
-        """Synchronously close the audio stream and release its resources."""
-        if self._track is not None:
-            self._track._unregister_audio_stream(self)
-        self._ffi_handle.dispose()
-        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
-        if self._processor is not None and self._processor_auto_close:
-            self._processor._close()
-
     async def aclose(self) -> None:
         """Asynchronously close the audio stream.
 
         This method cleans up resources associated with the audio stream and waits for
-        any pending operations to complete without blocking the event loop.
+        any pending operations to complete.
         """
         if self._track is not None:
             self._track._unregister_audio_stream(self)
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        self._ffi_handle.dispose()
         await self._task
         if self._processor is not None and self._processor_auto_close:
             self._processor._close()
diff --git a/livekit-rtc/livekit/rtc/video_source.py b/livekit-rtc/livekit/rtc/video_source.py
index ad34fb2d..6b2c3947 100644
--- a/livekit-rtc/livekit/rtc/video_source.py
+++ b/livekit-rtc/livekit/rtc/video_source.py
@@ -63,7 +63,7 @@ def capture_frame(
         metadata: proto_video.FrameMetadata | None = None,
     ) -> None:
         """Capture a frame, optionally attaching frame metadata."""
-        if self._ffi_handle.disposed:
+        if getattr(self._ffi_handle, "disposed", False):
             return
 
         req = proto_ffi.FfiRequest()
@@ -87,4 +87,5 @@ async def aclose(self) -> None:
 
         This method cleans up resources associated with the video source.
         """
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        if not getattr(self._ffi_handle, "disposed", False):
+            await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py
index ca0ef675..07938fe5 100644
--- a/livekit-rtc/livekit/rtc/video_stream.py
+++ b/livekit-rtc/livekit/rtc/video_stream.py
@@ -170,14 +170,8 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
-    def close(self) -> None:
-        """Synchronously close the video stream and release its resources."""
-        self._ffi_handle.dispose()
-        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
-
     async def aclose(self) -> None:
-        """Asynchronously close the video stream without blocking the event loop."""
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        self._ffi_handle.dispose()
         await self._task
 
     def _is_event(self, e: proto_ffi.FfiEvent) -> bool:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index cd84ac1d..3b23dab7 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -14,6 +14,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import threading
 
 import pytest
 
@@ -24,22 +26,22 @@
 async def test_audio_source_aclose_releases_waiter_and_cancels_timer() -> None:
     source = rtc.AudioSource(48000, 1)
     frame = rtc.AudioFrame.create(48000, 1, 48000)
-    await source.capture_frame(frame)
+    await asyncio.wait_for(source.capture_frame(frame), timeout=5.0)
 
     assert source._join_fut is not None
     assert source._join_handle is not None
 
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=5.0)
 
     assert source._ffi_handle.disposed
     assert source._join_fut is None
     assert source._join_handle is None
 
     # After aclose, wait_for_playout should return immediately without hanging
-    await source.wait_for_playout()
+    await asyncio.wait_for(source.wait_for_playout(), timeout=2.0)
 
     # Repeated calls are idempotent
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=2.0)
 
 
 def test_audio_source_sync_close() -> None:
@@ -53,6 +55,17 @@ def test_audio_source_sync_close() -> None:
     source.close()
 
 
+def test_audio_source_cross_thread_close() -> None:
+    source = rtc.AudioSource(48000, 1)
+    assert not source._ffi_handle.disposed
+
+    t = threading.Thread(target=source.close)
+    t.start()
+    t.join()
+
+    assert source._ffi_handle.disposed
+
+
 @pytest.mark.asyncio
 async def test_audio_source_capture_after_close_is_noop() -> None:
     source = rtc.AudioSource(48000, 1)
@@ -60,7 +73,7 @@ async def test_audio_source_capture_after_close_is_noop() -> None:
 
     frame = rtc.AudioFrame.create(48000, 1, 480)
     # Should safely return without error
-    await source.capture_frame(frame)
+    await asyncio.wait_for(source.capture_frame(frame), timeout=2.0)
 
 
 def test_video_source_sync_close() -> None:
@@ -79,7 +92,7 @@ async def test_video_source_aclose_and_capture_after_close() -> None:
     source = rtc.VideoSource(640, 480)
     assert not source._ffi_handle.disposed
 
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=5.0)
     assert source._ffi_handle.disposed
 
     # Capture after close is safely ignored
@@ -92,4 +105,4 @@ async def test_video_source_aclose_and_capture_after_close() -> None:
     source.capture_frame(frame)
 
     # Repeated aclose is idempotent
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=2.0)

From 08e4deb1adc4f2502e536920b75bd2bcb838034b Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 9 Nov 2026 20:15:30 +0500
Subject: [PATCH 3/6] fix(rtc): dynamically bind active event loop in
 AudioSource to avoid capture hang

# Conflicts:
#	livekit-rtc/livekit/rtc/audio_source.py
---
 livekit-rtc/livekit/rtc/audio_source.py       | 48 +++++++++++--------
 livekit-rtc/tests/test_source_stream_close.py | 33 +++++++++++++
 2 files changed, 60 insertions(+), 21 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index 93574a5c..15e35e4c 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -53,17 +53,12 @@ def __init__(
         """
         self._sample_rate = sample_rate
         self._num_channels = num_channels
-
-        if loop is not None:
-            self._loop = loop
-        else:
+        self._loop: asyncio.AbstractEventLoop | None = loop
+        if self._loop is None:
             try:
                 self._loop = asyncio.get_running_loop()
             except RuntimeError:
-                try:
-                    self._loop = asyncio.get_event_loop()
-                except RuntimeError:
-                    self._loop = asyncio.new_event_loop()
+                self._loop = None
 
         req = proto_ffi.FfiRequest()
         req.new_audio_source.type = proto_audio_frame.AudioSourceType.AUDIO_SOURCE_NATIVE
@@ -107,7 +102,12 @@ def clear_queue(self) -> None:
         req = proto_ffi.FfiRequest()
         req.clear_audio_buffer.source_handle = self._ffi_handle.handle
         _ = FfiClient.instance.request(req)
-        self._release_waiter()
+        self._release_waiter_threadsafe()
+
+    def _get_loop(self) -> asyncio.AbstractEventLoop:
+        if self._loop is None or self._loop.is_closed():
+            self._loop = asyncio.get_running_loop()
+        return self._loop
 
     async def capture_frame(self, frame: AudioFrame) -> None:
         """
@@ -129,6 +129,8 @@ async def capture_frame(self, frame: AudioFrame) -> None:
         if frame.samples_per_channel == 0 or getattr(self._ffi_handle, "disposed", False):
             return
 
+        loop = self._get_loop()
+
         now = time.monotonic()
         elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
         self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
@@ -137,10 +139,14 @@ async def capture_frame(self, frame: AudioFrame) -> None:
         if self._join_handle:
             self._join_handle.cancel()
 
-        if self._join_fut is None:
-            self._join_fut = self._loop.create_future()
+        if (
+            self._join_fut is None
+            or self._join_fut.get_loop() is not loop
+            or self._join_fut.get_loop().is_closed()
+        ):
+            self._join_fut = loop.create_future()
 
-        self._join_handle = self._loop.call_later(self._q_size, self._release_waiter)
+        self._join_handle = loop.call_later(self._q_size, self._release_waiter)
 
         req = proto_ffi.FfiRequest()
         req.capture_audio_frame.source_handle = self._ffi_handle.handle
@@ -150,7 +156,7 @@ async def capture_frame(self, frame: AudioFrame) -> None:
         # callback is let through, and wait_for drops the ones of other captures.
         async_id: int | None = None
         queue = FfiClient.instance.queue.subscribe(
-            loop=self._loop,
+            loop=loop,
             filter_fn=lambda e: (
                 e.WhichOneof("message") == "capture_audio_frame"
                 and (async_id is None or e.capture_audio_frame.async_id == async_id)
@@ -197,23 +203,23 @@ def _release_waiter(self) -> None:
         self._q_size = 0.0
         self._join_fut = None
 
-    def close(self) -> None:
-        """Close the audio source.
-
-        This method cleans up resources associated with the audio source.
-        """
+    def _release_waiter_threadsafe(self) -> None:
         try:
             running_loop = asyncio.get_running_loop()
         except RuntimeError:
             running_loop = None
 
-        if running_loop is self._loop:
-            self._release_waiter()
-        elif self._loop.is_running():
+        if self._loop is not None and self._loop.is_running() and running_loop is not self._loop:
             self._loop.call_soon_threadsafe(self._release_waiter)
         else:
             self._release_waiter()
 
+    def close(self) -> None:
+        """Close the audio source.
+
+        This method cleans up resources associated with the audio source.
+        """
+        self._release_waiter_threadsafe()
         self._ffi_handle.dispose()
 
     async def aclose(self) -> None:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index 3b23dab7..ea51b46d 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -106,3 +106,36 @@ async def test_video_source_aclose_and_capture_after_close() -> None:
 
     # Repeated aclose is idempotent
     await asyncio.wait_for(source.aclose(), timeout=2.0)
+
+
+def test_audio_source_init_outside_loop_and_capture_in_asyncio_run() -> None:
+    source = rtc.AudioSource(48000, 1)
+    assert source._loop is None
+    frame = rtc.AudioFrame.create(48000, 1, 480)
+
+    async def _capture() -> None:
+        await asyncio.wait_for(source.capture_frame(frame), timeout=5.0)
+        assert source._loop is not None
+        assert source._loop.is_running()
+        await asyncio.wait_for(source.aclose(), timeout=5.0)
+
+    asyncio.run(_capture())
+
+
+def test_audio_source_cross_thread_clear_queue() -> None:
+    source = rtc.AudioSource(48000, 1)
+    frame = rtc.AudioFrame.create(48000, 1, 48000)
+
+    async def _run() -> None:
+        await source.capture_frame(frame)
+        assert source._join_fut is not None
+
+        t = threading.Thread(target=source.clear_queue)
+        t.start()
+        t.join()
+
+        # Playout waiter should be released by cross-thread clear_queue
+        await asyncio.wait_for(source.wait_for_playout(), timeout=2.0)
+        await source.aclose()
+
+    asyncio.run(_run())

From da44bea2bf74bdbe24b0f6aa600a834fa88074a6 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 19 Oct 2026 02:05:47 +0500
Subject: [PATCH 4/6] fix(rtc): track capture generation in AudioSource to
 isolate clear_queue

---
 livekit-rtc/livekit/rtc/audio_source.py       | 19 ++++++++----
 livekit-rtc/tests/test_source_stream_close.py | 30 +++++++++++++++++++
 2 files changed, 43 insertions(+), 6 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index 15e35e4c..179f9972 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -72,6 +72,7 @@ def __init__(
 
         self._last_capture = 0.0
         self._q_size = 0.0
+        self._capture_gen = 0
         self._join_handle: asyncio.TimerHandle | None = None
         self._join_fut: asyncio.Future[None] | None = None
 
@@ -102,7 +103,7 @@ def clear_queue(self) -> None:
         req = proto_ffi.FfiRequest()
         req.clear_audio_buffer.source_handle = self._ffi_handle.handle
         _ = FfiClient.instance.request(req)
-        self._release_waiter_threadsafe()
+        self._release_waiter_threadsafe(gen=self._capture_gen)
 
     def _get_loop(self) -> asyncio.AbstractEventLoop:
         if self._loop is None or self._loop.is_closed():
@@ -131,6 +132,9 @@ async def capture_frame(self, frame: AudioFrame) -> None:
 
         loop = self._get_loop()
 
+        self._capture_gen += 1
+        current_gen = self._capture_gen
+
         now = time.monotonic()
         elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
         self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
@@ -146,7 +150,7 @@ async def capture_frame(self, frame: AudioFrame) -> None:
         ):
             self._join_fut = loop.create_future()
 
-        self._join_handle = loop.call_later(self._q_size, self._release_waiter)
+        self._join_handle = loop.call_later(self._q_size, self._release_waiter, current_gen)
 
         req = proto_ffi.FfiRequest()
         req.capture_audio_frame.source_handle = self._ffi_handle.handle
@@ -188,7 +192,10 @@ async def wait_for_playout(self) -> None:
 
         await asyncio.shield(self._join_fut)
 
-    def _release_waiter(self) -> None:
+    def _release_waiter(self, gen: int | None = None) -> None:
+        if gen is not None and gen != self._capture_gen:
+            return
+
         if self._join_handle is not None:
             self._join_handle.cancel()
             self._join_handle = None
@@ -203,16 +210,16 @@ def _release_waiter(self) -> None:
         self._q_size = 0.0
         self._join_fut = None
 
-    def _release_waiter_threadsafe(self) -> None:
+    def _release_waiter_threadsafe(self, gen: int | None = None) -> None:
         try:
             running_loop = asyncio.get_running_loop()
         except RuntimeError:
             running_loop = None
 
         if self._loop is not None and self._loop.is_running() and running_loop is not self._loop:
-            self._loop.call_soon_threadsafe(self._release_waiter)
+            self._loop.call_soon_threadsafe(self._release_waiter, gen)
         else:
-            self._release_waiter()
+            self._release_waiter(gen)
 
     def close(self) -> None:
         """Close the audio source.
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index ea51b46d..cdd7277d 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -139,3 +139,33 @@ async def _run() -> None:
         await source.aclose()
 
     asyncio.run(_run())
+
+
+def test_audio_source_clear_queue_does_not_cancel_subsequent_capture() -> None:
+    source = rtc.AudioSource(48000, 1, queue_size_ms=10000)
+    frame1 = rtc.AudioFrame.create(48000, 1, 48000)
+    frame2 = rtc.AudioFrame.create(48000, 1, 48000)
+
+    async def _run() -> None:
+        await source.capture_frame(frame1)
+        gen1 = source._capture_gen
+
+        # Simulate clear_queue scheduled from another thread
+        loop = asyncio.get_running_loop()
+        loop.call_soon(source._release_waiter, gen1)
+
+        # Capture a subsequent frame before yielding
+        await source.capture_frame(frame2)
+        assert source._capture_gen > gen1
+        assert source._join_fut is not None
+
+        # Allow queued clear callback to execute
+        await asyncio.sleep(0.01)
+
+        # The subsequent frame's playout future and timer must not be cleared
+        assert source._join_fut is not None
+        assert not source._join_fut.done()
+
+        await source.aclose()
+
+    asyncio.run(_run())

From 6601c3811e143d91acb2efc43ddfaf7d39558711 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 19 Oct 2026 02:11:22 +0500
Subject: [PATCH 5/6] fix(rtc): synchronize audio source queue metrics with
 thread lock on clear

---
 livekit-rtc/livekit/rtc/audio_source.py       | 70 +++++++++++--------
 livekit-rtc/tests/test_source_stream_close.py | 27 +++++++
 2 files changed, 67 insertions(+), 30 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index 179f9972..376550b7 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -14,8 +14,9 @@
 
 from __future__ import annotations
 
-import time
 import asyncio
+import threading
+import time
 
 from ._ffi_client import FfiHandle, FfiClient
 from ._proto import audio_frame_pb2 as proto_audio_frame
@@ -73,6 +74,7 @@ def __init__(
         self._last_capture = 0.0
         self._q_size = 0.0
         self._capture_gen = 0
+        self._lock = threading.Lock()
         self._join_handle: asyncio.TimerHandle | None = None
         self._join_fut: asyncio.Future[None] | None = None
 
@@ -89,7 +91,8 @@ def num_channels(self) -> int:
     @property
     def queued_duration(self) -> float:
         """The current duration (in seconds) of audio data queued for playback."""
-        return max(self._q_size - time.monotonic() + self._last_capture, 0.0)
+        with self._lock:
+            return max(self._q_size - time.monotonic() + self._last_capture, 0.0)
 
     def clear_queue(self) -> None:
         """
@@ -103,7 +106,11 @@ def clear_queue(self) -> None:
         req = proto_ffi.FfiRequest()
         req.clear_audio_buffer.source_handle = self._ffi_handle.handle
         _ = FfiClient.instance.request(req)
-        self._release_waiter_threadsafe(gen=self._capture_gen)
+        with self._lock:
+            self._last_capture = 0.0
+            self._q_size = 0.0
+            target_gen = self._capture_gen
+        self._release_waiter_threadsafe(gen=target_gen)
 
     def _get_loop(self) -> asyncio.AbstractEventLoop:
         if self._loop is None or self._loop.is_closed():
@@ -132,25 +139,27 @@ async def capture_frame(self, frame: AudioFrame) -> None:
 
         loop = self._get_loop()
 
-        self._capture_gen += 1
-        current_gen = self._capture_gen
+        with self._lock:
+            self._capture_gen += 1
+            current_gen = self._capture_gen
 
-        now = time.monotonic()
-        elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
-        self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
-        self._last_capture = now
+            now = time.monotonic()
+            elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
+            self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
+            self._last_capture = now
+            q_size = self._q_size
 
-        if self._join_handle:
-            self._join_handle.cancel()
+            if self._join_handle:
+                self._join_handle.cancel()
 
-        if (
-            self._join_fut is None
-            or self._join_fut.get_loop() is not loop
-            or self._join_fut.get_loop().is_closed()
-        ):
-            self._join_fut = loop.create_future()
+            if (
+                self._join_fut is None
+                or self._join_fut.get_loop() is not loop
+                or self._join_fut.get_loop().is_closed()
+            ):
+                self._join_fut = loop.create_future()
 
-        self._join_handle = loop.call_later(self._q_size, self._release_waiter, current_gen)
+            self._join_handle = loop.call_later(q_size, self._release_waiter, current_gen)
 
         req = proto_ffi.FfiRequest()
         req.capture_audio_frame.source_handle = self._ffi_handle.handle
@@ -193,22 +202,23 @@ async def wait_for_playout(self) -> None:
         await asyncio.shield(self._join_fut)
 
     def _release_waiter(self, gen: int | None = None) -> None:
-        if gen is not None and gen != self._capture_gen:
-            return
+        with self._lock:
+            if gen is not None and gen != self._capture_gen:
+                return
 
-        if self._join_handle is not None:
-            self._join_handle.cancel()
-            self._join_handle = None
+            if self._join_handle is not None:
+                self._join_handle.cancel()
+                self._join_handle = None
 
-        if self._join_fut is None:
-            return  # could be None when clear_queue is called
+            if self._join_fut is None:
+                return  # could be None when clear_queue is called
 
-        if not self._join_fut.done():
-            self._join_fut.set_result(None)
+            if not self._join_fut.done():
+                self._join_fut.set_result(None)
 
-        self._last_capture = 0.0
-        self._q_size = 0.0
-        self._join_fut = None
+            self._last_capture = 0.0
+            self._q_size = 0.0
+            self._join_fut = None
 
     def _release_waiter_threadsafe(self, gen: int | None = None) -> None:
         try:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index cdd7277d..ab97b5db 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -169,3 +169,30 @@ async def _run() -> None:
         await source.aclose()
 
     asyncio.run(_run())
+
+
+def test_audio_source_cross_thread_clear_queue_resets_queued_duration() -> None:
+    source = rtc.AudioSource(48000, 1, queue_size_ms=10000)
+    frame1 = rtc.AudioFrame.create(48000, 1, 48000)
+    frame2 = rtc.AudioFrame.create(48000, 1, 48000)
+
+    async def _run() -> None:
+        await source.capture_frame(frame1)
+        assert source.queued_duration > 0.8
+
+        t = threading.Thread(target=source.clear_queue)
+        t.start()
+        t.join()
+
+        # Queued duration should be reset immediately
+        assert source.queued_duration == 0.0
+
+        # Capture a new frame immediately after cross-thread clear
+        await source.capture_frame(frame2)
+
+        # Queued duration should reflect only the new frame (~1.0s), not ~2.0s
+        assert 0.8 < source.queued_duration <= 1.05
+
+        await source.aclose()
+
+    asyncio.run(_run())

From dcf6dcb82c68facfd273fda153d64aa58b2eb3e3 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 9 Nov 2026 20:25:19 +0500
Subject: [PATCH 6/6] test(rtc): exit cleanly in ctypes array garbage test to
 avoid SIGABRT on Linux

---
 livekit-rtc/tests/test_ffi_client.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py
index 53873884..bededd77 100644
--- a/livekit-rtc/tests/test_ffi_client.py
+++ b/livekit-rtc/tests/test_ffi_client.py
@@ -22,15 +22,20 @@
 _COUNT_ARRAY_TYPES_IN_GARBAGE = """
 import ctypes
 import gc
+import os
+import sys
 
 from livekit import rtc
 
 gc.collect()
 gc.set_debug(gc.DEBUG_SAVEALL)
-for _ in range(50):
+for _ in range(10):
     rtc.AudioProcessingModule()
 gc.collect()
-print(sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage))
+count = sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage)
+sys.stdout.write(f"{count}\\n")
+sys.stdout.flush()
+os._exit(0)
 """