fix(rtc): emit events to handlers in registration order

Author: avionicharshit-byte

.github/workflows/build-protocol.yml

@@ -31,6 +31,8 @@ jobs:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           submodules: true
+          # A fork PR's head branch only exists in the contributor's repo.
+          repository: ${{ github.event.pull_request.head.repo.full_name }}
           ref: ${{ github.event.pull_request.head.ref }}
 
       - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
@@ -50,6 +52,8 @@ jobs:
         run: ./generate_proto.sh
 
       - name: Add changes
+        # A fork PR's GITHUB_TOKEN is read-only; the verify step below covers it.
+        if: github.event.pull_request.head.repo.full_name == github.repository
         uses: EndBug/add-and-commit@cc9c08ba6c8df3b93a8f2db63e89b98368ae2ae8 # v11
         with:
           add: '["livekit-protocol/"]'
@@ -60,6 +64,18 @@ jobs:
           # generated stubs onto the PR head branch doesn't need a full fetch.
           fetch: false
 
+      - name: Verify generated stubs are current
+        if: github.event.pull_request.head.repo.full_name != github.repository
+        # `status`, not `diff`: a newly generated stub is untracked.
+        run: |
+          drift="$(git status --porcelain --ignore-submodules=all -- .)"
+          if [ -n "$drift" ]; then
+            echo "Generated stubs are out of date. Run ./generate_proto.sh and commit:"
+            echo "$drift"
+            git diff --ignore-submodules=all -- .
+            exit 1
+          fi
+
   build_wheels:
     name: Build Protocol wheel/sdist
     runs-on: ubuntu-latest

.github/workflows/build-rtc.yml

@@ -31,6 +31,8 @@ jobs:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           submodules: true
+          # A fork PR's head branch only exists in the contributor's repo.
+          repository: ${{ github.event.pull_request.head.repo.full_name }}
           ref: ${{ github.event.pull_request.head.ref }}
 
       - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
@@ -49,6 +51,8 @@ jobs:
         run: ./generate_proto.sh
 
       - name: Add changes
+        # A fork PR's GITHUB_TOKEN is read-only; the verify step below covers it.
+        if: github.event.pull_request.head.repo.full_name == github.repository
         uses: EndBug/add-and-commit@cc9c08ba6c8df3b93a8f2db63e89b98368ae2ae8 # v11
         with:
           add: '["livekit-rtc/"]'
@@ -59,6 +63,18 @@ jobs:
           # generated stubs onto the PR head branch doesn't need a full fetch.
           fetch: false
 
+      - name: Verify generated stubs are current
+        if: github.event.pull_request.head.repo.full_name != github.repository
+        # `status`, not `diff`: a newly generated stub is untracked.
+        run: |
+          drift="$(git status --porcelain --ignore-submodules=all -- .)"
+          if [ -n "$drift" ]; then
+            echo "Generated stubs are out of date. Run ./generate_proto.sh and commit:"
+            echo "$drift"
+            git diff --ignore-submodules=all -- .
+            exit 1
+          fi
+
   build_wheels:
     name: Build RTC wheels (${{ matrix.archs }})
     runs-on: ${{ matrix.os }}

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)

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"]