Files
livekit/pkg/rtc/participant_sdp.go
T
Raja SubramanianandClaude Opus 5.5 308554daf1 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>
2026-10-13 16:40:46 +05:30

420 lines
11 KiB
Go

// Copyright 2023 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 rtc
import (
"fmt"
"slices"
"strconv"
"strings"
"github.com/pion/sdp/v3"
"github.com/pion/webrtc/v4"
"google.golang.org/protobuf/proto"
"github.com/livekit/livekit-server/pkg/rtc/types"
"github.com/livekit/protocol/codecs/mime"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
lksdp "github.com/livekit/protocol/sdp"
)
func (p *ParticipantImpl) populateSdpCid(parsedOffer *sdp.SessionDescription) ([]*sdp.MediaDescription, []*sdp.MediaDescription) {
processUnmatch := func(unmatches []*sdp.MediaDescription, trackType livekit.TrackType) {
for _, unmatch := range unmatches {
streamID, ok := lksdp.ExtractStreamID(unmatch)
if !ok {
continue
}
sdpCodecs, err := lksdp.CodecsFromMediaDescription(unmatch)
if err != nil || len(sdpCodecs) == 0 {
p.pubLogger.Errorw(
"extract codecs from media section failed", err,
"media", unmatch,
"parsedOffer", parsedOffer,
)
continue
}
p.pendingTracksLock.Lock()
if p.isReleasedTrackCidLocked(streamID) {
p.pendingTracksLock.Unlock()
continue
}
signalCid, info, _, migrated, _ := p.getPendingTrack(streamID, trackType, false)
if migrated {
p.pendingTracksLock.Unlock()
continue
}
if info == nil {
p.pendingTracksLock.Unlock()
// could be already published track and the unmatch could be a back up codec publish
numUnmatchedTracks := 0
var unmatchedTrack types.MediaTrack
var unmatchedSdpMimeType mime.MimeType
found := false
for _, sdpCodec := range sdpCodecs {
sdpMimeType := mime.NormalizeMimeTypeCodec(sdpCodec.Name).ToMimeType()
for _, publishedTrack := range p.GetPublishedTracks() {
if sigCid, sdpCid := publishedTrack.(*MediaTrack).GetCidsForMimeType(sdpMimeType); sigCid != "" && sdpCid == "" {
// a back up codec has a SDP cid match
if sigCid == streamID {
found = true
break
} else {
numUnmatchedTracks++
unmatchedTrack = publishedTrack
unmatchedSdpMimeType = sdpMimeType
}
}
}
if found {
break
}
}
if !found && unmatchedTrack != nil {
if numUnmatchedTracks != 1 {
p.pubLogger.Warnw(
"too many unmatched tracks", nil,
"media", unmatch,
"parsedOffer", parsedOffer,
)
}
unmatchedTrack.(*MediaTrack).UpdateCodecSdpCid(unmatchedSdpMimeType, streamID)
p.pubLogger.Debugw(
"published track SDP cid updated",
"trackID", unmatchedTrack.ID(),
"track", logger.Proto(unmatchedTrack.ToProto()),
)
}
continue
}
if len(info.Codecs) == 0 {
p.pendingTracksLock.Unlock()
p.pubLogger.Warnw(
"track without codecs", nil,
"trackID", info.Sid,
"pendingTrack", p.pendingTracks[signalCid],
"media", unmatch,
"parsedOffer", parsedOffer,
)
continue
}
found := false
updated := false
for _, sdpCodec := range sdpCodecs {
if mime.NormalizeMimeTypeCodec(sdpCodec.Name) == mime.GetMimeTypeCodec(info.Codecs[0].MimeType) {
// set SdpCid only if different from SignalCid
if streamID != info.Codecs[0].Cid {
info.Codecs[0].SdpCid = streamID
updated = true
}
found = true
break
}
if found {
break
}
}
if !found {
// not using SimulcastCodec, i. e. mime type not available till track publish
if len(info.Codecs) == 1 {
// set SdpCid only if different from SignalCid
if streamID != info.Codecs[0].Cid {
info.Codecs[0].SdpCid = streamID
updated = true
}
}
}
if updated {
p.pendingTracks[signalCid].trackInfos[0] = proto.CloneOf(info)
p.pubLogger.Debugw(
"pending track SDP cid updated",
"signalCid", signalCid,
"trackID", info.Sid,
"pendingTrack", p.pendingTracks[signalCid],
)
}
p.pendingTracksLock.Unlock()
}
}
unmatchAudios, err := p.TransportManager.GetUnmatchMediaForOffer(parsedOffer, "audio")
if err != nil {
p.pubLogger.Warnw("could not get unmatch audios", err)
return nil, nil
}
unmatchVideos, err := p.TransportManager.GetUnmatchMediaForOffer(parsedOffer, "video")
if err != nil {
p.pubLogger.Warnw("could not get unmatch videos", err)
return nil, nil
}
processUnmatch(unmatchAudios, livekit.TrackType_AUDIO)
processUnmatch(unmatchVideos, livekit.TrackType_VIDEO)
return unmatchAudios, unmatchVideos
}
func (p *ParticipantImpl) setCodecPreferencesForPublisher(
parsedOffer *sdp.SessionDescription,
unmatchAudios []*sdp.MediaDescription,
unmatchVideos []*sdp.MediaDescription,
) {
unprocessedUnmatchAudios := p.setCodecPreferencesForPublisherMedia(
parsedOffer,
unmatchAudios,
livekit.TrackType_AUDIO,
)
p.setCodecPreferencesOpusRedForPublisher(parsedOffer, unprocessedUnmatchAudios)
_ = p.setCodecPreferencesForPublisherMedia(
parsedOffer,
unmatchVideos,
livekit.TrackType_VIDEO,
)
}
func (p *ParticipantImpl) setCodecPreferencesForPublisherMedia(
parsedOffer *sdp.SessionDescription,
unmatches []*sdp.MediaDescription,
trackType livekit.TrackType,
) []*sdp.MediaDescription {
unprocessed := make([]*sdp.MediaDescription, 0, len(unmatches))
for _, unmatch := range unmatches {
var ti *livekit.TrackInfo
var mimeType string
mid := lksdp.GetMidValue(unmatch)
if mid == "" {
unprocessed = append(unprocessed, unmatch)
continue
}
transceiver := p.TransportManager.GetPublisherRTPTransceiver(mid)
if transceiver == nil {
unprocessed = append(unprocessed, unmatch)
continue
}
streamID, ok := lksdp.ExtractStreamID(unmatch)
if !ok {
unprocessed = append(unprocessed, unmatch)
continue
}
p.pendingTracksLock.RLock()
mt := p.getPublishedTrackBySdpCid(streamID)
if mt != nil {
ti = mt.ToProto()
} else {
_, ti, _, _, _ = p.getPendingTrack(streamID, trackType, false)
}
p.pendingTracksLock.RUnlock()
if ti == nil {
unprocessed = append(unprocessed, unmatch)
continue
}
for _, c := range ti.Codecs {
if c.Cid == streamID || c.SdpCid == streamID {
mimeType = c.MimeType
break
}
}
if mimeType == "" && len(ti.Codecs) > 0 {
mimeType = ti.Codecs[0].MimeType
}
if mimeType == "" {
unprocessed = append(unprocessed, unmatch)
continue
}
configureReceiverCodecs(
transceiver,
mimeType,
p.params.ClientInfo.ComplyWithCodecOrderInSDPAnswer(),
)
}
return unprocessed
}
func (p *ParticipantImpl) setCodecPreferencesOpusRedForPublisher(
parsedOffer *sdp.SessionDescription,
unmatchAudios []*sdp.MediaDescription,
) {
for _, unmatchAudio := range unmatchAudios {
mid := lksdp.GetMidValue(unmatchAudio)
if mid == "" {
continue
}
transceiver := p.TransportManager.GetPublisherRTPTransceiver(mid)
if transceiver == nil {
continue
}
streamID, ok := lksdp.ExtractStreamID(unmatchAudio)
if !ok {
continue
}
p.pendingTracksLock.RLock()
_, ti, _, _, _ := p.getPendingTrack(streamID, livekit.TrackType_AUDIO, false)
p.pendingTracksLock.RUnlock()
if ti == nil {
continue
}
codecs, err := lksdp.CodecsFromMediaDescription(unmatchAudio)
if err != nil {
p.pubLogger.Errorw(
"extract codecs from media section failed", err,
"media", unmatchAudio,
"parsedOffer", parsedOffer,
)
continue
}
var opusPayload uint8
for _, codec := range codecs {
if mime.IsMimeTypeCodecStringOpus(codec.Name) {
opusPayload = codec.PayloadType
break
}
}
if opusPayload == 0 {
continue
}
preferRED := IsRedEnabled(ti)
// if RED is enabled for this track, prefer RED codec in offer
for _, codec := range codecs {
// codec contain opus/red
if preferRED &&
mime.IsMimeTypeCodecStringRED(codec.Name) &&
strings.Contains(codec.Fmtp, strconv.FormatInt(int64(opusPayload), 10)) {
configureReceiverCodecs(transceiver, "audio/red", true)
break
}
}
}
}
// configure publisher answer for audio track's dtx and stereo settings
func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescription) webrtc.SessionDescription {
parsedOffer := p.TransportManager.LastPublisherOfferParsed()
if parsedOffer == nil {
return answer
}
parsedAnswer, err := answer.Unmarshal()
if err != nil {
return answer
}
// payload type lookup scans the whole answer, resolve it once
opusPT, opusPTErr := parsedAnswer.GetPayloadTypeForCodec(sdp.Codec{Name: mime.MimeTypeCodecOpus.String()})
changed := false
for _, m := range parsedAnswer.MediaDescriptions {
switch m.MediaName.Media {
case "audio":
_, ok := m.Attribute(sdp.AttrKeyInactive)
if ok {
continue
}
mid, ok := m.Attribute(sdp.AttrKeyMID)
if !ok {
continue
}
// find track info from offer's stream id
var ti *livekit.TrackInfo
for _, om := range parsedOffer.MediaDescriptions {
_, ok := om.Attribute(sdp.AttrKeyInactive)
if ok {
continue
}
omid, ok := om.Attribute(sdp.AttrKeyMID)
if ok && omid == mid {
streamID, ok := lksdp.ExtractStreamID(om)
if !ok {
continue
}
track, _ := p.getPublishedTrackBySdpCid(streamID).(*MediaTrack)
if track == nil {
p.pendingTracksLock.RLock()
_, ti, _, _, _ = p.getPendingTrack(streamID, livekit.TrackType_AUDIO, false)
p.pendingTracksLock.RUnlock()
} else {
ti = track.ToProto()
}
break
}
}
if ti == nil {
// no need to configure
continue
}
if opusPTErr != nil {
p.pubLogger.Infow("failed to get opus payload type", "error", opusPTErr, "trackID", ti.Sid)
continue
}
for i, attr := range m.Attributes {
if strings.HasPrefix(attr.String(), fmt.Sprintf("fmtp:%d", opusPT)) {
if !slices.Contains(ti.AudioFeatures, livekit.AudioTrackFeature_TF_NO_DTX) {
attr.Value += ";usedtx=1"
} else {
attr.Value = strings.ReplaceAll(attr.Value, ";usedtx=1", "")
}
if slices.Contains(ti.AudioFeatures, livekit.AudioTrackFeature_TF_STEREO) {
attr.Value += ";stereo=1;maxaveragebitrate=510000"
} else {
attr.Value = strings.ReplaceAll(attr.Value, ";stereo=1", "")
}
m.Attributes[i] = attr
changed = true
}
}
default:
continue
}
}
if !changed {
return answer
}
bytes, err := parsedAnswer.Marshal()
if err != nil {
p.pubLogger.Infow("failed to marshal answer", "error", err)
return answer
}
answer.SDP = string(bytes)
return answer
}