Room.connect FFI subscription is unfiltered; audio_stream_event fans out O(rooms × events) and starves the asyncio loop

[wizd]

Summary

PR #564 added filter_fn to FfiQueue.subscribe() and applied it to AudioStream / VideoStream. Room.connect() still does an unfiltered long-lived subscribe:

self._ffi_queue = FfiClient.instance.queue.subscribe(self._loop)

Room._listen_task only consumes room_event and rpc_method_invocation, then always does put_nowait + await self._room_queue.join(). Every audio_stream_event is therefore delivered to every connected Room and costs a full event-loop round-trip each time.

Production impact

We run many 1:1 voice rooms in one Python process (device WebSocket gateway → rtc.Room per session). With default 10 ms audio frames:

• 8 rooms: ~7.5k listen-task wakeups/s — barely holds
• 16 rooms: ~30k wakeups/s — event loop saturates

Symptoms match #564 and #390: room.connect() futures convoy-resolve after 10–60 s, WebSocket keepalive pings time out, CPU is idle (GIL / single loop), not compute-bound.

AudioSource.capture_frame() is a short-lived subscribe waiting on capture_audio_frame (the request callback), so it should not be filtered to audio_source_event.

Suggested fix

In Room.connect(), subscribe with a filter matching what _listen_task actually reads:

self._ffi_queue = FfiClient.instance.queue.subscribe(
    self._loop,
    filter_fn=lambda e: e.WhichOneof(message) in (room_event, rpc_method_invocation),
)

This is the same pattern as #564, just completing it for Room. Happy to send a PR if that helps.

Environment

• livekit==1.1.8 / livekit-api==1.1.0
• one process, many rtc.Room instances
• no frame_size_ms on AudioStream (defaults to 10 ms)

[wizd]

Correction

The original suggestion — filter Room.connect()'s long-lived FfiQueue.subscribe(self._loop) to {room_event, rpc_method_invocation} — is wrong and will deadlock publish_track.

Room._listen_task does not only consume those two types. After handling rpc_method_invocation / room_event, it unconditionally forwards every delivered FfiEvent into Room._room_queue:

self._room_queue.put_nowait(event)
await self._room_queue.join()

LocalParticipant.publish_track and unpublish_track wait on that queue for their request callbacks (e.publish_track.async_id / e.unpublish_track.async_id). A whitelist that drops publish_track means await room.local_participant.publish_track(...) never returns.

We hit this in production: local_track_published arrived (Rust side succeeded) while Python was still blocked in publish_track, then room.py did:

lpublication = self.local_participant.track_publications[sid]

and raised KeyError because publish_track had not yet written _track_publications[sid]. Same pattern at room.py ~727 / ~732 / ~751.

What is actually safe to drop

audio_stream_event / video_stream_event only.

• AudioStream / VideoStream already subscribe independently with filter_fn (PR feat(rtc): Add event_types filtering to FfiQueue.subscribe() to reduce memory allocations #564).
• _listen_task does not handle those types (event.room_event.room_handle is 0, the elif does not match). Relaying them is a no-op plus put_nowait + join() — that is the O(rooms × frames) cost.

Do not filter AudioSource.capture_frame(): that subscribe is short-lived and waits on capture_audio_frame.

Suggested Room filter

self._ffi_queue = FfiClient.instance.queue.subscribe(
    self._loop,
    filter_fn=lambda e: e.WhichOneof(message) not in (
        audio_stream_event,
        video_stream_event,
    ),
)

Separate robustness bug

room.py _on_room_event for local_track_published / local_track_unpublished should use .get(sid) and skip if missing. local_track_published can race publish_track's dict write even without a bad filter.