fix: keep a paused upstream paused when the track is republished

.changeset/paused-upstream-republish.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+A track whose upstream is paused stays paused when it is republished, e.g. after a full reconnect: the new sender starts without the track and the server is told the track is muted.

src/room/RTCEngine.test.ts

@@ -16,6 +16,7 @@ import { roomOptionDefaults } from './defaults';
 import { ConnectionError, PublishDataError, UnexpectedConnectionState } from './errors';
 import { EngineEvent } from './events';
 import CriticalTimers from './timers';
+import type LocalVideoTrack from './track/LocalVideoTrack';
 import { Future } from './utils';
 
 describe('RTCEngine', () => {
@@ -623,6 +624,68 @@ describe('RTCEngine', () => {
     });
   });
 
+  describe('createSender', () => {
+    // a full reconnect republishes every track over a new sender, including one whose upstream
+    // the application paused
+    function stubPublisher(engine: RTCEngine) {
+      vi.stubGlobal(
+        'RTCPeerConnection',
+        class {
+          addTransceiver() {}
+        },
+      );
+      const sender = { replaceTrack: vi.fn(async () => {}) };
+      Object.assign(engine as unknown as Record<string, unknown>, {
+        pcManager: { addPublisherTransceiver: vi.fn(async () => ({ sender })) },
+      });
+      return sender;
+    }
+
+    function makeTrack(isUpstreamPaused: boolean) {
+      return {
+        kind: 'video',
+        isUpstreamPaused,
+        mediaStreamTrack: { id: 'camera' },
+        setSimulcastTrackSender: vi.fn(async () => {}),
+      } as unknown as LocalVideoTrack;
+    }
+
+    afterEach(() => {
+      vi.unstubAllGlobals();
+    });
+
+    it('keeps a paused track off the new sender', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const sender = stubPublisher(engine);
+
+      await engine.createSender(makeTrack(true), {});
+
+      expect(sender.replaceTrack).toHaveBeenCalledExactlyOnceWith(null);
+    });
+
+    it('keeps a paused track off a new backup codec sender', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const sender = stubPublisher(engine);
+
+      await engine.createSimulcastSender(
+        makeTrack(true),
+        { codec: 'vp8', mediaStreamTrack: { id: 'camera-clone' } as MediaStreamTrack },
+        { videoCodec: 'vp8' },
+      );
+
+      expect(sender.replaceTrack).toHaveBeenCalledExactlyOnceWith(null);
+    });
+
+    it('leaves a track that is not paused on the new sender', async () => {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const sender = stubPublisher(engine);
+
+      await engine.createSender(makeTrack(false), {});
+
+      expect(sender.replaceTrack).not.toHaveBeenCalled();
+    });
+  });
+
   describe('local connection quality Lost handling', () => {
     // The engine reacts to the server's own verdict: a sustained local `LOST` while
     // connected and publishing means our media isn't reaching the server, so it forces

src/room/RTCEngine.ts

@@ -1095,6 +1095,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       throw new UnexpectedConnectionState('Required webRTC APIs not supported on this device');
     }
     this.setupFrameMetadataSender(sender, opts);
+    await this.keepPausedTrackOffSender(track, sender);
     return sender;
   }
 
@@ -1115,10 +1116,19 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     }
     if (sender) {
       this.setupFrameMetadataSender(sender, opts);
+      await this.keepPausedTrackOffSender(track, sender);
     }
     return sender;
   }
 
+  // a track republished while its upstream is paused (e.g. after a full reconnect) gets a new
+  // sender, which pauseUpstream() never saw: the sender has to start without the track
+  private async keepPausedTrackOffSender(track: LocalTrack, sender: RTCRtpSender) {
+    if (track.isUpstreamPaused) {
+      await sender.replaceTrack(null);
+    }
+  }
+
   private get frameMetadataWorker(): Worker | undefined {
     return this.frameMetadataOptions?.worker;
   }

src/room/participant/LocalParticipant.test.ts

@@ -217,6 +217,16 @@ describe('LocalParticipant.publishAdditionalCodecForTrack', () => {
     expect(req.simulcastCodecs[0].layers).toEqual(req.layers);
   });
 
+  it('announces the backup codec of a paused track as muted', async () => {
+    const track = makeBackupCodecTrack({ width: 1280, height: 720 });
+    Object.assign(track, { isUpstreamPaused: true });
+    const participant = makeBackupCodecParticipant(track);
+
+    await publishBackupCodec(participant, track);
+
+    expect(capturedRequest(participant).muted).toBe(true);
+  });
+
   it('omits layers when the capture dimensions are unknown', async () => {
     const track = makeBackupCodecTrack({});
     const participant = makeBackupCodecParticipant(track);
@@ -231,3 +241,51 @@ describe('LocalParticipant.publishAdditionalCodecForTrack', () => {
     expect(participant.log.warn).toHaveBeenCalledOnce();
   });
 });
+
+describe('LocalParticipant republishing a track whose upstream is paused', () => {
+  // a full reconnect unpublishes every track and publishes it again over a new sender, while
+  // the track keeps its paused upstream state
+  function makePausedMicrophone() {
+    return {
+      isLocal: true,
+      kind: Track.Kind.Audio,
+      source: Track.Source.Microphone,
+      isMuted: false,
+      isUpstreamPaused: true,
+      mediaStreamTrack: { id: 'microphone-cid' },
+      on: vi.fn(),
+      getSourceTrackSettings: () => ({}),
+    } as unknown as LocalTrack;
+  }
+
+  async function capturePublishRequest(track: LocalTrack) {
+    const addTrack = vi.fn(async () => {
+      throw new Error('stop after the add track request');
+    });
+    const participant = Object.create(LocalParticipant.prototype) as Record<string, unknown>;
+    Object.assign(participant, {
+      encryptionType: Encryption_Type.NONE,
+      trackPublications: new Map(),
+      enabledPublishVideoCodecs: [],
+      hasPermissionsToPublish: () => true,
+      canPublishFrameMetadata: () => false,
+      engine: { isClosed: false, addTrack },
+      log: { warn: vi.fn(), debug: vi.fn(), info: vi.fn() },
+    });
+    const publish = (
+      participant as unknown as {
+        publish: (track: LocalTrack, opts: TrackPublishOptions, isStereo: boolean) => Promise<void>;
+      }
+    ).publish.bind(participant);
+
+    await expect(publish(track, {}, false)).rejects.toThrow('stop after the add track request');
+    expect(addTrack).toHaveBeenCalledOnce();
+    return addTrack.mock.calls[0][0] as AddTrackRequest;
+  }
+
+  it('announces the track as muted to the server', async () => {
+    const req = await capturePublishRequest(makePausedMicrophone());
+
+    expect(req.muted).toBe(true);
+  });
+});

src/room/participant/LocalParticipant.ts

@@ -929,7 +929,8 @@ export class LocalParticipant extends Participant {
       cid: track.mediaStreamTrack.id,
       name: opts.name,
       type: Track.kindToProto(track.kind),
-      muted: track.isMuted,
+      // the server sees a paused upstream as muted, see onTrackUpstreamPaused
+      muted: track.isMuted || track.isUpstreamPaused,
       source: Track.sourceToProto(track.source),
       disableDtx,
       encryption: this.encryptionType,

src/room/video/publisher.ts

@@ -403,7 +403,8 @@ export async function publishAdditionalCodecForTrack(
   const req = new AddTrackRequest({
     cid: simulcastTrack.mediaStreamTrack.id,
     type: Track.kindToProto(track.kind),
-    muted: track.isMuted,
+    // match the primary publish request, the backup sender starts without a paused track
+    muted: track.isMuted || track.isUpstreamPaused,
     source: Track.sourceToProto(track.source),
     sid: track.sid,
     width: dims?.width,