feat: port in updates from livekit/python-sdks#679

Author: 1egoman

packages/livekit-rtc/src/audio_stream.ts

@@ -14,11 +14,12 @@ import type { Track } from './track.js';
 export interface AudioStreamOptions {
   noiseCancellation?: NoiseCancellationOptions | FrameProcessor<AudioFrame>;
   /**
-   * If true and `noiseCancellation` is a {@link FrameProcessor}, leaves the
-   * processor open when the stream closes so the same processor can be reused
-   * with another {@link AudioStream}. Defaults to `false`.
+   * When the audio stream closes, whether to run the {@link FrameProcessor}'s
+   * `close()` method. If `false`, the processor is left open so it can be
+   * reused with another {@link AudioStream}. Only relevant when
+   * `noiseCancellation` is a {@link FrameProcessor}. Defaults to `true`.
    */
-  noiseCancellationLeaveOpen?: boolean;
+  autoCloseNoiseCancellation?: boolean;
   sampleRate?: number;
   numChannels?: number;
   frameSizeMs?: number;
@@ -38,7 +39,7 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
   private numChannels: number;
   private legacyNcOptions?: NoiseCancellationOptions;
   private frameProcessor: FrameProcessor<AudioFrame> | null = null;
-  private leaveProcessorOpen = false;
+  private autoCloseProcessor = true;
   private frameSizeMs?: number;
   private track: Track;
 
@@ -53,7 +54,7 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
       this.numChannels = sampleRateOrOptions.numChannels ?? 1;
       if (isFrameProcessor(sampleRateOrOptions.noiseCancellation)) {
         this.frameProcessor = sampleRateOrOptions.noiseCancellation;
-        this.leaveProcessorOpen = sampleRateOrOptions.noiseCancellationLeaveOpen ?? false;
+        this.autoCloseProcessor = sampleRateOrOptions.autoCloseNoiseCancellation ?? true;
       } else {
         this.legacyNcOptions = sampleRateOrOptions.noiseCancellation;
       }
@@ -131,7 +132,7 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
           this.disposed = true;
           this.track.unregisterAudioStream(this);
           this.ffiHandle.dispose();
-          if (this.frameProcessor && !this.leaveProcessorOpen) {
+          if (this.frameProcessor && this.autoCloseProcessor) {
             this.frameProcessor.close();
           }
         }
@@ -151,7 +152,7 @@ export class AudioStreamSource implements UnderlyingSource<AudioFrame> {
       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.leaveProcessorOpen) {
+      if (this.frameProcessor && this.autoCloseProcessor) {
         this.frameProcessor.close();
       }
     }

packages/livekit-rtc/src/participant.ts

@@ -767,6 +767,15 @@ export class LocalParticipant extends Participant {
 
       const pub = this.trackPublications.get(trackSid);
       if (pub) {
+        // Clear the processor's room context here too: this path races the
+        // localTrackUnpublished room event, and whichever loses finds the
+        // publication already gone and skips its own setRoom(null). Calling it
+        // from both paths guarantees the processor is cleared (and the
+        // tokenRefreshed listener detached); setRoom(null) is idempotent, so a
+        // double-clear when this path wins is safe.
+        if (pub.track) {
+          pub.track.setRoom(null);
+        }
         pub.track = undefined;
       }
       this.trackPublications.delete(trackSid);

packages/livekit-rtc/src/room.ts

@@ -553,11 +553,21 @@ export class Room extends (EventEmitter as new () => TypedEmitter<RoomCallbacks>
       this.emit(RoomEvent.LocalTrackPublished, publication!, this.localParticipant);
     } else if (ev.case == 'localTrackUnpublished') {
       const publication = this.localParticipant.trackPublications.get(ev.value.publicationSid!);
-      if (publication?.track) {
-        publication.track.setRoom(null);
+      const track = publication?.track;
+      if (track) {
+        track.setRoom(null);
       }
       this.localParticipant.trackPublications.delete(ev.value.publicationSid!);
+      // Emit while `publication.track` is still set, preserving the pre-existing
+      // payload for callbacks. The handler is synchronous, so nulling the track
+      // right after still completes before any other turn can observe it.
       this.emit(RoomEvent.LocalTrackUnpublished, publication!, this.localParticipant!);
+      // Mirror trackUnsubscribed: drop the publication's track reference. This
+      // also makes unpublishTrack's own setRoom(null) a no-op when it loses the
+      // race (its `pub.track` guard short-circuits), avoiding a redundant clear.
+      if (track && publication) {
+        publication.track = undefined;
+      }
     } else if ((ev.case as string) == 'localTrackRepublished') {
       const value = (ev as any).value;
       const previousSid: string = value.previousSid!;
@@ -567,6 +577,15 @@ export class Room extends (EventEmitter as new () => TypedEmitter<RoomCallbacks>
         publication.updateInfo(newInfo);
         this.localParticipant.trackPublications.delete(previousSid);
         this.localParticipant.trackPublications.set(publication.sid!, publication);
+        if (publication.track?.info) {
+          // Keep the local-track invariant (track.sid == publication.sid, set at
+          // publishTrack) intact across republish, then re-push metadata so any
+          // attached FrameProcessor learns the new publication SID / credentials.
+          // setRoom with the same room is a no-op for the tokenRefreshed listener
+          // but re-fans the metadata to every registered AudioStream.
+          publication.track.info.sid = publication.sid;
+          publication.track.setRoom(this);
+        }
         this.emit(RoomEvent.LocalTrackRepublished, publication, previousSid, this.localParticipant);
       } else {
         log.warn(`RoomEvent.LocalTrackRepublished: previous publication not found: ${previousSid}`);

packages/livekit-rtc/src/track.ts

@@ -56,6 +56,13 @@ export abstract class Track {
   /** @internal */
   setRoom(room: Room | null): void {
     const oldRoom = this.resolveRoom();
+    if (!oldRoom && !room) {
+      // Already roomless — nothing to detach and nothing to re-clear. Without
+      // this guard a second setRoom(null) (e.g. the unpublishTrack /
+      // localTrackUnpublished race calling it from both paths) would re-fire
+      // onStreamInfoCleared / onCredentialsCleared on every registered processor.
+      return;
+    }
     if (oldRoom && oldRoom !== room) {
       oldRoom.off('tokenRefreshed', this.onRoomTokenRefreshed);
     }