Fix panic and lost EOS when kill races pipeline startup

pkg/media/pipeline.go

@@ -89,9 +89,12 @@ 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,
+		// SendEOS can reach the loop before Run does, so it is built here
+		// rather than on the way into the loop, and is never nil.
+		loop:        glib.NewMainLoop(glib.MainContextDefault(), false),
 		pipelineErr: make(chan error, 1),
 		eos:         newEOSDispatcher(),
 		established: make(map[types.StreamKind]string),
@@ -104,9 +107,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 +214,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 +237,9 @@ func (p *Pipeline) Run(ctx context.Context) error {
 	logger.Infow("starting GST pipeline")
 
 	// run main loop
+	if p.closed.IsBroken() {
+		logger.Warnw("shutdown requested before the main loop started, queued until the loop starts", nil)
+	}
 	p.loop.Run()
 
 	logger.Infow("GST pipeline stopped")
@@ -382,11 +385,23 @@ 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, and is safe to call before Run has started it:
+// the quit is queued on the main context and dispatched when the loop runs.
+// Calling Quit directly is not safe there, because g_main_loop_run sets
+// is_running=TRUE on entry and overwrites it. Assumes this is the only main
+// loop on the default context.
+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,12 +15,19 @@
 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"
 
 	"github.com/livekit/ingress/pkg/types"
+	"github.com/livekit/protocol/logger"
 )
 
 const testSystemMemoryCaps = "video/x-raw,format=NV12,width=1280,height=720,framerate=30/1"
@@ -76,3 +83,274 @@ 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"
+
+// The warning Run emits when the shutdown beat it to the loop.
+const raceWarning = "shutdown requested before the main loop started"
+
+// 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)
+}
+
+// waitRunning blocks until the loop is running, and gives up rather than
+// spinning forever so a pipeline that never starts one fails instead of hanging.
+func waitRunning(t *testing.T, l *glib.MainLoop) {
+	t.Helper()
+
+	deadline := time.Now().Add(5 * time.Second)
+	for !l.IsRunning() {
+		if time.Now().After(deadline) {
+			t.Fatal("loop never started")
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// stubSource is the smallest Source that lets Run reach its main loop. The real
+// ones all want a network peer.
+type stubSource struct{}
+
+func (stubSource) GetSources() []*gst.Element          { return nil }
+func (stubSource) ValidateCaps(*gst.Caps) error        { return nil }
+func (stubSource) Start(context.Context, func()) error { return nil }
+func (stubSource) Close() error                        { return nil }
+
+// runnableTestPipeline adds the two collaborators Run walks through, stubbed to
+// the minimum that lets it both reach and leave the loop.
+func runnableTestPipeline(t *testing.T) *Pipeline {
+	t.Helper()
+
+	p := newTestPipeline(t)
+	p.input = &Input{source: stubSource{}}
+
+	sink := &WebRTCSink{}
+	sink.sdkReady.Break() // Close blocks on this
+	p.sink = sink
+
+	return p
+}
+
+// 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")
+}
+
+// Run itself, on the racing path: the shutdown lands first, and Run has to
+// reach the loop, be stopped by the queued quit, and report the race. Driving
+// the real Run is what makes this a regression test for the loop moving back
+// out of New, which the tests above cannot see. Child process, because the
+// assertion is on log output and the default logger discards.
+func TestRunReportsAndSurvivesTheStartupRace(t *testing.T) {
+	if os.Getenv(eosChildEnv) == "1" {
+		logger.InitFromConfig(&logger.Config{Level: "debug"}, "ingress")
+
+		p := runnableTestPipeline(t)
+
+		p.SendEOS(context.Background())
+		time.Sleep(500 * time.Millisecond)
+		require.False(t, p.loop.IsRunning(), "loop must not have started yet")
+
+		done := make(chan error, 1)
+		go func() { done <- p.Run(context.Background()) }()
+
+		select {
+		case err := <-done:
+			require.NoError(t, err)
+		case <-time.After(10 * time.Second):
+			t.Fatal("Run did not return: the queued quit never reached its loop")
+		}
+		return
+	}
+
+	out, err := runChild(t, "TestRunReportsAndSurvivesTheStartupRace")
+
+	require.NoError(t, err, "child process failed:\n%s", out)
+	require.Contains(t, string(out), raceWarning,
+		"the startup race should be reported:\n%s", out)
+}
+
+// The warning has to be specific to the race or it is noise: a shutdown that
+// arrives once Run is already in the loop is ordinary, and must stay quiet.
+func TestRunIsQuietWhenShutdownArrivesLater(t *testing.T) {
+	if os.Getenv(eosChildEnv) == "1" {
+		logger.InitFromConfig(&logger.Config{Level: "debug"}, "ingress")
+
+		p := runnableTestPipeline(t)
+
+		done := make(chan error, 1)
+		go func() { done <- p.Run(context.Background()) }()
+
+		waitRunning(t, p.loop)
+		p.SendEOS(context.Background())
+
+		select {
+		case err := <-done:
+			require.NoError(t, err)
+		case <-time.After(10 * time.Second):
+			t.Fatal("Run did not return")
+		}
+		return
+	}
+
+	out, err := runChild(t, "TestRunIsQuietWhenShutdownArrivesLater")
+
+	require.NoError(t, err, "child process failed:\n%s", out)
+	require.NotContains(t, string(out), raceWarning,
+		"a shutdown after the loop started is not the race:\n%s", out)
+}
+
+// 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()
+}