Handle data track SID reassignment

.changeset/handle-data-track-sid-reassignment.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+Handle data track SID reassignment

src/room/data-track/incoming/IncomingDataTrackManager.test.ts

@@ -86,6 +86,37 @@ describe('DataTrackIncomingManager', () => {
         DataTrackHandle.fromNumber(5),
       ]);
     });
+
+    it('should not republish when a track sid is reassigned', async () => {
+      const manager = new IncomingDataTrackManager();
+      const managerEvents = subscribeToEvents<DataTrackIncomingManagerCallbacks>(manager, [
+        'sfuUpdateSubscription',
+        'trackPublished',
+        'trackUnpublished',
+      ]);
+
+      const pubHandle = DataTrackHandle.fromNumber(5);
+      const oldSid = 'old sid';
+      const newSid = 'new sid';
+
+      // Simulate track published
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+
+      const { track } = await managerEvents.waitFor('trackPublished');
+      expect(track.info.sid).toStrictEqual(oldSid);
+
+      // Simulate publisher full reconnect: same track, new SID
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+
+      // No publish/unpublish should appear
+      expect(managerEvents.areThereBufferedEvents('trackPublished')).toStrictEqual(false);
+      expect(managerEvents.areThereBufferedEvents('trackUnpublished')).toStrictEqual(false);
+      expect(track.info.sid).toStrictEqual(newSid);
+    });
   });
 
   describe('Track subscription', () => {
@@ -1064,6 +1095,217 @@ describe('DataTrackIncomingManager', () => {
       expect(onlyFrame.done).toStrictEqual(false);
       expect(onlyFrame.value?.payload).toStrictEqual(new Uint8Array([0xb1, 0xb2]));
     });
+
+    it('should resubscribe an active subscription when a track sid is reassigned', async () => {
+      const manager = new IncomingDataTrackManager();
+      const managerEvents = subscribeToEvents<DataTrackIncomingManagerCallbacks>(manager, [
+        'sfuUpdateSubscription',
+        'trackPublished',
+      ]);
+
+      const pubHandle = DataTrackHandle.fromNumber(5);
+      const oldSid = 'old sid';
+      const newSid = 'new sid';
+      const oldSubHandle = DataTrackHandle.fromNumber(6);
+      const newSubHandle = DataTrackHandle.fromNumber(7);
+
+      // Simulate track published
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+      const { track } = await managerEvents.waitFor('trackPublished');
+
+      // Subscribe to the track
+      const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(oldSid);
+      const reader = stream.getReader();
+      await managerEvents.waitFor('sfuUpdateSubscription');
+
+      // Simulate SFU assigning subscriber handle
+      manager.receivedSfuSubscriberHandles(new Map([[oldSubHandle, oldSid]]));
+
+      await sfuSubscriptionComplete;
+
+      // Simulate publisher full reconnect: same track, new SID
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+
+      // Manager should re-subscribe under the new SID
+      const event = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(event.subscribe).toStrictEqual(true);
+      expect(event.sid).toStrictEqual(newSid);
+      expect(track.info.sid).toStrictEqual(newSid);
+
+      // Simulate SFU assigning a new subscriber handle
+      manager.receivedSfuSubscriberHandles(new Map([[newSubHandle, newSid]]));
+
+      // Frames received on the new handle reach the existing subscriber
+      manager.packetReceived(
+        new DataTrackPacket(
+          new DataTrackPacketHeader({
+            extensions: new DataTrackExtensions(),
+            frameNumber: WrapAroundUnsignedInt.u16(0),
+            marker: FrameMarker.Single,
+            sequence: WrapAroundUnsignedInt.u16(0),
+            timestamp: DataTrackTimestamp.fromRtpTicks(0),
+            trackHandle: newSubHandle,
+          }),
+          new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05]),
+        ).toBinary(),
+      );
+
+      const { value, done } = await reader.read();
+      expect(done).toStrictEqual(false);
+      expect(value?.payload).toStrictEqual(new Uint8Array([0x01, 0x02, 0x03, 0x04, 0x05]));
+    });
+
+    it('should terminate the sfu subscription when the stream is cancelled after a sid reassignment', async () => {
+      const manager = new IncomingDataTrackManager();
+      const managerEvents = subscribeToEvents<DataTrackIncomingManagerCallbacks>(manager, [
+        'sfuUpdateSubscription',
+        'trackPublished',
+      ]);
+
+      const pubHandle = DataTrackHandle.fromNumber(5);
+      const oldSid = 'old sid';
+      const newSid = 'new sid';
+      const oldSubHandle = DataTrackHandle.fromNumber(6);
+      const newSubHandle = DataTrackHandle.fromNumber(7);
+
+      // Simulate track published
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+      await managerEvents.waitFor('trackPublished');
+
+      // Subscribe to the track
+      const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(oldSid);
+      const reader = stream.getReader();
+      await managerEvents.waitFor('sfuUpdateSubscription');
+
+      // Simulate SFU assigning subscriber handle
+      manager.receivedSfuSubscriberHandles(new Map([[oldSubHandle, oldSid]]));
+      await sfuSubscriptionComplete;
+
+      // Simulate publisher full reconnect: same track, new SID
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+
+      // Manager should re-subscribe under the new SID
+      const resubscribeEvent = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(resubscribeEvent.subscribe).toStrictEqual(true);
+      expect(resubscribeEvent.sid).toStrictEqual(newSid);
+
+      // Simulate SFU assigning a new subscriber handle
+      manager.receivedSfuSubscriberHandles(new Map([[newSubHandle, newSid]]));
+
+      // Cancel the stream; the SFU unsubscribe must go out under the new SID
+      await reader.cancel();
+      const endEvent = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(endEvent.sid).toStrictEqual(newSid);
+      expect(endEvent.subscribe).toStrictEqual(false);
+    });
+
+    it('should terminate the sfu subscription when the abortsignal fires after a sid reassignment', async () => {
+      const manager = new IncomingDataTrackManager();
+      const managerEvents = subscribeToEvents<DataTrackIncomingManagerCallbacks>(manager, [
+        'sfuUpdateSubscription',
+        'trackPublished',
+      ]);
+
+      const pubHandle = DataTrackHandle.fromNumber(5);
+      const oldSid = 'old sid';
+      const newSid = 'new sid';
+      const oldSubHandle = DataTrackHandle.fromNumber(6);
+      const newSubHandle = DataTrackHandle.fromNumber(7);
+
+      // Simulate track published
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+      await managerEvents.waitFor('trackPublished');
+
+      // Subscribe to the track
+      const controller = new AbortController();
+      const [stream, sfuSubscriptionComplete] = manager.openSubscriptionStream(
+        oldSid,
+        controller.signal,
+      );
+      const reader = stream.getReader();
+      await managerEvents.waitFor('sfuUpdateSubscription');
+
+      // Simulate SFU assigning subscriber handle
+      manager.receivedSfuSubscriberHandles(new Map([[oldSubHandle, oldSid]]));
+      await sfuSubscriptionComplete;
+
+      // Simulate publisher full reconnect: same track, new SID
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+
+      // Manager should re-subscribe under the new SID
+      const resubscribeEvent = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(resubscribeEvent.subscribe).toStrictEqual(true);
+      expect(resubscribeEvent.sid).toStrictEqual(newSid);
+
+      // Simulate SFU assigning a new subscriber handle
+      manager.receivedSfuSubscriberHandles(new Map([[newSubHandle, newSid]]));
+
+      // Abort the subscription; the SFU unsubscribe must go out under the new SID
+      const inFlightReadPromise = reader.read();
+      controller.abort();
+      await expect(inFlightReadPromise).rejects.toThrowError(
+        'Subscription to data track cancelled by caller',
+      );
+      const endEvent = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(endEvent.sid).toStrictEqual(newSid);
+      expect(endEvent.subscribe).toStrictEqual(false);
+    });
+
+    it('should cancel a pending subscription under the new sid after a sid reassignment', async () => {
+      const manager = new IncomingDataTrackManager();
+      const managerEvents = subscribeToEvents<DataTrackIncomingManagerCallbacks>(manager, [
+        'sfuUpdateSubscription',
+        'trackPublished',
+      ]);
+
+      const pubHandle = DataTrackHandle.fromNumber(5);
+      const oldSid = 'old sid';
+      const newSid = 'new sid';
+
+      // Simulate track published
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: oldSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+      await managerEvents.waitFor('trackPublished');
+
+      // Begin subscribing to the track, leaving the subscription pending
+      const controller = new AbortController();
+      const [, sfuSubscriptionComplete] = manager.openSubscriptionStream(oldSid, controller.signal);
+      const startEvent = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(startEvent.sid).toStrictEqual(oldSid);
+      expect(startEvent.subscribe).toStrictEqual(true);
+
+      // Simulate publisher full reconnect: same track, new SID
+      await manager.receiveSfuPublicationUpdates(
+        new Map([['id', [{ sid: newSid, pubHandle, name: 'test', usesE2ee: false }]]]),
+      );
+
+      // Manager should re-subscribe under the new SID
+      const resubscribeEvent = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(resubscribeEvent.subscribe).toStrictEqual(true);
+      expect(resubscribeEvent.sid).toStrictEqual(newSid);
+
+      // Abort the pending subscription; the SFU unsubscribe must go out under the new SID
+      controller.abort();
+      await expect(sfuSubscriptionComplete).rejects.toStrictEqual(
+        DataTrackSubscribeError.cancelled(),
+      );
+      const endEvent = await managerEvents.waitFor('sfuUpdateSubscription');
+      expect(endEvent.sid).toStrictEqual(newSid);
+      expect(endEvent.subscribe).toStrictEqual(false);
+    });
   });
 });
 

src/room/data-track/incoming/IncomingDataTrackManager.ts

@@ -147,6 +147,10 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () =>
     let streamController: ReadableStreamDefaultController<DataTrackFrame> | null = null;
     const sfuSubscriptionComplete = new Future<void, DataTrackSubscribeError>();
 
+    // Hold the descriptor by reference: SID reassignment re-keys the map, so lookups
+    // with the SID captured at subscribe time would fail.
+    const descriptor = this.descriptors.get(sid);
+
     const detachSignal = () => {
       signal?.removeEventListener('abort', onAbort);
     };
@@ -158,8 +162,7 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () =>
         log.warn(`ReadableStream subscribed to ${sid} was not started.`);
         return;
       }
-      const descriptor = this.descriptors.get(sid);
-      if (!descriptor) {
+      if (!descriptor || this.descriptors.get(descriptor.info.sid) !== descriptor) {
         log.warn(`Unknown track ${sid}, skipping cancel...`);
         return;
       }
@@ -180,9 +183,8 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () =>
       if (!streamController) {
         return;
       }
-      const currentDescriptor = this.descriptors.get(sid);
-      if (currentDescriptor?.subscription.type === 'active') {
-        currentDescriptor.subscription.streamControllers.delete(streamController);
+      if (descriptor?.subscription.type === 'active') {
+        descriptor.subscription.streamControllers.delete(streamController);
       }
 
       streamController.error(DataTrackSubscribeError.cancelled());
@@ -198,8 +200,7 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () =>
 
           this.subscribeRequest(sid, signal)
             .then(async () => {
-              const descriptor = this.descriptors.get(sid);
-              if (!descriptor) {
+              if (!descriptor || this.descriptors.get(descriptor.info.sid) !== descriptor) {
                 log.error(`Unknown track ${sid}`);
                 const err = DataTrackSubscribeError.disconnected();
                 controller.error(err);
@@ -330,7 +331,7 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () =>
             descriptor.subscription = { type: 'none' };
 
             // Let the SFU know that the subscribe has been cancelled
-            this.emit('sfuUpdateSubscription', { sid, subscribe: false });
+            this.emit('sfuUpdateSubscription', { sid: descriptor.info.sid, subscribe: false });
 
             if (previousDescriptorSubscription.type === 'pending') {
               previousDescriptorSubscription.completionFuture.reject?.(
@@ -458,6 +459,9 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () =>
         if (this.descriptors.has(info.sid)) {
           continue;
         }
+        if (this.handleSidReassigned(publisherIdentity, info)) {
+          continue;
+        }
         await this.handleTrackPublished(publisherIdentity, info);
       }
       publisherParticipantToSidsInUpdate.set(publisherIdentity, sidsInUpdate);
@@ -499,6 +503,62 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () =>
     this.emit('trackPublished', { track });
   }
 
+  /**
+   * Detects and handles SID reassignment, which occurs when the publisher
+   * republishes its tracks after a full reconnect.
+   *
+   * Returns `true` if an SID reassignment occurred, `false` otherwise.
+   */
+  private handleSidReassigned(
+    publisherIdentity: Participant['identity'],
+    info: DataTrackInfo,
+  ): boolean {
+    // Publisher identity and pub handle are stable across republications.
+    const existingEntry = Array.from(this.descriptors.entries()).find(
+      ([_sid, descriptor]) =>
+        descriptor.publisherIdentity === publisherIdentity &&
+        descriptor.info.pubHandle === info.pubHandle,
+    );
+    if (!existingEntry) {
+      return false;
+    }
+    const [oldSid, descriptor] = existingEntry;
+
+    // Invariant: other than SID, info should not have changed.
+    // TODO: consider refactoring to move SID out of info to allow for direct comparison.
+    const { name, usesE2ee } = descriptor.info;
+    if (name !== info.name || usesE2ee !== info.usesE2ee) {
+      log.warn(`Info mismatch for ${oldSid}, treating as new publication`);
+      return false;
+    }
+
+    const newSid = info.sid;
+    log.debug(`SID reassigned: ${oldSid} -> ${newSid}`);
+
+    if (!this.descriptors.delete(oldSid)) {
+      return false;
+    }
+    descriptor.info.sid = newSid;
+
+    switch (descriptor.subscription.type) {
+      case 'none':
+        break;
+      case 'pending':
+      case 'active':
+        // The SFU does not carry subscriptions across a publisher's full
+        // reconnect; re-request the subscription under the new SID.
+        this.emit('sfuUpdateSubscription', { sid: newSid, subscribe: true });
+        break;
+    }
+    if (descriptor.subscription.type === 'active') {
+      // Keep the routing index consistent until the SFU assigns a new handle
+      // (see `registerSubscriberHandle`).
+      this.subscriptionHandles.set(descriptor.subscription.subcriptionHandle, newSid);
+    }
+    this.descriptors.set(newSid, descriptor);
+    return true;
+  }
+
   handleTrackUnpublished(sid: DataTrackSid) {
     const descriptor = this.descriptors.get(sid);
     if (!descriptor) {