livekit
diff --git a/‎pkg/media/pipeline.go‎
Lines changed: 30 additions & 1 deletion b/‎pkg/media/pipeline.go‎
Lines changed: 30 additions & 1 deletion
diff --git a/‎pkg/media/pipeline_test.go‎
Lines changed: 78 additions & 0 deletions b/‎pkg/media/pipeline_test.go‎
Lines changed: 78 additions & 0 deletions
@@ -17,6 +17,7 @@ package media
 import (
 	"context"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -56,6 +57,11 @@ type Pipeline struct {
 	pipelineErr chan error
 
 	eos *eosDispatcher
+
+	trackLock sync.Mutex
+
+	// The caps each stream kind's output was built from.
+	established map[types.StreamKind]string
 }
 
 func New(ctx context.Context, params *params.Params, g *stats.LocalMediaStatsGatherer) (*Pipeline, error) {
@@ -88,6 +94,7 @@ func New(ctx context.Context, params *params.Params, g *stats.LocalMediaStatsGat
 		input:       input,
 		pipelineErr: make(chan error, 1),
 		eos:         newEOSDispatcher(),
+		established: make(map[types.StreamKind]string),
 	}
 
 	input.SetOnEOS(p.eos.Fire)
@@ -137,6 +144,24 @@ func (p *Pipeline) onParamsReady(kind types.StreamKind, gPad *gst.GhostPad) {
 		return
 	}
 
+	newCaps := caps.(*gst.Caps)
+
+	// The audio and video pads notify on separate GStreamer streaming threads,
+	// so this map is shared mutable state and every access takes the lock.
+	p.trackLock.Lock()
+	builtCaps, built := p.established[kind]
+	p.trackLock.Unlock()
+
+	if built {
+		// Rebuilding is not an option -- it adds a second output of the same name
+		// and cannot restructure a published track -- so the session continues on
+		// the output it has. Caps the pipeline genuinely cannot take fail
+		// negotiation downstream and surface on the bus.
+		logger.Warnw("caps renegotiated after the output was built, continuing on the existing output", nil,
+			"kind", kind, "establishedCaps", builtCaps, "newCaps", newCaps.String())
+		return
+	}
+
 	defer func() {
 		if err != nil {
 			p.SetStatus(livekit.IngressState_ENDPOINT_ERROR, err)
@@ -149,7 +174,7 @@ func (p *Pipeline) onParamsReady(kind types.StreamKind, gPad *gst.GhostPad) {
 		p.SendStateUpdate(context.Background())
 	}()
 
-	bin, err := p.sink.AddTrack(kind, caps.(*gst.Caps))
+	bin, err := p.sink.AddTrack(kind, newCaps)
 	if err != nil {
 		return
 	}
@@ -159,6 +184,10 @@ func (p *Pipeline) onParamsReady(kind types.StreamKind, gPad *gst.GhostPad) {
 		return
 	}
 
+	p.trackLock.Lock()
+	p.established[kind] = newCaps.String()
+	p.trackLock.Unlock()
+
 	gPad.AddProbe(gst.PadProbeTypeBlockDownstream, func(pad *gst.Pad, _ *gst.PadProbeInfo) gst.PadProbeReturn {
 		// link
 		if linkReturn := pad.Link(bin.GetStaticPad("sink")); linkReturn != gst.PadLinkOK {
 
@@ -0,0 +1,78 @@
+// 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 media
+
+import (
+	"testing"
+
+	"github.com/go-gst/go-gst/gst"
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/ingress/pkg/types"
+)
+
+const testSystemMemoryCaps = "video/x-raw,format=NV12,width=1280,height=720,framerate=30/1"
+
+// newCapsHoldingGhostPad returns a src ghost pad carrying capsStr, the shape
+// Input surfaces. A pad's caps property is its sticky CAPS event, and an
+// inactive pad is flushing, so the pad is activated before the event is stored.
+func newCapsHoldingGhostPad(t *testing.T, capsStr string) *gst.GhostPad {
+	t.Helper()
+
+	ghost := gst.NewGhostPadNoTarget("video", gst.PadDirectionSource)
+	require.True(t, ghost.SetActive(true))
+	require.Equal(t, gst.FlowOK,
+		ghost.StoreStickyEvent(gst.NewCapsEvent(gst.NewCapsFromString(capsStr))))
+	require.NotNil(t, ghost.GetCurrentCaps())
+
+	return ghost
+}
+
+// The renegotiation this fix exists for. On GStreamer 1.28 the video pad is
+// advertised as memory:GLMemory and then renegotiated to system memory, so
+// onParamsReady runs twice for one pad. The second run must not build another
+// output: the bin name is hardcoded, so gst_bin_add would reject it.
+//
+// sink is deliberately left nil. Building an output dereferences it, so a
+// regression here fails loudly instead of silently rebuilding.
+func TestSecondCapsNotificationDoesNotRebuild(t *testing.T) {
+	gst.Init(nil)
+
+	const builtCaps = "video/x-raw(memory:GLMemory),format=NV12,width=1280,height=720,texture-target=rectangle"
+
+	p := &Pipeline{
+		established: map[types.StreamKind]string{types.Video: builtCaps},
+	}
+
+	p.onParamsReady(types.Video, newCapsHoldingGhostPad(t, testSystemMemoryCaps))
+
+	require.Equal(t, builtCaps, p.established[types.Video],
+		"the established caps must survive a renegotiation")
+	require.Len(t, p.established, 1)
+}
+
+// A notification carrying no caps is not a renegotiation and must not record
+// anything, or the real caps that follow would be treated as the second one.
+func TestCapsNotificationWithoutCapsIsIgnored(t *testing.T) {
+	gst.Init(nil)
+
+	p := &Pipeline{
+		established: make(map[types.StreamKind]string),
+	}
+
+	p.onParamsReady(types.Video, gst.NewGhostPadNoTarget("video", gst.PadDirectionSource))
+
+	require.Empty(t, p.established)
+}