diff --git a/pkg/media/mist_mkv_ingest_test.go b/pkg/media/mist_mkv_ingest_test.go index e0bc22cd..6b67609d 100644 --- a/pkg/media/mist_mkv_ingest_test.go +++ b/pkg/media/mist_mkv_ingest_test.go @@ -27,6 +27,14 @@ import ( // stack dump). With transcode=true the node keys are supplied so the worker // completes to dual-codec, as production does. func runMKVThroughIngestWorker(t *testing.T, mkv []byte, transcode bool) (int, error) { + segs, err := runMKVThroughIngestWorkerSegments(t, mkv, transcode) + return len(segs), err +} + +// runMKVThroughIngestWorkerSegments is runMKVThroughIngestWorker returning the +// raw segment payloads, for tests that validate the emitted segments rather +// than just count them. +func runMKVThroughIngestWorkerSegments(t *testing.T, mkv []byte, transcode bool) ([][]byte, error) { t.Helper() old := ingestWorkerWatchdog ingestWorkerWatchdog = 10 * time.Second @@ -59,15 +67,15 @@ func runMKVThroughIngestWorker(t *testing.T, mkv []byte, transcode bool) (int, e runErr := RunMKVIngestWorker(ctx, cfg, bytes.NewReader(mkv), ingestframe.NewWriter(&buf), func() []byte { return cfg.Manifest }) r := ingestframe.NewReader(&buf) - segs := 0 + var segs [][]byte for { - typ, _, rerr := r.ReadFrame() + typ, payload, rerr := r.ReadFrame() if errors.Is(rerr, io.EOF) { break } require.NoError(t, rerr) if typ == ingestframe.Segment { - segs++ + segs = append(segs, payload) } } return segs, runErr @@ -187,6 +195,71 @@ func TestMKVIngestMistTail(t *testing.T) { require.GreaterOrEqual(t, segs, 20, "tail sample emits segments past the keyframe-only transition") } +// makeBFrameAACMKV synthesizes an H264 stream WITH B-frames (PTS ≠ DTS) +// alongside AAC audio in a streamable MKV — the shape a hardware encoder or +// non-zerolatency x264 push produces. Matroska blocks carry only presentation +// timestamps, so on demux the reordered video arrives with dts=none; without +// DTS reconstruction the fMP4 muxer stretches the video track and every +// segment fails validation downstream (see buildMKVIngestPipeline's +// h264timestamper). b-adapt=false forces x264 to actually emit the configured +// B-frames rather than deciding per-scene. +func makeBFrameAACMKV(t *testing.T, ctx context.Context, seconds int) []byte { + t.Helper() + gstinit.InitGST() + desc := strings.Join([]string{ + fmt.Sprintf("videotestsrc num-buffers=%d ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc bframes=2 b-adapt=false key-int-max=30 speed-preset=veryfast ! h264parse ! matroskamux name=mux streamable=true ! appsink name=sink", seconds*30), + fmt.Sprintf("audiotestsrc num-buffers=%d samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mux.", seconds*47), + }, "\n") + pipeline, err := gst.NewPipelineFromString(desc) + require.NoError(t, err) + + sinkEle, err := pipeline.GetElementByName("sink") + require.NoError(t, err) + var buf bytes.Buffer + app.SinkFromElement(sinkEle).SetCallbacks(&app.SinkCallbacks{ + NewSampleFunc: WriterNewSample(ctx, &buf), + }) + + busErr := make(chan error, 1) + go func() { busErr <- HandleBusMessages(ctx, pipeline) }() + require.NoError(t, pipeline.SetState(gst.StatePlaying)) + defer func() { _ = pipeline.SetState(gst.StateNull) }() + require.NoError(t, <-busErr, "synthesize B-frame MKV") + require.NotEmpty(t, buf.Bytes()) + return buf.Bytes() +} + +// TestMKVIngestBFramesValidate is the regression test for B-frame MKV ingest: +// every emitted segment must survive the full ValidateMP4Media chokepoint. +// Before DTS reconstruction, mp4mux treated the reordered (dts=none) B-frame +// PTS as monotonic timing, stretching the video track ~2.2×; the segments +// LOOKED fine (both tracks present, signatures valid) but the video/audio +// duration mismatch made push-mode qtdemux EOS the audio pad before the audio +// bytes arrived — muxl's flat wrap is non-interleaved, video first — and every +// segment was rejected with "no audio in segment". +func TestMKVIngestBFramesValidate(t *testing.T) { + ctx := context.Background() + mkv := makeBFrameAACMKV(t, ctx, 8) + + segs, err := runMKVThroughIngestWorkerSegments(t, mkv, false) + require.NoError(t, err, "B-frame stream ingests cleanly") + require.GreaterOrEqual(t, len(segs), 6, "B-frame stream emits its segments") + + sawBFrames := false + for i, seg := range segs { + res, verr := ValidateMP4Media(ctx, seg) + require.NoError(t, verr, "segment %d validates (video+audio both present)", i) + dur := time.Duration(res.MediaData.Duration) + require.Greater(t, dur, 500*time.Millisecond, "segment %d duration sane", i) + require.Less(t, dur, 2*time.Second, "segment %d duration not stretched", i) + if res.MediaData.Video[0].BFrames { + sawBFrames = true + } + } + require.True(t, sawBFrames, "synthesized stream actually contains B-frames — if this fails the test no longer exercises the reorder path") + t.Logf("B-frame stream: %d segments, all validated", len(segs)) +} + // TestMKVIngestMistFullSampleNoTranscode is TestMKVIngestMistFullSample // without node keys (segment+sign only). During diagnosis this proved the // wedge was in the core ingest pipeline, not the dual-codec transcode stage — diff --git a/pkg/media/mkv_ingest.go b/pkg/media/mkv_ingest.go index 03ae5d88..64182fcf 100644 --- a/pkg/media/mkv_ingest.go +++ b/pkg/media/mkv_ingest.go @@ -82,9 +82,18 @@ func buildMKVIngestPipeline(ctx context.Context, input io.Reader, signerElem *gs // no EOS — a live stream wedges until the watchdog kills it. Use the shared // Queue2Big preset (no time/buffer cap, generous byte cap) like the other // demux-fed pipelines (transcode, rtmp_push, packetize, media_data_parser). + // + // h264timestamper: Matroska blocks carry only presentation timestamps, so + // for B-frame streams (PTS ≠ DTS) matroskademux emits reordered PTS with + // dts=none — and h264parse does not reconstruct DTS. The fMP4 muxer needs + // DTS to mux a reordered stream; without it it treats the jumbled PTS as + // monotonic timing and stretches the video track (~2.2× on a real capture), + // which downstream makes qtdemux EOS the audio pad early and every segment + // fails validation with "no audio in segment". h264timestamper rebuilds + // DTS from the H264 picture order count. pipelineSlice := []string{ "appsrc name=streamsrc ! matroskademux name=demux", - "demux. ! " + constants.Queue2Big + " ! h264parse name=parse", + "demux. ! " + constants.Queue2Big + " ! h264parse ! h264timestamper name=videoout", "demux. ! " + constants.Queue2Big + " ! fdkaacdec ! audioresample ! opusenc name=audioenc", } pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) @@ -98,7 +107,7 @@ func buildMKVIngestPipeline(ctx context.Context, input io.Reader, signerElem *gs app.SrcFromElement(srcele).SetCallbacks(&app.SourceCallbacks{ NeedDataFunc: ReaderNeedDataIncremental(ctx, input), }) - parseEle, err := pipeline.GetElementByName("parse") + parseEle, err := pipeline.GetElementByName("videoout") if err != nil { return nil, err }