diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index b0c452545..cfc53eaaf 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -147,6 +147,25 @@ func selectAlternativeAudioCodec(enabledCodecs []*livekit.Codec) string { return mime.MimeTypeOpus.String() } +// mergeCodecsByMime returns a union b, deduplicated by mime type. +func mergeCodecsByMime(a, b []*livekit.Codec) []*livekit.Codec { + merged := make([]*livekit.Codec, 0, len(a)+len(b)) + merged = append(merged, a...) + for _, c := range b { + seen := false + for _, existing := range a { + if mime.IsMimeTypeStringEqual(c.Mime, existing.Mime) { + seen = true + break + } + } + if !seen { + merged = append(merged, c) + } + } + return merged +} + func filterCodecs( codecs []webrtc.RTPCodecParameters, enabledCodecs []*livekit.Codec, diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 6c0e34ac0..13b3276ad 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -307,7 +307,8 @@ type TransportParams struct { Twcc *lktwcc.Responder DirectionConfig DirectionConfig CongestionControlConfig config.CongestionControlConfig - EnabledCodecs []*livekit.Codec + EnabledPublishCodecs []*livekit.Codec + EnabledSubscribeCodecs []*livekit.Codec Logger logger.Logger Transport livekit.SignalTarget SimTracks map[uint32]sfuinterceptor.SimulcastTrackInfo @@ -339,7 +340,11 @@ func newPeerConnection( // Some of the browser clients do not handle H.264 High Profile in signalling properly. // They still decode if the actual stream is H.264 High Profile, but do not handle it well in signalling. // So, disable H.264 High Profile for SUBSCRIBER peer connection to ensure it is not offered. - me, err := createMediaEngine(params.EnabledCodecs, directionConfig, params.IsOfferer) + // + // Single-PC mode registers the union of publish and subscribe codecs so subscriptions + // can negotiate subscribe-only codecs; per-direction filtering happens at the transceiver + // level (configureSenderCodecs, restrictReceiverCodecsToPublishList). + me, err := createMediaEngine(mergeCodecsByMime(params.EnabledPublishCodecs, params.EnabledSubscribeCodecs), directionConfig, params.IsOfferer) if err != nil { return nil, nil, nil, err } @@ -2736,6 +2741,8 @@ func (t *PCTransport) createAndSendAnswer() error { t.numRequestSentAudios, t.numRequestSentVideos = 0, 0 t.lock.Unlock() + t.restrictReceiverCodecsToPublishList() + answer, err := t.pc.CreateAnswer(nil) if err != nil { if errors.Is(err, webrtc.ErrConnectionClosed) { @@ -3122,6 +3129,34 @@ func configureSenderCodecs( tr.SetCodecPreferences(filteredCodecs) } +// restrictReceiverCodecsToPublishList narrows recv-side transceiver codec +// preferences to the publish list, so the answer doesn't advertise +// subscribe-only codecs as receivable. No-op in dual-PC mode. +func (t *PCTransport) restrictReceiverCodecsToPublishList() { + for _, tr := range t.pc.GetTransceivers() { + if tr.Direction() != webrtc.RTPTransceiverDirectionRecvonly && + tr.Direction() != webrtc.RTPTransceiverDirectionSendrecv { + continue + } + receiver := tr.Receiver() + if receiver == nil { + continue + } + filtered := filterCodecs( + receiver.GetParameters().Codecs, + t.params.EnabledPublishCodecs, + t.params.DirectionConfig.RTCPFeedback, + false, + ) + if len(filtered) == 0 { + continue + } + if err := tr.SetCodecPreferences(filtered); err != nil { + t.params.Logger.Warnw("failed to set recv codec preferences", err, "mid", tr.Mid()) + } + } +} + func configureReceiverCodecs( tr *webrtc.RTPTransceiver, preferredMimeType string, diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index c48a2e4d2..ab328d7c5 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -400,7 +400,7 @@ func TestNegotiationFailed(t *testing.T) { func TestFilteringCandidates(t *testing.T) { params := TransportParams{ Config: &WebRTCConfig{}, - EnabledCodecs: []*livekit.Codec{ + EnabledPublishCodecs: []*livekit.Codec{ {Mime: mime.MimeTypeOpus.String()}, {Mime: mime.MimeTypeVP8.String()}, {Mime: mime.MimeTypeH264.String()}, @@ -635,3 +635,145 @@ func TestConfigureAudioTransceiver(t *testing.T) { }) } } + +// In single-PC mode the publisher PC carries both publish and subscribe +// directions. If the MediaEngine were built only from the publish codec list, +// the SDP offer would not advertise some codecs in the m-section even though +// the subscribe direction is supposed to support it. This regression-tests +// the union behavior in newPeerConnection: build the MediaEngine from publish + +// subscribe codec lists. +func TestSinglePCMediaEngineUnionsCodecs(t *testing.T) { + videoMSectionCodecs := func(transport *PCTransport) []string { + _, err := transport.pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo) + require.NoError(t, err) + offer, err := transport.pc.CreateOffer(nil) + require.NoError(t, err) + parsed, err := offer.Unmarshal() + require.NoError(t, err) + var rtpmaps []string + for _, m := range parsed.MediaDescriptions { + if m.MediaName.Media != "video" { + continue + } + for _, a := range m.Attributes { + if a.Key == "rtpmap" { + rtpmaps = append(rtpmaps, a.Value) + } + } + } + return rtpmaps + } + + sdpHasH264 := func(rtpmaps []string) bool { + for _, r := range rtpmaps { + if strings.Contains(r, "H264/") { + return true + } + } + return false + } + + publishOnly := []*livekit.Codec{ + {Mime: mime.MimeTypeOpus.String()}, + {Mime: mime.MimeTypeVP8.String()}, + } + subscribeOnly := []*livekit.Codec{ + {Mime: mime.MimeTypeOpus.String()}, + {Mime: mime.MimeTypeVP8.String()}, + {Mime: mime.MimeTypeH264.String()}, + } + + // Control: only publish codecs set (dual-PC publisher PC). H.264 absent. + dualPC, err := NewPCTransport(TransportParams{ + Config: &WebRTCConfig{}, + EnabledPublishCodecs: publishOnly, + Handler: &transportfakes.FakeHandler{}, + }) + require.NoError(t, err) + require.False(t, sdpHasH264(videoMSectionCodecs(dualPC)), + "dual-PC publisher must not advertise H.264 when it's stripped from the publish list") + + // Single-PC publisher PC: both lists set. H.264 must appear. + singlePC, err := NewPCTransport(TransportParams{ + Config: &WebRTCConfig{}, + EnabledPublishCodecs: publishOnly, + EnabledSubscribeCodecs: subscribeOnly, + IsSendSide: true, + Handler: &transportfakes.FakeHandler{}, + }) + require.NoError(t, err) + require.True(t, sdpHasH264(videoMSectionCodecs(singlePC)), + "single-PC publisher must advertise H.264 from the subscribe list even when it's stripped from the publish list") +} + +// Regression test for restrictReceiverCodecsToPublishList: subscribe-only +// codecs (e.g., H.264) registered for subscriptions must not leak into the +// recv-side m-section of an answer, or the peer could publish them. +func TestSinglePCAnswerStripsSubscribeOnlyCodecsFromRecvSide(t *testing.T) { + publishCodecs := []*livekit.Codec{ + {Mime: mime.MimeTypeOpus.String()}, + {Mime: mime.MimeTypeVP8.String()}, + } + subscribeCodecs := []*livekit.Codec{ + {Mime: mime.MimeTypeOpus.String()}, + {Mime: mime.MimeTypeVP8.String()}, + {Mime: mime.MimeTypeH264.String()}, + } + + handler := &transportfakes.FakeHandler{} + server, err := NewPCTransport(TransportParams{ + Config: &WebRTCConfig{}, + EnabledPublishCodecs: publishCodecs, + EnabledSubscribeCodecs: subscribeCodecs, + IsSendSide: true, + Handler: handler, + }) + require.NoError(t, err) + defer server.Close() + + var clientME webrtc.MediaEngine + require.NoError(t, registerCodecs(&clientME, subscribeCodecs, RTCPFeedbackConfig{}, false)) + client, err := webrtc.NewAPI(webrtc.WithMediaEngine(&clientME)).NewPeerConnection(webrtc.Configuration{}) + require.NoError(t, err) + defer client.Close() + + _, err = client.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionSendonly, + }) + require.NoError(t, err) + offer, err := client.CreateOffer(nil) + require.NoError(t, err) + require.Contains(t, offer.SDP, "H264/", "offer must advertise H.264") + require.NoError(t, client.SetLocalDescription(offer)) + + var answer atomic.Pointer[webrtc.SessionDescription] + handler.OnAnswerCalls(func(sd webrtc.SessionDescription, _ uint32, _ map[string]string) error { + answer.Store(&sd) + return nil + }) + require.NoError(t, server.HandleRemoteDescription(*client.LocalDescription(), 1)) + + require.Eventually(t, func() bool { + return answer.Load() != nil + }, 5*time.Second, 10*time.Millisecond, "server did not produce answer") + + parsed, err := answer.Load().Unmarshal() + require.NoError(t, err) + + var videoSection *sdp.MediaDescription + for _, m := range parsed.MediaDescriptions { + if m.MediaName.Media == "video" { + videoSection = m + break + } + } + require.NotNil(t, videoSection, "answer missing video m-section") + + for _, a := range videoSection.Attributes { + if a.Key != "rtpmap" { + continue + } + require.NotContains(t, a.Value, "H264/", + "answer must not advertise H.264 in recv-side m-section: %s", a.Value) + } +} diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 25cacf8fe..528cd72c7 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -140,13 +140,21 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro t.mediaLossProxy.OnMediaLossUpdate(t.onMediaLossUpdate) lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER) + // In single-PC mode the one PC carries both directions, so it needs both + // codec lists registered on its MediaEngine. In dual-PC mode the publisher + // PC is recvonly, so subscribe codecs are left nil. + var publisherSubscribeCodecs []*livekit.Codec + if params.UseSinglePeerConnection || params.UseOneShotSignallingMode { + publisherSubscribeCodecs = params.EnabledSubscribeCodecs + } publisher, err := NewPCTransport(TransportParams{ ProtocolVersion: params.ProtocolVersion, Config: params.Config, Twcc: params.Twcc, DirectionConfig: params.Config.Publisher, CongestionControlConfig: params.CongestionControlConfig, - EnabledCodecs: params.EnabledPublishCodecs, + EnabledPublishCodecs: params.EnabledPublishCodecs, + EnabledSubscribeCodecs: publisherSubscribeCodecs, Logger: lgr, SimTracks: params.SimTracks, ClientInfo: params.ClientInfo, @@ -173,7 +181,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro Config: params.Config, DirectionConfig: params.Config.Subscriber, CongestionControlConfig: params.CongestionControlConfig, - EnabledCodecs: params.EnabledSubscribeCodecs, + EnabledSubscribeCodecs: params.EnabledSubscribeCodecs, Logger: lgr, ClientInfo: params.ClientInfo, IsOfferer: true, diff --git a/test/client/client.go b/test/client/client.go index 10362dece..f44c884d8 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -304,7 +304,8 @@ func (c *RTCClient) createTransport(rtcconf webrtc.Configuration) error { c.publisher, err = rtc.NewPCTransport(rtc.TransportParams{ Config: &conf, DirectionConfig: conf.Subscriber, - EnabledCodecs: c.enabledCodecs, + EnabledPublishCodecs: c.enabledCodecs, + EnabledSubscribeCodecs: c.enabledCodecs, IsOfferer: true, IsSendSide: true, Handler: publisherHandler, @@ -389,7 +390,8 @@ func (c *RTCClient) createTransport(rtcconf webrtc.Configuration) error { c.subscriber, err = rtc.NewPCTransport(rtc.TransportParams{ Config: &conf, DirectionConfig: conf.Publisher, - EnabledCodecs: c.enabledCodecs, + EnabledPublishCodecs: c.enabledCodecs, + EnabledSubscribeCodecs: c.enabledCodecs, Handler: subscriberHandler, DatachannelMaxReceiverBufferSize: 1500, DatachannelSlowThreshold: 1024 * 1024 * 1024,