Send EOS for every appsrc linked into the pipeline (CS-1547)

A track subscribed during the CloseWriters() window has its appsrc linked
into the running pipeline by OnTrackAdded(), but never sends EOS: cleanup
guards on playing.IsBroken(), and the PLAYING notification that would break
that fuse is dropped by submitOp() once closing is set. The mixer waits on
that input pad forever and shutdown hangs - "endStreamProcessed not broken
after 3 seconds", then ErrPipelineFrozen 30s later.

Guard cleanup on a new addedToPipeline fuse instead, broken where the appsrc
actually becomes part of the pipeline: after OnTrackAdded() for post-init
subscriptions, and on BuildReady for tracks present at startup. Playing()
breaks it too, since reaching PLAYING implies being linked.

This is cfce5d0 re-landed with the hole that forced its revert closed. There,
addedToPipeline was broken during source init, so a track that EOF'd during
the build phase reached OnEOSSent() -> SendEOS() -> c.p.Stop() with c.p still
nil - a panic, and an unsynchronized read of a field BuildPipeline was still
writing. Now the fuse is never broken before BuildReady closes, which is what
publishes c.p, and onEOSSent() returns early if the pipeline is not built yet.

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

Author: pksgit

pkg/pipeline/controller.go

@@ -489,6 +489,17 @@ func (c *Controller) trackStreamRetry(ctx context.Context, stream *config.Stream
 }
 
 func (c *Controller) onEOSSent() {
+	// A writer can finish before BuildPipeline() has published the pipeline - a
+	// track that EOFs during the build phase. Touching c.p then is both a nil
+	// dereference and an unsynchronized read of a field BuildPipeline is still
+	// writing; closing BuildReady is what publishes it safely. Nothing to stop if
+	// the pipeline does not exist yet.
+	select {
+	case <-c.callbacks.BuildReady:
+	default:
+		return
+	}
+
 	// for video-only track/track composite, EOS might have already
 	// made it through the pipeline by the time endRecording is closed
 	if (c.Passthrough || c.RequestType == types.RequestTypeTrackComposite) && !c.AudioEnabled {

pkg/pipeline/controller_test.go

@@ -22,13 +22,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	"github.com/livekit/egress/pkg/config"
 	"github.com/livekit/egress/pkg/gstreamer"
 	"github.com/livekit/egress/pkg/ipc"
 	"github.com/livekit/egress/pkg/pipeline/source"
+	"github.com/livekit/egress/pkg/types"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/rpc"
 )
@@ -166,3 +168,33 @@ func TestUnsolicitedEOSBeforeSendEOS(t *testing.T) {
 		t.Fatal("output file is empty")
 	}
 }
+
+// TestOnEOSSentBeforePipelineBuilt covers the failure that forced the revert of
+// cfce5d0, the first attempt at fixing CS-1547.
+//
+// A writer whose track EOFs during the build phase runs its cleanup while
+// BuildPipeline() is still running. Sending EOS from there reaches OnEOSSent()
+// -> SendEOS() -> c.p.Stop() with c.p not yet assigned: a nil dereference, and
+// an unsynchronized read of a field BuildPipeline is concurrently writing. A
+// production panic was observed roughly 1ms into the build.
+//
+// The config below is the one combination that makes onEOSSent() forward to
+// SendEOS(): passthrough / track composite with no audio.
+func TestOnEOSSentBeforePipelineBuilt(t *testing.T) {
+	c := &Controller{
+		PipelineConfig: &config.PipelineConfig{
+			RequestType: types.RequestTypeTrackComposite,
+			Passthrough: true,
+			AudioConfig: config.AudioConfig{AudioEnabled: false},
+		},
+		// BuildReady still open and p still nil: BuildPipeline has not published
+		// the pipeline yet.
+		callbacks: &gstreamer.Callbacks{BuildReady: make(chan struct{})},
+	}
+
+	require.NotPanics(t, c.onEOSSent,
+		"cleanup running during the build phase must not dereference the pipeline")
+	require.False(t, c.eosSent.IsBroken(),
+		"onEOSSent must not start the EOS sequence before the pipeline exists")
+	require.Nil(t, c.p, "sanity: the test covers the pre-build window")
+}

pkg/pipeline/source/sdk/appwriter.go

@@ -104,6 +104,7 @@ type AppWriter struct {
 	lastReceived             atomic.Time
 	lastPushed               atomic.Time
 	playing                  core.Fuse
+	addedToPipeline          core.Fuse
 	draining                 core.Fuse
 	unsubscribed             core.Fuse
 	endStreamSignaled        core.Fuse
@@ -260,6 +261,7 @@ func (w *AppWriter) start() {
 		w.logger.Errorw("endStreamProcessed not broken after 3 seconds, bug in the draining logic!", nil,
 			"endStreamSourceProcessed", w.endStreamSourceProcessed.IsBroken(),
 			"playing", w.playing.IsBroken(),
+			"addedToPipeline", w.addedToPipeline.IsBroken(),
 			"active", w.active.Load(),
 			"lastReceived", w.lastReceived.Load(),
 			"lastPushed", w.lastPushed.Load(),
@@ -268,7 +270,22 @@ func (w *AppWriter) start() {
 	}
 
 	// clean up
-	if w.playing.IsBroken() {
+	if w.shouldSendEOS() {
+		if !w.playing.IsBroken() {
+			// The appsrc was linked into the pipeline but its PLAYING notification
+			// never arrived - the CloseWriters race (CS-1547). Under the old
+			// playing.IsBroken() guard this writer would have skipped EOS and hung
+			// the shutdown, so this line means the race happened and was handled.
+			// Expect it to be rare and to coincide with a shutdown; if it is common,
+			// or appears outside one, the PLAYING notification is being lost far more
+			// often than the race alone would explain.
+			w.logger.Warnw("appsrc never reported PLAYING, sending EOS anyway", nil,
+				"active", w.active.Load(),
+				"draining", w.draining.IsBroken(),
+				"unsubscribed", w.unsubscribed.IsBroken(),
+				"lastReceived", w.lastReceived.Load(),
+			)
+		}
 		w.callbacks.OnEOSSent()
 		flow := w.src.EndStream()
 		if flow == gst.FlowFlushing {
@@ -665,9 +682,31 @@ func (w *AppWriter) maybeCheckPipelineLag(pts time.Duration) {
 }
 
 func (w *AppWriter) Playing() {
+	// reaching PLAYING implies the appsrc is linked into the pipeline
+	w.addedToPipeline.Break()
 	w.playing.Break()
 }
 
+// MarkAddedToPipeline signals that the appsrc has been linked into the GStreamer
+// pipeline. From this point the pipeline has an input pad for this writer and
+// will block EOS aggregation until the writer ends its stream.
+func (w *AppWriter) MarkAddedToPipeline() {
+	w.addedToPipeline.Break()
+}
+
+// shouldSendEOS reports whether cleanup must push EOS into this writer's appsrc.
+//
+// The condition that matters is whether the appsrc was linked into the pipeline,
+// NOT whether we were told it reached PLAYING. GStreamer propagates the state
+// change to a newly added element asynchronously, and the resulting notification
+// is dropped outright once CloseWriters() sets closing (submitOp returns early,
+// and the worker may already be gone). A writer added moments before shutdown
+// would therefore never send EOS, and the mixer would wait on its pad forever -
+// the frozen pipeline in CS-1547.
+func (w *AppWriter) shouldSendEOS() bool {
+	return w.addedToPipeline.IsBroken()
+}
+
 // Drain blocks until finished
 func (w *AppWriter) Drain(force bool) {
 	w.draining.Once(func() {

pkg/pipeline/source/sdk/appwriter_test.go

@@ -0,0 +1,243 @@
+// 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 sdk
+
+import (
+	"testing"
+	"time"
+
+	"github.com/go-gst/go-gst/gst"
+	"github.com/go-gst/go-gst/gst/app"
+	"github.com/stretchr/testify/require"
+)
+
+// TestShouldSendEOS pins the cleanup decision in AppWriter.start().
+//
+// CS-1547: the guard used to be playing.IsBroken(), which asks "did GStreamer's
+// PLAYING notification reach us?" rather than "is this appsrc linked into the
+// pipeline?". Those differ during shutdown, because CloseWriters() sets closing
+// before GStreamer finishes the async state change and submitOp() then drops the
+// notification - permanently. The middle case below is the frozen pipeline.
+func TestShouldSendEOS(t *testing.T) {
+	t.Run("never added to the pipeline", func(t *testing.T) {
+		w := &AppWriter{}
+
+		require.False(t, w.shouldSendEOS(),
+			"no appsrc in the pipeline means nothing downstream is waiting for EOS")
+	})
+
+	t.Run("added to the pipeline but PLAYING never delivered", func(t *testing.T) {
+		w := &AppWriter{}
+		w.MarkAddedToPipeline()
+
+		require.False(t, w.playing.IsBroken(),
+			"precondition: this is the race - the PLAYING notification was dropped")
+		require.True(t, w.shouldSendEOS(),
+			"CS-1547: the appsrc is linked into the pipeline, so cleanup must send EOS "+
+				"even though we were never told it reached PLAYING")
+
+		// this state - owing EOS without ever having been told PLAYING - is exactly
+		// what AppWriter.start() logs as "appsrc never reported PLAYING, sending EOS
+		// anyway", the signal that the race occurred and was handled
+	})
+
+	t.Run("PLAYING implies added to the pipeline", func(t *testing.T) {
+		w := &AppWriter{}
+		w.Playing()
+
+		require.True(t, w.addedToPipeline.IsBroken(),
+			"an appsrc cannot reach PLAYING without being linked into the pipeline")
+		require.True(t, w.shouldSendEOS(),
+			"the normal path must keep sending EOS exactly as before")
+	})
+}
+
+const (
+	testAudioCaps     = "audio/x-raw,format=S16LE,layout=interleaved,rate=48000,channels=1"
+	testFrameSize     = 1920 // 20ms of 48kHz mono S16LE
+	testFrameDuration = 20 * time.Millisecond
+	eosWaitTimeout    = 5 * time.Second
+)
+
+type mixerTestPipeline struct {
+	pipeline *gst.Pipeline
+	mixer    *gst.Element
+}
+
+func newMixerTestPipeline(t *testing.T) *mixerTestPipeline {
+	t.Helper()
+	gst.Init(nil)
+
+	pipeline, err := gst.NewPipeline("cs-1547")
+	require.NoError(t, err)
+
+	mixer, err := gst.NewElement("audiomixer")
+	require.NoError(t, err)
+	sink, err := gst.NewElement("fakesink")
+	require.NoError(t, err)
+	require.NoError(t, sink.SetProperty("sync", false))
+
+	require.NoError(t, pipeline.AddMany(mixer, sink))
+	require.NoError(t, mixer.Link(sink))
+
+	p := &mixerTestPipeline{pipeline: pipeline, mixer: mixer}
+	t.Cleanup(func() { _ = pipeline.SetState(gst.StateNull) })
+	return p
+}
+
+// addTrack links a new appsrc into the pipeline, exactly like OnTrackAdded().
+func (p *mixerTestPipeline) addTrack(t *testing.T, name string) *app.Source {
+	t.Helper()
+
+	elem, err := gst.NewElementWithName("appsrc", "app_"+name)
+	require.NoError(t, err)
+	conv, err := gst.NewElement("audioconvert")
+	require.NoError(t, err)
+
+	src := app.SrcFromElement(elem)
+	src.SetCaps(gst.NewCapsFromString(testAudioCaps))
+	src.SetArg("format", "time")
+	require.NoError(t, elem.SetProperty("is-live", false))
+
+	require.NoError(t, p.pipeline.AddMany(elem, conv))
+	require.NoError(t, gst.ElementLinkMany(elem, conv, p.mixer))
+
+	// dynamic add: bring the new branch up to the pipeline's state
+	require.True(t, elem.SyncStateWithParent())
+	require.True(t, conv.SyncStateWithParent())
+
+	return src
+}
+
+func pushSilence(t *testing.T, src *app.Source, frames int, startPTS time.Duration) time.Duration {
+	t.Helper()
+
+	pts := startPTS
+	silence := make([]byte, testFrameSize)
+	for i := 0; i < frames; i++ {
+		b := gst.NewBufferFromBytes(silence)
+		b.SetPresentationTimestamp(gst.ClockTime(uint64(pts)))
+		b.SetDuration(gst.ClockTime(uint64(testFrameDuration)))
+		require.Equal(t, gst.FlowOK, src.PushBuffer(b))
+		pts += testFrameDuration
+	}
+	return pts
+}
+
+// waitForEOS returns true if the pipeline reached EOS within eosWaitTimeout.
+func (p *mixerTestPipeline) waitForEOS(t *testing.T) bool {
+	t.Helper()
+
+	bus := p.pipeline.GetPipelineBus()
+	deadline := time.Now().Add(eosWaitTimeout)
+	for time.Now().Before(deadline) {
+		msg := bus.TimedPopFiltered(gst.ClockTime(uint64(500*time.Millisecond)), gst.MessageEOS|gst.MessageError)
+		if msg == nil {
+			continue
+		}
+		switch msg.Type() {
+		case gst.MessageEOS:
+			return true
+		case gst.MessageError:
+			t.Fatalf("pipeline error: %s", msg.String())
+		}
+	}
+	return false
+}
+
+// TestMissingEOSFreezesPipeline: the late writer never sends EOS
+// (cleanup guard skipped), so the mixer never forwards EOS and shutdown hangs.
+func TestMissingEOSFreezesPipeline(t *testing.T) {
+	p := newMixerTestPipeline(t)
+
+	srcA := p.addTrack(t, "A")
+	require.NoError(t, p.pipeline.SetState(gst.StatePlaying))
+	pts := pushSilence(t, srcA, 10, 0)
+
+	// a subscription lands inside the CloseWriters() window: the appsrc is
+	// linked into the running pipeline
+	srcB := p.addTrack(t, "B")
+	pushSilence(t, srcB, 2, pts)
+
+	// shutdown: A drains and sends EOS, B's cleanup skipped EndStream()
+	// because playing.IsBroken() == false
+	require.Equal(t, gst.FlowOK, srcA.EndStream())
+
+	require.False(t, p.waitForEOS(t),
+		"CS-1547: expected the pipeline to hang - it should NOT reach EOS while app_B never sent EOS")
+}
+
+// TestEOSFromAllWritersCompletes is the control: with EndStream() on
+// every appsrc that was linked into the pipeline, EOS propagates immediately.
+func TestEOSFromAllWritersCompletes(t *testing.T) {
+	p := newMixerTestPipeline(t)
+
+	srcA := p.addTrack(t, "A")
+	require.NoError(t, p.pipeline.SetState(gst.StatePlaying))
+	pts := pushSilence(t, srcA, 10, 0)
+
+	srcB := p.addTrack(t, "B")
+	pushSilence(t, srcB, 2, pts)
+
+	require.Equal(t, gst.FlowOK, srcA.EndStream())
+	require.Equal(t, gst.FlowOK, srcB.EndStream())
+
+	require.True(t, p.waitForEOS(t), "pipeline should reach EOS once every appsrc sent EOS")
+}
+
+// TestEOSFromNotYetPlayingAppsrc is the premise the fix rests on: EOS
+// from an appsrc that was linked and asked to start, but has not been reported
+// as PLAYING, still reaches the mixer.
+//
+// This is the state a track is in during the race. AddSourceBin links the new
+// bin and calls SyncStateWithParent (gstreamer/bin.go), so the element is on its
+// way to PLAYING via PAUSED while our Go code is still waiting for a bus message
+// that CloseWriters will cause to be dropped. Sending EOS from there works.
+//
+// Boundary worth knowing: an appsrc left in NULL - never asked to change state -
+// accepts EndStream() with FlowOK but never propagates it, and the pipeline hangs.
+// egress never links a bin without syncing its state; if that ever changes, EOS
+// alone will not be enough to unblock shutdown.
+func TestEOSFromNotYetPlayingAppsrc(t *testing.T) {
+	p := newMixerTestPipeline(t)
+
+	srcA := p.addTrack(t, "A")
+	require.NoError(t, p.pipeline.SetState(gst.StatePlaying))
+	pts := pushSilence(t, srcA, 10, 0)
+
+	// linked and started, but only as far as PAUSED - PLAYING was never confirmed
+	elem, err := gst.NewElementWithName("appsrc", "app_B")
+	require.NoError(t, err)
+	conv, err := gst.NewElement("audioconvert")
+	require.NoError(t, err)
+
+	srcB := app.SrcFromElement(elem)
+	srcB.SetCaps(gst.NewCapsFromString(testAudioCaps))
+	srcB.SetArg("format", "time")
+	require.NoError(t, elem.SetProperty("is-live", false))
+	require.NoError(t, p.pipeline.AddMany(elem, conv))
+	require.NoError(t, gst.ElementLinkMany(elem, conv, p.mixer))
+	require.NoError(t, elem.SetState(gst.StatePaused))
+	require.NoError(t, conv.SetState(gst.StatePaused))
+	require.Equal(t, gst.StatePaused, elem.GetCurrentState(), "precondition: never reported PLAYING")
+
+	pushSilence(t, srcB, 2, pts)
+	require.Equal(t, gst.FlowOK, srcA.EndStream())
+	require.Equal(t, gst.FlowOK, srcB.EndStream())
+
+	require.True(t, p.waitForEOS(t),
+		"EOS from a linked-but-not-yet-PLAYING appsrc must still complete the pipeline - "+
+			"this is what makes addedToPipeline a valid guard")
+}