fix(rtc): avoid AssertionError in FfiHandle.__del__ after global FFI disposal (#857)

Author: NabiBukhsh-AI

livekit-rtc/livekit/rtc/_ffi_client.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
 import signal
 import asyncio
 import sys
@@ -24,6 +26,7 @@
 import platform
 import atexit
 import threading
+import weakref
 from typing import Callable, Generic, List, Optional, TypeVar
 
 from ._proto import ffi_pb2 as proto_ffi
@@ -68,10 +71,14 @@ def get_ffi_lib() -> ctypes.CDLL:
 INVALID_HANDLE = 0
 
 
+_live_handles: weakref.WeakSet[FfiHandle] = weakref.WeakSet()
+
+
 class FfiHandle:
     def __init__(self, handle: int) -> None:
         self.handle = handle
         self._disposed = False
+        _live_handles.add(self)
 
     def __del__(self) -> None:
         self.dispose()
@@ -93,7 +100,7 @@ def dispose(self) -> None:
         if self.handle != INVALID_HANDLE and not self._disposed:
             self._disposed = True
             ffi = FfiClient._instance
-            if ffi is None or ffi._pid != os.getpid():
+            if ffi is None or ffi._pid != os.getpid() or getattr(FfiClient, "_ffi_disposed", False):
                 return
             dropped = ffi._ffi_lib.livekit_ffi_drop_handle(ctypes.c_uint64(self.handle))
             assert dropped
@@ -223,6 +230,7 @@ def to_python_level(level: proto_ffi.LogLevel.ValueType) -> Optional[int]:
 
 class FfiClient:
     _instance: Optional["FfiClient"] = None
+    _ffi_disposed: bool = False
 
     @classproperty
     def instance(cls) -> "FfiClient":
@@ -231,6 +239,7 @@ def instance(cls) -> "FfiClient":
         return cls._instance
 
     def __init__(self) -> None:
+        FfiClient._ffi_disposed = False
         self._pid = os.getpid()
         self._lock = threading.RLock()
         self._queue = FfiQueue[proto_ffi.FfiEvent]()
@@ -266,14 +275,17 @@ def __init__(self) -> None:
             ffi_event_callback, True, b"python", __version__.encode("ascii")
         )
 
-        ffi_lib = self._ffi_lib
-        init_pid = self._pid
-
         @atexit.register
         def _dispose_lk_ffi() -> None:
-            if os.getpid() != init_pid:
-                return
-            ffi_lib.livekit_ffi_dispose()
+            self._dispose()
+
+    def _dispose(self) -> None:
+        if self._pid != os.getpid() or FfiClient._ffi_disposed:
+            return
+        FfiClient._ffi_disposed = True
+        for handle in list(_live_handles):
+            handle.mark_consumed()
+        self._ffi_lib.livekit_ffi_dispose()
 
     @property
     def queue(self) -> FfiQueue[proto_ffi.FfiEvent]:

livekit-rtc/tests/test_ffi_handle_lifecycle.py

@@ -0,0 +1,91 @@
+# 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 gc
+import subprocess
+import sys
+
+import pytest
+
+from livekit.rtc._ffi_client import FfiClient, FfiHandle
+from livekit.rtc._proto import audio_frame_pb2, ffi_pb2
+
+
+def _allocate_real_handle() -> int:
+    req = ffi_pb2.FfiRequest()
+    req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE
+    req.new_audio_source.sample_rate = 48000
+    req.new_audio_source.num_channels = 1
+    req.new_audio_source.queue_size_ms = 100
+    resp = FfiClient.instance.request(req)
+    handle_id = resp.new_audio_source.source.handle.id
+    assert handle_id != 0
+    return int(handle_id)
+
+
+def test_active_runtime_invalid_handle_asserts() -> None:
+    FfiClient.instance
+    with pytest.raises(AssertionError):
+        FfiHandle(999999).dispose()
+
+
+def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None:
+    handle_id = _allocate_real_handle()
+    handle = FfiHandle(handle_id)
+    assert not handle.disposed
+
+    # Global dispose marks all live handles consumed and sets _ffi_disposed
+    FfiClient.instance._dispose()
+    assert handle.disposed
+
+    # Explicit dispose or __del__ after global dispose should be a no-op and not assert
+    handle.dispose()
+    handle.__del__()
+
+    # Newly instantiated handle after global dispose also safely no-ops without assert
+    late_handle = FfiHandle(12345)
+    late_handle.dispose()
+
+    del handle
+    del late_handle
+    gc.collect()
+
+
+def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None:
+    code = """
+import sys
+from livekit.rtc._proto import audio_frame_pb2, ffi_pb2
+from livekit.rtc._ffi_client import FfiClient, FfiHandle
+
+req = ffi_pb2.FfiRequest()
+req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE
+req.new_audio_source.sample_rate = 48000
+req.new_audio_source.num_channels = 1
+req.new_audio_source.queue_size_ms = 100
+resp = FfiClient.instance.request(req)
+handle_id = resp.new_audio_source.source.handle.id
+# Leave handle alive at process shutdown
+global_handle = FfiHandle(handle_id)
+sys.exit(0)
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 0
+    assert "AssertionError" not in result.stderr
+    assert "Exception ignored" not in result.stderr