Fix panic and lost EOS when kill races pipeline startup

[pksgit]

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.

The reported panic

Run created p.loop, so a SendEOS that got there first dereferenced a nil loop:

panic: runtime error: invalid memory address or nil pointer dereference
glib.(*MainLoop).Quit(...)          gmainloop.go:65
media.(*Pipeline).SendEOS.func1.2() pipeline.go:385
created by media.(*Pipeline).SendEOS.func1 in goroutine 51

It is raised on a goroutine SendEOS spawns, where nothing can recover it, so it took the whole handler process down rather than failing one ingress. The if p.loop != nil guard already sitting in the sink's close callback is the fingerprint of someone hitting this earlier on that path.

The failure underneath it

Fixing the nil is not enough. Once the loop exists, a quit issued before Run reaches loop.Run() is still lost, and this half is silent.

g_main_loop_quit has no running check — it unconditionally sets is_running = FALSE. g_main_loop_run then sets TRUE on entry, unconditionally, and only afterwards reads it:

g_atomic_int_set (&loop->is_running, TRUE);       /* gmain.c:4875 */
while (g_atomic_int_get (&loop->is_running))
  g_main_context_iterate_unlocked (...);

So the quit is not rejected, it is overwritten. FALSE is also the loop's initial value, so the flag cannot distinguish "not started yet" from "asked to stop" — the request was never representable. The only reader of that flag is the while inside Run.

Nothing reaps a handler stuck there. 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. The process outlives its ingress, holds the room participant open (neither input.Close nor sink.Close is reached), fails the DeleteIngress, and blocks the instance from draining — for !s.sm.IsIdle() never exits.

Its window spans pipeline.Start and input.Start, so it is considerably wider than the panic window and likely more common than what got reported.

The change

• Build the loop in New so it is never nil.
• quitLoop queues 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 two Quit() calls in messageWatch are untouched: those run on the loop's own thread while it is already spinning, where a direct quit is correct.

Measured on GStreamer 1.28, all four timings:

	timing	loop stopped
A	direct Quit before the loop starts	no
B	IdleAdd(Quit) before the loop starts	yes
C	direct Quit while running	yes
D	IdleAdd(Quit) while running	yes

The change converts A into B. C and D are unaffected, so steady-state shutdown behaviour does not change.

Tests

Five tests, verified to fail without the fix rather than merely to pass with it. Reverting quitLoop to a direct quit fails TestSendEOSBeforeRunIsHonored, TestQuitLoopBeforeRunIsHonored and TestQuitLoopIsSafeConcurrently.

TestDirectQuitBeforeRunIsLost pins the GStreamer behaviour this fix exists for: if a future version honours an early quit, it fails and says the indirection may no longer be needed.

Green under -race and -count=3.

Reviewer notes

• The nil-loop half is not covered by the suite. The tests build their own Pipeline value, so moving the loop back into Run does not fail anything. Covering it needs either Input/WebRTCSink behind interfaces so a test can call Run, or a test that asserts on the source. Neither is in this PR.
• IdleAdd targets the default main context. That coupling is load bearing: a second main loop in this process would share the context, dispatch the source before our loop starts, and lose the quit again — silently. Nothing else in ingress creates a main loop today and the handler builds one pipeline, so this is latent, not live. A private context is not practical here, since gst_bus_add_watch also targets the default one.
• The fallback direct Quit cannot help in the race. IdleAdd fails only on a bad callback type or a failed allocation, so it is a can't-happen branch, and if it ever fired during the window it is exactly the call that does not work. Happy to drop it to log-only.
• Not included: an early bail in Run when EOS was already requested. Correctness does not need it, but without it an ingress killed during startup still connects to the room before tearing down.

The same shape exists in egress at pkg/gstreamer/pipeline.go (Stop quits a loop Run has not started yet), though Stop does OnStop plus a full state change to NULL before quitting, so the window is far narrower there. Egress already builds its loop in the constructor, so it has no nil exposure. Raised separately; not addressed here.

🤖 Generated with Claude Code

[devin-ai-integration]

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no bugs or issues to report.

[milos-lk]

It is raised on a goroutine SendEOS spawns, where nothing can recover it, so it took the whole handler process down rather than failing one ingress.

we launch handler per ingress - handler process going down is actually causing failing egress

[biglittlebigben]

Would using counterfeiter on the Source interface work here instead?

[pksgit]

Thanks I wasn't aware of counterfeiter. It seems this pattern is used in egress and livekit/livekit repos and not in ingress yet. To use counterfeiter for this, the change will be a bit involved, mainly around moving the test to a different package to avoid circular dependency error and then also requires writing testing seams for Pipeline to make its internals accessible in the test. That's a similar arrangement as pkg/sfu/export_test.go which seems to exist for the same reason. Here is a preview of the changes needed https://claude.ai/code/artifact/1f9398cb-033d-44b7-b621-69c0066edeab
Let me know if this makes sense and we still want to do it for consistency purpose and I will go ahead and make this change. Thx

[biglittlebigben]

May not be worth it, but what is the circular dependency in this case?

[milos-lk]

I like the idea but there might be one subtle issue - quitting the loop on the default context requires the loop to be eventually running and that might stall under some conditions. The scenario I am worried about is Kill landing after the pipeline New invocation but before loop.Run(). In that case quitting will be added to the main context from the kill watcher goroutine. In a meanwhile pipeline run goroutine goes through the sequence New() -> Run() -> pipeline.Start() -> input.Start() -> loop.Run(). So if anything stalls before loop.Run() killing will be pending and process still running. The critical part is input.Start() which based on what I can see can block inside RTMPRelaySource on http.DefaultClient.Do until response headers arrive and that might depend on publisher (at least it has context passed for rtmp though - for whip even the context isn't used).

An alternative idea coming to my mind might be having Run to store the cancel first and then check the closed fuse, bailing with a sink only close before starting anything. Since SendEOS breaks the fuse before loading the cancel, a kill always either hits the bail or finds a live cancel that aborts a blocked input.Start. On top of we might add arming a 60s deadline when the kill fires, and if Run still isn't back, dumpping goroutines and exiting the process (just as a safety net).

[milos-lk]

the test does not guarantee the intended ordering. Sleeping 500 ms does not prove the SendEOS goroutine completed BlockSetState and queued the idle callback. Under load, the callback may occur after the loop starts, making the test pass without exercising “quit before Run.”
Can we use maybe something like:

require.Eventually(t, glib.MainContextDefault().Pending, 2*time.Second, 5 * time.Millisecond,
      "quit was never queued")

[milos-lk]

it's harmless but just pointing it out - the warning has check/run race - SendEOS can occur after checking the fuse but before running p.loop.Run()