Revert PR #4838 in favor of a better solution (#4840)

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() {

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

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)
 	}
 }
 

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")