From 308554daf1d52c397f318d680bc9d8954b4b97d1 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 13 Oct 2026 16:40:46 +0530 Subject: [PATCH] rtc: release timed out pending tracks at the pending track limit (#4998) * rtc: reclaim abandoned pending tracks A pending track is freed only when its first media arrives or the participant leaves. A client that calls AddTrack and then gives up before any media flows leaves the slot held for the whole session. After 20 such slots, every later AddTrack is rejected. Reclaim these slots when the cap is hit, if no media arrived and the slot is older than the publish timeout. Keep the supervisor monitor in step for queued tracks that share a track id. Add a publish_timeout config value, default 30s. Fixes: #4970 Co-Authored-By: Claude Opus 4.8 * rtc: fix publish_timeout comment to not name an internal constant Co-Authored-By: Claude Opus 4.8 * rtc: release supervisor-timed-out pending tracks at the limit Replace the createdAt sweep with the supervisor's publish timeout. The supervisor counts time only while the publisher is connected and the track is unmuted, and never reports a live track, so a successor queued behind a live track is not released. Restart the publish timer when a queued publish reaches the front of the queue, so it does not inherit the clock of the track ahead of it. Write the publication error under the write lock, since it is now read from another goroutine. Drop the publish_timeout config value. Co-Authored-By: Claude Opus 5.5 * rtc: skip stale supervisor timeouts and keep live monitors TimedOutPublications re-checks the monitor, because the cached error is cleared only on the next tick. A track that published since is not reported, so its queued successor is not released. RemovePublication keeps a monitor that still holds a published track. Co-Authored-By: Claude Opus 5.5 * rtc: drop late media for released pending tracks Media or an offer with no matching pending track falls back to the first pending track of the same type. After a release, late media for the released track could take another pending track's identity. Record the released track's signal and SDP client ids, the 32 most recent, and drop media or offers for them. An AddTrack that reuses the client id clears it. Co-Authored-By: Claude Opus 5.5 * rtc: clear all ids of a released track when it is re-added A re-add cleared only the signal client id, so a Firefox republish that keeps the old SDP id had its offer and media dropped. Keep the ids of each released track together, and clear the whole track when a new AddTrack reuses its signal or simulcast codec client id. The limit now counts released tracks, and a track ages out with all of its ids. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 4.8 --- pkg/rtc/participant.go | 118 +++++++++++++++ pkg/rtc/participant_internal_test.go | 116 ++++++++++++++ pkg/rtc/participant_sdp.go | 5 + pkg/rtc/supervisor/participant_supervisor.go | 55 ++++++- .../supervisor/participant_supervisor_test.go | 112 ++++++++++++++ pkg/rtc/supervisor/publication_monitor.go | 17 +++ .../supervisor/publication_monitor_test.go | 142 ++++++++++++++++++ 7 files changed, 561 insertions(+), 4 deletions(-) create mode 100644 pkg/rtc/supervisor/participant_supervisor_test.go create mode 100644 pkg/rtc/supervisor/publication_monitor_test.go diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index bf7420d28..b0bc076fb 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -83,6 +83,7 @@ const ( cMaxPendingTracks = 20 cMaxPendingQueuedTracks = 3 + cMaxReleasedTracks = 32 // unsequenced reliable data (server API sends) cannot be recovered from the // data message cache, so it is held here until the reliable data channel is writable @@ -296,6 +297,10 @@ type ParticipantImpl struct { pendingTracks map[string]*pendingTrackInfo pendingPublishingTracks map[livekit.TrackID]*pendingTrackInfo pendingRemoteTracks []*pendingRemoteTrack + // client ids (signal, codec and SDP) of each pending track released without media, so that + // media or an offer for them is dropped instead of being matched by type to another pending + // track, bounded to cMaxReleasedTracks, oldest out first + releasedTrackCids [][]string // supported codecs enabledPublishCodecs []*livekit.Codec @@ -2976,6 +2981,94 @@ func (p *ParticipantImpl) onSubscribedAudioCodecChange( return p.sendSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate) } +// releasePendingTrackLocked drops every pending entry of a track whose media never arrived, +// so it stops counting against cMaxPendingTracks. Must be called with pendingTracksLock held. +func (p *ParticipantImpl) releasePendingTrackLocked(trackID livekit.TrackID) { + var ( + released *livekit.TrackInfo + releasedCids []string + ) + for cid, pti := range p.pendingTracks { + remaining := pti.trackInfos[:0] + for _, ti := range pti.trackInfos { + if livekit.TrackID(ti.Sid) == trackID { + released = ti + releasedCids = append(releasedCids, cid) + for _, c := range ti.Codecs { + releasedCids = append(releasedCids, c.Cid, c.SdpCid) + } + continue + } + remaining = append(remaining, ti) + } + pti.trackInfos = remaining + if len(pti.trackInfos) == 0 { + delete(p.pendingTracks, cid) + } + } + if released == nil { + return + } + p.addReleasedTrackCidsLocked(releasedCids) + + p.pubLogger.Infow( + "releasing pending track that never received media", + "trackID", trackID, + "trackInfo", logger.Proto(released), + "numPendingTracks", len(p.pendingTracks), + ) + if p.supervisor != nil { + p.supervisor.RemovePublication(trackID) + } + if p.ProtocolVersion().SupportsUnpublish() { + p.sendTrackUnpublished(trackID) + } else { + p.sendTrackMuted(trackID, true) + } +} + +// must be called with pendingTracksLock held +func (p *ParticipantImpl) addReleasedTrackCidsLocked(cids []string) { + cids = slices.DeleteFunc(cids, func(cid string) bool { return cid == "" }) + slices.Sort(cids) + cids = slices.Compact(cids) + if len(cids) == 0 { + return + } + p.releasedTrackCids = append(p.releasedTrackCids, cids) + if len(p.releasedTrackCids) > cMaxReleasedTracks { + p.releasedTrackCids = p.releasedTrackCids[1:] + } +} + +// must be called with pendingTracksLock held +func (p *ParticipantImpl) isReleasedTrackCidLocked(cid string) bool { + for _, cids := range p.releasedTrackCids { + if slices.Contains(cids, cid) { + return true + } + } + return false +} + +// clearReleasedTrackCidsLocked drops every released track that shares a client id with +// this request. A client re-adding a released track intends to publish on it again, so all +// of its ids are cleared, including an SDP id a reused transceiver keeps. +// Must be called with pendingTracksLock held. +func (p *ParticipantImpl) clearReleasedTrackCidsLocked(req *livekit.AddTrackRequest) { + p.releasedTrackCids = slices.DeleteFunc(p.releasedTrackCids, func(cids []string) bool { + if slices.Contains(cids, req.Cid) { + return true + } + for _, sc := range req.SimulcastCodecs { + if sc.Cid != "" && slices.Contains(cids, sc.Cid) { + return true + } + } + return false + }) +} + func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit.TrackInfo { p.pendingTracksLock.Lock() if req.Sid != "" { @@ -2998,6 +3091,14 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit return ti } + if len(p.pendingTracks) >= cMaxPendingTracks && p.supervisor != nil { + // a client can abandon a publish without any signal the server acts on, + // so before rejecting, release pending tracks the supervisor reports as timed out + for _, trackID := range p.supervisor.TimedOutPublications() { + p.releasePendingTrackLocked(trackID) + } + } + if len(p.pendingTracks) >= cMaxPendingTracks { p.pubLogger.Infow("too many pending tracks, rejecting new track", "numPendingTracks", len(p.pendingTracks)) p.sendRequestResponse(&livekit.RequestResponse{ @@ -3011,6 +3112,8 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit return nil } + p.clearReleasedTrackCidsLocked(req) + backupCodecPolicy := req.BackupCodecPolicy // enable simulcast codec for audio by default @@ -3368,6 +3471,16 @@ func (p *ParticipantImpl) mediaTrackReceived( ) mt, ok := p.getPublishedTrackBySdpCid(track.ID()).(*MediaTrack) if !ok { + if p.isReleasedTrackCidLocked(track.ID()) { + p.pendingTracksLock.Unlock() + p.pubLogger.Infow( + "dropping media for a released pending track", + "trackID", track.ID(), + "mid", mid, + ) + return nil, false, false, buffer.VideoLayersRid{} + } + var ( signalCid string ti *livekit.TrackInfo @@ -3715,6 +3828,11 @@ func (p *ParticipantImpl) onUpTrackManagerClose() { func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackType, skipQueued bool) (string, *livekit.TrackInfo, buffer.VideoLayersRid, bool, time.Time) { signalCid := clientId + if p.isReleasedTrackCidLocked(clientId) { + // do not fall back to matching by type, it would hand this to another pending track + return signalCid, nil, buffer.VideoLayersRid{}, false, time.Time{} + } + pendingInfo := p.pendingTracks[clientId] if pendingInfo == nil { track_loop: diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index f3271eea4..c32b8a643 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -256,6 +256,122 @@ func TestTrackPublishing(t *testing.T) { require.Equal(t, p.pendingTracks["cid"].trackInfos[0].Sid, p.pendingTracks["cid"].trackInfos[1].Sid) }) + t.Run("should release timed out pending tracks when at the pending track limit", func(t *testing.T) { + p := newParticipantForTestWithOpts("test", &participantOpts{protocolVersion: 7}) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + // one slot holds two queued entries that share a track id, both before any media + queued := p.addPendingTrack(&livekit.AddTrackRequest{Cid: "cidq", Name: "q0", Type: livekit.TrackType_VIDEO}) + require.NotNil(t, queued) + p.addPendingTrack(&livekit.AddTrackRequest{Cid: "cidq", Name: "q1", Type: livekit.TrackType_VIDEO}) // queued, returns nil + require.Len(t, p.pendingTracks["cidq"].trackInfos, 2) + + var sids []livekit.TrackID + for i := 0; i < cMaxPendingTracks-1; i++ { + ti := p.addPendingTrack(&livekit.AddTrackRequest{ + Cid: fmt.Sprintf("cid%d", i), + Name: fmt.Sprintf("track%d", i), + Type: livekit.TrackType_VIDEO, + }) + require.NotNil(t, ti) + sids = append(sids, livekit.TrackID(ti.Sid)) + } + require.Len(t, p.pendingTracks, cMaxPendingTracks) + + // nothing timed out, so a new publish is rejected + require.Nil(t, p.addPendingTrack(&livekit.AddTrackRequest{Cid: "fresh", Name: "fresh", Type: livekit.TrackType_VIDEO})) + + // one lone track times out, a new publish takes its slot + p.supervisor.SetPublicationTimedOutForTest(sids[0]) + numMessages := sink.WriteMessageCallCount() + require.NotNil(t, p.addPendingTrack(&livekit.AddTrackRequest{Cid: "fresh", Name: "fresh", Type: livekit.TrackType_VIDEO})) + require.Nil(t, p.pendingTracks["cid0"]) + require.NotNil(t, p.pendingTracks["fresh"]) + require.Len(t, p.pendingTracks, cMaxPendingTracks) + require.Empty(t, p.supervisor.TimedOutPublications()) + + require.Equal(t, numMessages+1, sink.WriteMessageCallCount()) + res := sink.WriteMessageArgsForCall(numMessages).(*livekit.SignalResponse) + require.IsType(t, &livekit.SignalResponse_TrackUnpublished{}, res.Message) + require.Equal(t, string(sids[0]), res.Message.(*livekit.SignalResponse_TrackUnpublished).TrackUnpublished.TrackSid) + + // the queued track id times out, both entries are released together + p.supervisor.SetPublicationTimedOutForTest(livekit.TrackID(queued.Sid)) + require.NotNil(t, p.addPendingTrack(&livekit.AddTrackRequest{Cid: "fresh2", Name: "fresh2", Type: livekit.TrackType_VIDEO})) + require.Nil(t, p.pendingTracks["cidq"]) + require.NotNil(t, p.pendingTracks["fresh2"]) + require.Len(t, p.pendingTracks, cMaxPendingTracks) + }) + + t.Run("should not match late media of a released pending track to another pending track", func(t *testing.T) { + p := newParticipantForTestWithOpts("test", &participantOpts{protocolVersion: 7}) + + released := p.addPendingTrack(&livekit.AddTrackRequest{Cid: "cidA", Name: "a", Type: livekit.TrackType_VIDEO}) + require.NotNil(t, released) + // SDP client id that an offer would set, as Firefox does + p.pendingTracks["cidA"].trackInfos[0].Codecs[0].SdpCid = "sdpA" + + p.pendingTracksLock.Lock() + p.releasePendingTrackLocked(livekit.TrackID(released.Sid)) + p.pendingTracksLock.Unlock() + + other := p.addPendingTrack(&livekit.AddTrackRequest{Cid: "cidB", Name: "b", Type: livekit.TrackType_VIDEO}) + require.NotNil(t, other) + + // media for the released track, by signal or SDP client id, does not fall back to the other track by type + for _, cid := range []string{"cidA", "sdpA"} { + _, ti, _, _, _ := p.getPendingTrack(cid, livekit.TrackType_VIDEO, true) + require.Nil(t, ti, cid) + } + // an unknown client id still falls back by type, as before + _, ti, _, _, _ := p.getPendingTrack("unknown", livekit.TrackType_VIDEO, true) + require.NotNil(t, ti) + require.Equal(t, other.Sid, ti.Sid) + + // reusing the released client id publishes on it again + reused := p.addPendingTrack(&livekit.AddTrackRequest{Cid: "cidA", Name: "a2", Type: livekit.TrackType_VIDEO}) + require.NotNil(t, reused) + _, ti, _, _, _ = p.getPendingTrack("cidA", livekit.TrackType_VIDEO, true) + require.NotNil(t, ti) + require.Equal(t, reused.Sid, ti.Sid) + // the old SDP id is cleared too, a reused transceiver keeps it + p.pendingTracksLock.Lock() + require.False(t, p.isReleasedTrackCidLocked("sdpA")) + p.pendingTracksLock.Unlock() + }) + + t.Run("should clear a released track when a new request reuses one of its codec client ids", func(t *testing.T) { + p := newParticipantForTest("test") + + p.pendingTracksLock.Lock() + p.addReleasedTrackCidsLocked([]string{"cidA", "codecA", "sdpA"}) + p.addReleasedTrackCidsLocked([]string{"cidB"}) + p.clearReleasedTrackCidsLocked(&livekit.AddTrackRequest{ + Cid: "cidC", + SimulcastCodecs: []*livekit.SimulcastCodec{{Codec: "vp8", Cid: "codecA"}}, + }) + for _, cid := range []string{"cidA", "codecA", "sdpA"} { + require.False(t, p.isReleasedTrackCidLocked(cid), cid) + } + require.True(t, p.isReleasedTrackCidLocked("cidB")) + p.pendingTracksLock.Unlock() + }) + + t.Run("should bound released track client ids", func(t *testing.T) { + p := newParticipantForTest("test") + + p.pendingTracksLock.Lock() + for i := 0; i <= cMaxReleasedTracks; i++ { + p.addReleasedTrackCidsLocked([]string{fmt.Sprintf("cid%d", i), fmt.Sprintf("sdp%d", i)}) + } + require.Len(t, p.releasedTrackCids, cMaxReleasedTracks) + // the oldest track ages out with all of its ids + require.False(t, p.isReleasedTrackCidLocked("cid0")) + require.False(t, p.isReleasedTrackCidLocked("sdp0")) + require.True(t, p.isReleasedTrackCidLocked(fmt.Sprintf("sdp%d", cMaxReleasedTracks))) + p.pendingTracksLock.Unlock() + }) + t.Run("should not allow adding disallowed sources", func(t *testing.T) { p := newParticipantForTest("test") p.SetPermission(&livekit.ParticipantPermission{ diff --git a/pkg/rtc/participant_sdp.go b/pkg/rtc/participant_sdp.go index 5896e99df..26b338777 100644 --- a/pkg/rtc/participant_sdp.go +++ b/pkg/rtc/participant_sdp.go @@ -50,6 +50,11 @@ func (p *ParticipantImpl) populateSdpCid(parsedOffer *sdp.SessionDescription) ([ } p.pendingTracksLock.Lock() + if p.isReleasedTrackCidLocked(streamID) { + p.pendingTracksLock.Unlock() + continue + } + signalCid, info, _, migrated, _ := p.getPendingTrack(streamID, trackType, false) if migrated { p.pendingTracksLock.Unlock() diff --git a/pkg/rtc/supervisor/participant_supervisor.go b/pkg/rtc/supervisor/participant_supervisor.go index 1dc6b9d32..cb960aadf 100644 --- a/pkg/rtc/supervisor/participant_supervisor.go +++ b/pkg/rtc/supervisor/participant_supervisor.go @@ -108,6 +108,56 @@ func (p *ParticipantSupervisor) AddPublication(trackID livekit.TrackID) { p.lock.Unlock() } +// RemovePublication drops the monitor of a pending publication. A monitor that still +// holds a published track is kept, so a live track never loses its monitor. +func (p *ParticipantSupervisor) RemovePublication(trackID livekit.TrackID) { + p.lock.Lock() + defer p.lock.Unlock() + + pm, ok := p.publications[trackID] + if !ok { + return + } + if pub, ok := pm.opMon.(*PublicationMonitor); ok && pub.hasPublishedTrack() { + p.params.Logger.Warnw( + "supervisor: not removing publication with a published track", nil, + "trackID", trackID, + ) + return + } + delete(p.publications, trackID) +} + +// TimedOutPublications returns the track IDs whose pending publish timed out. +func (p *ParticipantSupervisor) TimedOutPublications() []livekit.TrackID { + p.lock.RLock() + defer p.lock.RUnlock() + + var trackIDs []livekit.TrackID + for trackID, pm := range p.publications { + // pm.err is refreshed once per tick, check again so a track that published since is skipped + if pm.err != nil && pm.opMon.Check() != nil { + trackIDs = append(trackIDs, trackID) + } + } + return trackIDs +} + +// SetPublicationTimedOutForTest puts a publication in the state checkPublications reports +// once the publish wait passes. Only for tests outside this package. +func (p *ParticipantSupervisor) SetPublicationTimedOutForTest(trackID livekit.TrackID) { + p.lock.Lock() + if pm, ok := p.publications[trackID]; ok { + if pub, ok := pm.opMon.(*PublicationMonitor); ok { + pub.lock.Lock() + pub.unmutedAt = time.Now().Add(-2 * publishWaitDuration) + pub.lock.Unlock() + } + pm.err = errPublishTimeout + } + p.lock.Unlock() +} + func (p *ParticipantSupervisor) SetPublicationMute(trackID livekit.TrackID, isMuted bool) { p.lock.Lock() pm, ok := p.publications[trackID] @@ -149,7 +199,7 @@ func (p *ParticipantSupervisor) checkState() { func (p *ParticipantSupervisor) checkPublications() { var erroredPublications []livekit.TrackID var removablePublications []livekit.TrackID - p.lock.RLock() + p.lock.Lock() for trackID, pm := range p.publications { if err := pm.opMon.Check(); err != nil { if pm.err == nil { @@ -167,9 +217,6 @@ func (p *ParticipantSupervisor) checkPublications() { } } } - p.lock.RUnlock() - - p.lock.Lock() for _, trackID := range removablePublications { delete(p.publications, trackID) } diff --git a/pkg/rtc/supervisor/participant_supervisor_test.go b/pkg/rtc/supervisor/participant_supervisor_test.go new file mode 100644 index 000000000..c07cbda41 --- /dev/null +++ b/pkg/rtc/supervisor/participant_supervisor_test.go @@ -0,0 +1,112 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package supervisor + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +// newSupervisorForTest skips the ticker goroutine, tests call checkPublications directly +func newSupervisorForTest() *ParticipantSupervisor { + s := &ParticipantSupervisor{ + params: ParticipantSupervisorParams{Logger: logger.GetLogger()}, + publications: make(map[livekit.TrackID]*trackMonitor), + } + s.SetPublisherPeerConnectionConnected(true) + return s +} + +func addUnmutedPublication(s *ParticipantSupervisor, trackID livekit.TrackID) { + s.AddPublication(trackID) + s.SetPublicationMute(trackID, false) +} + +// agePublication moves the publish window start back past the publish wait +func agePublication(s *ParticipantSupervisor, trackID livekit.TrackID) { + pm := s.publications[trackID].opMon.(*PublicationMonitor) + pm.lock.Lock() + pm.unmutedAt = pm.unmutedAt.Add(-2 * publishWaitDuration) + pm.lock.Unlock() +} + +func TestTimedOutPublications(t *testing.T) { + const trackID = livekit.TrackID("TR_test") + + t.Run("abandoned publication is reported and can be removed", func(t *testing.T) { + s := newSupervisorForTest() + addUnmutedPublication(s, trackID) + + s.checkPublications() + require.Empty(t, s.TimedOutPublications()) + + agePublication(s, trackID) + s.checkPublications() + require.Equal(t, []livekit.TrackID{trackID}, s.TimedOutPublications()) + + s.RemovePublication(trackID) + require.Empty(t, s.TimedOutPublications()) + require.Empty(t, s.publications) + }) + + t.Run("muted publication is not reported", func(t *testing.T) { + s := newSupervisorForTest() + addUnmutedPublication(s, trackID) + s.SetPublicationMute(trackID, true) + + agePublication(s, trackID) + s.checkPublications() + require.Empty(t, s.TimedOutPublications()) + }) + + t.Run("stale timeout is not reported once the track publishes", func(t *testing.T) { + s := newSupervisorForTest() + addUnmutedPublication(s, trackID) + addUnmutedPublication(s, trackID) + + agePublication(s, trackID) + s.checkPublications() + require.Equal(t, []livekit.TrackID{trackID}, s.TimedOutPublications()) + + // media arrives late, before the next tick clears the cached error + s.SetPublishedTrack(trackID, &typesfakes.FakeLocalMediaTrack{}) + require.Empty(t, s.TimedOutPublications()) + }) + + t.Run("publication with a published track is not removed", func(t *testing.T) { + s := newSupervisorForTest() + addUnmutedPublication(s, trackID) + s.SetPublishedTrack(trackID, &typesfakes.FakeLocalMediaTrack{}) + + s.RemovePublication(trackID) + require.Contains(t, s.publications, trackID) + }) + + t.Run("live publication with a queued successor is not reported", func(t *testing.T) { + s := newSupervisorForTest() + addUnmutedPublication(s, trackID) + addUnmutedPublication(s, trackID) // queued successor shares the track id + s.SetPublishedTrack(trackID, &typesfakes.FakeLocalMediaTrack{}) + + agePublication(s, trackID) + s.checkPublications() + require.Empty(t, s.TimedOutPublications()) + }) +} diff --git a/pkg/rtc/supervisor/publication_monitor.go b/pkg/rtc/supervisor/publication_monitor.go index 199d2188f..572c2e2a7 100644 --- a/pkg/rtc/supervisor/publication_monitor.go +++ b/pkg/rtc/supervisor/publication_monitor.go @@ -163,6 +163,13 @@ func (p *PublicationMonitor) Check() error { return nil } +func (p *PublicationMonitor) hasPublishedTrack() bool { + p.lock.RLock() + defer p.lock.RUnlock() + + return p.publishedTrack != nil +} + func (p *PublicationMonitor) IsIdle() bool { p.lock.RLock() defer p.lock.RUnlock() @@ -170,7 +177,10 @@ func (p *PublicationMonitor) IsIdle() bool { return p.desiredPublishes.Len() == 0 && p.publishedTrack == nil } +// update is the only place the front of desiredPublishes moves. +// Any other pop must also re-arm the publish timer. func (p *PublicationMonitor) update() { + popped := false for { var pub *publish if p.desiredPublishes.Len() > 0 { @@ -184,7 +194,14 @@ func (p *PublicationMonitor) update() { if (pub.isStart && p.publishedTrack == nil) || (!pub.isStart && p.publishedTrack != nil) { // put it back as the condition is not satisfied p.desiredPublishes.PushFront(pub) + + // a queued publish reached the front, start its publish window from now + if popped && pub.isStart { + p.unmutedAt = time.Time{} + p.maybeStartMonitor() + } return } + popped = true } } diff --git a/pkg/rtc/supervisor/publication_monitor_test.go b/pkg/rtc/supervisor/publication_monitor_test.go new file mode 100644 index 000000000..721c2da69 --- /dev/null +++ b/pkg/rtc/supervisor/publication_monitor_test.go @@ -0,0 +1,142 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package supervisor + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" + "github.com/livekit/protocol/logger" +) + +func newMonitorForTest() *PublicationMonitor { + return NewPublicationMonitor(PublicationMonitorParams{ + TrackID: "TR_test", + IsPeerConnectionConnected: true, + Logger: logger.GetLogger(), + }) +} + +// addUnmuted adds a pending publish the way the participant does: add, then set mute +func addUnmuted(m *PublicationMonitor) { + m.PostEvent(types.OperationMonitorEventAddPendingPublication, nil) + m.PostEvent(types.OperationMonitorEventSetPublicationMute, false) +} + +// age moves the publish window start back past the publish wait and returns it +func age(m *PublicationMonitor) time.Time { + m.lock.Lock() + defer m.lock.Unlock() + m.unmutedAt = m.unmutedAt.Add(-2 * publishWaitDuration) + return m.unmutedAt +} + +func getUnmutedAt(m *PublicationMonitor) time.Time { + m.lock.RLock() + defer m.lock.RUnlock() + return m.unmutedAt +} + +// Each subtest is one row of the re-arm case table: the event, and whether the +// publish timer restarts because a queued start reached the front. +func TestPublicationMonitorRearm(t *testing.T) { + t.Run("first add: no re-arm", func(t *testing.T) { + m := newMonitorForTest() + m.PostEvent(types.OperationMonitorEventAddPendingPublication, nil) + require.True(t, getUnmutedAt(m).IsZero()) + }) + + t.Run("successor queued before the track publishes: no re-arm", func(t *testing.T) { + m := newMonitorForTest() + addUnmuted(m) + aged := age(m) + + m.PostEvent(types.OperationMonitorEventAddPendingPublication, nil) + require.Equal(t, aged, getUnmutedAt(m)) + require.ErrorIs(t, m.Check(), errPublishTimeout) + }) + + t.Run("track publishes: no re-arm", func(t *testing.T) { + m := newMonitorForTest() + addUnmuted(m) + addUnmuted(m) + aged := age(m) + + m.PostEvent(types.OperationMonitorEventSetPublishedTrack, &typesfakes.FakeLocalMediaTrack{}) + require.Equal(t, aged, getUnmutedAt(m)) + require.NoError(t, m.Check()) // front is the live track's end + }) + + t.Run("successor queued while the track is live: no re-arm", func(t *testing.T) { + m := newMonitorForTest() + addUnmuted(m) + m.PostEvent(types.OperationMonitorEventSetPublishedTrack, &typesfakes.FakeLocalMediaTrack{}) + aged := age(m) + + m.PostEvent(types.OperationMonitorEventAddPendingPublication, nil) + require.Equal(t, aged, getUnmutedAt(m)) + require.NoError(t, m.Check()) + }) + + t.Run("track closes with a successor queued: re-arm", func(t *testing.T) { + m := newMonitorForTest() + addUnmuted(m) + addUnmuted(m) + track := &typesfakes.FakeLocalMediaTrack{} + m.PostEvent(types.OperationMonitorEventSetPublishedTrack, track) + age(m) + + m.PostEvent(types.OperationMonitorEventClearPublishedTrack, track) + require.Less(t, time.Since(getUnmutedAt(m)), publishWaitDuration) + require.NoError(t, m.Check()) + + // the successor still times out once its own window passes without media + age(m) + require.ErrorIs(t, m.Check(), errPublishTimeout) + }) + + t.Run("track closes with a muted successor queued: clock cleared", func(t *testing.T) { + m := newMonitorForTest() + addUnmuted(m) + addUnmuted(m) + track := &typesfakes.FakeLocalMediaTrack{} + m.PostEvent(types.OperationMonitorEventSetPublishedTrack, track) + m.PostEvent(types.OperationMonitorEventSetPublicationMute, true) + age(m) + + m.PostEvent(types.OperationMonitorEventClearPublishedTrack, track) + require.True(t, getUnmutedAt(m).IsZero()) + + // the clock starts on unmute + m.PostEvent(types.OperationMonitorEventSetPublicationMute, false) + require.Less(t, time.Since(getUnmutedAt(m)), publishWaitDuration) + }) + + t.Run("track closes with nothing queued: no re-arm", func(t *testing.T) { + m := newMonitorForTest() + addUnmuted(m) + track := &typesfakes.FakeLocalMediaTrack{} + m.PostEvent(types.OperationMonitorEventSetPublishedTrack, track) + aged := age(m) + + m.PostEvent(types.OperationMonitorEventClearPublishedTrack, track) + require.Equal(t, aged, getUnmutedAt(m)) + require.True(t, m.IsIdle()) + }) +}