Add live stream recovery gating

Drop stale H.264 packets until a recent keyframe arrives, with lifecycle logging and slow MoQ write diagnostics. Add focused FrameGate tests and configure the UI package registry.

machinery/src/cloud/livemoq/recovery.go

@@ -0,0 +1,45 @@
+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
+}
+
+// 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.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

@@ -17,9 +17,12 @@ import (
 )
 
 const (
-	defaultMoQRelayURL = "https://relay.uug.ai/anon"
+	defaultMoQRelayURL      = "https://relay.uug.ai/anon"
-	minMoQRetryDelay   = time.Second
+	minMoQRetryDelay        = time.Second
-	maxMoQRetryDelay   = 30 * time.Second
+	maxMoQRetryDelay        = 30 * time.Second
+	maxMoQLivePacketAge     = 1500 * time.Millisecond
+	slowMoQWriteThreshold   = 100 * time.Millisecond
+	moQWriteWarningInterval = 10 * time.Second
 )
 
 type liveMoQConfig struct {
@@ -136,7 +139,8 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 	defer stream.Finish()
 
 	cursor := config.queue.Latest()
-	writing := false
+	gate := livemoq.FrameGate{}
+	var lastSlowWriteWarning time.Time
 	for {
 		packet, err := cursor.ReadPacket()
 		if err != nil {
@@ -145,12 +149,17 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 		if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") {
 			continue
 		}
-		if !writing {
+		allowed, event := gate.Allow(packet.IsKeyFrame, packet.CurrentTime, time.Now(), maxMoQLivePacketAge)
-			if !packet.IsKeyFrame {
+		switch event {
-				continue
+		case livemoq.FrameGateEventStarted:
-			}
-			writing = true
 			log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; broadcast is live")
+		case livemoq.FrameGateEventLagging:
+			log.Log.Warning("cloud.publishLiveStreamMoQ(): stream is lagging; dropping packets until a recent keyframe")
+		case livemoq.FrameGateEventRecovered:
+			log.Log.Info("cloud.publishLiveStreamMoQ(): caught up with live stream at a recent keyframe")
+		}
+		if !allowed {
+			continue
 		}
 		payload, err := livemoq.NormalizeH264AccessUnit(packet.Data)
 		if err != nil {
@@ -160,8 +169,24 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 			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(): WriteFrame blocked for %s (packet_age=%s keyframe=%t)",
+				writeDuration.Round(time.Millisecond), packetAge.Round(time.Millisecond), packet.IsKeyFrame,
+			))
+			lastSlowWriteWarning = time.Now()
+		}
 	}
 }