From e544206932df8675be8d83da2522d5fd84359b7a Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Thu, 9 Jul 2026 10:30:35 -0700
Subject: [PATCH] Send demand updates

---
 pkg/rtc/datatrack.go                          | 18 +++++++++++++++++-
 pkg/rtc/participant.go                        |  3 +++
 pkg/rtc/participant_data_track.go             |  3 +++
 pkg/rtc/participant_signal.go                 |  7 +++++++
 pkg/rtc/signalling/interfaces.go              |  1 +
 pkg/rtc/signalling/signalling.go              |  8 ++++++++
 pkg/rtc/signalling/signallingunimplemented.go |  4 ++++
 7 files changed, 43 insertions(+), 1 deletion(-)

diff --git a/pkg/rtc/datatrack.go b/pkg/rtc/datatrack.go
index fd22f6258..ce5b26e1c 100644
--- a/pkg/rtc/datatrack.go
+++ b/pkg/rtc/datatrack.go
@@ -40,6 +40,8 @@ type DataTrackParams struct {
 	ParticipantID       func() livekit.ParticipantID
 	ParticipantIdentity livekit.ParticipantIdentity
 	BytesTrackStats     *BytesTrackStats
+	// invoked when the number of subscribers changes on a dynacasted track
+	OnSubscriberCountChanged func(subscriberCount uint32)
 }
 
 type subscribedDataTrack struct {
@@ -115,9 +117,9 @@ func (d *DataTrack) Name() string {
 
 func (d *DataTrack) AddSubscriber(sub types.LocalParticipant) (types.DataDownTrack, error) {
 	d.lock.Lock()
-	defer d.lock.Unlock()
 
 	if _, ok := d.subscribedTracks[sub.ID()]; ok {
+		d.lock.Unlock()
 		return nil, errAlreadySubscribed
 	}
 
@@ -139,6 +141,7 @@ func (d *DataTrack) AddSubscriber(sub types.LocalParticipant) (types.DataDownTra
 		BytesTrackStats:  bytesStats,
 	})
 	if err != nil {
+		d.lock.Unlock()
 		bytesStats.Stop()
 		return nil, err
 	}
@@ -147,6 +150,10 @@ func (d *DataTrack) AddSubscriber(sub types.LocalParticipant) (types.DataDownTra
 		subscriber:    sub,
 		dataDownTrack: dataDownTrack,
 	}
+	subscriberCount := len(d.subscribedTracks)
+	d.lock.Unlock()
+
+	d.notifySubscriberCountChanged(subscriberCount)
 	return dataDownTrack, nil
 }
 
@@ -154,13 +161,22 @@ func (d *DataTrack) RemoveSubscriber(subID livekit.ParticipantID) {
 	d.lock.Lock()
 	subscribedTrack, ok := d.subscribedTracks[subID]
 	delete(d.subscribedTracks, subID)
+	subscriberCount := len(d.subscribedTracks)
 	d.lock.Unlock()
 
 	if ok {
 		subscribedTrack.dataDownTrack.Close()
+		d.notifySubscriberCountChanged(subscriberCount)
 	}
 }
 
+func (d *DataTrack) notifySubscriberCountChanged(subscriberCount int) {
+	if !d.dti.IsDynacasted || d.params.OnSubscriberCountChanged == nil {
+		return
+	}
+	d.params.OnSubscriberCountChanged(uint32(subscriberCount))
+}
+
 func (d *DataTrack) IsSubscriber(subID livekit.ParticipantID) bool {
 	d.lock.Lock()
 	defer d.lock.Unlock()
diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go
index 9cba516cf..add486bc4 100644
--- a/pkg/rtc/participant.go
+++ b/pkg/rtc/participant.go
@@ -1403,6 +1403,9 @@ func (p *ParticipantImpl) SetMigrateInfo(
 					p.params.TelemetryListener,
 					p.params.Reporter,
 				),
+				OnSubscriberCountChanged: func(subscriberCount uint32) {
+					p.sendDataTrackDemandUpdate(dti.PubHandle, subscriberCount)
+				},
 			},
 			dti,
 		)
diff --git a/pkg/rtc/participant_data_track.go b/pkg/rtc/participant_data_track.go
index df9854744..93265f6eb 100644
--- a/pkg/rtc/participant_data_track.go
+++ b/pkg/rtc/participant_data_track.go
@@ -124,6 +124,9 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData
 				p.params.TelemetryListener,
 				p.params.Reporter,
 			),
+			OnSubscriberCountChanged: func(subscriberCount uint32) {
+				p.sendDataTrackDemandUpdate(dti.PubHandle, subscriberCount)
+			},
 		},
 		dti,
 	)
diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go
index 8f5c3dd2b..2b0b949b0 100644
--- a/pkg/rtc/participant_signal.go
+++ b/pkg/rtc/participant_signal.go
@@ -369,6 +369,13 @@ func (p *ParticipantImpl) SendDataTrackSubscriberHandles(handles map[uint32]*liv
 	}))
 }
 
+func (p *ParticipantImpl) sendDataTrackDemandUpdate(pubHandle uint32, subscriberCount uint32) error {
+	return p.signaller.WriteMessage(p.signalling.SignalDataTrackDemandUpdate(&livekit.DataTrackDemandUpdate{
+		PubHandle:       pubHandle,
+		SubscriberCount: subscriberCount,
+	}))
+}
+
 func (p *ParticipantImpl) sendStoreDataBlobResponse(requestId uint32, key *livekit.DataBlobKey) error {
 	return p.signaller.WriteMessage(p.signalling.SignalStoreDataBlobResponse(&livekit.StoreDataBlobResponse{
 		RequestId: requestId,
diff --git a/pkg/rtc/signalling/interfaces.go b/pkg/rtc/signalling/interfaces.go
index 99c3a0acc..1cdb011b7 100644
--- a/pkg/rtc/signalling/interfaces.go
+++ b/pkg/rtc/signalling/interfaces.go
@@ -62,6 +62,7 @@ type ParticipantSignalling interface {
 	SignalPublishDataTrackResponse(publishDataTrackResponse *livekit.PublishDataTrackResponse) proto.Message
 	SignalUnpublishDataTrackResponse(unpublishDataTrackResponse *livekit.UnpublishDataTrackResponse) proto.Message
 	SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles *livekit.DataTrackSubscriberHandles) proto.Message
+	SignalDataTrackDemandUpdate(dataTrackDemandUpdate *livekit.DataTrackDemandUpdate) proto.Message
 	SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message
 	SignalGetDataBlobResponse(getDataBlobResponse *livekit.GetDataBlobResponse) proto.Message
 }
diff --git a/pkg/rtc/signalling/signalling.go b/pkg/rtc/signalling/signalling.go
index c13cee1be..fe0107d5b 100644
--- a/pkg/rtc/signalling/signalling.go
+++ b/pkg/rtc/signalling/signalling.go
@@ -259,6 +259,14 @@ func (s *signalling) SignalDataTrackSubscriberHandles(dataTrackSubscriberHandles
 	}
 }
 
+func (s *signalling) SignalDataTrackDemandUpdate(dataTrackDemandUpdate *livekit.DataTrackDemandUpdate) proto.Message {
+	return &livekit.SignalResponse{
+		Message: &livekit.SignalResponse_DataTrackDemandUpdate{
+			DataTrackDemandUpdate: dataTrackDemandUpdate,
+		},
+	}
+}
+
 func (s *signalling) SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message {
 	return &livekit.SignalResponse{
 		Message: &livekit.SignalResponse_StoreDataBlobResponse{
diff --git a/pkg/rtc/signalling/signallingunimplemented.go b/pkg/rtc/signalling/signallingunimplemented.go
index 8210c6aac..36ad12791 100644
--- a/pkg/rtc/signalling/signallingunimplemented.go
+++ b/pkg/rtc/signalling/signallingunimplemented.go
@@ -128,6 +128,10 @@ func (u *signallingUnimplemented) SignalDataTrackSubscriberHandles(dataTrackSubs
 	return nil
 }
 
+func (u *signallingUnimplemented) SignalDataTrackDemandUpdate(dataTrackDemandUpdate *livekit.DataTrackDemandUpdate) proto.Message {
+	return nil
+}
+
 func (u *signallingUnimplemented) SignalStoreDataBlobResponse(storeDataBlobResponse *livekit.StoreDataBlobResponse) proto.Message {
 	return nil
 }