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 <noreply@anthropic.com>

* rtc: fix publish_timeout comment to not name an internal constant

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

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:

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{

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

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

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

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

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