From a8ce3985ca7553185de4b43ba1f468defb6c13e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20Fuente=20P=C3=A9rez?= Date: Fri, 4 Sep 2026 21:32:48 +0200 Subject: [PATCH] Revert PR #4838 in favor of a better solution (#4840) --- pkg/rtc/participant.go | 3 +++ pkg/rtc/room.go | 38 ----------------------------- pkg/rtc/subscriptionmanager.go | 2 +- pkg/rtc/subscriptionmanager_test.go | 1 - 4 files changed, 4 insertions(+), 40 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 17b07dac3..01b14479d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -2204,6 +2204,9 @@ func (p *ParticipantImpl) setupSubscriptionManager() { SubscriptionLimitAudio: p.params.SubscriptionLimitAudio, UseOneShotSignallingMode: p.params.UseOneShotSignallingMode, }) + p.SubscriptionManager.OnSubscribeStatusChanged(func(publisherID livekit.ParticipantID, subscribed bool) { + p.listener().OnSubscribeStatusChanged(p, publisherID, subscribed) + }) } func (p *ParticipantImpl) MetricsCollectorTimeToCollectMetrics() { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 0585fe48a..76224e731 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1681,8 +1681,6 @@ func (r *Room) connectionQualityWorker() { defer ticker.Stop() prevConnectionInfos := make(map[livekit.ParticipantID]*livekit.ConnectionQualityInfo) - // per participant, the participants whose quality it was told about in its last update - toldQualityOf := make(map[livekit.ParticipantID]map[livekit.ParticipantID]struct{}) // send updates to only users that are subscribed to each other for !r.IsClosed() { <-ticker.C @@ -1720,32 +1718,6 @@ func (r *Room) connectionQualityWorker() { } } - // also send if a participant is subscribed to someone whose quality it has never been told about: - // its subscriptions may have been established after the tick that announced it as a new entrant, - // and without a later quality change it would never receive their quality - for pID := range toldQualityOf { - if _, nowOk := nowConnectionInfos[pID]; !nowOk { - // participant is not ACTIVE any more - delete(toldQualityOf, pID) - } - } - if !sendUpdate { - sendUpdate = slices.ContainsFunc(participants, func(op types.LocalParticipant) bool { - if !op.ProtocolVersion().SupportsConnectionQuality() || op.State() != livekit.ParticipantInfo_ACTIVE { - return false - } - for _, sid := range op.GetSubscribedParticipants() { - if _, nowOk := nowConnectionInfos[sid]; !nowOk { - continue - } - if _, told := toldQualityOf[op.ID()][sid]; !told { - return true - } - } - return false - }) - } - if !sendUpdate { prevConnectionInfos = nowConnectionInfos continue @@ -1774,20 +1746,10 @@ func (r *Room) connectionQualityWorker() { // no change continue } - if err := op.SendConnectionQualityUpdate(update); err != nil { r.logger.Warnw("could not send connection quality update", err, "participant", op.Identity()) - continue } - - // remember which participants' quality this participant has been sent, - // only after a successful send so that a failed one is retried on next tick - told := make(map[livekit.ParticipantID]struct{}, len(update.Updates)) - for _, info := range update.Updates { - told[livekit.ParticipantID(info.ParticipantSid)] = struct{}{} - } - toldQualityOf[op.ID()] = told } prevConnectionInfos = nowConnectionInfos diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 3b5ddd849..773989e51 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -1222,7 +1222,7 @@ func (m *SubscriptionManager) unmarkSubscribedTo(publisherID livekit.Participant } m.lock.Unlock() if changedCB != nil && lastSubscription { - go changedCB(publisherID, false) + changedCB(publisherID, false) } } diff --git a/pkg/rtc/subscriptionmanager_test.go b/pkg/rtc/subscriptionmanager_test.go index 28b1a4864..9dd32f68c 100644 --- a/pkg/rtc/subscriptionmanager_test.go +++ b/pkg/rtc/subscriptionmanager_test.go @@ -113,7 +113,6 @@ func TestSubscribe(t *testing.T) { require.Eventually(t, func() bool { return numParticipantSubscribed.Load() == 2 }, subSettleTimeout, subCheckInterval, "participant subscribe status was not updated twice") - // the unsubscribed callback is delivered on a goroutine of its own require.Eventually(t, func() bool { return numParticipantUnsubscribed.Load() == 1 }, subSettleTimeout, subCheckInterval, "participant unsubscribe status was not updated")