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]

Please update the comment to mention what the code does, not reviewer only relevant info about what the code used to do.

[biglittlebigben]

Would using counterfeiter on the Source interface work here instead?