From da00d20b448a1da16d5c608469503451a1fa5abf Mon Sep 17 00:00:00 2001 From: apoorva-01 Date: Fri, 3 Jul 2026 11:07:37 +0530 Subject: [PATCH] Start AudioStream _run task after _ffi_handle is set If owned-stream creation raises, __init__ never assigns _ffi_handle, but the already-scheduled _run task still dereferences it and leaks an uncatchable AttributeError on the livekit logger. Creating the task last leaves no orphan on that path. --- livekit-rtc/livekit/rtc/audio_stream.py | 8 ++-- livekit-rtc/tests/test_audio_stream_init.py | 53 +++++++++++++++++++++ 2 files changed, 58 insertions(+), 3 deletions(-) create mode 100644 livekit-rtc/tests/test_audio_stream_init.py diff --git a/livekit-rtc/livekit/rtc/audio_stream.py b/livekit-rtc/livekit/rtc/audio_stream.py index 237b0d7f..97772f4f 100644 --- a/livekit-rtc/livekit/rtc/audio_stream.py +++ b/livekit-rtc/livekit/rtc/audio_stream.py @@ -125,9 +125,6 @@ def __init__( self._processor = noise_cancellation self._processor_auto_close = auto_close_noise_cancellation - self._task = self._loop.create_task(self._run()) - self._task.add_done_callback(task_done_logger) - stream: Any = None if "participant" in kwargs: stream = self._create_owned_stream_from_participant( @@ -138,6 +135,11 @@ def __init__( self._ffi_handle = FfiHandle(stream.handle.id) self._info = stream.info + # Start _run only after _ffi_handle is set, so a failure above doesn't orphan + # a task that dereferences an unassigned _ffi_handle. + self._task = self._loop.create_task(self._run()) + self._task.add_done_callback(task_done_logger) + if self._track is not None: self._track._register_audio_stream(self) diff --git a/livekit-rtc/tests/test_audio_stream_init.py b/livekit-rtc/tests/test_audio_stream_init.py new file mode 100644 index 00000000..80da28ab --- /dev/null +++ b/livekit-rtc/tests/test_audio_stream_init.py @@ -0,0 +1,53 @@ +# 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. + +"""Regression tests for AudioStream.__init__ error handling.""" + +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock, patch + +import pytest + +from livekit import rtc + + +async def test_failed_stream_creation_does_not_orphan_run_task() -> None: + """If owned-stream creation fails, __init__ must not leave a running _run task. + + _run dereferences self._ffi_handle, which is only assigned once stream creation + succeeds. Scheduling the task before that assignment leaves an orphaned task that + raises AttributeError from the event loop, uncatchable by the caller. + """ + track = MagicMock(spec=rtc.Track) + before = asyncio.all_tasks() + + with patch.object( + rtc.AudioStream, + "_create_owned_stream", + side_effect=RuntimeError("track already closed"), + ): + with pytest.raises(RuntimeError, match="track already closed"): + rtc.AudioStream(track) + + orphaned = [ + t + for t in asyncio.all_tasks() - before + if getattr(t.get_coro(), "__qualname__", "") == "AudioStream._run" + ] + for t in orphaned: + t.cancel() + + assert not orphaned