diff --git a/machinery/src/cloud/livemoq/annexb.go b/machinery/src/cloud/livemoq/annexb.go index 7f4e4f5..f6444e7 100644 --- a/machinery/src/cloud/livemoq/annexb.go +++ b/machinery/src/cloud/livemoq/annexb.go @@ -10,6 +10,11 @@ import ( var annexBStartCode = []byte{0x00, 0x00, 0x00, 0x01} +// H264NormalizationStats reports malformed duplication removed from an access unit. +type H264NormalizationStats struct { + DuplicateIDRNALUs int +} + // EnsureAnnexB restores the start code stripped by the Agent capture queue. func EnsureAnnexB(payload []byte) []byte { if hasAnnexBStartCode(payload) { @@ -21,20 +26,28 @@ func EnsureAnnexB(payload []byte) []byte { return append(framed, payload...) } -// NormalizeH264AccessUnit removes delimiters and duplicate parameter sets that -// can make older MoQ splitters emit a parameter-only frame before the IDR. +// NormalizeH264AccessUnit removes delimiters and exact duplicate parameter-set +// or IDR NALUs that can confuse older MoQ splitters and decoders. func NormalizeH264AccessUnit(payload []byte) ([]byte, error) { + normalized, _, err := NormalizeH264AccessUnitWithStats(payload) + return normalized, err +} + +// NormalizeH264AccessUnitWithStats also reports exact duplicate IDR NALUs. +func NormalizeH264AccessUnitWithStats(payload []byte) ([]byte, H264NormalizationStats, error) { nalus, err := h264.AnnexBUnmarshal(EnsureAnnexB(payload)) if err != nil { - return nil, err + return nil, H264NormalizationStats{}, err } + stats := H264NormalizationStats{} normalized := make([][]byte, 0, len(nalus)) for _, nalu := range nalus { if len(nalu) == 0 || nalu[0]&0x1f == 9 { continue } - if nalu[0]&0x1f == 7 || nalu[0]&0x1f == 8 { + naluType := nalu[0] & 0x1f + if naluType == 7 || naluType == 8 || naluType == 5 { duplicate := false for _, existing := range normalized { if bytes.Equal(existing, nalu) { @@ -43,13 +56,17 @@ func NormalizeH264AccessUnit(payload []byte) ([]byte, error) { } } if duplicate { + if naluType == 5 { + stats.DuplicateIDRNALUs++ + } continue } } normalized = append(normalized, nalu) } - return h264.AnnexBMarshal(normalized) + result, err := h264.AnnexBMarshal(normalized) + return result, stats, err } // BroadcastPath returns the relay path a quality tier is published on. Every diff --git a/machinery/src/cloud/livemoq/annexb_test.go b/machinery/src/cloud/livemoq/annexb_test.go index a361d6f..a929a23 100644 --- a/machinery/src/cloud/livemoq/annexb_test.go +++ b/machinery/src/cloud/livemoq/annexb_test.go @@ -66,6 +66,34 @@ func TestNormalizeH264AccessUnit(t *testing.T) { } } +func TestNormalizeH264AccessUnitRemovesOnlyExactDuplicateIDRSlices(t *testing.T) { + startCode := []byte{0x00, 0x00, 0x00, 0x01} + idrSlice1 := []byte{0x65, 0x88, 0x84} + idrSlice2 := []byte{0x65, 0x44, 0x22} + + payload := make([]byte, 0) + for _, nalu := range [][]byte{idrSlice1, idrSlice1, idrSlice2} { + payload = append(payload, startCode...) + payload = append(payload, nalu...) + } + + got, stats, err := NormalizeH264AccessUnitWithStats(payload) + if err != nil { + t.Fatal(err) + } + want := make([]byte, 0) + for _, nalu := range [][]byte{idrSlice1, idrSlice2} { + want = append(want, startCode...) + want = append(want, nalu...) + } + if !bytes.Equal(got, want) { + t.Fatalf("NormalizeH264AccessUnitWithStats() = %x, want %x", got, want) + } + if stats.DuplicateIDRNALUs != 1 { + t.Fatalf("DuplicateIDRNALUs = %d, want 1", stats.DuplicateIDRNALUs) + } +} + func TestBroadcastPath(t *testing.T) { if got := BroadcastPath("/devices/", "/camera-1/", models.StreamQualityHigh); got != "devices/camera-1/live.hang" { t.Fatalf("BroadcastPath() high = %q", got) diff --git a/machinery/src/cloud/livemoq/dedup.go b/machinery/src/cloud/livemoq/dedup.go new file mode 100644 index 0000000..71d2d48 --- /dev/null +++ b/machinery/src/cloud/livemoq/dedup.go @@ -0,0 +1,41 @@ +package livemoq + +import ( + "crypto/sha256" + "time" +) + +type KeyframeDeduplicator struct { + hasPrevious bool + timestampMs int64 + capturedAtMs int64 + observedAt time.Time + digest [sha256.Size]byte +} + +func (d *KeyframeDeduplicator) Reset() { + *d = KeyframeDeduplicator{} +} + +// IsDuplicate reports exact repeated keyframe access units observed close +// together. Distinct IDR slices within one access unit remain untouched. +func (d *KeyframeDeduplicator) IsDuplicate(timestampMs int64, capturedAtMs int64, payload []byte, observedAt time.Time, window time.Duration) bool { + digest := sha256.Sum256(payload) + duplicate := d.hasPrevious && d.timestampMs == timestampMs && d.digest == digest + if duplicate { + if capturedAtMs > 0 && d.capturedAtMs > 0 { + gap := time.Duration(capturedAtMs-d.capturedAtMs) * time.Millisecond + duplicate = gap >= 0 && gap <= window + } else { + gap := observedAt.Sub(d.observedAt) + duplicate = gap >= 0 && gap <= window + } + } + + d.hasPrevious = true + d.timestampMs = timestampMs + d.capturedAtMs = capturedAtMs + d.observedAt = observedAt + d.digest = digest + return duplicate +} diff --git a/machinery/src/cloud/livemoq/dedup_test.go b/machinery/src/cloud/livemoq/dedup_test.go new file mode 100644 index 0000000..76d45aa --- /dev/null +++ b/machinery/src/cloud/livemoq/dedup_test.go @@ -0,0 +1,60 @@ +package livemoq + +import ( + "testing" + "time" +) + +func TestKeyframeDeduplicator(t *testing.T) { + now := time.UnixMilli(10_000) + window := 500 * time.Millisecond + payload := []byte{0x00, 0x00, 0x00, 0x01, 0x65, 0x88} + deduplicator := KeyframeDeduplicator{} + + if deduplicator.IsDuplicate(1_000, 10_000, payload, now, window) { + t.Fatal("first keyframe reported as duplicate") + } + if !deduplicator.IsDuplicate(1_000, 10_020, payload, now.Add(20*time.Millisecond), window) { + t.Fatal("exact repeated keyframe was not reported as duplicate") + } + if deduplicator.IsDuplicate(2_000, 11_000, payload, now.Add(time.Second), window) { + t.Fatal("same payload with a new timestamp reported as duplicate") + } + if deduplicator.IsDuplicate(2_000, 11_020, append(payload, 0x01), now.Add(1020*time.Millisecond), window) { + t.Fatal("different payload with the same timestamp reported as duplicate") + } +} + +func TestKeyframeDeduplicatorAllowsTimestampReuseOutsideWindow(t *testing.T) { + now := time.UnixMilli(10_000) + payload := []byte{0x00, 0x00, 0x00, 0x01, 0x65, 0x88} + deduplicator := KeyframeDeduplicator{} + + deduplicator.IsDuplicate(1_000, 10_000, payload, now, 500*time.Millisecond) + if deduplicator.IsDuplicate(1_000, 20_000, payload, now.Add(10*time.Second), 500*time.Millisecond) { + t.Fatal("later keyframe after timestamp reset reported as duplicate") + } +} + +func TestKeyframeDeduplicatorFallsBackToObservationTime(t *testing.T) { + now := time.UnixMilli(10_000) + payload := []byte{0x65, 0x88} + deduplicator := KeyframeDeduplicator{} + + deduplicator.IsDuplicate(1_000, 0, payload, now, 500*time.Millisecond) + if !deduplicator.IsDuplicate(1_000, 0, payload, now.Add(20*time.Millisecond), 500*time.Millisecond) { + t.Fatal("duplicate without capture time was not reported") + } +} + +func TestKeyframeDeduplicatorReset(t *testing.T) { + now := time.UnixMilli(10_000) + payload := []byte{0x65, 0x88} + deduplicator := KeyframeDeduplicator{} + + deduplicator.IsDuplicate(1_000, 10_000, payload, now, 500*time.Millisecond) + deduplicator.Reset() + if deduplicator.IsDuplicate(1_000, 10_020, payload, now.Add(20*time.Millisecond), 500*time.Millisecond) { + t.Fatal("first keyframe after reset reported as duplicate") + } +} diff --git a/machinery/src/cloud/livemoq_enabled.go b/machinery/src/cloud/livemoq_enabled.go index 92e3f6d..8aa371d 100644 --- a/machinery/src/cloud/livemoq_enabled.go +++ b/machinery/src/cloud/livemoq_enabled.go @@ -24,6 +24,7 @@ const ( maxMoQLivePacketAge = 1500 * time.Millisecond slowMoQWriteThreshold = 100 * time.Millisecond moQWriteWarningInterval = 10 * time.Second + duplicateKeyframeWindow = 500 * time.Millisecond ) type liveMoQConfig struct { @@ -174,7 +175,9 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { cursor := config.queue.Latest() gate := livemoq.FrameGate{} + deduplicator := livemoq.KeyframeDeduplicator{} var lastSlowWriteWarning time.Time + var lastDuplicateKeyframeWarning time.Time idle := false for { packet, err := cursor.ReadPacket() @@ -186,6 +189,7 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { // nothing. The gate is closed so the next viewer resumes on a keyframe. if !idle { gate.Reset() + deduplicator.Reset() idle = true } continue @@ -206,10 +210,27 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { if !allowed { continue } - payload, err := livemoq.NormalizeH264AccessUnit(packet.Data) + payload, normalizationStats, err := livemoq.NormalizeH264AccessUnitWithStats(packet.Data) if err != nil { return fmt.Errorf("normalize H.264 access unit: %w", err) } + if normalizationStats.DuplicateIDRNALUs > 0 && time.Since(lastDuplicateKeyframeWarning) >= moQWriteWarningInterval { + log.Log.Warning(fmt.Sprintf( + "cloud.publishLiveStreamMoQ(): %s removed %d duplicate IDR NALU(s) from H.264 keyframe (timestamp_ms=%d)", + config.label(), normalizationStats.DuplicateIDRNALUs, packet.Time, + )) + lastDuplicateKeyframeWarning = time.Now() + } + if packet.IsKeyFrame && deduplicator.IsDuplicate(packet.Time, packet.CurrentTime, payload, time.Now(), duplicateKeyframeWindow) { + if time.Since(lastDuplicateKeyframeWarning) >= moQWriteWarningInterval { + log.Log.Warning(fmt.Sprintf( + "cloud.publishLiveStreamMoQ(): %s dropping duplicate H.264 keyframe (timestamp_ms=%d)", + config.label(), packet.Time, + )) + lastDuplicateKeyframeWarning = time.Now() + } + continue + } frame := moq.Frame{ Payload: payload, TimestampUs: livemoq.TimestampUs(packet.Time),