Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
13 kB · 296 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297package media
import ( "bytes" "context" "errors" "fmt" "io" "os" "strings" "testing" "time"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/crypto/signers" "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/ingestframe" "stream.place/streamplace/pkg/muxl" "stream.place/streamplace/test/remote")
// runMP4ThroughIngestWorker feeds a fragmented-MP4 byte stream through the// isolated ingest worker and returns how many signed canonical segments it// emitted. The worker watchdog is shortened so a wedged pipeline returns// promptly instead of hanging the test (override via SP_TEST_WATCHDOG to e.g.// park a wedge for a stack dump). With transcode=true the node keys are// supplied so the worker completes to dual-codec, as production does.func runMP4ThroughIngestWorker(t *testing.T, mp4 []byte, transcode bool) (int, error) { segs, err := runMP4ThroughIngestWorkerSegments(t, mp4, transcode) return len(segs), err}
// runMP4ThroughIngestWorkerSegments is runMP4ThroughIngestWorker returning the// raw segment payloads, for tests that validate the emitted segments rather// than just count them.func runMP4ThroughIngestWorkerSegments(t *testing.T, mp4 []byte, transcode bool) ([][]byte, error) { t.Helper() old := ingestWorkerWatchdog ingestWorkerWatchdog = 10 * time.Second if wd := os.Getenv("SP_TEST_WATCHDOG"); wd != "" { d, perr := time.ParseDuration(wd) require.NoError(t, perr) ingestWorkerWatchdog = d } defer func() { ingestWorkerWatchdog = old }()
ctx := context.Background() ms := newBareSegmentSigner(t) keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) require.NoError(t, err) manifest, err := ms.buildManifest(ctx, time.Now().UnixMilli()) require.NoError(t, err) cfg := IngestWorkerConfig{ StreamerDID: ms.Streamer(), KeyPEM: keyPEM, CertPEM: ms.Cert, Manifest: manifest, BroadcasterHost: "test.example.com", } if transcode { cfg.NodeCertPEM = ms.Cert cfg.NodeKeyPEM = keyPEM }
var buf bytes.Buffer runErr := RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), ingestframe.NewWriter(&buf), func() []byte { return cfg.Manifest })
r := ingestframe.NewReader(&buf) var segs [][]byte for { typ, payload, rerr := r.ReadFrame() if errors.Is(rerr, io.EOF) { break } require.NoError(t, rerr) if typ == ingestframe.Segment { segs = append(segs, payload) } } return segs, runErr}
// runSynthPipeline runs a gst-launch description whose sink is an appsink// named "sink" and returns everything the sink produced.func runSynthPipeline(t *testing.T, ctx context.Context, desc string) []byte { t.Helper() gstinit.InitGST() 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 test stream") require.NotEmpty(t, buf.Bytes()) return buf.Bytes()}
// makeSparseVideoAACFMP4 synthesizes the stream shape that wedged production// ingest back when the bridge format was MKV: video that degrades to// keyframe-only at a low rate (here 0.5fps — MistServer drops all delta frames// when a push falls behind, leaving ~1s-apart keyframes) alongside continuous// 48kHz AAC audio, in a fragmented MP4. The wedge mechanism is// demux-agnostic — the audio branch must buffer a full video-frame gap while// the demux walks the byte stream to the next video frame, and the// aggregator-based fMP4 muxer downstream consumes nothing until every pad has// data — so the regression carries over to the qtdemux pipeline (see// buildMP4IngestPipeline's Queue2Big comment).func makeSparseVideoAACFMP4(t *testing.T, ctx context.Context, seconds int) []byte { t.Helper() desc := strings.Join([]string{ fmt.Sprintf("videotestsrc num-buffers=%d ! video/x-raw,width=320,height=240,framerate=1/2 ! x264enc key-int-max=1 tune=zerolatency speed-preset=ultrafast ! h264parse ! mp4mux name=mux fragment-duration=500 ! appsink name=sink", (seconds+1)/2), fmt.Sprintf("audiotestsrc num-buffers=%d samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mux.", seconds*47), }, "\n") return runSynthPipeline(t, ctx, desc)}
// TestMP4IngestSparseVideoNoWedge is the regression test for a production// ingest wedge: a stream whose video goes sparse (keyframe-only, ≥1s between// video frames) deadlocked the ingest pipeline — audio backpressure through// the default 1s-capped queue blocked the demux, the fMP4 aggregator starved// on its video pad, and the stream hung with no EOS until the watchdog killed// it. With Queue2Big on both ingest branches the same stream must segment to// completion.func TestMP4IngestSparseVideoNoWedge(t *testing.T) { ctx := context.Background() mp4 := makeSparseVideoAACFMP4(t, ctx, 12)
segs, err := runMP4ThroughIngestWorker(t, mp4, false) require.NoError(t, err, "sparse-video stream ingests cleanly (wedge → watchdog → context canceled)") // 12s of 2s-apart keyframes ≈ 6 GoPs; wedging yields 0-1 segments. require.GreaterOrEqual(t, segs, 4, "sparse-video stream emits its segments") t.Logf("sparse-video stream: %d segments", segs)}
// videoPTSDTSOffsets flat-wraps a canonical segment and returns every video// sample's PTS−DTS composition offset. This is the direct probe for the class// of bug that motivated the fMP4 ingest rewrite: MKV ingest reconstructed DTS// with h264timestamper, whose SPS fallback minted a constant spurious offset// for streams that declare no reorder window (VideoToolbox) — pushing every// GoP's presentation past its segment's declared window and breaking WebRTC// playback at each keyframe. fMP4 ingest reads the container's real DTS, so a// no-reorder stream must come out with PTS == DTS on every sample.func videoPTSDTSOffsets(t *testing.T, ctx context.Context, segment []byte) []time.Duration { t.Helper() gstinit.InitGST() var flat bytes.Buffer require.NoError(t, muxl.RunMuxlWrap(ctx, bytes.NewReader(segment), "flat", &flat))
desc := strings.Join([]string{ "appsrc name=src ! qtdemux name=demux", "demux.video_0 ! queue ! h264parse ! appsink name=sink sync=false", "demux.audio_0 ! queue ! fakesink sync=false", }, "\n") pipeline, err := gst.NewPipelineFromString(desc) require.NoError(t, err)
srcEle, err := pipeline.GetElementByName("src") require.NoError(t, err) app.SrcFromElement(srcEle).SetCallbacks(&app.SourceCallbacks{ NeedDataFunc: ReaderNeedDataIncremental(ctx, bytes.NewReader(flat.Bytes())), })
sinkEle, err := pipeline.GetElementByName("sink") require.NoError(t, err) var offsets []time.Duration app.SinkFromElement(sinkEle).SetCallbacks(&app.SinkCallbacks{ NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { sample := sink.PullSample() if sample == nil { return gst.FlowEOS } buf := sample.GetBuffer() pts, dts := buf.PresentationTimestamp(), buf.DecodingTimestamp() if pts != gst.ClockTimeNone && dts != gst.ClockTimeNone { offsets = append(offsets, time.Duration(int64(pts)-int64(dts))) } return gst.FlowOK }, })
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, "demux flat-wrapped segment") require.NotEmpty(t, offsets, "segment has video samples with timestamps") return offsets}
// TestMP4IngestMistRealSample runs a real MistServer live-MP4 capture through// the worker: a VideoToolbox (macOS hardware encoder) 720p H264 + AAC RTMP// push, pulled from Mist's HTTP .mp4 output — exactly what production ingest// consumes since the MKV→fMP4 bridge rewrite. VideoToolbox is the interesting// encoder here because its SPS declares no reorder window, which is what sent// the old MKV path's h264timestamper into its spurious-offset fallback; this// stream must instead come through with its real timestamps: PTS == DTS on// every video sample of every signed segment.func TestMP4IngestMistRealSample(t *testing.T) { ctx := context.Background() mp4, err := os.ReadFile(remote.RemoteFixture("ee4d7f8f9b267ba8229314162ef268186048f91ac4242b13fce3f5ee955b97ae/mist-vt-720p.mp4")) require.NoError(t, err)
segs, err := runMP4ThroughIngestWorkerSegments(t, mp4, false) require.NoError(t, err, "real Mist fMP4 capture ingests cleanly") require.GreaterOrEqual(t, len(segs), 3, "capture emits its segments")
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, 200*time.Millisecond, "segment %d duration sane", i) require.Less(t, dur, 6*time.Second, "segment %d duration not stretched", i) require.False(t, res.MediaData.Video[0].BFrames, "VideoToolbox capture has no B-frames") for _, off := range videoPTSDTSOffsets(t, ctx, seg) { require.Equal(t, time.Duration(0), off, "segment %d: no-reorder stream must keep PTS == DTS — a nonzero offset means ingest invented a reorder delay", i) } } t.Logf("real Mist capture: %d segments, all validated with PTS == DTS", len(segs))}
// makeBFrameAACFMP4 synthesizes an H264 stream WITH B-frames (PTS ≠ DTS)// alongside AAC audio in a fragmented MP4 — the shape a hardware encoder or// non-zerolatency x264 push produces, as delivered by MistServer's live .mp4// output. Unlike Matroska, MP4 track fragments carry real decode timestamps,// so the ingest pipeline needs no DTS reconstruction for the fMP4 muxer to mux// the reordered stream correctly. b-adapt=false forces x264 to actually emit// the configured B-frames rather than deciding per-scene.func makeBFrameAACFMP4(t *testing.T, ctx context.Context, seconds int) []byte { t.Helper() 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 ! mp4mux name=mux fragment-duration=500 ! 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") return runSynthPipeline(t, ctx, desc)}
// TestMP4IngestBFramesValidate is the regression test for B-frame ingest:// every emitted segment must survive the full ValidateMP4Media chokepoint with// a sane duration, and the stream's real reorder offsets must be preserved.// (On the old MKV path this scenario originally lost DTS entirely — mp4mux// stretched the video track ~2.2× and every segment failed validation with// "no audio in segment"; the h264timestamper fix for that in turn minted// spurious offsets on no-reorder streams. fMP4's container DTS sidesteps the// whole trade-off, and this test pins the B-frame half of it.)func TestMP4IngestBFramesValidate(t *testing.T) { ctx := context.Background() mp4 := makeBFrameAACFMP4(t, ctx, 8)
segs, err := runMP4ThroughIngestWorkerSegments(t, mp4, false) require.NoError(t, err, "B-frame stream ingests cleanly") require.GreaterOrEqual(t, len(segs), 6, "B-frame stream emits its segments")
sawBFrames := false sawReorderOffset := 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 } for _, off := range videoPTSDTSOffsets(t, ctx, seg) { if off > 0 { sawReorderOffset = true } } } require.True(t, sawBFrames, "synthesized stream actually contains B-frames — if this fails the test no longer exercises the reorder path") require.True(t, sawReorderOffset, "B-frame stream keeps its real PTS−DTS reorder offsets through ingest") t.Logf("B-frame stream: %d segments, all validated", len(segs))}
// TestMP4IngestRejectsMatroskaWithDiagnosis: an MKV stream landing on the// fMP4 ingest (a MistServer still running the legacy MKVExec process config)// must fail immediately with a message that names the actual problem — not a// generic qtdemux parse error on an endless Mist-side restart loop.func TestMP4IngestRejectsMatroskaWithDiagnosis(t *testing.T) { mkvish := append(append([]byte{}, matroskaMagic...), make([]byte, 1024)...) _, err := runMP4ThroughIngestWorker(t, mkvish, false) require.Error(t, err) require.Contains(t, err.Error(), "Matroska") require.Contains(t, err.Error(), "MKVExec")}