close audio streams after unsubscribe, bump FFI to 0.12.68

.changeset/lucky-pandas-listen.md

@@ -0,0 +1,5 @@
+---
+'@livekit/rtc-node': patch
+---
+
+Close audio streams when their track is unsubscribed. An unsubscribed track never receives `eos` from the FFI, so any `AudioStream` attached to it kept delivering frames — after a reconnect that meant the stale stream and the new subscription's stream both delivered the publisher's audio.

.changeset/sixty-trains-win.md

@@ -0,0 +1,5 @@
+---
+"@livekit/rtc-node": patch
+---
+
+bump FFI to 0.12.68

packages/livekit-rtc/package.json

@@ -32,7 +32,7 @@
   "dependencies": {
     "@datastructures-js/deque": "1.0.8",
     "@livekit/mutex": "^1.0.0",
-    "@livekit/rtc-ffi-bindings": "0.12.60",
+    "@livekit/rtc-ffi-bindings": "0.12.68",
     "@livekit/typed-emitter": "^3.0.0",
     "pino": "^9.0.0",
     "pino-pretty": "^13.0.0"

packages/livekit-rtc/src/audio_stream.ts

@@ -123,20 +123,9 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
         this.controller.enqueue(frame);
         break;
       case 'eos':
-        FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-        this.controller.close();
-        // Dispose the native handle so the FD is released on stream end,
-        // not just when cancel() is called explicitly by the consumer.
-        // Guard against double-dispose if cancel() is called after EOS
-        // while buffered frames are still in the ReadableStream queue.
-        if (!this.disposed) {
-          this.disposed = true;
-          this.track.unregisterAudioStream(this);
-          this.ffiHandle.dispose();
-          if (this.frameProcessor && this.autoCloseProcessor) {
-            this.frameProcessor.close();
-          }
-        }
+        // Disposes the native handle so the FD is released on stream end, not
+        // just when cancel() is called explicitly by the consumer.
+        this.teardown();
         break;
     }
   };
@@ -145,19 +134,43 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
     this.controller = controller;
   }
 
-  cancel() {
+  /**
+   * Detach from the FFI and release resources: on `eos`, on `cancel()`, or
+   * because the track went away (e.g. it was unsubscribed — which never
+   * produces an `eos`, so without this the stream would keep delivering
+   * frames). Already-buffered frames stay readable and the consumer sees `done`
+   * after draining them.
+   *
+   * @remarks
+   * Idempotent, so `eos` arriving after a `cancel()` (or vice versa) doesn't
+   * double-dispose the handle while buffered frames are still queued.
+   *
+   * @internal
+   */
+  teardown() {
     FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-    if (!this.disposed) {
-      this.disposed = true;
-      this.track.unregisterAudioStream(this);
-      this.ffiHandle.dispose();
-      // Also close the frame processor on cancel for symmetry with the EOS path,
-      // so resources are released regardless of how the stream ends.
-      if (this.frameProcessor && this.autoCloseProcessor) {
-        this.frameProcessor.close();
-      }
+    if (this.disposed) {
+      return;
+    }
+    this.disposed = true;
+    this.track.unregisterAudioStream(this);
+    this.ffiHandle.dispose();
+    // Close the frame processor on every teardown path so resources are
+    // released regardless of how the stream ended.
+    if (this.frameProcessor && this.autoCloseProcessor) {
+      this.frameProcessor.close();
+    }
+    try {
+      this.controller?.close();
+    } catch {
+      // Already closed — e.g. cancel(), where the consumer has torn the
+      // ReadableStream down before the underlying source is notified.
     }
   }
+
+  cancel() {
+    this.teardown();
+  }
 }
 
 export class AudioStream extends ReadableStream<AudioFrame> {

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

@@ -125,10 +125,23 @@ function makeLocalAudioTrack(sid: string): LocalAudioTrack {
 
 function makeStream(processor: FrameProcessor<AudioFrame> | null): AudioStreamSource {
   // Minimal stub exercising only the surface the Track touches: the `processor`
-  // getter and a no-op `cancel()`. Keeping cancel inert isolates the
-  // metadata-push assertions from the real teardown path, which is covered
+  // getter and no-op `cancel()` / `teardown()`. Keeping teardown inert isolates
+  // the metadata-push assertions from the real teardown path, which is covered
   // separately via simulateStreamClose.
-  return { processor, cancel: () => {} } as unknown as AudioStreamSource;
+  return { processor, cancel: () => {}, teardown: () => {} } as unknown as AudioStreamSource;
+}
+
+/** A stream stub that records whether the room tore it down. */
+function makeEndTrackingStream(): { stream: AudioStreamSource; endCount: () => number } {
+  let ended = 0;
+  const stream = {
+    processor: null,
+    cancel: () => {},
+    teardown: () => {
+      ended += 1;
+    },
+  } as unknown as AudioStreamSource;
+  return { stream, endCount: () => ended };
 }
 
 function makeLocalParticipant(identity: string): LocalParticipant {
@@ -577,6 +590,40 @@ describe('AudioStream room lifecycle', () => {
     });
   });
 
+  it('trackUnsubscribed ends the audio streams attached to the track', async () => {
+    // Regression: an unsubscribed track never receives `eos` from the FFI, so
+    // its AudioStreams kept delivering frames. After a reconnect that means the
+    // stale stream and the new subscription's stream both deliver the
+    // publisher's audio.
+    const room = makeRoom({ name: 'room-1', token: 'tok-1', serverUrl: 'wss://r' });
+    attachRemoteParticipant(room, 'alice', [{ publicationSid: TRACK_SID, trackSid: TRACK_SID }]);
+    const track = makeTrack(TRACK_SID);
+    const publication = room.remoteParticipants.get('alice')!.trackPublications.get(TRACK_SID)!;
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    (publication as any).track = track;
+    track.setRoom(room);
+
+    const first = makeEndTrackingStream();
+    const second = makeEndTrackingStream();
+    track.registerAudioStream(first.stream);
+    track.registerAudioStream(second.stream);
+
+    // The handler still sees a live track — teardown happens after the event.
+    const seenDuringEvent: Array<number> = [];
+    room.on('trackUnsubscribed', () => seenDuringEvent.push(first.endCount()));
+
+    await dispatchRoomEvent(room, {
+      case: 'trackUnsubscribed',
+      value: { participantIdentity: 'alice', trackSid: TRACK_SID },
+    });
+
+    expect(seenDuringEvent).toEqual([0]);
+    expect(first.endCount()).toBe(1);
+    expect(second.endCount()).toBe(1);
+    expect(publication.track).toBeUndefined();
+    expect(publication.subscribed).toBe(false);
+  });
+
   it('localTrackUnpublished event nulls publication track', async () => {
     const room = makeRoom({ name: 'room-1', token: 'tok-1', serverUrl: 'wss://r' });
     const track = makeTrack(TRACK_SID);

packages/livekit-rtc/src/room.ts

@@ -679,6 +679,10 @@ export class Room extends (EventEmitter as new () => TypedEmitter<RoomCallbacks>
         publication.track = undefined;
         publication.subscribed = false;
         this.emit(RoomEvent.TrackUnsubscribed, track, publication, participant);
+        // An unsubscribed track never gets an `eos`, so any AudioStream on it
+        // keeps delivering frames. Tear them down here — after the event, so
+        // handlers still see a live track.
+        track.closeAudioStreams();
       } catch (e: unknown) {
         log.warn(`RoomEvent.TrackUnsubscribed: ${(e as Error).message}`);
       }