diff --git a/livekit-rtc/livekit/rtc/event_emitter.py b/livekit-rtc/livekit/rtc/event_emitter.py index 91161a7d..064e3f76 100644 --- a/livekit-rtc/livekit/rtc/event_emitter.py +++ b/livekit-rtc/livekit/rtc/event_emitter.py @@ -1,6 +1,6 @@ import asyncio import inspect -from typing import Any, Callable, Dict, Set, Optional, Generic, TypeVar +from typing import Any, Callable, Dict, Optional, Generic, TypeVar from .log import logger @@ -21,7 +21,13 @@ def __init__(self) -> None: """ Initialize a new instance of EventEmitter. """ - self._events: Dict[T_contra, Set[Callable]] = dict() + # A dict is used as an ordered set: handlers are dispatched in the order they + # were registered. A set would order them by hash, which is stable for the life + # of a process but arbitrary between runs, so a handler that reads a field a + # peer handler mutates during the same emit would see a different value from + # one process to the next. Insertion order is also what the JS and Rust SDKs + # give, and what "on" reads like it promises. + self._events: Dict[T_contra, Dict[Callable, None]] = dict() def emit(self, event: T_contra, *args: Any) -> None: """ @@ -45,7 +51,9 @@ def greet(name): ``` """ if event in self._events: - callables = self._events[event].copy() + # Copied, so a handler may call `off` (or `once` can remove itself) without + # mutating the collection being iterated. + callables = list(self._events[event]) for callback in callables: try: sig = inspect.signature(callback) @@ -175,8 +183,10 @@ def greet(name): ) if event not in self._events: - self._events[event] = set() - self._events[event].add(callback) + self._events[event] = {} + # Re-registering an existing callback keeps its original position, which is + # what `set.add` did, so registering twice still dispatches once. + self._events[event].setdefault(callback, None) return callback else: @@ -209,4 +219,4 @@ def greet(name): ``` """ if event in self._events: - self._events[event].discard(callback) + self._events[event].pop(callback, None) diff --git a/livekit-rtc/tests/test_event_emitter_order.py b/livekit-rtc/tests/test_event_emitter_order.py new file mode 100644 index 00000000..80c907e6 --- /dev/null +++ b/livekit-rtc/tests/test_event_emitter_order.py @@ -0,0 +1,145 @@ +"""Dispatch order for `EventEmitter`. + +Handlers used to live in a `set`, so `emit` walked them in hash order. That is +stable for the life of a process and arbitrary between runs, which is the worst +shape a bug can have: a handler that reads a field a peer handler writes during +the same emit is correct on the machine it was written on and wrong somewhere +else, all of the time rather than some of the time. + +The ordering tests below pin hashes deliberately. Registering three handlers and +hoping a set shuffles them would pass on the broken implementation roughly one +run in six, and a test that only usually fails is not a regression test. +""" + +from __future__ import annotations + +from livekit.rtc.event_emitter import EventEmitter + + +class Handler: + """A callable whose hash is chosen, so set order is known rather than lucky. + + CPython places small integer hashes by value, so handlers registered with + descending hashes come back out of a `set` in ascending order, which is the + exact reverse of how they went in. + """ + + def __init__(self, name: str, hash_value: int, seen: list, action=None) -> None: + self.name = name + self._hash = hash_value + self._seen = seen + self._action = action + + def __call__(self) -> None: + self._seen.append(self.name) + if self._action is not None: + self._action() + + def __hash__(self) -> int: + return self._hash + + def __eq__(self, other: object) -> bool: + return self is other + + +def test_handlers_run_in_the_order_they_were_registered() -> None: + seen: list = [] + emitter = EventEmitter[str]() + for name, h in (("first", 3), ("second", 2), ("third", 1)): + emitter.on("event", Handler(name, h, seen)) + + emitter.emit("event") + + assert seen == ["first", "second", "third"], seen + + +def test_a_handler_observes_what_an_earlier_handler_wrote() -> None: + """The reported failure, reduced. + + livekit-agents stamps `speech_id` onto a metrics object by mutating it in one + handler and expects later subscribers to see the stamped value. Under hash + order the reporter measured the pre-stamp state in 7 of 12 fresh processes. + """ + record: dict = {"speech_id": None} + observed: list = [] + + # hashes chosen so a set would dispatch the reader first, before the stamp + emitter = EventEmitter[str]() + emitter.on( + "metrics", + Handler("stamp", 2, [], lambda: record.__setitem__("speech_id", "abc123")), + ) + emitter.on( + "metrics", + Handler("read", 1, [], lambda: observed.append(record["speech_id"])), + ) + emitter.emit("metrics") + + assert observed == ["abc123"], observed + + +def test_registering_the_same_handler_twice_dispatches_it_once() -> None: + """`set.add` was idempotent and callers rely on that.""" + seen: list = [] + emitter = EventEmitter[str]() + handler = Handler("only", 1, seen) + + emitter.on("event", handler) + emitter.on("event", handler) + emitter.emit("event") + + assert seen == ["only"], seen + + +def test_re_registering_does_not_move_a_handler_to_the_back() -> None: + """Otherwise `on` would quietly double as a reordering operation, and a + caller re-subscribing an existing handler would change what its peers see.""" + seen: list = [] + emitter = EventEmitter[str]() + first, second = Handler("first", 3, seen), Handler("second", 2, seen) + emitter.on("event", first) + emitter.on("event", second) + + emitter.on("event", first) + emitter.emit("event") + + assert seen == ["first", "second"], seen + + +def test_unsubscribing_during_dispatch_takes_effect_from_the_next_emit() -> None: + """`emit` walks a copy, so the emit that removed a handler still runs it. + + That is deliberate and `once` depends on it, but it is worth pinning: moving + to an ordered container would be an easy place to start iterating the live + collection instead, which turns an `off` inside a handler into a mutation + during iteration. + """ + seen: list = [] + emitter = EventEmitter[str]() + + def first() -> None: + seen.append("first") + emitter.off("event", second) + + def second() -> None: + seen.append("second") + + emitter.on("event", first) + emitter.on("event", second) + emitter.emit("event") + emitter.emit("event") + + assert seen == ["first", "second", "first"], seen + + +def test_once_fires_once_and_leaves_the_order_alone() -> None: + seen: list = [] + emitter = EventEmitter[str]() + emitter.on("event", Handler("before", 3, seen)) + emitter.once("event", Handler("once", 2, seen)) + emitter.on("event", Handler("after", 1, seen)) + + emitter.emit("event") + emitter.emit("event") + + assert seen == ["before", "once", "after", "before", "after"], seen