Merge pull request #313 from kerberos-io/fix/moq-recovery-strategy

fix/moq-recovery-strategy

README.md

@@ -429,10 +429,20 @@ uses Debian Trixie. The publisher is disabled unless explicitly enabled at runti
       -e AGENT_LIVE_MOQ_URL=https://relay.uug.ai/anon \
       kerberos/agent
 
-`AGENT_LIVE_MOQ_BROADCAST_PREFIX` defaults to `devices`, producing the broadcast
-`devices/<agent-key>/live.hang`. `AGENT_LIVE_MOQ_QUALITY` accepts `auto` (the
-default), `high`, or `low` and selects the main or sub camera stream when the
-Agent starts. The initial implementation publishes H.264 video only.
+`AGENT_LIVE_MOQ_BROADCAST_PREFIX` defaults to `devices`. MoQ viewers subscribe to
+a relay and never negotiate with the Agent, so every quality tier is published as
+its own broadcast and switching quality is simply a resubscribe:
+
+| Tier   | Broadcast                             | Source                                     |
+| ------ | ------------------------------------- | ------------------------------------------ |
+| `high` | `devices/<agent-key>/live.hang`       | highest-resolution camera stream           |
+| `low`  | `devices/<agent-key>/live-low.hang`   | sub stream (main stream when none is set)  |
+
+Each tier only uploads while it has at least one subscriber, so the tier nobody
+watches costs virtually no bandwidth. `AGENT_LIVE_MOQ_QUALITY` accepts `high` or
+`low` to pin the Agent to a single tier; viewers requesting the other tier then
+find no broadcast. Any other value (including the default) publishes both. The
+initial implementation publishes H.264 video only.
 
 The `/anon` relay route is intended for interoperability testing. Production
 deployments must set `AGENT_LIVE_MOQ_URL` to a short-lived, device-scoped

machinery/src/cloud/livemoq/annexb.go

@@ -5,10 +5,16 @@ import (
 	"strings"
 
 	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
+	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
 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) {
@@ -20,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) {
@@ -42,21 +56,35 @@ 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
 }
 
-func BroadcastPath(prefix string, deviceKey string) string {
+// BroadcastPath returns the relay path a quality tier is published on. Every
+// tier gets its own broadcast so a viewer switches between the camera's main and
+// sub stream by resubscribing to another path, without any control channel back
+// to the Agent. The high tier keeps the historical ".../live.hang" path so
+// existing viewers keep working; the low tier lives next to it on
+// ".../live-low.hang".
+func BroadcastPath(prefix string, deviceKey string, quality string) string {
 	prefix = strings.Trim(prefix, "/")
 	if prefix == "" {
 		prefix = "devices"
 	}
-	return prefix + "/" + strings.Trim(deviceKey, "/") + "/live.hang"
+	name := "live.hang"
+	if quality == models.StreamQualityLow {
+		name = "live-low.hang"
+	}
+	return prefix + "/" + strings.Trim(deviceKey, "/") + "/" + name
 }
 
 // TimestampUs converts the capture presentation timestamp from milliseconds.

machinery/src/cloud/livemoq/annexb_test.go

@@ -3,6 +3,8 @@ package livemoq
 import (
 	"bytes"
 	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
 func TestEnsureAnnexB(t *testing.T) {
@@ -64,13 +66,44 @@ func TestNormalizeH264AccessUnit(t *testing.T) {
 	}
 }
 
-func TestBroadcastPath(t *testing.T) {
-	if got := BroadcastPath("/devices/", "/camera-1/"); got != "devices/camera-1/live.hang" {
-		t.Fatalf("BroadcastPath() = %q", got)
+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...)
 	}
-	if got := BroadcastPath("", "camera-1"); got != "devices/camera-1/live.hang" {
+
+	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)
+	}
+	if got := BroadcastPath("", "camera-1", models.StreamQualityHigh); got != "devices/camera-1/live.hang" {
 		t.Fatalf("BroadcastPath() default = %q", got)
 	}
+	if got := BroadcastPath("", "camera-1", models.StreamQualityLow); got != "devices/camera-1/live-low.hang" {
+		t.Fatalf("BroadcastPath() low = %q", got)
+	}
 }
 
 func TestTimestampUs(t *testing.T) {

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/recovery.go

@@ -0,0 +1,55 @@
+package livemoq
+
+import "time"
+
+type FrameGateEvent uint8
+
+const (
+	FrameGateEventNone FrameGateEvent = iota
+	FrameGateEventStarted
+	FrameGateEventLagging
+	FrameGateEventRecovered
+)
+
+// FrameGate keeps publication on a decodable, recent GOP.
+type FrameGate struct {
+	started    bool
+	recovering bool
+}
+
+// Reset closes the gate so publication resumes on the next keyframe. It is used
+// when the publisher stopped writing for a reason unrelated to the stream health
+// (no subscribers), so the next viewer never receives a partial GOP.
+func (g *FrameGate) Reset() {
+	g.started = false
+	g.recovering = false
+}
+
+// Allow rejects stale frames and waits for a fresh keyframe before reopening.
+func (g *FrameGate) Allow(isKeyFrame bool, capturedAtMs int64, now time.Time, maxAge time.Duration) (bool, FrameGateEvent) {
+	if capturedAtMs > 0 && now.Sub(time.UnixMilli(capturedAtMs)) > maxAge {
+		event := FrameGateEventNone
+		if g.started {
+			if !g.recovering {
+				event = FrameGateEventLagging
+			}
+			g.started = false
+			g.recovering = true
+		}
+		return false, event
+	}
+
+	if !g.started {
+		if !isKeyFrame {
+			return false, FrameGateEventNone
+		}
+		g.started = true
+		if g.recovering {
+			g.recovering = false
+			return true, FrameGateEventRecovered
+		}
+		return true, FrameGateEventStarted
+	}
+
+	return true, FrameGateEventNone
+}

machinery/src/cloud/livemoq/recovery_test.go

@@ -0,0 +1,49 @@
+package livemoq
+
+import (
+	"testing"
+	"time"
+)
+
+func TestFrameGateRecoversAtFreshKeyframe(t *testing.T) {
+	now := time.UnixMilli(10_000)
+	maxAge := 1500 * time.Millisecond
+	gate := FrameGate{}
+
+	tests := []struct {
+		name         string
+		isKeyFrame   bool
+		capturedAtMs int64
+		wantAllowed  bool
+		wantEvent    FrameGateEvent
+	}{
+		{name: "waits for initial keyframe", capturedAtMs: 10_000},
+		{name: "starts at initial keyframe", isKeyFrame: true, capturedAtMs: 10_000, wantAllowed: true, wantEvent: FrameGateEventStarted},
+		{name: "publishes fresh delta", capturedAtMs: 10_020, wantAllowed: true},
+		{name: "detects stale packet", capturedAtMs: 8_000, wantEvent: FrameGateEventLagging},
+		{name: "rejects fresh delta while recovering", capturedAtMs: 10_040},
+		{name: "rejects stale keyframe without duplicate event", isKeyFrame: true, capturedAtMs: 8_000},
+		{name: "recovers at fresh keyframe", isKeyFrame: true, capturedAtMs: 10_060, wantAllowed: true, wantEvent: FrameGateEventRecovered},
+		{name: "publishes delta after recovery", capturedAtMs: 10_080, wantAllowed: true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			allowed, event := gate.Allow(test.isKeyFrame, test.capturedAtMs, now, maxAge)
+			if allowed != test.wantAllowed {
+				t.Fatalf("Allow() allowed = %t, want %t", allowed, test.wantAllowed)
+			}
+			if event != test.wantEvent {
+				t.Fatalf("Allow() event = %d, want %d", event, test.wantEvent)
+			}
+		})
+	}
+}
+
+func TestFrameGateAllowsMissingCaptureTime(t *testing.T) {
+	gate := FrameGate{}
+	allowed, event := gate.Allow(true, 0, time.Now(), time.Second)
+	if !allowed || event != FrameGateEventStarted {
+		t.Fatalf("Allow() = (%t, %d), want (true, %d)", allowed, event, FrameGateEventStarted)
+	}
+}

machinery/src/cloud/livemoq_enabled.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/kerberos-io/agent/machinery/src/cloud/livemoq"
@@ -17,9 +18,13 @@ import (
 )
 
 const (
-	defaultMoQRelayURL = "https://relay.uug.ai/anon"
-	minMoQRetryDelay   = time.Second
-	maxMoQRetryDelay   = 30 * time.Second
+	defaultMoQRelayURL      = "https://relay.uug.ai/anon"
+	minMoQRetryDelay        = time.Second
+	maxMoQRetryDelay        = 30 * time.Second
+	maxMoQLivePacketAge     = 1500 * time.Millisecond
+	slowMoQWriteThreshold   = 100 * time.Millisecond
+	moQWriteWarningInterval = 10 * time.Second
+	duplicateKeyframeWindow = 500 * time.Millisecond
 )
 
 type liveMoQConfig struct {
@@ -30,8 +35,23 @@ type liveMoQConfig struct {
 	queue       *packets.Queue
 }
 
+// label identifies the tier in log lines, since one Agent runs a publisher per
+// quality tier.
+func (c liveMoQConfig) label() string {
+	return c.quality + " (" + c.sourceLabel + " stream)"
+}
+
 // StartLiveStreamMoQ starts the publisher only in the dedicated MoQ build and
 // only when explicitly enabled by the deployment.
+//
+// Unlike WebRTC and HLS — where a viewer negotiates a session with the Agent and
+// can therefore ask for another quality on the fly — MoQ viewers subscribe to a
+// relay and never talk to the Agent. The quality selector is honoured by
+// publishing each tier as its OWN broadcast (see livemoq.BroadcastPath): the
+// high tier from the camera's highest-resolution stream and the low tier from
+// its sub stream, so switching quality in the frontend is a resubscribe to the
+// other path. Each tier only uploads while it actually has subscribers, so the
+// second broadcast is close to free when nobody watches it.
 func StartLiveStreamMoQ(configuration *models.Configuration, communication *models.Communication, subStreamEnabled bool) {
 	if os.Getenv("AGENT_LIVE_MOQ_ENABLED") != "true" {
 		return
@@ -47,39 +67,47 @@ func StartLiveStreamMoQ(configuration *models.Configuration, communication *mode
 		return
 	}
 
-	quality := os.Getenv("AGENT_LIVE_MOQ_QUALITY")
-	if quality == "" {
-		quality = models.StreamQualityAuto
-	}
-	useSub := models.SelectSubStreamForQuality(config, quality, subStreamEnabled)
-	queue := communication.Queue
-	sourceLabel := "main"
-	if useSub && communication.SubQueue != nil {
-		queue = communication.SubQueue
-		sourceLabel = "sub"
-	}
-	if queue == nil {
-		log.Log.Warning("cloud.StartLiveStreamMoQ(): selected packet queue is unavailable")
-		return
+	// Both tiers are published by default. AGENT_LIVE_MOQ_QUALITY pins the Agent
+	// to a single tier for deployments that must never publish the other one
+	// (viewers asking for the pinned-away tier then find no broadcast).
+	qualities := []string{models.StreamQualityHigh, models.StreamQualityLow}
+	switch strings.ToLower(strings.TrimSpace(os.Getenv("AGENT_LIVE_MOQ_QUALITY"))) {
+	case models.StreamQualityHigh:
+		qualities = []string{models.StreamQualityHigh}
+	case models.StreamQualityLow:
+		qualities = []string{models.StreamQualityLow}
 	}
 
 	relayURL := os.Getenv("AGENT_LIVE_MOQ_URL")
 	if relayURL == "" {
 		relayURL = defaultMoQRelayURL
 	}
-	publisherConfig := liveMoQConfig{
-		relayURL:    relayURL,
-		broadcast:   livemoq.BroadcastPath(os.Getenv("AGENT_LIVE_MOQ_BROADCAST_PREFIX"), config.Key),
-		quality:     quality,
-		sourceLabel: sourceLabel,
-		queue:       queue,
-	}
+	broadcastPrefix := os.Getenv("AGENT_LIVE_MOQ_BROADCAST_PREFIX")
 
 	ctx := context.Background()
 	if communication.Context != nil {
 		ctx = *communication.Context
 	}
-	go runLiveStreamMoQ(ctx, publisherConfig)
+
+	for _, quality := range qualities {
+		queue := communication.Queue
+		sourceLabel := "main"
+		if models.SelectSubStreamForQuality(config, quality, subStreamEnabled) && communication.SubQueue != nil {
+			queue = communication.SubQueue
+			sourceLabel = "sub"
+		}
+		if queue == nil {
+			log.Log.Warning("cloud.StartLiveStreamMoQ(): packet queue for the " + quality + " tier is unavailable")
+			continue
+		}
+		go runLiveStreamMoQ(ctx, liveMoQConfig{
+			relayURL:    relayURL,
+			broadcast:   livemoq.BroadcastPath(broadcastPrefix, config.Key, quality),
+			quality:     quality,
+			sourceLabel: sourceLabel,
+			queue:       queue,
+		})
+	}
 }
 
 func runLiveStreamMoQ(ctx context.Context, config liveMoQConfig) {
@@ -135,33 +163,123 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 	}
 	defer stream.Finish()
 
+	// Only upload while this tier is actually being watched. `publishing` starts
+	// true so the track becomes discoverable on the relay even before the first
+	// subscriber ever arrives; from the moment a viewer has attached once, the
+	// subscriber watcher takes over and idles the tier again when everybody left.
+	watchCtx, cancelWatch := context.WithCancel(ctx)
+	defer cancelWatch()
+	publishing := &atomic.Bool{}
+	publishing.Store(true)
+	go watchLiveStreamMoQSubscribers(watchCtx, stream, publishing, config)
+
 	cursor := config.queue.Latest()
-	writing := false
+	gate := livemoq.FrameGate{}
+	deduplicator := livemoq.KeyframeDeduplicator{}
+	var lastSlowWriteWarning time.Time
+	var lastDuplicateKeyframeWarning time.Time
+	idle := false
 	for {
 		packet, err := cursor.ReadPacket()
 		if err != nil {
 			return fmt.Errorf("read packet: %w", err)
 		}
+		if !publishing.Load() {
+			// Keep draining the cursor so we stay at the live edge, but publish
+			// nothing. The gate is closed so the next viewer resumes on a keyframe.
+			if !idle {
+				gate.Reset()
+				deduplicator.Reset()
+				idle = true
+			}
+			continue
+		}
+		idle = false
 		if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") {
 			continue
 		}
-		if !writing {
-			if !packet.IsKeyFrame {
-				continue
-			}
-			writing = true
-			log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; broadcast is live")
+		allowed, event := gate.Allow(packet.IsKeyFrame, packet.CurrentTime, time.Now(), maxMoQLivePacketAge)
+		switch event {
+		case livemoq.FrameGateEventStarted:
+			log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; " + config.label() + " broadcast is live")
+		case livemoq.FrameGateEventLagging:
+			log.Log.Warning("cloud.publishLiveStreamMoQ(): " + config.label() + " stream is lagging; dropping packets until a recent keyframe")
+		case livemoq.FrameGateEventRecovered:
+			log.Log.Info("cloud.publishLiveStreamMoQ(): caught up with the " + config.label() + " live stream at a recent keyframe")
 		}
-		payload, err := livemoq.NormalizeH264AccessUnit(packet.Data)
+		if !allowed {
+			continue
+		}
+		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),
 		}
+		writeStartedAt := time.Now()
 		if err := stream.WriteFrame(frame); err != nil {
 			return fmt.Errorf("write H.264 access unit: %w", err)
 		}
+		writeDuration := time.Since(writeStartedAt)
+		if writeDuration >= slowMoQWriteThreshold && time.Since(lastSlowWriteWarning) >= moQWriteWarningInterval {
+			packetAge := time.Duration(0)
+			if packet.CurrentTime > 0 {
+				packetAge = time.Since(time.UnixMilli(packet.CurrentTime))
+				if packetAge < 0 {
+					packetAge = 0
+				}
+			}
+			log.Log.Warning(fmt.Sprintf(
+				"cloud.publishLiveStreamMoQ(): %s WriteFrame blocked for %s (packet_age=%s keyframe=%t)",
+				config.label(), writeDuration.Round(time.Millisecond), packetAge.Round(time.Millisecond), packet.IsKeyFrame,
+			))
+			lastSlowWriteWarning = time.Now()
+		}
+	}
+}
+
+// watchLiveStreamMoQSubscribers flips the publisher between uploading and idling
+// as viewers subscribe to and leave this tier's broadcast. Used and Unused both
+// block, so they are followed from their own goroutine.
+//
+// It deliberately never turns publishing off before the first subscriber has
+// been observed: the relay catalog is only complete once media has flowed, so
+// going idle up front could keep the tier undiscoverable. On any error it fails
+// open (keeps publishing) — a stalled watcher must never take the live view down.
+func watchLiveStreamMoQSubscribers(ctx context.Context, stream *moq.MediaProducer, publishing *atomic.Bool, config liveMoQConfig) {
+	for ctx.Err() == nil {
+		if err := stream.Used(ctx); err != nil {
+			publishing.Store(true)
+			return
+		}
+		if publishing.CompareAndSwap(false, true) {
+			log.Log.Info("cloud.watchLiveStreamMoQSubscribers(): viewer subscribed, resuming the " + config.label() + " broadcast")
+		}
+
+		if err := stream.Unused(ctx); err != nil {
+			publishing.Store(true)
+			return
+		}
+		publishing.Store(false)
+		log.Log.Info("cloud.watchLiveStreamMoQSubscribers(): no viewers left, idling the " + config.label() + " broadcast")
 	}
 }