diff --git a/machinery/src/capture/gortsplib.go b/machinery/src/capture/gortsplib.go index 9d6653c..d41e954 100644 --- a/machinery/src/capture/gortsplib.go +++ b/machinery/src/capture/gortsplib.go @@ -953,12 +953,31 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets pkt.Data = pkt.Data[4:] if pkt.IsKeyFrame { - annexbNALUStartCode := func() []byte { return []byte{0x00, 0x00, 0x00, 0x01} } - pkt.Data = append(annexbNALUStartCode(), pkt.Data...) - pkt.Data = append(g.VideoH264Forma.PPS, pkt.Data...) - pkt.Data = append(annexbNALUStartCode(), pkt.Data...) - pkt.Data = append(g.VideoH264Forma.SPS, pkt.Data...) - pkt.Data = append(annexbNALUStartCode(), pkt.Data...) + // Prepend SPS/PPS (when available) in front of every keyframe so the + // access unit is self-contained. Downstream decoders (and the MP4 writer's + // in-band parameter-set recovery) rely on this; a recording whose first + // frame lacks SPS/PPS produces an MP4 with an empty avcC, which makes FFmpeg + // report "non-existing PPS 0 referenced". + // + // Build the payload in a freshly allocated buffer. The previous code + // did append(g.VideoH264Forma.PPS, pkt.Data...): because the SPS/PPS + // slices are sub-slices of the RTP reassembly buffer (spare capacity), + // that append wrote into - and corrupted - the shared parameter-set + // backing arrays, occasionally poisoning the SPS/PPS stored for the + // recording. + startCode := []byte{0x00, 0x00, 0x00, 0x01} + out := make([]byte, 0, len(g.VideoH264Forma.SPS)+len(g.VideoH264Forma.PPS)+len(pkt.Data)+12) + if len(g.VideoH264Forma.SPS) > 0 { + out = append(out, startCode...) + out = append(out, g.VideoH264Forma.SPS...) + } + if len(g.VideoH264Forma.PPS) > 0 { + out = append(out, startCode...) + out = append(out, g.VideoH264Forma.PPS...) + } + out = append(out, startCode...) + out = append(out, pkt.Data...) + pkt.Data = out } writeStart := time.Now() diff --git a/machinery/src/video/mp4.go b/machinery/src/video/mp4.go index c63281d..791bc6f 100644 --- a/machinery/src/video/mp4.go +++ b/machinery/src/video/mp4.go @@ -32,6 +32,16 @@ const MacEpochOffset uint64 = 2082844800 // resulting in ~3 second fragments (assuming a typical GOP interval). const FragmentDurationMs = 3000 +// SeamGapDivisor controls loop-seam detection. A keyframe is treated as an +// upstream loop/restart seam when it arrives in less than (previous keyframe +// interval / SeamGapDivisor) — i.e. far sooner than the established keyframe +// cadence. Comparing against the *previous* interval (rather than a fixed +// millisecond threshold) makes the check scale automatically with the camera's +// configured GOP size: it works the same whether keyframes are 0.5s, 1s, 2s or +// more apart, and does not misfire on legitimately short-GOP or all-intra +// streams (where every interval is similar, so none looks anomalously short). +const SeamGapDivisor = 2 + type MP4 struct { // FileName is the name of the file FileName string @@ -74,6 +84,8 @@ type MP4 struct { TotalKeyframesWritten int // Total keyframes written to trun boxes FragmentKeyframeCount int // Keyframes in the current fragment PendingSampleIsKeyframe bool // Whether the pending video sample is a keyframe + LastKeyframeRawPTS uint64 // Raw PTS of the most recently seen keyframe (across fragments) + LastKeyframeGapMs uint64 // Interval (ms) between the two most recent keyframes; reference cadence for seam detection } // NewMP4 creates a new MP4 object. @@ -298,6 +310,42 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p } shouldFlush := !mp4.Start || elapsed >= FragmentDurationMs + // Detect an upstream source-loop / restart discontinuity. When a source + // MP4 is looped through virtual-rtsp (ffmpeg `-stream_loop -1 -re`) the + // loop seam emits a fresh IDR much sooner than a normal GOP would. PTS + // keeps growing monotonically, so the timing-only `elapsed` check above + // does not catch it and the seam IDR ends up as a mid-fragment sync + // sample. macOS VideoToolbox rejects such a fragment with + // kVTVideoDecoderBadDataErr (-12909) and MSE players (Video.js / + // Chromium / Firefox) report media corruption, because the inner IDR + // resets frame_num/POC inside what they expect to be a single GOP. This + // is exactly what freezes playback around the seam (e.g. the ~10s mark). + // Force a fragment boundary so the seam IDR starts its own fragment. + // + // The GOP size is configurable per camera, so we do NOT compare against a + // fixed millisecond threshold. Instead we compare this keyframe interval + // to the previous one and only flag a *sudden* shortening: a seam IDR + // arrives in less than (previous interval / SeamGapDivisor). Deriving the + // threshold from the observed cadence keeps detection correct for any + // configured GOP (0.5s, 1s, 2s, ...) and avoids false positives on + // steady short-GOP / all-intra streams (where consecutive intervals are + // similar, so none looks anomalously short). Because the reference is the + // immediately preceding interval, a burst of close keyframes only forces + // a single flush instead of cascading into many tiny fragments. + if trackID == uint32(mp4.VideoTrack) && mp4.Start && + mp4.LastKeyframeRawPTS > 0 && pts > mp4.LastKeyframeRawPTS { + gap := pts - mp4.LastKeyframeRawPTS + if !shouldFlush && mp4.LastKeyframeGapMs > 0 && + gap*SeamGapDivisor < mp4.LastKeyframeGapMs { + log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): forcing fragment flush at unexpectedly close keyframe (interval=%d ms, previous interval=%d ms, fragment elapsed=%d ms) - likely upstream loop/restart discontinuity", gap, mp4.LastKeyframeGapMs, elapsed)) + shouldFlush = true + } + mp4.LastKeyframeGapMs = gap + } + if trackID == uint32(mp4.VideoTrack) { + mp4.LastKeyframeRawPTS = pts + } + if shouldFlush { // Write the previous segment to the file if mp4.Start { @@ -571,6 +619,13 @@ func (mp4 *MP4) Close(config *models.Config) { includePS := true spsNALUs, ppsNALUs := normalizeH264ParameterSets(mp4.SPSNALUs, mp4.PPSNALUs) log.Log.Debug("mp4.Close(): AVC parameter sets: SPS=" + formatNaluDebug(spsNALUs) + ", PPS=" + formatNaluDebug(ppsNALUs)) + if len(spsNALUs) == 0 || len(ppsNALUs) == 0 { + // An avcC without both SPS and PPS is invalid: downstream FFmpeg-based + // pipelines decoding this file will report "non-existing PPS 0 referenced" + // and fail to extract any frame. Surface it loudly so the capture-side + // parameter-set handling can be diagnosed. + log.Log.Error(fmt.Sprintf("mp4.Close(): incomplete H264 parameter sets (SPS=%d, PPS=%d) - the avcC will be invalid and downstream decoders will report 'non-existing PPS 0 referenced'", len(spsNALUs), len(ppsNALUs))) + } err := init.Moov.Traks[0].SetAVCDescriptor("avc1", spsNALUs, ppsNALUs, includePS) if err != nil { log.Log.Error("mp4.Close(): error setting AVC descriptor: " + err.Error()) @@ -597,6 +652,11 @@ func (mp4 *MP4) Close(config *models.Config) { includePS := true vpsNALUs, spsNALUs, ppsNALUs := normalizeH265ParameterSets(mp4.VPSNALUs, mp4.SPSNALUs, mp4.PPSNALUs) log.Log.Debug("mp4.Close(): HEVC parameter sets: VPS=" + formatNaluDebug(vpsNALUs) + ", SPS=" + formatNaluDebug(spsNALUs) + ", PPS=" + formatNaluDebug(ppsNALUs)) + if len(vpsNALUs) == 0 || len(spsNALUs) == 0 || len(ppsNALUs) == 0 { + // An hvcC missing VPS/SPS/PPS is invalid and downstream FFmpeg-based + // pipelines will fail to decode the recording. Surface it loudly. + log.Log.Error(fmt.Sprintf("mp4.Close(): incomplete H265 parameter sets (VPS=%d, SPS=%d, PPS=%d) - the hvcC will be invalid and downstream decoders will fail to process the recording", len(vpsNALUs), len(spsNALUs), len(ppsNALUs))) + } err := init.Moov.Traks[0].SetHEVCDescriptor("hvc1", vpsNALUs, spsNALUs, ppsNALUs, [][]byte{}, includePS) if err != nil { log.Log.Error("mp4.Close(): error setting HEVC descriptor: " + err.Error()) diff --git a/machinery/src/video/mp4_loopseam_test.go b/machinery/src/video/mp4_loopseam_test.go new file mode 100644 index 0000000..7b843a9 --- /dev/null +++ b/machinery/src/video/mp4_loopseam_test.go @@ -0,0 +1,163 @@ +package video + +import ( + "os" + "testing" + + mp4ff "github.com/Eyevinn/mp4ff/mp4" + "github.com/kerberos-io/agent/machinery/src/models" +) + +// runLoopSeamScenario builds a fragmented MP4 that reproduces the loop-seam +// pattern observed in the failing virtual-rtsp recordings (see +// thales_1781183923_3-758_2top_0-0-0-0_-1_30221.mp4): a steady GOP cadence, +// but at the source-MP4 loop boundary an IDR arrives prematurely - far sooner +// than a normal GOP. Without the fix this seam IDR ends up bunched into the +// same fragment as the prior GOP's IDR, producing a mid-fragment sync sample +// that trips macOS VideoToolbox (kVTVideoDecoderBadDataErr / -12909) and +// MSE-based players, freezing playback around the seam (~10s in the original +// file). +// +// gopFrames is the number of frames per GOP, so the same scenario can be +// exercised at different (configurable) camera GOP sizes. The fix derives its +// threshold from the observed keyframe cadence, so the seam must be isolated +// into its own fragment regardless of GOP size. +func runLoopSeamScenario(t *testing.T, gopFrames int) { + t.Helper() + + tmpFile, err := os.CreateTemp("", "test_loop_seam_*.mp4") + if err != nil { + t.Fatalf("create temp: %v", err) + } + tmpFile.Close() + defer os.Remove(tmpFile.Name()) + + sps := []byte{0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0xa0, 0x47, 0xfe, 0xc8} + pps := []byte{0x68, 0xce, 0x38, 0x80} + mp4Video := NewMP4(tmpFile.Name(), [][]byte{sps}, [][]byte{pps}, nil, 60) + mp4Video.SetWidth(1920) + mp4Video.SetHeight(1080) + v := mp4Video.AddVideoTrack("H264") + + mk := func(k bool) []byte { + nt := byte(0x01) + if k { + nt = 0x65 + } + f := []byte{0, 0, 0, 1, nt} + for i := 0; i < 200; i++ { + f = append(f, byte(i)) + } + return f + } + + frameDur := uint64(33) + normalGOPms := uint64(gopFrames) * frameDur + pts := uint64(0) + emitFrame := func(isKey bool) { + // compositionOffset is 0: synthetic stream has no B-frames. + mp4Video.AddSampleToTrack(v, isKey, mk(isKey), pts, 0) + pts += frameDur + } + emitP := func(n int) { + for i := 0; i < n; i++ { + emitFrame(false) + } + } + // emitGOP emits one GOP: a leading keyframe followed by gopFrames-1 P-frames. + emitGOP := func() { + emitFrame(true) + emitP(gopFrames - 1) + } + + // Several healthy GOPs to establish the cadence and fill a couple of + // fragments, then the last "good" keyframe followed by a few P-frames so we + // are clearly mid-fragment when the seam arrives. + for g := 0; g < 9; g++ { + emitGOP() + } + emitFrame(true) + seamLead := gopFrames / 5 // seam IDR arrives ~20% into the GOP + if seamLead < 1 { + seamLead = 1 + } + emitP(seamLead) + // Loop seam: the source recording restarts, emitting a fresh IDR far sooner + // than the normal GOP. Without the fix this lands as a mid-fragment sync + // sample and freezes playback at the seam. + emitFrame(true) + emitP(gopFrames - 1) // the seam IDR opens a fresh, healthy GOP + // The recording continues with normal GOPs to the end. + for g := 0; g < 10; g++ { + emitGOP() + } + + mp4Video.Close(&models.Config{Signing: &models.Signing{PrivateKey: ""}}) + + f, err := os.Open(tmpFile.Name()) + if err != nil { + t.Fatalf("open: %v", err) + } + defer f.Close() + parsed, err := mp4ff.DecodeFile(f) + if err != nil { + t.Fatalf("decode: %v", err) + } + + // A healthy fragment only ever contains keyframes spaced ~normalGOPms apart. + // If any fragment contains two keyframes closer than half a normal GOP, the + // premature seam IDR was not isolated and the file will freeze on playback. + maxBunchMs := normalGOPms / 2 + + fragIdx := 0 + for _, seg := range parsed.Segments { + for _, fr := range seg.Fragments { + for _, traf := range fr.Moof.Trafs { + if traf.Tfhd.TrackID != 1 { + continue + } + tfdt := traf.Tfdt.BaseMediaDecodeTime() + offset := uint64(0) + var keys []uint64 + for _, trun := range traf.Truns { + for _, s := range trun.Samples { + // sample_depends_on == 2 => "does not depend on others" => IDR/sync. + if (s.Flags>>24)&0x03 == 0x02 { + keys = append(keys, offset) + } + offset += uint64(s.Dur) + } + } + t.Logf("gop=%dframes frag %d tfdt=%d samples_dur=%d keys@%v", gopFrames, fragIdx, tfdt, offset, keys) + for i := 1; i < len(keys); i++ { + gap := keys[i] - keys[i-1] + if gap < maxBunchMs { + t.Errorf("gop=%dframes frag %d (tfdt=%d): two IDRs only %d ms apart in same fragment (< %d) - seam was not isolated", + gopFrames, fragIdx, tfdt, gap, maxBunchMs) + } + } + fragIdx++ + } + } + } +} + +// TestMP4LoopSeamIsolation exercises the ~1s GOP case (30 frames @ ~33ms), +// matching the original failing recording. +func TestMP4LoopSeamIsolation(t *testing.T) { + runLoopSeamScenario(t, 30) +} + +// TestMP4LoopSeamIsolationLargeGOP exercises a larger ~2s GOP (60 frames). The +// GOP size is configurable per camera; this guards against regressing to a +// fixed-millisecond threshold that would only work for ~1s GOPs. +func TestMP4LoopSeamIsolationLargeGOP(t *testing.T) { + runLoopSeamScenario(t, 60) +} + +// TestMP4LoopSeamIsolationShortGOP exercises a short ~0.5s GOP (15 frames), +// where a fixed ~1s threshold would misfire on every keyframe. The relative +// detection must only isolate the genuine premature seam IDR. +func TestMP4LoopSeamIsolationShortGOP(t *testing.T) { + runLoopSeamScenario(t, 15) +}