WIP - extending egress pipeline to support non live (faster than realtime) sources

pkg/config/base.go

@@ -59,7 +59,7 @@ type BaseConfig struct {
 	S3AssumeRoleExternalID string                  `yaml:"s3_assume_role_external_id"` // if set, this external ID is used by default for S3 uploads
 
 	// advanced
-	Insecure                      bool                                `yaml:"insecure"`                           // allow chrome to connect to an insecure websocket
+	Insecure                      bool                                `yaml:"insecure"`                           // allow chrome to connect to an insecure websocket, bypasses chrome LNA checks
 	Debug                         DebugConfig                         `yaml:"debug"`                              // create dot file on internal error
 	ChromeFlags                   map[string]interface{}              `yaml:"chrome_flags"`                       // additional flags to pass to Chrome
 	Latency                       LatencyConfig                       `yaml:"latency"`                            // gstreamer latencies, modifying these may break the service

pkg/config/output.go

@@ -222,6 +222,16 @@ func (p *PipelineConfig) updateOutputs(req *livekit.ExportReplayRequest) error {
 		return errors.ErrInvalidInput("output")
 	}
 
+	// Non-live pipelines produce data faster than realtime. Stream outputs
+	// (RTMP, WebSocket) cannot ingest faster than 1x playback speed.
+	if !p.Live {
+		for _, output := range req.Outputs {
+			if _, ok := output.Config.(*livekit.Output_Stream); ok {
+				return errors.ErrNotSupported("stream output for non-live pipeline")
+			}
+		}
+	}
+
 	var hasFile, hasStream, hasSegments bool
 	var fileCount, streamCount, segmentCount int
 

pkg/config/pipeline.go

@@ -59,7 +59,14 @@ type PipelineConfig struct {
 	Info            *livekit.EgressInfo        `yaml:"-"`
 	Manifest        *Manifest                  `yaml:"-"`
 	StorageReporter storageobs.ProjectReporter `yaml:"-"`
-	IsReplay        bool                       `yaml:"-"`
+	Live            bool                       `yaml:"-"`
+}
+
+// IsReplay returns true when this is a replay/export pipeline. Use this for
+// replay-specific integration points (IPC calls, storage access). For generic
+// pipeline behavior (is-live, leaky queues, backpressure) use the Live field.
+func (p *PipelineConfig) IsReplay() bool {
+	return !p.Live
 }
 
 var (
@@ -151,6 +158,7 @@ func NewPipelineConfig(confString string, req *rpc.StartEgressRequest) (*Pipelin
 		},
 		Outputs:         make(map[types.EgressType][]OutputConfig),
 		StorageReporter: storageobs.NewNoopProjectReporter(),
+		Live:            true,
 	}
 
 	if err := yaml.Unmarshal([]byte(confString), p); err != nil {
@@ -177,6 +185,7 @@ func GetValidatedPipelineConfig(conf *ServiceConfig, req *rpc.StartEgressRequest
 		BaseConfig: conf.BaseConfig,
 		TmpDir:     path.Join(TmpDir, req.EgressId),
 		Outputs:    make(map[types.EgressType][]OutputConfig),
+		Live:       true,
 	}
 
 	return p, p.Update(req)
@@ -417,7 +426,6 @@ func (p *PipelineConfig) Update(request *rpc.StartEgressRequest) error {
 		}
 
 	case *rpc.StartEgressRequest_Replay:
-		p.IsReplay = true
 		replayReq := req.Replay
 		clone := proto.Clone(replayReq).(*livekit.ExportReplayRequest)
 		p.Info.Request = &livekit.EgressInfo_Replay{

pkg/handler/handler.go

@@ -133,7 +133,7 @@ func (h *Handler) Run() {
 	}
 
 	// Replay coordination: signal ready and get timing
-	if h.conf.IsReplay {
+	if h.conf.IsReplay() {
 		rctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 		resp, err := h.ipcServiceClient.ReplayReady(rctx, &rpc.EgressReadyRequest{
 			EgressId: h.conf.Info.EgressId,

pkg/pipeline/builder/audio.go

@@ -145,7 +145,7 @@ func BuildAudioBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig) error
 			return err
 		}
 	} else {
-		queue, err := gstreamer.BuildQueue(fmt.Sprintf("%s_queue", audioBinName), p.Latency.PipelineLatency, leakyQueue)
+		queue, err := gstreamer.BuildQueue(fmt.Sprintf("%s_queue", audioBinName), p.Latency.PipelineLatency, p.Live)
 		if err != nil {
 			return errors.ErrGstPipelineError(err)
 		}
@@ -220,8 +220,10 @@ func (b *AudioBin) buildSDKInput() error {
 			return err
 		}
 	}
-	if err := b.addAudioTestSrcBin(); err != nil {
-		return err
+	if b.conf.Live {
+		if err := b.addAudioTestSrcBin(); err != nil {
+			return err
+		}
 	}
 	if err := b.addMixer(); err != nil {
 		return err
@@ -252,9 +254,14 @@ func (b *AudioBin) addAudioAppSrcBinLocked(ts *config.TrackSource) error {
 		return false
 	})
 	ts.AppSrc.SetArg("format", "time")
-	if err := ts.AppSrc.SetProperty("is-live", true); err != nil {
+	if err := ts.AppSrc.SetProperty("is-live", b.conf.Live); err != nil {
 		return err
 	}
+	if !b.conf.Live {
+		if err := ts.AppSrc.SetProperty("block", true); err != nil {
+			return err
+		}
+	}
 	if err := appSrcBin.AddElement(ts.AppSrc.Element); err != nil {
 		return err
 	}

pkg/pipeline/builder/file.go

@@ -58,6 +58,11 @@ func BuildFileBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig) (*gstr
 	if err = sink.SetProperty("sync", false); err != nil {
 		return nil, errors.ErrGstPipelineError(err)
 	}
+	if !p.Live {
+		if err = sink.SetProperty("async", false); err != nil {
+			return nil, errors.ErrGstPipelineError(err)
+		}
+	}
 
 	if err = b.AddElements(mux.GetElement(), sink); err != nil {
 		return nil, err

pkg/pipeline/builder/video.go

@@ -49,9 +49,11 @@ type VideoBin struct {
 	rawVideoTee *gst.Element
 }
 
-// buildLeakyVideoQueue creates a leaky queue and attaches a monitor to track dropped buffers
-func (b *VideoBin) buildLeakyVideoQueue(name string) (*gst.Element, error) {
-	queue, err := gstreamer.BuildQueue(name, b.conf.Latency.PipelineLatency, true)
+// buildVideoQueue creates a queue for the video pipeline. For live sources the
+// queue is leaky (drops old buffers when full) to handle real-time overrun. For
+// non-live replay the queue is blocking so backpressure throttles the source.
+func (b *VideoBin) buildVideoQueue(name string) (*gst.Element, error) {
+	queue, err := gstreamer.BuildQueue(name, b.conf.Latency.PipelineLatency, b.conf.Live)
 	if err != nil {
 		return nil, errors.ErrGstPipelineError(err)
 	}
@@ -97,7 +99,7 @@ func BuildVideoBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig) error
 			return tee.GetRequestPad("src_%u")
 		}
 	} else if len(p.GetEncodedOutputs()) > 0 {
-		queue, err := b.buildLeakyVideoQueue("video_queue")
+		queue, err := b.buildVideoQueue("video_queue")
 		if err != nil {
 			return err
 		}
@@ -281,7 +283,7 @@ func (b *VideoBin) buildWebInput() error {
 		return errors.ErrGstPipelineError(err)
 	}
 
-	videoQueue, err := b.buildLeakyVideoQueue("video_input_queue")
+	videoQueue, err := b.buildVideoQueue("video_input_queue")
 	if err != nil {
 		return err
 	}
@@ -384,9 +386,14 @@ func (b *VideoBin) buildAppSrcBin(ts *config.TrackSource, name string) (*gstream
 		return false
 	})
 	ts.AppSrc.SetArg("format", "time")
-	if err := ts.AppSrc.SetProperty("is-live", true); err != nil {
+	if err := ts.AppSrc.SetProperty("is-live", b.conf.Live); err != nil {
 		return nil, errors.ErrGstPipelineError(err)
 	}
+	if !b.conf.Live {
+		if err := ts.AppSrc.SetProperty("block", true); err != nil {
+			return nil, errors.ErrGstPipelineError(err)
+		}
+	}
 	if err := appSrcBin.AddElement(ts.AppSrc.Element); err != nil {
 		return nil, err
 	}
@@ -725,7 +732,7 @@ func (b *VideoBin) addDecodedVideoSink() error {
 }
 
 func (b *VideoBin) addVideoConverter(bin *gstreamer.Bin) error {
-	videoQueue, err := b.buildLeakyVideoQueue("video_input_queue")
+	videoQueue, err := b.buildVideoQueue("video_input_queue")
 	if err != nil {
 		return err
 	}

pkg/pipeline/controller.go

@@ -104,7 +104,70 @@ func New(ctx context.Context, conf *config.PipelineConfig, ipcServiceClient ipc.
 	ctx, span := tracer.Start(ctx, "Pipeline.New")
 	defer span.End()
 
+	c := newController(conf, ipcServiceClient)
+
+	// initialize gst
+	go func() {
+		_, span := tracer.Start(ctx, "gst.Init")
+		defer span.End()
+		gst.Init(nil)
+		gst.SetLogFunction(c.gstLog)
+		close(c.callbacks.GstReady)
+	}()
+
+	// create source
 	var err error
+	c.src, err = source.New(ctx, conf, c.callbacks)
+	if err != nil {
+		return nil, err
+	}
+
+	// create pipeline
+	<-c.callbacks.GstReady
+	if err = c.BuildPipeline(); err != nil {
+		c.src.Close()
+		return nil, err
+	}
+
+	return c, nil
+}
+
+// NewWithSource creates a Controller with a pre-built source. This is used when
+// the source is constructed externally (e.g. TestSource for non-live pipeline
+// testing, or ReplaySource for offline export). The source must have already
+// populated the PipelineConfig with track information before calling this.
+func NewWithSource(ctx context.Context, conf *config.PipelineConfig, src source.Source) (*Controller, error) {
+	ctx, span := tracer.Start(ctx, "Pipeline.NewWithSource")
+	defer span.End()
+	c := newController(conf, nil)
+	c.src = src
+
+	// initialize gst
+	go func() {
+		_, span := tracer.Start(ctx, "gst.Init")
+		defer span.End()
+		gst.Init(nil)
+		gst.SetLogFunction(c.gstLog)
+		close(c.callbacks.GstReady)
+	}()
+
+	// create pipeline
+	<-c.callbacks.GstReady
+	if err := c.BuildPipeline(); err != nil {
+		c.src.Close()
+		return nil, err
+	}
+
+	return c, nil
+}
+
+// Callbacks returns the pipeline callbacks. Sources that need to wait for
+// GstReady before creating appsrc elements can use this.
+func (c *Controller) Callbacks() *gstreamer.Callbacks {
+	return c.callbacks
+}
+
+func newController(conf *config.PipelineConfig, ipcServiceClient ipc.EgressServiceClient) *Controller {
 	c := &Controller{
 		PipelineConfig:   conf,
 		ipcServiceClient: ipcServiceClient,
@@ -129,30 +192,7 @@ func New(ctx context.Context, conf *config.PipelineConfig, ipcServiceClient ipc.
 		logger.Debugw("debug dot requested", "reason", reason)
 		c.generateDotFile(reason)
 	})
-
-	// initialize gst
-	go func() {
-		_, span := tracer.Start(ctx, "gst.Init")
-		defer span.End()
-		gst.Init(nil)
-		gst.SetLogFunction(c.gstLog)
-		close(c.callbacks.GstReady)
-	}()
-
-	// create source
-	c.src, err = source.New(ctx, conf, c.callbacks)
-	if err != nil {
-		return nil, err
-	}
-
-	// create pipeline
-	<-c.callbacks.GstReady
-	if err = c.BuildPipeline(); err != nil {
-		c.src.Close()
-		return nil, err
-	}
-
-	return c, nil
+	return c
 }
 
 func (c *Controller) BuildPipeline() error {
@@ -168,9 +208,9 @@ func (c *Controller) BuildPipeline() error {
 		c.stopped.Break()
 		return nil
 	})
-	if c.SourceType == types.SourceTypeSDK {
+	if sdkSrc, ok := c.src.(*source.SDKSource); ok {
 		p.SetEOSFunc(func() bool {
-			c.src.(*source.SDKSource).CloseWriters()
+			sdkSrc.CloseWriters()
 			return true
 		})
 	}
@@ -519,11 +559,11 @@ func (c *Controller) SendEOS(ctx context.Context, reason string) {
 
 		case livekit.EgressStatus_EGRESS_ACTIVE:
 			c.Info.UpdateStatus(livekit.EgressStatus_EGRESS_ENDING)
-			_, _ = c.ipcServiceClient.HandlerUpdate(ctx, c.Info)
+			c.sendHandlerUpdate(ctx, c.Info)
 			c.sendEOS()
 
 		case livekit.EgressStatus_EGRESS_ENDING:
-			_, _ = c.ipcServiceClient.HandlerUpdate(ctx, c.Info)
+			c.sendHandlerUpdate(ctx, c.Info)
 			c.sendEOS()
 
 		case livekit.EgressStatus_EGRESS_LIMIT_REACHED:
@@ -856,7 +896,7 @@ func (c *Controller) updateStartTime(startedAt int64) {
 
 	if c.Info.Status == livekit.EgressStatus_EGRESS_STARTING {
 		c.Info.UpdateStatus(livekit.EgressStatus_EGRESS_ACTIVE)
-		_, _ = c.ipcServiceClient.HandlerUpdate(context.Background(), c.Info)
+		c.sendHandlerUpdate(context.Background(), c.Info)
 	}
 }
 
@@ -894,7 +934,13 @@ func (c *Controller) streamUpdated(ctx context.Context) {
 		}
 	}
 
-	_, _ = c.ipcServiceClient.HandlerUpdate(ctx, c.Info)
+	c.sendHandlerUpdate(ctx, c.Info)
+}
+
+func (c *Controller) sendHandlerUpdate(ctx context.Context, info *livekit.EgressInfo) {
+	if c.ipcServiceClient != nil {
+		_, _ = c.ipcServiceClient.HandlerUpdate(ctx, info)
+	}
 }
 
 func (c *Controller) updateEndTime() {