From f2d4f81ce570e5b740b232ebe2c767da5518ba1a Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 14 Jul 2026 15:47:17 -0700 Subject: [PATCH] Handle data track SID reassignment (#2000) --- .../handle-data-track-sid-reassignment.md | 5 + .../incoming/IncomingDataTrackManager.test.ts | 242 ++++++++++++++++++ .../incoming/IncomingDataTrackManager.ts | 76 +++++- 3 files changed, 315 insertions(+), 8 deletions(-) create mode 100644 .changeset/handle-data-track-sid-reassignment.md diff --git a/.changeset/handle-data-track-sid-reassignment.md b/.changeset/handle-data-track-sid-reassignment.md new file mode 100644 index 0000000000..ba9b4129d1 --- /dev/null +++ b/.changeset/handle-data-track-sid-reassignment.md @@ -0,0 +1,5 @@ +--- +"livekit-client": patch +--- + +Handle data track SID reassignment diff --git a/src/room/data-track/incoming/IncomingDataTrackManager.test.ts b/src/room/data-track/incoming/IncomingDataTrackManager.test.ts index f94affa825..0d5e19822f 100644 --- a/src/room/data-track/incoming/IncomingDataTrackManager.test.ts +++ b/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(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(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(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(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(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); + }); }); }); diff --git a/src/room/data-track/incoming/IncomingDataTrackManager.ts b/src/room/data-track/incoming/IncomingDataTrackManager.ts index 0f78dfee00..c024d3ce91 100644 --- a/src/room/data-track/incoming/IncomingDataTrackManager.ts +++ b/src/room/data-track/incoming/IncomingDataTrackManager.ts @@ -147,6 +147,10 @@ export default class IncomingDataTrackManager extends (EventEmitter as new () => let streamController: ReadableStreamDefaultController | null = null; const sfuSubscriptionComplete = new Future(); + // 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) {