Fix panic and lost EOS when kill races pipeline startup

Handler.HandleIngress starts Pipeline.Run on a goroutine and calls
SendEOS from its kill watcher, so a DeleteIngress arriving during
startup runs both against the same Pipeline. Two things went wrong:

Run created p.loop, so a SendEOS that got there first dereferenced a
nil loop. That panic is raised on a goroutine SendEOS spawns, where
nothing can recover it, so it took the whole handler process down.

Once the loop existed, a quit issued before Run reached loop.Run() was
still lost: g_main_loop_run sets is_running=TRUE on entry, overwriting
the FALSE that g_main_loop_quit wrote. The flag cannot distinguish "not
started yet" from "asked to stop", so the request was never
representable rather than discarded. The loop then ran with nobody left
to stop it. Nothing reaps a handler in that state -- the process manager
only cleans up once cmd.Run returns, its SIGKILL backstop is guarded on
a fuse already broken by then, and the handler traps the SIGINT that
killAll sends -- so the process outlives its ingress, holds the room
participant open, and blocks the instance from draining. That failure
is silent, and its window spans pipeline.Start and input.Start, so it
is likely more common than the panic that got reported.

Build the loop in New so it is never nil, and queue quits as idle
sources, which live on the context rather than in a flag Run overwrites
and are dispatched as soon as the loop starts.

The fix also makes the race invisible: a kill during startup now tears
down cleanly and leaves no trace. Run warns when it finds the fuse
already broken on its way to the loop, so the race stays observable
after it stops being fatal.

Reverting the idle source fails three of the new tests. Two things are
not covered: the nil loop half, since the tests build their own
Pipeline value, and the warning, which sits past the point in Run that
a test without a real Input can reach.

Fixes CS-1376

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

Author: pksgit

pkg/media/pipeline.go

@@ -89,9 +89,13 @@ func New(ctx context.Context, params *params.Params, g *stats.LocalMediaStatsGat
 	}
 
 	p := &Pipeline{
-		Params:      params,
-		pipeline:    pipeline,
-		input:       input,
+		Params:   params,
+		pipeline: pipeline,
+		input:    input,
+		// Built here rather than in Run so that it is never nil: SendEOS can
+		// run before Run starts, and used to dereference a nil loop and take
+		// the handler process down.
+		loop:        glib.NewMainLoop(glib.MainContextDefault(), false),
 		pipelineErr: make(chan error, 1),
 		eos:         newEOSDispatcher(),
 		established: make(map[types.StreamKind]string),
@@ -104,9 +108,7 @@ func New(ctx context.Context, params *params.Params, g *stats.LocalMediaStatsGat
 			(*cancel)()
 		}
 
-		if p.loop != nil {
-			p.loop.Quit()
-		}
+		p.quitLoop()
 	}, g, p.eos)
 	if err != nil {
 		return nil, err
@@ -213,7 +215,6 @@ func (p *Pipeline) Run(ctx context.Context) error {
 	var err error
 
 	// add watch
-	p.loop = glib.NewMainLoop(glib.MainContextDefault(), false)
 	p.pipeline.GetPipelineBus().AddWatch(p.messageWatch)
 
 	// set state to playing (this does not start the pipeline)
@@ -237,6 +238,12 @@ func (p *Pipeline) Run(ctx context.Context) error {
 	logger.Infow("starting GST pipeline")
 
 	// run main loop
+	if p.closed.IsBroken() {
+		// A kill beat Run to the loop. Harmless, since SendEOS queued its quit
+		// as an idle source that the loop dispatches as soon as it starts, but
+		// it means this ingress was torn down before it ever ran.
+		logger.Warnw("shutdown requested before the main loop started", nil)
+	}
 	p.loop.Run()
 
 	logger.Infow("GST pipeline stopped")
@@ -382,11 +389,29 @@ func (p *Pipeline) SendEOS(ctx context.Context) {
 				logger.Errorw("pipeline frozen", psrpc.NewErrorf(psrpc.Internal, "pipeline frozen"))
 			}
 
-			p.loop.Quit()
+			p.quitLoop()
 		}()
 	})
 }
 
+// quitLoop stops the main loop from a goroutine that is not the loop's own.
+//
+// The quit is queued as an idle source rather than issued directly:
+// g_main_loop_run sets is_running=TRUE on entry, so a direct Quit landing
+// before Run is overwritten and lost, leaving the loop spinning with nobody to
+// stop it. A queued source lives on the context instead, and is dispatched once
+// the loop starts. This assumes nothing else runs a main loop on the default
+// context, which would dispatch the source early and lose the quit again.
+//
+// IdleAdd fails only on a bad callback type or a failed allocation, never
+// because no loop is running, so the direct Quit is a can't-happen fallback.
+func (p *Pipeline) quitLoop() {
+	if _, err := glib.IdleAdd(p.loop.Quit); err != nil {
+		logger.Errorw("failed to schedule loop quit, quitting directly", err)
+		p.loop.Quit()
+	}
+}
+
 func (p *Pipeline) GetGstPipelineDebugDot() string {
 	return p.pipeline.DebugBinToDotData(gst.DebugGraphShowAll)
 }

pkg/media/pipeline_test.go

@@ -15,8 +15,14 @@
 package media
 
 import (
+	"context"
+	"os"
+	"os/exec"
+	"sync"
 	"testing"
+	"time"
 
+	"github.com/go-gst/go-glib/glib"
 	"github.com/go-gst/go-gst/gst"
 	"github.com/stretchr/testify/require"
 
@@ -76,3 +82,169 @@ func TestCapsNotificationWithoutCapsIsIgnored(t *testing.T) {
 
 	require.Empty(t, p.established)
 }
+
+// CS-1376. Handler.HandleIngress starts Run on a goroutine and calls SendEOS
+// from its kill watcher, so a DeleteIngress arriving during startup runs both
+// against the same Pipeline. Two failures came out of that:
+//
+//   - Run used to create p.loop, so SendEOS could dereference a nil loop and
+//     take the handler process down with it.
+//   - Even once the loop existed, a direct Quit issued before Run reached
+//     loop.Run() was discarded, and the handler hung on a loop nobody would
+//     stop again. That one is silent, which makes it the worse of the two.
+//
+// New now builds the loop, and quitLoop queues the quit as an idle source so it
+// survives until the loop starts.
+
+const eosChildEnv = "INGRESS_EOS_RACE_CHILD"
+
+// newTestPipeline is the state New leaves a Pipeline in, minus the parts that
+// need a room connection. The loop matters here: it is what New now owns.
+func newTestPipeline(t *testing.T) *Pipeline {
+	t.Helper()
+
+	gst.Init(nil)
+
+	pipeline, err := gst.NewPipeline("pipeline")
+	require.NoError(t, err)
+
+	return &Pipeline{
+		pipeline:    pipeline,
+		loop:        glib.NewMainLoop(glib.MainContextDefault(), false),
+		pipelineErr: make(chan error, 1),
+		eos:         newEOSDispatcher(),
+	}
+}
+
+// startLoop starts the loop and returns a channel closed once it stops. The
+// channel is returned rather than waited on here so that a test can observe the
+// same run twice: a loop that is already running must not be started again.
+func startLoop(p *Pipeline) <-chan struct{} {
+	returned := make(chan struct{})
+	go func() {
+		p.loop.Run()
+		close(returned)
+	}()
+
+	return returned
+}
+
+func stoppedWithin(returned <-chan struct{}, timeout time.Duration) bool {
+	select {
+	case <-returned:
+		return true
+	case <-time.After(timeout):
+		return false
+	}
+}
+
+// runLoop starts the loop and reports whether it stopped within the timeout.
+func runLoop(p *Pipeline, timeout time.Duration) bool {
+	return stoppedWithin(startLoop(p), timeout)
+}
+
+// The regression for both halves of the bug, on the real SendEOS path. Runs in
+// a child process because the nil dereference happened on a goroutine SendEOS
+// spawns, and no parent can recover a panic raised on another goroutine: before
+// the fix the process died outright rather than failing an assertion.
+func TestSendEOSBeforeRunIsHonored(t *testing.T) {
+	if os.Getenv(eosChildEnv) == "1" {
+		p := newTestPipeline(t)
+
+		// The race: EOS lands before Run has started the loop.
+		p.SendEOS(context.Background())
+
+		// SendEOS issues its quit from a goroutine, once the pipeline has gone
+		// to NULL. Wait for that to have happened before starting the loop, so
+		// the quit is reliably the early one this test is about; without the
+		// wait the loop is often already running by then and the race is not
+		// exercised at all. Going to NULL on an empty pipeline takes
+		// microseconds, so this is a wide margin, not a tuned one.
+		time.Sleep(500 * time.Millisecond)
+		require.False(t, p.loop.IsRunning(), "loop must not have started yet")
+
+		require.True(t, runLoop(p, 10*time.Second),
+			"loop.Run did not return: the queued EOS was lost")
+		return
+	}
+
+	out, err := runChild(t, "TestSendEOSBeforeRunIsHonored")
+
+	require.NoError(t, err, "child process failed:\n%s", out)
+	require.NotContains(t, string(out), "panic:", "child panicked:\n%s", out)
+}
+
+// The queuing on its own, without SendEOS's timers deciding when the quit is
+// issued. This pins the ordering the fix depends on: quit first, loop second.
+func TestQuitLoopBeforeRunIsHonored(t *testing.T) {
+	p := newTestPipeline(t)
+
+	p.quitLoop()
+
+	require.True(t, runLoop(p, 5*time.Second),
+		"a quit queued before Run must still stop the loop")
+}
+
+// quitLoop is also reached from the sink's close callback and from SendEOS's
+// timeout goroutine, so it has to tolerate being called more than once and from
+// several goroutines. Meaningful under -race.
+func TestQuitLoopIsSafeConcurrently(t *testing.T) {
+	p := newTestPipeline(t)
+
+	var wg sync.WaitGroup
+	for range 4 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			p.quitLoop()
+		}()
+	}
+	wg.Wait()
+
+	require.True(t, runLoop(p, 5*time.Second), "loop must still stop")
+}
+
+// SendEOS is fused, but the fuse only stops the body running twice; it does not
+// order SendEOS against Run. Guards that the fused path still stops the loop.
+func TestSendEOSTwiceStillStopsTheLoop(t *testing.T) {
+	p := newTestPipeline(t)
+
+	p.SendEOS(context.Background())
+	p.SendEOS(context.Background())
+
+	require.True(t, runLoop(p, 10*time.Second), "loop must still stop")
+}
+
+// Why quitLoop queues rather than calling Quit directly. g_main_loop_run sets
+// is_running itself, so a direct Quit arriving first is overwritten and lost:
+// the flag records "not running", never "was asked to stop". This is GStreamer
+// behavior rather than ours, so it is pinned here to justify the indirection
+// and to catch it changing under us.
+func TestDirectQuitBeforeRunIsLost(t *testing.T) {
+	p := newTestPipeline(t)
+
+	p.loop.Quit()
+
+	returned := startLoop(p)
+
+	require.False(t, stoppedWithin(returned, 2*time.Second),
+		"a direct quit before Run is expected to be lost; if this now passes, "+
+			"quitLoop's idle source may no longer be needed")
+
+	// Confirms the loop really is running, rather than merely unscheduled.
+	require.True(t, p.loop.IsRunning())
+
+	// Stop the run started above; do not start a second one.
+	p.loop.Quit()
+	require.True(t, stoppedWithin(returned, 5*time.Second),
+		"loop did not stop on the second quit")
+}
+
+func runChild(t *testing.T, testName string) ([]byte, error) {
+	t.Helper()
+
+	cmd := exec.Command(os.Args[0], "-test.run=^"+testName+"$", "-test.v")
+	cmd.Env = append(os.Environ(), eosChildEnv+"=1")
+
+	return cmd.CombinedOutput()
+}