fix(rtc): reject async callbacks in EventEmitter.once

livekit-rtc/livekit/rtc/event_emitter.py

@@ -13,6 +13,17 @@
 # function in 3.16, and a bare attribute access would then fail at import time.
 _ASYNCIO_COROUTINE_MARKER = getattr(asyncio.coroutines, "_is_coroutine", None)
 
+
+def _check_async_callback(callback: Callable, method_name: str) -> None:
+    if inspect.iscoroutinefunction(callback) or (
+        _ASYNCIO_COROUTINE_MARKER is not None
+        and getattr(callback, "_is_coroutine", None) is _ASYNCIO_COROUTINE_MARKER
+    ):
+        raise ValueError(
+            f"Cannot register an async callback with `.{method_name}()`. Use `asyncio.create_task` within your synchronous callback instead."
+        )
+
+
 T_contra = TypeVar("T_contra", contravariant=True)
 
 
@@ -115,6 +126,7 @@ def greet_once(name):
             ```
         """
         if callback is not None:
+            _check_async_callback(callback, "once")
 
             def once_callback(*args: Any, **kwargs: Any) -> None:
                 self.off(event, once_callback)
@@ -169,13 +181,7 @@ def greet(name):
             ```
         """
         if callback is not None:
-            if inspect.iscoroutinefunction(callback) or (
-                _ASYNCIO_COROUTINE_MARKER is not None
-                and getattr(callback, "_is_coroutine", None) is _ASYNCIO_COROUTINE_MARKER
-            ):
-                raise ValueError(
-                    "Cannot register an async callback with `.on()`. Use `asyncio.create_task` within your synchronous callback instead."
-                )
+            _check_async_callback(callback, "on")
 
             if event not in self._events:
                 self._events[event] = {}

tests/rtc/test_emitter.py

@@ -186,6 +186,57 @@ def test_on_rejects_async_mock() -> None:
         emitter.on("connected", AsyncMock())
 
 
+def test_once_rejects_async_callback() -> None:
+    """`.once()` refuses coroutine functions, however they are spelled."""
+    EventTypes = Literal["connected"]
+
+    emitter = EventEmitter[EventTypes]()
+
+    async def on_connected() -> None:
+        pass
+
+    with pytest.raises(ValueError, match="Cannot register an async callback"):
+        emitter.once("connected", on_connected)
+
+    with pytest.raises(ValueError, match="Cannot register an async callback"):
+        emitter.once("connected", functools.partial(on_connected))
+
+    with pytest.raises(ValueError, match="Cannot register an async callback"):
+
+        @emitter.once("connected")
+        async def on_connected_decorated() -> None:
+            pass
+
+
+def test_once_rejects_sentinel_tagged_callback() -> None:
+    """A callable tagged with asyncio's private coroutine sentinel is rejected by `.once()`."""
+    marker = getattr(asyncio.coroutines, "_is_coroutine", None)
+    if marker is None:
+        pytest.skip("asyncio.coroutines._is_coroutine is not defined on this Python")
+
+    EventTypes = Literal["connected"]
+
+    emitter = EventEmitter[EventTypes]()
+
+    def on_connected() -> None:
+        pass
+
+    on_connected._is_coroutine = marker  # type: ignore[attr-defined]
+
+    with pytest.raises(ValueError, match="Cannot register an async callback"):
+        emitter.once("connected", on_connected)
+
+
+def test_once_rejects_async_mock() -> None:
+    """`AsyncMock` is rejected by `.once()` on every supported version."""
+    EventTypes = Literal["connected"]
+
+    emitter = EventEmitter[EventTypes]()
+
+    with pytest.raises(ValueError, match="Cannot register an async callback"):
+        emitter.once("connected", AsyncMock())
+
+
 class _OrderedHandler:
     """A callable whose hash is fixed, so a set orders it independently of registration."""