From 08770220cdc1350a3c879e7fdb797f58f6f0ee87 Mon Sep 17 00:00:00 2001 From: Mansur Akhmadi Date: Thu, 12 Nov 2026 14:09:51 +0700 Subject: [PATCH] sfu: don't add a closed down track in SetReceiver (#5179) Co-authored-by: Mansur Akhmadi <10739460+mansur-akhmadi@users.noreply.github.com> --- pkg/sfu/downtrack.go | 6 +++ pkg/sfu/downtrack_test.go | 93 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+) create mode 100644 pkg/sfu/downtrack_test.go diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index bd7235b34..766620405 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -809,6 +809,12 @@ func (d *DownTrack) SetReceiver(r TrackReceiver) { d.handleUpstreamCodecChange(r.Codec().MimeType) d.bindLock.Lock() + // a close that ran while bindLock was released may already have removed the down + // track from r, adding it now would leave a closed down track registered there + if d.IsClosed() { + d.bindLock.Unlock() + return + } if err := r.AddDownTrack(d); err != nil { d.params.Logger.Warnw("failed to add downtrack to receiver", err) } diff --git a/pkg/sfu/downtrack_test.go b/pkg/sfu/downtrack_test.go new file mode 100644 index 000000000..87b459340 --- /dev/null +++ b/pkg/sfu/downtrack_test.go @@ -0,0 +1,93 @@ +// 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 sfu_test + +import ( + "testing" + + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/logger" + + "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/sfu/pacer" + "github.com/livekit/livekit-server/pkg/sfu/sfufakes" +) + +func TestDownTrackSetReceiver(t *testing.T) { + newDownTrack := func(t *testing.T, r sfu.TrackReceiver) *sfu.DownTrack { + dt, err := sfu.NewDownTrack(sfu.DownTrackParams{ + Codecs: []webrtc.RTPCodecParameters{vp8CodecParams}, + Receiver: r, + BufferFactory: buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory(), + Pacer: pacer.NewPassThrough(logger.GetLogger(), newNullBWE()), + Logger: logger.GetLogger(), + Listener: &sfufakes.FakeDownTrackListener{}, + StreamID: "fake-stream", + SubID: "fake-sub", + MaxTrack: 500, + }) + require.NoError(t, err) + t.Cleanup(dt.Close) + return dt + } + + t.Run("moves to the new receiver", func(t *testing.T) { + r1 := newFakeTrackReceiver(vp8CodecParams) + dt := newDownTrack(t, r1) + + r2 := newFakeTrackReceiver(vp8CodecParams) + dt.SetReceiver(r2) + + require.Equal(t, sfu.TrackReceiver(r2), dt.Receiver()) + require.Equal(t, 1, r1.DeleteDownTrackCallCount()) + require.Equal(t, 1, r2.AddDownTrackCallCount()) + require.Same(t, dt, r2.AddDownTrackArgsForCall(0)) + }) + + t.Run("closed before the switch", func(t *testing.T) { + r1 := newFakeTrackReceiver(vp8CodecParams) + dt := newDownTrack(t, r1) + dt.Close() + + r2 := newFakeTrackReceiver(vp8CodecParams) + dt.SetReceiver(r2) + + require.Equal(t, sfu.TrackReceiver(r1), dt.Receiver()) + require.Zero(t, r2.AddOnReadyCallCount()) + require.Zero(t, r2.AddDownTrackCallCount()) + }) + + // SetReceiver releases bindLock while it registers for readiness and handles a codec + // change; a close in that window deletes the down track from the new receiver before + // it was added, so it must not be added afterwards + t.Run("closed during the switch", func(t *testing.T) { + r1 := newFakeTrackReceiver(vp8CodecParams) + dt := newDownTrack(t, r1) + + r2 := newFakeTrackReceiver(vp8CodecParams) + r2.AddOnReadyCalls(func(f func()) { + dt.Close() + f() + }) + dt.SetReceiver(r2) + + require.True(t, dt.IsClosed()) + require.Equal(t, 1, r2.DeleteDownTrackCallCount()) + require.Zero(t, r2.AddDownTrackCallCount(), "closed down track was added to the new receiver") + }) +}