From 358e7a7660e57954bedd54d4a1cad7f7be1df27e Mon Sep 17 00:00:00 2001 From: avionicharshit-byte Date: Wed, 2 Sep 2026 21:22:42 +0530 Subject: [PATCH] fix(rtc): emit events to handlers in registration order --- livekit-rtc/livekit/rtc/event_emitter.py | 15 ++-- tests/rtc/test_emitter.py | 103 +++++++++++++++++++++++ 2 files changed, 112 insertions(+), 6 deletions(-) diff --git a/livekit-rtc/livekit/rtc/event_emitter.py b/livekit-rtc/livekit/rtc/event_emitter.py index bc64d1d7..0d8390dd 100644 --- a/livekit-rtc/livekit/rtc/event_emitter.py +++ b/livekit-rtc/livekit/rtc/event_emitter.py @@ -1,6 +1,6 @@ import inspect import asyncio -from typing import Any, Callable, Dict, Set, Optional, Generic, TypeVar +from typing import Any, Callable, Dict, Optional, Generic, TypeVar from .log import logger @@ -12,7 +12,10 @@ def __init__(self) -> None: """ Initialize a new instance of EventEmitter. """ - self._events: Dict[T_contra, Set[Callable]] = dict() + # A dict keyed by callback is an insertion-ordered set: handlers run in the + # order they were registered, so one that mutates the event still runs before + # a peer that reads it. + self._events: Dict[T_contra, Dict[Callable, None]] = dict() def emit(self, event: T_contra, *args: Any) -> None: """ @@ -36,7 +39,7 @@ def greet(name): ``` """ if event in self._events: - callables = self._events[event].copy() + callables = list(self._events[event]) for callback in callables: try: sig = inspect.signature(callback) @@ -163,8 +166,8 @@ def greet(name): ) if event not in self._events: - self._events[event] = set() - self._events[event].add(callback) + self._events[event] = {} + self._events[event][callback] = None return callback else: @@ -197,4 +200,4 @@ def greet(name): ``` """ if event in self._events: - self._events[event].discard(callback) + self._events[event].pop(callback, None) diff --git a/tests/rtc/test_emitter.py b/tests/rtc/test_emitter.py index b5a1bf59..8a5df79d 100644 --- a/tests/rtc/test_emitter.py +++ b/tests/rtc/test_emitter.py @@ -102,3 +102,106 @@ def on_error_another() -> None: emitter.emit("error") assert len(calls) == 2 + + +class _OrderedHandler: + """A callable whose hash is fixed, so a set orders it independently of registration.""" + + def __init__(self, name: str, hash_value: int, sink: list[str]) -> None: + self._name = name + self._hash = hash_value + self._sink = sink + + def __hash__(self) -> int: + return self._hash + + def __eq__(self, other: object) -> bool: + return self is other + + def __call__(self) -> None: + self._sink.append(self._name) + + +def test_handlers_run_in_registration_order() -> None: + # Handlers were kept in a set, so dispatch order was hash-derived. The hashes here are + # picked so a set yields them in the opposite order to the one they were added in. + emitter = EventEmitter[str]() + order: list[str] = [] + + emitter.on("event", _OrderedHandler("first", 5, order)) + emitter.on("event", _OrderedHandler("second", 1, order)) + + emitter.emit("event") + assert order == ["first", "second"] + + +def test_a_mutating_handler_runs_before_a_peer_that_reads_it() -> None: + # The livekit-agents case: one handler stamps a field onto the emitted object and a + # user handler registered later reads it. Registration order has to decide. + class Event: + def __init__(self) -> None: + self.speech_id: Any = None + + class Stamp: + def __hash__(self) -> int: + return 5 + + def __eq__(self, other: object) -> bool: + return self is other + + def __call__(self, ev: Event) -> None: + ev.speech_id = "speech_1" + + class Read: + def __init__(self, sink: list[Any]) -> None: + self._sink = sink + + def __hash__(self) -> int: + return 1 + + def __eq__(self, other: object) -> bool: + return self is other + + def __call__(self, ev: Event) -> None: + self._sink.append(ev.speech_id) + + emitter = EventEmitter[str]() + seen: list[Any] = [] + emitter.on("metrics", Stamp()) + emitter.on("metrics", Read(seen)) + + for _ in range(5): + emitter.emit("metrics", Event()) + + assert seen == ["speech_1"] * 5 + + +def test_off_still_removes_a_handler() -> None: + emitter = EventEmitter[str]() + calls: list[str] = [] + + @emitter.on("event") + def keep() -> None: + calls.append("keep") + + @emitter.on("event") + def drop() -> None: + calls.append("drop") + + emitter.off("event", drop) + emitter.off("event", drop) # removing twice must not raise + emitter.emit("event") + assert calls == ["keep"] + + +def test_registering_the_same_handler_twice_keeps_one_entry() -> None: + emitter = EventEmitter[str]() + calls: list[str] = [] + + def handler() -> None: + calls.append("x") + + emitter.on("event", handler) + emitter.on("event", handler) + emitter.emit("event") + assert calls == ["x"]