Deduplicate repeated H.264 keyframes

Remove exact duplicate IDR NALUs during normalization and drop repeated keyframes within a short timestamp window. Add normalization statistics, logging, reset handling, and coverage for deduplication behavior.

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

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)

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
+}

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")
+	}
+}

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),