From d0df2aff50cd4e9bb9ac4afe62c6cc9d0117f853 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Tue, 15 Sep 2026 21:17:41 -0700 Subject: [PATCH] ingest: only an IDR starts a segment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The client's broadcast encoder emits open-GOP I-frames at scene cuts: I-slices that are not IDRs and carry no SPS/PPS, right after an IDR. h264parse flags every I-slice as a keyframe, mp4mux writes it as a sync sample, and the muxl segmenter cut a new signed segment there — one no decoder can start on. Downstream, that one segment a minute took everything with it: the transcoder's orchestrator failed it ("non- existing PPS 0 referenced", "Invalid data found when processing input"), the gateway's session cleanup panicked and left the manifest in "No sessions available" until the manifest rotated; the packetizer found no valid frame in it, so it was never published to the bus and syndicating nodes never received it; and an HLS fragment could start on it. A pad probe after h264parse on both ingest paths (MP4 from Mist, WHIP) now clears the keyframe flag on any access unit without an IDR slice, so such frames stay inside the GoP of the IDR before them (streamplace_ingest_non_idr_keyframes_total counts them). Regression test: x264 open-GOP input, every signed segment must start on an IDR (8 of 9 did not before). And the transcoder push refuses any TS without SPS, PPS and an IDR (ErrNoKeyframe, streamplace_transcode_skipped_no_keyframe_total) so no other source of such a segment can drop the orchestrator session again. Reproduced offline on the client's recording: 41 of 653 pushed segments had no IDR, every one a segment starting right after a one-frame IDR segment Claude-Session: https://claude.ai/code/session_014aPQ5yqG9QFxKnbwQfCYKa (cherry picked from commit f0c4f31356d98c8f527020b806772b01e01e96ec) --- pkg/director/stream_session.go | 5 ++ pkg/livepeer/keyframe_test.go | 21 ++++++ pkg/livepeer/livepeer.go | 37 +++++++++ pkg/livepeer/real_pipeline_test.go | 76 +++++++++++++++++++ pkg/media/ingest_worker.go | 2 +- pkg/media/keyframe_probe.go | 87 +++++++++++++++++++++ pkg/media/keyframe_probe_test.go | 32 ++++++++ pkg/media/mp4_ingest.go | 6 +- pkg/media/open_gop_ingest_test.go | 117 +++++++++++++++++++++++++++++ pkg/media/webrtc_ingest.go | 10 ++- pkg/spmetrics/spmetrics.go | 10 +++ 11 files changed, 399 insertions(+), 4 deletions(-) create mode 100644 pkg/livepeer/keyframe_test.go create mode 100644 pkg/media/keyframe_probe.go create mode 100644 pkg/media/keyframe_probe_test.go create mode 100644 pkg/media/open_gop_ingest_test.go diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index 2ea4d80ea..cd1be81d4 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -875,6 +875,11 @@ func (ss *StreamSession) Transcode(ctx context.Context, spseg *placestream.Segme log.Debug(ctx, "transcode: skipped, transcoder is behind") return nil } + if errors.Is(err, livepeer.ErrNoKeyframe) { + spmetrics.TranscodeSkippedNoKeyframeTotal.Inc() + log.Warn(ctx, "transcode: skipped, segment does not start with an IDR", "segment", notif.Segment.ID) + return nil + } if err != nil { spmetrics.TranscodeErrorsTotal.Inc() return err diff --git a/pkg/livepeer/keyframe_test.go b/pkg/livepeer/keyframe_test.go new file mode 100644 index 000000000..5de41b5e7 --- /dev/null +++ b/pkg/livepeer/keyframe_test.go @@ -0,0 +1,21 @@ +package livepeer + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestTSStartsDecodable(t *testing.T) { + nal := func(types ...byte) []byte { + var b []byte + for _, ty := range types { + b = append(b, 0x47, 0x41, 0x00, 0x10, 0, 0, 0, 1, ty, 0xaa) + } + return b + } + require.True(t, tsStartsDecodable(nal(9, 7, 8, 5, 9, 1))) + require.False(t, tsStartsDecodable(nal(9, 1, 9, 1)), "P-slices only") + require.False(t, tsStartsDecodable(nal(9, 5)), "IDR without its parameter sets") + require.False(t, tsStartsDecodable(nil)) +} diff --git a/pkg/livepeer/livepeer.go b/pkg/livepeer/livepeer.go index ab58ef89c..d2f8d4734 100644 --- a/pkg/livepeer/livepeer.go +++ b/pkg/livepeer/livepeer.go @@ -35,6 +35,40 @@ const SegmentsInFlight = 2 // Skipping keeps the renditions near live at the cost of a gap in them. const MaxWaiting = 2 +// ErrNoKeyframe is returned for a segment whose video has no IDR with its +// parameter sets: a transcoder decodes each pushed segment from scratch, so +// it cannot decode such a segment and, worse, a go-livepeer gateway drops +// its orchestrator session over the failure. The segment is skipped (a +// gap in the renditions) rather than pushed. The ingest path keeps such +// frames inside their GoP (media.installIDRKeyframeProbe); this guards the +// transcoder from any other source of them. +var ErrNoKeyframe = errors.New("segment does not start with an IDR and its parameter sets") + +// tsStartsDecodable reports whether an MPEG-TS video segment carries SPS, +// PPS and an IDR slice — what a fresh decoder needs. +func tsStartsDecodable(ts []byte) bool { + var sps, pps, idr bool + for i := 0; i+3 < len(ts); { + if ts[i] == 0 && ts[i+1] == 0 && ts[i+2] == 1 { + switch ts[i+3] & 31 { + case 7: + sps = true + case 8: + pps = true + case 5: + idr = true + } + if sps && pps && idr { + return true + } + i += 4 + continue + } + i++ + } + return false +} + // ErrBacklog is returned for a segment skipped because too many are already // waiting for a transcode slot. var ErrBacklog = errors.New("transcode backlog: segment skipped") @@ -140,6 +174,9 @@ func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, if audioSeg.Len() == 0 { return nil, fmt.Errorf("no audio in segment") } + if !tsStartsDecodable(tsSeg.Bytes()) { + return nil, ErrNoKeyframe + } if ls.waiting.Add(1) > MaxWaiting { ls.waiting.Add(-1) return nil, ErrBacklog diff --git a/pkg/livepeer/real_pipeline_test.go b/pkg/livepeer/real_pipeline_test.go index b9df55b21..0b17aaba7 100644 --- a/pkg/livepeer/real_pipeline_test.go +++ b/pkg/livepeer/real_pipeline_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "crypto" + "encoding/binary" "fmt" "os" "path/filepath" @@ -221,6 +222,17 @@ func TestRealBroadcastPipeline(t *testing.T) { r.dup = true } if scanOnly { + if os.Getenv("SP_REAL_LIST") != "" { + for _, ev := range evs { + if ev.Type == "segment" || ev.Type == "signed-segment" { + if vb := ev.Tracks["1"]; len(vb) > 0 { + nals, sync := firstSampleInfo(vb) + t.Logf("LIST %d dur=%.3f samples=%d tfdt=%d firstNALs=%v firstSync=%v", i, r.dur, r.samples, r.srcTfdt, nals, sync) + } + break + } + } + } rows = append(rows, r) continue } @@ -441,3 +453,67 @@ func walkBoxRange(b []byte, start, end int, fn func(typ string, s, e int) bool) off += size } } + +// firstSampleInfo returns the NAL types of a track fragment's first sample +// and whether its trun flags mark it a sync sample (the flag muxl cuts +// segments on). +func firstSampleInfo(b []byte) (nals []int, sync bool) { + var size int + var flags uint32 + haveFlags := false + var mdatOff int + walkBoxRange(b, 0, len(b), func(typ string, start, end int) bool { + switch typ { + case "moof", "traf": + walkBoxRange(b, start+8, end, func(t2 string, s2, e2 int) bool { + if t2 == "traf" { + walkBoxRange(b, s2+8, e2, func(t3 string, s3, e3 int) bool { + if t3 == "trun" && size == 0 { + body := b[s3+8 : e3] + tflags := binary.BigEndian.Uint32(body[0:4]) & 0xffffff + p := 8 + if tflags&1 != 0 { + p += 4 + } + if tflags&4 != 0 { + flags = binary.BigEndian.Uint32(body[p:]) + haveFlags = true + p += 4 + } + if tflags&0x100 != 0 { + p += 4 + } + if tflags&0x200 != 0 { + size = int(binary.BigEndian.Uint32(body[p:])) + p += 4 + } + if tflags&0x400 != 0 && !haveFlags { + flags = binary.BigEndian.Uint32(body[p:]) + haveFlags = true + } + } + return true + }) + } + return true + }) + case "mdat": + if mdatOff == 0 { + mdatOff = start + 8 + } + } + return true + }) + if size == 0 || mdatOff == 0 { + return nil, false + } + q := mdatOff + for q+4 < len(b) && q < mdatOff+size { + l := int(binary.BigEndian.Uint32(b[q:])) + nals = append(nals, int(b[q+4]&31)) + q += 4 + l + } + // sample_is_non_sync_sample is bit 16 of the sample flags + sync = haveFlags && flags&0x10000 == 0 + return nals, sync +} diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index d22ef0239..f2f54d4b5 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -250,7 +250,7 @@ func RunMP4IngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re if err != nil { return fmt.Errorf("build signer element: %w", err) } - pipeline, err := buildMP4IngestPipeline(ctx, media, signerElem) + pipeline, err := buildMP4IngestPipeline(ctx, media, signerElem, cfg.StreamerDID) if err != nil { return fmt.Errorf("build pipeline: %w", err) } diff --git a/pkg/media/keyframe_probe.go b/pkg/media/keyframe_probe.go new file mode 100644 index 000000000..21d0fb505 --- /dev/null +++ b/pkg/media/keyframe_probe.go @@ -0,0 +1,87 @@ +package media + +import ( + "context" + "encoding/binary" + + "github.com/go-gst/go-gst/gst" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/spmetrics" +) + +// h264NALTypes lists the NAL unit types of one H.264 access unit, in either +// Annex B (start codes) or AVC (4-byte length prefixes) form. It returns nil +// when the bytes are neither. +func h264NALTypes(b []byte) []int { + if len(b) < 5 { + return nil + } + if b[0] == 0 && b[1] == 0 && (b[2] == 1 || (b[2] == 0 && b[3] == 1)) { + var out []int + for i := 0; i+3 < len(b); { + if b[i] == 0 && b[i+1] == 0 && b[i+2] == 1 { + out = append(out, int(b[i+3]&31)) + i += 4 + continue + } + i++ + } + return out + } + var out []int + for i := 0; i+4 <= len(b); { + l := int(binary.BigEndian.Uint32(b[i:])) + if l <= 0 || i+4+l > len(b) { + return nil // not length-prefixed after all + } + out = append(out, int(b[i+4]&31)) + i += 4 + l + } + return out +} + +// auHasIDR reports whether an H.264 access unit contains an IDR slice: the +// only kind of frame a segment may start on, since only an IDR resets the +// decoder and only an IDR is guaranteed to have the parameter sets a +// decoder joining the stream needs. +func auHasIDR(b []byte) bool { + for _, t := range h264NALTypes(b) { + if t == 5 { + return true + } + } + return false +} + +// installIDRKeyframeProbe makes the H.264 buffers passing pad carry the +// keyframe flag only when they hold an IDR slice. h264parse marks every +// I-slice as a keyframe, and a broadcast encoder that emits open-GOP +// I-frames (non-IDR, no SPS/PPS, seen at scene cuts) then gets those +// written as sync samples, on which the segmenter starts a new signed +// segment — one no decoder can start on: the transcoder rejects it, the +// packetizer finds no frame in it, and an HLS player joining there stalls. +// Clearing the flag keeps such frames inside the GoP of the IDR before +// them. +func installIDRKeyframeProbe(ctx context.Context, pad *gst.Pad, streamer string) { + if pad == nil { + return + } + pad.AddProbe(gst.PadProbeTypeBuffer, func(_ *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn { + buf := info.GetBuffer() + if buf == nil || buf.HasFlags(gst.BufferFlagDeltaUnit) { + return gst.PadProbeOK + } + m := buf.Map(gst.MapRead) + if m == nil { + return gst.PadProbeOK + } + idr := auHasIDR(m.Bytes()) + buf.Unmap() + if !idr { + buf.SetFlags(gst.BufferFlagDeltaUnit) + spmetrics.IngestNonIDRKeyframesTotal.WithLabelValues(streamer).Inc() + log.Debug(ctx, "ingest: keyframe-flagged frame without an IDR slice kept inside its GoP", "streamer", streamer) + } + return gst.PadProbeOK + }) +} diff --git a/pkg/media/keyframe_probe_test.go b/pkg/media/keyframe_probe_test.go new file mode 100644 index 000000000..a86d75ec6 --- /dev/null +++ b/pkg/media/keyframe_probe_test.go @@ -0,0 +1,32 @@ +package media + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestAUHasIDR(t *testing.T) { + annexB := func(types ...byte) []byte { + var b []byte + for _, ty := range types { + b = append(b, 0, 0, 0, 1, ty, 0xaa, 0xbb) + } + return b + } + avc := func(types ...byte) []byte { + var b []byte + for _, ty := range types { + b = append(b, 0, 0, 0, 3, ty, 0xaa, 0xbb) + } + return b + } + require.True(t, auHasIDR(annexB(9, 7, 8, 5)), "IDR with headers, Annex B") + require.True(t, auHasIDR(avc(7, 8, 5)), "IDR with headers, length-prefixed") + require.False(t, auHasIDR(annexB(9, 1)), "a non-IDR I- or P-slice") + require.False(t, auHasIDR(avc(1)), "length-prefixed non-IDR") + require.False(t, auHasIDR(avc(6, 1)), "SEI then non-IDR slice") + require.False(t, auHasIDR(nil)) + require.False(t, auHasIDR([]byte{1, 2, 3, 4, 5, 6}), "garbage is not an IDR") + require.Equal(t, []int{7, 8, 5, 1}, h264NALTypes(avc(7, 8, 5, 1))) +} diff --git a/pkg/media/mp4_ingest.go b/pkg/media/mp4_ingest.go index 7332246bb..c82479e1d 100644 --- a/pkg/media/mp4_ingest.go +++ b/pkg/media/mp4_ingest.go @@ -39,7 +39,7 @@ func (mm *MediaManager) MP4Ingest(ctx context.Context, input io.Reader, ms Media if err != nil { return err } - pipeline, err := buildMP4IngestPipeline(ctx, input, signer) + pipeline, err := buildMP4IngestPipeline(ctx, input, signer, ms.Streamer()) if err != nil { return err } @@ -104,7 +104,7 @@ func rejectMatroska(input io.Reader) (io.Reader, error) { return br, nil } -func buildMP4IngestPipeline(ctx context.Context, input io.Reader, signerElem *gst.Element) (*gst.Pipeline, error) { +func buildMP4IngestPipeline(ctx context.Context, input io.Reader, signerElem *gst.Element, streamer string) (*gst.Pipeline, error) { input, err := rejectMatroska(input) if err != nil { return nil, err @@ -140,6 +140,8 @@ func buildMP4IngestPipeline(ctx context.Context, input io.Reader, signerElem *gs if err != nil { return nil, err } + // Only an IDR starts a segment (see installIDRKeyframeProbe). + installIDRKeyframeProbe(ctx, parseEle.GetStaticPad("src"), streamer) if err := pipeline.Add(signerElem); err != nil { return nil, err } diff --git a/pkg/media/open_gop_ingest_test.go b/pkg/media/open_gop_ingest_test.go new file mode 100644 index 000000000..923fa29f2 --- /dev/null +++ b/pkg/media/open_gop_ingest_test.go @@ -0,0 +1,117 @@ +package media + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/muxl" +) + +// makeOpenGOPFMP4 synthesizes what a broadcast encoder sends at scene cuts: +// open-GOP I-frames, I-slices that are not IDRs (x264 open-gop makes every +// keyframe after the first one of those), between IDRs, alongside AAC +// audio, in a fragmented MP4. h264parse flags them as keyframes. +func makeOpenGOPFMP4(t *testing.T, ctx context.Context) []byte { + t.Helper() + desc := "videotestsrc num-buffers=90 pattern=ball ! video/x-raw,width=320,height=240,framerate=30/1 ! " + + "x264enc key-int-max=10 bframes=2 b-adapt=false speed-preset=ultrafast option-string=open-gop=1:scenecut=0 ! h264parse ! mp4mux name=mux fragment-duration=500 ! appsink name=sink " + + "audiotestsrc num-buffers=130 samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! fdkaacenc ! aacparse ! mux." + return runSynthPipeline(t, ctx, desc) +} + +// videoFirstSampleNALs returns the NAL types of the first video sample of +// each signed segment's first event. +func videoFirstSampleNALs(t *testing.T, ctx context.Context, seg []byte) []int { + t.Helper() + evs, err := unwrapMuxlEvents(ctx, seg) + require.NoError(t, err) + for _, ev := range evs { + if ev.Type != "segment" && ev.Type != "signed-segment" { + continue + } + vb := ev.Tracks["1"] + if len(vb) == 0 { + return nil + } + return firstSampleNALTypes(vb) + } + return nil +} + +// firstSampleNALTypes walks a canonical track fragment to its first sample +// and lists that sample's NAL types (length-prefixed). +func firstSampleNALTypes(b []byte) []int { + var size int + var mdat []byte + walkBoxes(b, func(typ string, body []byte) bool { + switch typ { + case "moof": + walkBoxes(body, func(t2 string, traf []byte) bool { + if t2 != "traf" { + return true + } + walkBoxes(traf, func(t3 string, trun []byte) bool { + if t3 != "trun" || size != 0 || len(trun) < 8 { + return true + } + flags := uint32(trun[1])<<16 | uint32(trun[2])<<8 | uint32(trun[3]) + p := 8 + if flags&1 != 0 { + p += 4 + } + if flags&4 != 0 { + p += 4 + } + if flags&0x100 != 0 { + p += 4 + } + if flags&0x200 != 0 && p+4 <= len(trun) { + size = int(uint32(trun[p])<<24 | uint32(trun[p+1])<<16 | uint32(trun[p+2])<<8 | uint32(trun[p+3])) + } + return true + }) + return true + }) + case "mdat": + if mdat == nil { + mdat = body + } + } + return true + }) + if size == 0 || len(mdat) < size { + return nil + } + return h264NALTypes(mdat[:size]) +} + +var _ = muxl.RunMuxlUnwrapEvents + +// Every signed segment starts on an IDR, even when the encoder's keyframes +// include open-GOP I-frames: those stay inside the GoP of the IDR before +// them rather than starting a segment no decoder could begin on. +func TestIngestOpenGOPSegmentsStartOnIDR(t *testing.T) { + ctx := context.Background() + mp4 := makeOpenGOPFMP4(t, ctx) + segs, err := runMP4ThroughIngestWorkerSegments(t, mp4, false) + require.NoError(t, err) + require.NotEmpty(t, segs) + var nonIDR int + for i, seg := range segs { + nals := videoFirstSampleNALs(t, ctx, seg) + require.NotEmpty(t, nals, "segment %d has a first video sample", i) + hasIDR := false + for _, n := range nals { + if n == 5 { + hasIDR = true + } + } + if !hasIDR { + nonIDR++ + t.Logf("segment %d starts with NALs %v", i, nals) + } + } + require.Zero(t, nonIDR, "segments starting without an IDR") + require.LessOrEqual(t, len(segs), 2, "one IDR in 90 frames (the other 8 keyframes are open-GOP I-frames): one GoP, not a segment per I-frame") +} diff --git a/pkg/media/webrtc_ingest.go b/pkg/media/webrtc_ingest.go index 70b6b2bfd..995fcd7ac 100644 --- a/pkg/media/webrtc_ingest.go +++ b/pkg/media/webrtc_ingest.go @@ -50,7 +50,7 @@ func (mm *MediaManager) webRTCIngestPipeline(ctx context.Context, cancel context pipelineSlice := []string{ "multiqueue name=queue", - "appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse disable-passthrough=true config-interval=-1 ! h264timestamper ! identity ! queue.sink_0", + "appsrc format=time is-live=true do-timestamp=true name=videosrc ! capsfilter caps=application/x-rtp ! rtph264depay ! capsfilter caps=video/x-h264,stream-format=byte-stream,alignment=nal ! h264parse name=videoparse disable-passthrough=true config-interval=-1 ! h264timestamper ! identity ! queue.sink_0", "appsrc format=time do-timestamp=true name=audiosrc ! capsfilter caps=application/x-rtp,media=audio,encoding-name=OPUS,payload=111 ! rtpopusdepay ! opusparse ! queue.sink_1", } @@ -59,6 +59,14 @@ func (mm *MediaManager) webRTCIngestPipeline(ctx context.Context, cancel context return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err) } + // Only an IDR starts a segment (see installIDRKeyframeProbe). + if parseEle, err := pipeline.GetElementByName("videoparse"); err == nil { + streamer := "whip" + if keyRevSigner != nil { + streamer = keyRevSigner.Streamer() + } + installIDRKeyframeProbe(ctx, parseEle.GetStaticPad("src"), streamer) + } queue, err := pipeline.GetElementByName("queue") if err != nil { return nil, fmt.Errorf("failed to get queue element from pipeline: %w", err) diff --git a/pkg/spmetrics/spmetrics.go b/pkg/spmetrics/spmetrics.go index dbcf4d060..137c21152 100644 --- a/pkg/spmetrics/spmetrics.go +++ b/pkg/spmetrics/spmetrics.go @@ -56,6 +56,16 @@ var TranscodeErrorsTotal = promauto.NewCounter(prometheus.CounterOpts{ Help: "total number of transcode errors", }) +var IngestNonIDRKeyframesTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "streamplace_ingest_non_idr_keyframes_total", + Help: "frames the parser flagged as keyframes that hold no IDR slice (open-GOP I-frames), kept inside their GoP instead of starting a segment", +}, []string{"streamer"}) + +var TranscodeSkippedNoKeyframeTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "streamplace_transcode_skipped_no_keyframe_total", + Help: "segments not sent to the transcoder because their video does not start with an IDR and its parameter sets (a transcoder cannot decode them and drops the session)", +}) + var TranscodeManifestRotationsTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_transcode_manifest_rotations_total", Help: "times a stream moved to a fresh gateway manifest after the gateway refused every recent push (see livepeer.RotateAfterFailures)", -- knotmirror 0.0.1