From c7c6bcbdf2a2bbe5a877c108ad90a6c6aff5426a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Verstraeten?= Date: Fri, 7 Aug 2026 13:23:44 +0200 Subject: [PATCH] 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 | 45 ++++++++++++++++++ machinery/src/cloud/livemoq/recovery_test.go | 49 ++++++++++++++++++++ machinery/src/cloud/livemoq_enabled.go | 43 +++++++++++++---- 3 files changed, 128 insertions(+), 9 deletions(-) create mode 100644 machinery/src/cloud/livemoq/recovery.go create mode 100644 machinery/src/cloud/livemoq/recovery_test.go diff --git a/machinery/src/cloud/livemoq/recovery.go b/machinery/src/cloud/livemoq/recovery.go new file mode 100644 index 0000000..67bc72e --- /dev/null +++ b/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 +} diff --git a/machinery/src/cloud/livemoq/recovery_test.go b/machinery/src/cloud/livemoq/recovery_test.go new file mode 100644 index 0000000..d4c22a2 --- /dev/null +++ b/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) + } +} diff --git a/machinery/src/cloud/livemoq_enabled.go b/machinery/src/cloud/livemoq_enabled.go index 64d7418..c970da2 100644 --- a/machinery/src/cloud/livemoq_enabled.go +++ b/machinery/src/cloud/livemoq_enabled.go @@ -17,9 +17,12 @@ 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 ) 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 { - if !packet.IsKeyFrame { - continue - } - writing = true + 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; 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() + } } }