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