From 5aa3ddcd7b1cf93ceca29f5f9e35ed64137419fd Mon Sep 17 00:00:00 2001 From: Milos Pesic Date: Tue, 28 Jul 2026 00:05:35 +0200 Subject: [PATCH 1/2] Don't declare the pipeline frozen when EOS arrived before it was requested MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Logs are showing recorded a VP8 tracks which sometimes fail with "pipeline frozen", but the pipeline wasn't frozen, it had already finished. When the publisher left the room, their encoder emitted a degenerate final keyframe (8x8), the muxer refused the caps renegotiation, and the appsrc died with not-negotiated, pushing EOS downstream as its last act. That EOS reached the file sink and posted on the bus before the controller's own end path ran. sendEOS() then installed its sink EOS probes after the EOS event had already passed (so they could never fire) and armed the 30-second watchdog after the bus handler had already tried to stop a timer that didn't exist yet (so nothing could ever stop it). The timer fired, declared the pipeline frozen, and a correctly finalized recording was reported as failed. Logs show this class of shutdown is common and always benign: end-of-track placeholder frames (8x8 VP8, 2x2 H264) trigger the same not-negotiated error-EOS a few times a day across unrelated customers, and every sampled instance coincides with the publisher leaving the room, the frozen verdict is just the unlucky ordering of a race the shutdown usually wins. The fix makes sendEOS() respect the eosReceived fuse: if the bus EOS was already processed, the pipeline is already stopping and there is nothing to arm. The watchdog callback also checks the fuse, closing the narrow window where the EOS lands between the guard and the timer assignment; in that case it falls through to the idempotent Stop() instead of failing the egress. A genuinely frozen pipeline — EOS sent but never reaching the sinks, still fails after eosTimeout exactly as before. The new controller test reproduces the scenario end-to-end with a real pipeline (silence generator → opus → ogg file sink): it sends EOS directly through the pipeline, invokes SendEOS the moment the bus EOS is processed, and asserts the egress completes with a finalized file. Without the guard it fails deterministically with the production signature, EGRESS_FAILED (error: "pipeline frozen"). --- pkg/pipeline/controller.go | 10 +- pkg/pipeline/controller_test.go | 168 ++++++++++++++++++++++++++++++++ 2 files changed, 176 insertions(+), 2 deletions(-) create mode 100644 pkg/pipeline/controller_test.go diff --git a/pkg/pipeline/controller.go b/pkg/pipeline/controller.go index a1deaca4..d25301ee 100644 --- a/pkg/pipeline/controller.go +++ b/pkg/pipeline/controller.go @@ -47,11 +47,13 @@ import ( const ( pipelineName = "pipeline" - eosTimeout = time.Second * 30 streamRetryUpdateInterval = time.Minute ) +// var to allow tests to shorten it +var eosTimeout = time.Second * 30 + type Controller struct { *config.PipelineConfig ipcServiceClient ipc.EgressServiceClient @@ -575,6 +577,10 @@ func (c *Controller) SendEOS(ctx context.Context, reason string) { } func (c *Controller) sendEOS() { + if c.eosReceived.IsBroken() { + return + } + for _, sinks := range c.sinks { for _, s := range sinks { s.AddEOSProbe() @@ -587,7 +593,7 @@ func (c *Controller) sendEOS() { switch egressType { case types.EgressTypeFile, types.EgressTypeSegments, types.EgressTypeImages: for _, s := range si { - if !s.EOSReceived() { + if !c.eosReceived.IsBroken() && !s.EOSReceived() { c.OnError(errors.ErrPipelineFrozen) return } diff --git a/pkg/pipeline/controller_test.go b/pkg/pipeline/controller_test.go new file mode 100644 index 00000000..386064c7 --- /dev/null +++ b/pkg/pipeline/controller_test.go @@ -0,0 +1,168 @@ +// 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 pipeline + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "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/protocol/livekit" + "github.com/livekit/protocol/rpc" +) + +type nopIPCClient struct{} + +func (nopIPCClient) HandlerReady(context.Context, *ipc.HandlerReadyRequest, ...grpc.CallOption) (*emptypb.Empty, error) { + return &emptypb.Empty{}, nil +} +func (nopIPCClient) HandlerUpdate(context.Context, *livekit.EgressInfo, ...grpc.CallOption) (*emptypb.Empty, error) { + return &emptypb.Empty{}, nil +} +func (nopIPCClient) HandlerFinished(context.Context, *ipc.HandlerFinishedRequest, ...grpc.CallOption) (*emptypb.Empty, error) { + return &emptypb.Empty{}, nil +} +func (nopIPCClient) ReplayReady(context.Context, *rpc.EgressReadyRequest, ...grpc.CallOption) (*rpc.EgressReadyResponse, error) { + return &rpc.EgressReadyResponse{}, nil +} +func (nopIPCClient) StorageEvent(context.Context, *ipc.StorageEventRequest, ...grpc.CallOption) (*emptypb.Empty, error) { + return &emptypb.Empty{}, nil +} + +// staticSource satisfies source.Source without connecting to a room. The +// pipeline input is the audio bin's silence generator (audiotestsrc). +type staticSource struct { + startedAt int64 +} + +func (s *staticSource) StartRecording() <-chan struct{} { return nil } +func (s *staticSource) EndRecording() <-chan struct{} { return nil } +func (s *staticSource) GetStartedAt() int64 { return s.startedAt } +func (s *staticSource) GetEndedAt() int64 { return time.Now().UnixNano() } +func (s *staticSource) Close() {} + +// TestUnsolicitedEOSBeforeSendEOS reproduces the shutdown race where EOS +// reaches the sinks before the controller requests it — e.g. a source erroring +// at end of stream pushes EOS downstream (see the not-negotiated caps +// rejection on end-of-track placeholder frames). SendEOS arriving after the +// bus EOS must not arm the frozen-pipeline watchdog: the EOS probes it would +// install can never fire, the timer can never be stopped, and a healthy +// shutdown would be reported as "pipeline frozen" after eosTimeout. +func TestUnsolicitedEOSBeforeSendEOS(t *testing.T) { + prevTimeout := eosTimeout + eosTimeout = 2 * time.Second + t.Cleanup(func() { eosTimeout = prevTimeout }) + + tmpDir := t.TempDir() + outputPath := filepath.Join(tmpDir, "out.ogg") + + // audio-only room composite: SDK source type, silence generator input + req := &rpc.StartEgressRequest{ + EgressId: "EG_controllerTest", + Token: "token", + WsUrl: "wss://test", + Request: &rpc.StartEgressRequest_RoomComposite{ + RoomComposite: &livekit.RoomCompositeEgressRequest{ + RoomName: "controller-test", + AudioOnly: true, + FileOutputs: []*livekit.EncodedFileOutput{{ + Filepath: outputPath, + }}, + }, + }, + } + confString := fmt.Sprintf("tmp_dir: %s\ntemplate_base: http://localhost", tmpDir) + conf, err := config.NewPipelineConfig(confString, req) + if err != nil { + t.Fatalf("failed to build pipeline config: %v", err) + } + + src := &staticSource{startedAt: time.Now().UnixNano()} + c, err := NewWithSource(context.Background(), conf, nopIPCClient{}, func(*gstreamer.Callbacks) (source.Source, error) { + return src, nil + }) + if err != nil { + t.Fatalf("failed to build controller: %v", err) + } + + infoCh := make(chan *livekit.EgressInfo, 1) + go func() { + infoCh <- c.Run(context.Background()) + }() + + select { + case <-c.playing.Watch(): + case <-time.After(10 * time.Second): + t.Fatal("pipeline did not reach PLAYING") + } + + // As soon as the bus EOS is processed (before Close marks the egress + // complete), invoke the normal end path — mirroring the source-side + // shutdown that raced the unsolicited EOS in production. + sendEOSDone := make(chan struct{}) + go func() { + defer close(sendEOSDone) + <-c.eosReceived.Watch() + c.SendEOS(context.Background(), "test end") + }() + + // let some media reach the sink, then send EOS through the pipeline, + // bypassing the controller + time.Sleep(500 * time.Millisecond) + c.p.SendEOS() + + select { + case <-sendEOSDone: + case <-time.After(10 * time.Second): + t.Fatal("bus EOS not received") + } + + var info *livekit.EgressInfo + select { + case info = <-infoCh: + case <-time.After(10 * time.Second): + t.Fatal("Run did not return") + } + + // a mistakenly armed watchdog would misfire within eosTimeout and mark + // the egress failed with "pipeline frozen" + time.Sleep(eosTimeout + time.Second) + + if info.Status != livekit.EgressStatus_EGRESS_COMPLETE { + t.Fatalf("expected status %s, got %s (error: %q)", + livekit.EgressStatus_EGRESS_COMPLETE, info.Status, info.Error) + } + if info.Error != "" { + t.Fatalf("expected no error, got %q", info.Error) + } + + stat, err := os.Stat(outputPath) + if err != nil { + t.Fatalf("output file missing: %v", err) + } + if stat.Size() == 0 { + t.Fatal("output file is empty") + } +} From 7e42c799f49ec8aa6697e8329446393c68782027 Mon Sep 17 00:00:00 2001 From: Milos Pesic Date: Tue, 28 Jul 2026 10:43:59 +0200 Subject: [PATCH 2/2] Update comment --- pkg/pipeline/controller_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/pipeline/controller_test.go b/pkg/pipeline/controller_test.go index 386064c7..aeeb5782 100644 --- a/pkg/pipeline/controller_test.go +++ b/pkg/pipeline/controller_test.go @@ -120,7 +120,7 @@ func TestUnsolicitedEOSBeforeSendEOS(t *testing.T) { // As soon as the bus EOS is processed (before Close marks the egress // complete), invoke the normal end path — mirroring the source-side - // shutdown that raced the unsolicited EOS in production. + // shutdown path. sendEOSDone := make(chan struct{}) go func() { defer close(sendEOSDone)