fix(rtc): offload FFI handle disposal to thread in aclose and add sync close

livekit-rtc/livekit/rtc/audio_source.py

@@ -163,6 +163,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
 
@@ -173,9 +177,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)

livekit-rtc/livekit/rtc/audio_stream.py

@@ -315,15 +315,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()

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)

livekit-rtc/livekit/rtc/video_stream.py

@@ -164,8 +164,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:

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()