diff --git a/pkg/api/rtmp_server.go b/pkg/api/rtmp_server.go index 2a4a2295d..4db61a11a 100644 --- a/pkg/api/rtmp_server.go +++ b/pkg/api/rtmp_server.go @@ -65,6 +65,13 @@ func (a *StreamplaceAPI) HandleRTMPPublisher(ctx context.Context, sc *gortmplib. // eRTMP multitrack (OBS "Multitrack Video") track count // Note that this currently only supports H.264 video and 1 track of AAC audio videoTrackCount := 0 + // relayDone is closed the moment RTMPIngest (the consumer of EventChan via + // the internal playback relay) returns. The data callbacks below send into + // EventChan synchronously inside r.Read(); if the relay dies nobody drains + // that channel, an unbounded send blocks r.Read() forever, so the 10s read + // deadline never fires and the session never tears down. Guarding every send + // on relayDone turns a permanent hang into a quick drop-and-exit. + relayDone := make(chan struct{}) for _, track := range r.Tracks() { log.Log(ctx, "get track", "track", track) @@ -78,11 +85,16 @@ func (a *StreamplaceAPI) HandleRTMPPublisher(ctx context.Context, sc *gortmplib. videoTrackCount++ session.VideoTracks[trackID] = track r.OnDataH264(track, func(pts time.Duration, dts time.Duration, au [][]byte) { - session.EventChan <- &media.RTMPH264Data{ + // Guarded send: if the relay is gone, drop the frame instead of + // blocking inside r.Read() forever (see relayDone below). + select { + case session.EventChan <- &media.RTMPH264Data{ TrackID: trackID, AU: au, PTS: pts, DTS: dts, + }: + case <-relayDone: } }) @@ -92,14 +104,21 @@ func (a *StreamplaceAPI) HandleRTMPPublisher(ctx context.Context, sc *gortmplib. } session.AudioTrack = track r.OnDataMPEG4Audio(track, func(pts time.Duration, au []byte) { - session.EventChan <- &media.RTMPAACData{ + select { + case session.EventChan <- &media.RTMPAACData{ AU: au, PTS: pts, + }: + case <-relayDone: } }) default: - return fmt.Errorf("unsupported track type: %T (multitrack ingest supports H.264 video + AAC audio only)", track) + // Unsupported track type: drop and log it, and keep the session + // running on the remaining tracks, rather than rejecting the whole + // stream over one unsupported track. If this drops the only video, + // the videoTrackCount == 0 check after the loop fails the session. + log.Log(ctx, "dropping unsupported track", "type", fmt.Sprintf("%T", track)) } } if videoTrackCount == 0 { @@ -107,11 +126,17 @@ func (a *StreamplaceAPI) HandleRTMPPublisher(ctx context.Context, sc *gortmplib. } g, ctx := errgroup.WithContext(ctx) + g.Go(func() error { for { if ctx.Err() != nil { return ctx.Err() } + select { + case <-relayDone: + return fmt.Errorf("relay stopped, ending publish session") + default: + } err = sc.RW.(net.Conn).SetReadDeadline(time.Now().Add(RTMPTimeout)) if err != nil { return err @@ -124,6 +149,7 @@ func (a *StreamplaceAPI) HandleRTMPPublisher(ctx context.Context, sc *gortmplib. }) g.Go(func() error { + defer close(relayDone) return a.MediaManager.RTMPIngest(ctx, fmt.Sprintf("rtmp://%s/live/%s", a.rtmpInternalPlaybackAddr, streamer), mediaSigner, videoTrackCount) }) diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index ad287d214..81ab9af07 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -222,13 +222,24 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment // their bitrate and reconnects clean — at which point findProblems clears it. // The client dedupes place.stream.error by code, so repeated kicks surface as // a single persistent problem. - if bitrate, exceeded := exceedsMaxBitrate(len(notif.Data), notif.Segment.MediaData.Duration, ss.cli.MaximumLiveBitrate); exceeded { + // The kick is measured over the WHOLE segment — for a multitrack source that + // is the sum of every rendition, which is naturally N× a single track. + // Scaling the cap by the number of video tracks keeps the intent (bound a + // streamer's total egress) without kicking someone who was fine before for + // enabling multitrack. trackCount stays 1 whenever Video is nil or holds a + // single entry (single-track sources), so single-track behavior is unchanged. + trackCount := 1 + if notif.Segment.MediaData != nil && len(notif.Segment.MediaData.Video) > 1 { + trackCount = len(notif.Segment.MediaData.Video) + } + maxBitrate := ss.cli.MaximumLiveBitrate * trackCount + if bitrate, exceeded := exceedsMaxBitrate(len(notif.Data), notif.Segment.MediaData.Duration, maxBitrate); exceeded { log.Log(ctx, "live bitrate exceeded maximum, disconnecting stream", - "streamer", notif.Segment.RepoDID, "bitrate", bitrate, "max", ss.cli.MaximumLiveBitrate) + "streamer", notif.Segment.RepoDID, "bitrate", bitrate, "max", maxBitrate, "tracks", trackCount) ss.bus.Publish(notif.Segment.RepoDID, media.NewStreamKick( "bitrate", fmt.Sprintf("Your stream's bitrate (%d kbps) exceeds this server's maximum of %d kbps. Lower your encoder's bitrate to keep streaming.", - bitrate/1000, ss.cli.MaximumLiveBitrate/1000), + bitrate/1000, maxBitrate/1000), )) return nil } diff --git a/pkg/media/ingest_supervisor.go b/pkg/media/ingest_supervisor.go index 8fd05e10b..3f84c3486 100644 --- a/pkg/media/ingest_supervisor.go +++ b/pkg/media/ingest_supervisor.go @@ -27,6 +27,20 @@ import ( // shorten it. var ingestWorkerWatchdog = 30 * time.Second +// Bounds how long probing may wait for qtdemux to expose the input's track +// layout while PAUSED. A valid fMP4 source should expose its tracks after the +// init segment arrives; otherwise the ingest cannot be constructed. +var mp4TrackProbeTimeout = 10 * time.Second + +// Prevents multitrack RTMP ingest from waiting indefinitely for all declared +// tracks to link after entering PLAYING. A missing track can leave mp4mux +// stalled without otherwise failing the pipeline. +var multitrackIngestWatchdog = 10 * time.Second + +// Keeps worker shutdown bounded if the signer fails to drain after its input +// has been closed. +var ingestWorkerDrainTimeout = 30 * time.Second + // MP4IngestIsolated is the process-isolated counterpart to MP4Ingest. Instead of // running the demux + sign pipeline in this process — where a native gst fault, // OOM, or deadlock would take the whole node down — it spawns a dedicated diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index 53ebd6d14..b1b35474e 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -7,6 +7,7 @@ import ( "io" "net/http/httputil" "sync" + "time" "github.com/go-gst/go-gst/gst" "stream.place/streamplace/pkg/config" @@ -265,9 +266,7 @@ func RunMP4IngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re return fmt.Errorf("set playing: %w", err) } defer func() { - if err := pipeline.SetState(gst.StateNull); err != nil { - log.Error(ctx, "ingest worker: set null", "error", err) - } + teardownPipeline(ctx, pipeline) }() // Pipeline done (EOS/error) → drain the signer (cancel flushes the final GoP; @@ -275,7 +274,16 @@ func RunMP4IngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re // so the last dual-codec completions are framed before we return. pipeErr := <-busErr cancel() - <-done + // Bound the signer drain: a wedged signer (e.g. the muxl wasm deadlock + // against its unread stdout pipe that drainCtx protects against) would never + // close done, and an unbounded wait here would defeat the self-watchdog — the + // worker must exit even if the signer refuses to. The final GoP is lost in + // that case, which is the right trade: the worker is dying anyway. + select { + case <-done: + case <-time.After(ingestWorkerDrainTimeout): + log.Error(ctx, "ingest worker: signer drain timed out, exiting anyway") + } flush() return pipeErr } diff --git a/pkg/media/mp4_ingest.go b/pkg/media/mp4_ingest.go index 4213df3bd..3fb408a0f 100644 --- a/pkg/media/mp4_ingest.go +++ b/pkg/media/mp4_ingest.go @@ -163,14 +163,22 @@ func buildMP4IngestPipeline(ctx context.Context, input io.Reader, makeSigner mp4 defer func() { // capture so we don't prevent cleanup if err != nil { - _ = pausedPipeline.SetState(gst.StateNull) + padsMu.Lock() + for pad, id := range blockProbes { + pad.RemoveProbe(id) + pad.AddProbe(gst.PadProbeTypeBuffer, func(p *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn { + return gst.PadProbeDrop + }) + } + padsMu.Unlock() + teardownPipeline(ctx, pausedPipeline) } }() select { case <-noMorePads: case <-ctx.Done(): return nil, nil, ctx.Err() - case <-time.After(10 * time.Second): + case <-time.After(mp4TrackProbeTimeout): return nil, nil, fmt.Errorf("timed out waiting for qtdemux to expose MP4 tracks") } diff --git a/pkg/media/packetize.go b/pkg/media/packetize.go index bcbc88f1b..5bedf350c 100644 --- a/pkg/media/packetize.go +++ b/pkg/media/packetize.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "runtime" "strings" "time" @@ -236,14 +237,12 @@ func Packetize(ctx context.Context, cli *config.CLI, seg *bus.Seg, preferHeight } defer func() { - err = pipeline.SetState(gst.StateNull) - if err != nil { - log.Error(ctx, "failed to set pipeline to null state", "error", err) - } + teardownPipeline(ctx, pipeline) err = pipeline.Remove(demuxBin.Element) if err != nil { log.Error(ctx, "failed to remove demux bin from bin", "error", err) } + runtime.KeepAlive(demuxBin.Element) }() err = <-busErr diff --git a/pkg/media/pipeline_teardown.go b/pkg/media/pipeline_teardown.go new file mode 100644 index 000000000..1ecb5f27f --- /dev/null +++ b/pkg/media/pipeline_teardown.go @@ -0,0 +1,25 @@ +package media + +import ( + "context" + "runtime" + + "github.com/go-gst/go-gst/gst" + "stream.place/streamplace/pkg/log" +) + +// Shuts the pipeline down completely before allowing its Go wrapper to become +// unreachable. Call this only when the caller is finished with the pipeline, and +// it must not be used again after teardown returns. +func teardownPipeline(ctx context.Context, pipeline *gst.Pipeline) { + // Teardown must complete synchronously. SetState may return while GStreamer + // is still transitioning, allowing the Go wrapper's finalizer to release the + // underlying object before shutdown finishes. + if err := pipeline.BlockSetState(gst.StateNull); err != nil { + log.Error(ctx, "failed to set pipeline state to null", "error", err) + } + + // Keep the wrapper reachable through the full state transition so its + // finalizer cannot release the underlying GStreamer object mid-teardown. + runtime.KeepAlive(pipeline) +} diff --git a/pkg/media/rtmp_ingest.go b/pkg/media/rtmp_ingest.go index a5aefbb8b..b605379ad 100644 --- a/pkg/media/rtmp_ingest.go +++ b/pkg/media/rtmp_ingest.go @@ -3,7 +3,9 @@ package media import ( "context" "fmt" + "sort" "strings" + "sync" "time" "github.com/bluenviron/gortsplib/v5/pkg/format" @@ -80,6 +82,11 @@ func (mm *MediaManager) RTMPIngest(ctx context.Context, rtmpURL string, ms Media return err } + // linked records which flvdemux pads have linked. Shared between the + // pad-added handler and the multitrack pad watchdog. + linkedMu := sync.Mutex{} + linked := map[string]bool{} + handle, err := demux.Connect("pad-added", func(self *gst.Element, pad *gst.Pad) { name := pad.GetName() chain, ok := queues[name] @@ -91,6 +98,9 @@ func (mm *MediaManager) RTMPIngest(ctx context.Context, rtmpURL string, ms Media log.Error(ctx, "error linking flvdemux pad", "pad", name, "error", ret) return } + linkedMu.Lock() + linked[name] = true + linkedMu.Unlock() // Elements were created before the pipeline started, but a pad can // appear while parts of the graph are still mid-rollout; a push into a // not-yet-PLAYING branch returns FLUSHING and silently stalls the @@ -120,13 +130,51 @@ func (mm *MediaManager) RTMPIngest(ctx context.Context, rtmpURL string, ms Media return err } - defer func() { - err := pipeline.SetState(gst.StateNull) - if err != nil { - log.Error(ctx, "error setting pipeline to null state", "error", err) + // Prevents multitrack ingest from hanging indefinitely when a declared track + // never links. mp4mux waits for every requested pad before emitting output, so + // a missing track can otherwise stall the pipeline without producing a bus + // error. After a grace period from PLAYING, fail the pipeline and report the + // unlinked pads. + // + // Note that this only verifies pad linkage and not continued data flow. A track that links + // and later goes silent will require a separate no-output watchdog. + go func() { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + deadline := time.After(multitrackIngestWatchdog) + for { + linkedMu.Lock() + allLinked := len(linked) == len(queues) + var missing []string + if !allLinked { + for name := range queues { + if !linked[name] { + missing = append(missing, name) + } + } + } + linkedMu.Unlock() + if allLinked { + return + } + select { + case <-ctx.Done(): + return + case <-deadline: + sort.Strings(missing) + err := fmt.Errorf("multitrack ingest wedged: declared track(s) never linked: %s", strings.Join(missing, ",")) + log.Error(ctx, "multitrack ingest pad watchdog fired", "unlinked", missing) + pipeline.GetPipelineBus().Post(gst.NewErrorMessage(pipeline, err, "", nil)) + return + case <-ticker.C: + } } }() + defer func() { + teardownPipeline(ctx, pipeline) + }() + err = <-busErr return err diff --git a/pkg/media/rtmp_push.go b/pkg/media/rtmp_push.go index 00bf2e19d..f52fa301d 100644 --- a/pkg/media/rtmp_push.go +++ b/pkg/media/rtmp_push.go @@ -249,9 +249,7 @@ func (mm *MediaManager) runRTMPPushPipeline(ctx context.Context, source io.Reade defer func() { log.Log(ctx, "shutting down RTMP push pipeline") - if err := pipeline.SetState(gst.StateNull); err != nil { - log.Error(ctx, "failed to set pipeline state to null", "error", err) - } + teardownPipeline(ctx, pipeline) }() return <-errCh