perf(rtc-node): route audio stream events and capture callbacks by handle

.changeset/route-ffi-events-by-handle.md

@@ -0,0 +1,5 @@
+---
+'@livekit/rtc-node': patch
+---
+
+Audio stream events and capture callbacks go straight to the stream or waiter they belong to instead of to every FFI listener, and rooms skip events that are neither room events nor RPC invocations before taking their event lock. Dispatch cost no longer grows with the number of audio streams and rooms in the process.

packages/livekit-rtc/src/audio_source.ts

@@ -159,9 +159,9 @@ export class AudioSource {
       message: { case: 'captureAudioFrame', value: req },
     });
 
-    const cb = await FfiClient.instance.waitFor<CaptureAudioFrameCallback>((ev) => {
-      return ev.message.case == 'captureAudioFrame' && ev.message.value.asyncId == res.asyncId;
-    });
+    const cb: CaptureAudioFrameCallback = await FfiClient.instance.waitForCaptureAudioFrame(
+      res.asyncId!,
+    );
 
     if (cb.error) {
       throw new Error(cb.error);

packages/livekit-rtc/src/audio_stream.ts

@@ -6,7 +6,7 @@ import { AudioStreamType, NewAudioStreamRequest } from '@livekit/rtc-ffi-binding
 import type { UnderlyingSource } from 'node:stream/web';
 import { AudioFrame } from './audio_frame.js';
 import type { FfiEvent } from './ffi_client.js';
-import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
+import { FfiClient, FfiHandle } from './ffi_client.js';
 import { type FrameProcessor, isFrameProcessor } from './frame_processor.js';
 import { log } from './log.js';
 import type { Track } from './track.js';
@@ -35,6 +35,9 @@ export interface NoiseCancellationOptions {
 export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
   private controller?: ReadableStreamDefaultController<AudioFrame>;
   private ffiHandle: FfiHandle;
+  // The plain value events carry, which routes them here; the native `handle` getter costs a call
+  // across into native code on every read.
+  private ffiHandleId: bigint;
   private disposed = false;
   private sampleRate: number;
   private numChannels: number;
@@ -86,9 +89,10 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
       },
     });
 
-    this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
+    this.ffiHandleId = res.stream!.handle!.id!;
+    this.ffiHandle = new FfiHandle(this.ffiHandleId);
 
-    FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
+    FfiClient.instance.onAudioStreamEvent(this.ffiHandleId, this.onEvent);
     track.registerAudioStream(this);
   }
 
@@ -104,7 +108,7 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
 
     if (
       ev.message.case != 'audioStreamEvent' ||
-      ev.message.value.streamHandle != this.ffiHandle.handle
+      ev.message.value.streamHandle != this.ffiHandleId
     ) {
       return;
     }
@@ -148,7 +152,7 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
    * @internal
    */
   teardown() {
-    FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+    FfiClient.instance.offAudioStreamEvent(this.ffiHandleId);
     if (this.disposed) {
       return;
     }

packages/livekit-rtc/src/ffi_client.test.ts

@@ -0,0 +1,100 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { FfiEvent } from '@livekit/rtc-ffi-bindings';
+import type { PartialMessage } from '@livekit/rtc-ffi-bindings';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import { FfiClient, FfiClientEvent } from './ffi_client.js';
+
+const native = vi.hoisted(() => ({
+  onEvent: undefined as ((data: Uint8Array) => void) | undefined,
+}));
+
+vi.mock('@livekit/rtc-ffi-bindings', async () => {
+  const actual = await vi.importActual<typeof import('@livekit/rtc-ffi-bindings')>(
+    '@livekit/rtc-ffi-bindings',
+  );
+  return {
+    ...actual,
+    livekitInitialize: (onEvent: (data: Uint8Array) => void) => {
+      native.onEvent = onEvent;
+    },
+  };
+});
+
+function deliver(event: PartialMessage<FfiEvent>) {
+  native.onEvent!(new FfiEvent(event).toBinary());
+}
+
+function audioStreamEos(streamHandle: bigint): PartialMessage<FfiEvent> {
+  return {
+    message: {
+      case: 'audioStreamEvent',
+      value: { streamHandle, message: { case: 'eos', value: {} } },
+    },
+  };
+}
+
+function captureAudioFrameCallback(asyncId: bigint): PartialMessage<FfiEvent> {
+  return { message: { case: 'captureAudioFrame', value: { asyncId } } };
+}
+
+describe('FfiClient event routing', () => {
+  let client: FfiClient;
+  let emitted: FfiEvent[];
+
+  beforeEach(() => {
+    globalThis._ffiClientInstance = undefined;
+    client = FfiClient.instance;
+    emitted = [];
+    client.on(FfiClientEvent.FfiEvent, (ev) => emitted.push(ev));
+  });
+
+  it('hands an audio stream event to its own stream only, without emitting it', () => {
+    const first = vi.fn();
+    const second = vi.fn();
+    client.onAudioStreamEvent(1n, first);
+    client.onAudioStreamEvent(2n, second);
+
+    deliver(audioStreamEos(2n));
+
+    expect(first).not.toHaveBeenCalled();
+    expect(second).toHaveBeenCalledOnce();
+    expect(second.mock.calls[0]![0].message.value.streamHandle).toBe(2n);
+    expect(emitted).toEqual([]);
+  });
+
+  it('drops an audio stream event once its stream is removed', () => {
+    const listener = vi.fn();
+    client.onAudioStreamEvent(1n, listener);
+    client.offAudioStreamEvent(1n);
+
+    deliver(audioStreamEos(1n));
+
+    expect(listener).not.toHaveBeenCalled();
+    expect(emitted).toEqual([]);
+  });
+
+  it('resolves the waiter for a capture callback, without emitting it', async () => {
+    const callback = client.waitForCaptureAudioFrame(7n);
+
+    deliver(captureAudioFrameCallback(7n));
+
+    await expect(callback).resolves.toMatchObject({ asyncId: 7n });
+    expect(emitted).toEqual([]);
+  });
+
+  it('still emits a capture callback that nothing waits for', () => {
+    deliver(captureAudioFrameCallback(8n));
+
+    expect(emitted).toHaveLength(1);
+    expect(emitted[0]!.message.case).toBe('captureAudioFrame');
+  });
+
+  it('emits every other event as before', () => {
+    deliver({ message: { case: 'disconnect', value: { asyncId: 9n } } });
+
+    expect(emitted).toHaveLength(1);
+    expect(emitted[0]!.message.case).toBe('disconnect');
+  });
+});

packages/livekit-rtc/src/ffi_client.ts

@@ -2,6 +2,7 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import {
+  type CaptureAudioFrameCallback,
   FfiEvent,
   FfiHandle,
   FfiRequest,
@@ -43,20 +44,61 @@ export class FfiClient extends (EventEmitter as new () => TypedEmitter<FfiClient
 
   private _nextRequestAsyncId = BigInt(0);
 
+  // Audio frames are most of the events and each has exactly one reader, its stream; emitting them
+  // would make every stream and room in the process check every other call's frames.
+  private audioStreamListeners = new Map<bigint, (event: FfiEvent) => void>();
+
+  // Every audio frame an agent speaks gets its own callback. Waiting on the emitter would add and
+  // remove a listener per frame and make every other waiter test it.
+  private captureAudioFrameWaiters = new Map<
+    bigint,
+    (callback: CaptureAudioFrameCallback) => void
+  >();
+
   constructor() {
     super();
     this.setMaxListeners(0);
 
     livekitInitialize(
       (event_data: Uint8Array) => {
         const event = FfiEvent.fromBinary(event_data);
+        if (event.message.case === 'audioStreamEvent') {
+          this.audioStreamListeners.get(event.message.value.streamHandle!)?.(event);
+          return;
+        }
+        if (event.message.case === 'captureAudioFrame') {
+          const asyncId = event.message.value.asyncId!;
+          const waiter = this.captureAudioFrameWaiters.get(asyncId);
+          if (waiter) {
+            this.captureAudioFrameWaiters.delete(asyncId);
+            waiter(event.message.value);
+            return;
+          }
+        }
         this.emit(FfiClientEvent.FfiEvent, event);
       },
       true,
       SDK_VERSION,
     );
   }
 
+  /** @internal */
+  onAudioStreamEvent(streamHandle: bigint, listener: (event: FfiEvent) => void) {
+    this.audioStreamListeners.set(streamHandle, listener);
+  }
+
+  /** @internal */
+  offAudioStreamEvent(streamHandle: bigint) {
+    this.audioStreamListeners.delete(streamHandle);
+  }
+
+  /** @internal Call in the same tick as the request, so its callback cannot arrive first. */
+  waitForCaptureAudioFrame(asyncId: bigint): Promise<CaptureAudioFrameCallback> {
+    return new Promise((resolve) => {
+      this.captureAudioFrameWaiters.set(asyncId, resolve);
+    });
+  }
+
   request<T>(req: PartialMessage<FfiRequest>): T {
     const request = new FfiRequest(req);
     const req_data = request.toBinary();

packages/livekit-rtc/src/room.ts

@@ -549,6 +549,11 @@ export class Room extends (EventEmitter as new () => TypedEmitter<RoomCallbacks>
   }
 
   private onFfiEvent = async (ffiEvent: FfiEvent) => {
+    // Every room in the process sees every FFI event, most of them other streams' audio frames.
+    // processFfiEvent acts on these two cases only, so the rest return before waiting on the lock.
+    if (ffiEvent.message.case != 'roomEvent' && ffiEvent.message.case != 'rpcMethodInvocation') {
+      return;
+    }
     const unlock = await this.ffiEventLock.lock();
     try {
       if (!this.localParticipant || !this.ffiHandle || !this.info) {