diff --git a/README.md b/README.md index c3d07f5..f89b3bb 100644 --- a/README.md +++ b/README.md @@ -574,11 +574,14 @@ its own broadcast and switching quality is simply a resubscribe: | `high` | `devices//live.hang` | highest-resolution camera stream | | `low` | `devices//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. +Each tier uploads every frame while it has subscribers. While idle, it continues +to upload fresh keyframes only. This keeps the relay's latest cached GOP current, +so a new viewer does not initially render the final frame from the previous +viewer session, while avoiding the bandwidth cost of continuously sending every +delta frame. `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. Two bounded duration settings tune recovery for unusual network conditions: diff --git a/machinery/src/cloud/livemoq/recovery.go b/machinery/src/cloud/livemoq/recovery.go index d2f9f26..55bc9f3 100644 --- a/machinery/src/cloud/livemoq/recovery.go +++ b/machinery/src/cloud/livemoq/recovery.go @@ -20,6 +20,24 @@ type FrameGate struct { recovering bool } +// AudienceGate publishes every frame while watched and only keyframes while +// idle. The idle keyframes keep the relay's latest cached group current without +// paying the bandwidth cost of a full stream when nobody is watching. +type AudienceGate struct { + idle bool +} + +func (g *AudienceGate) Allow(hasSubscribers bool, isKeyFrame bool) (allowed bool, enteredIdle bool) { + if hasSubscribers { + g.idle = false + return true, false + } + + enteredIdle = !g.idle + g.idle = true + return isKeyFrame, enteredIdle +} + // WriteWatchdog tracks the single synchronous frame write performed by a MoQ // publisher so another goroutine can interrupt a wedged native call. type WriteWatchdog struct { @@ -46,9 +64,8 @@ func (w *WriteWatchdog) Elapsed(now time.Time) (time.Duration, bool) { return elapsed, true } -// 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. +// Reset closes the gate so publication resumes on the next keyframe. Entering +// idle mode uses it to start the relay-cache refresh on a complete GOP. func (g *FrameGate) Reset() { g.started = false g.recovering = false diff --git a/machinery/src/cloud/livemoq/recovery_test.go b/machinery/src/cloud/livemoq/recovery_test.go index 0cc0fa2..747a3a8 100644 --- a/machinery/src/cloud/livemoq/recovery_test.go +++ b/machinery/src/cloud/livemoq/recovery_test.go @@ -48,6 +48,29 @@ func TestFrameGateAllowsMissingCaptureTime(t *testing.T) { } } +func TestAudienceGateRefreshesOnlyKeyframesWhileIdle(t *testing.T) { + gate := AudienceGate{} + + if allowed, enteredIdle := gate.Allow(true, false); !allowed || enteredIdle { + t.Fatalf("active delta = (%t, %t), want (true, false)", allowed, enteredIdle) + } + if allowed, enteredIdle := gate.Allow(false, false); allowed || !enteredIdle { + t.Fatalf("first idle delta = (%t, %t), want (false, true)", allowed, enteredIdle) + } + if allowed, enteredIdle := gate.Allow(false, true); !allowed || enteredIdle { + t.Fatalf("idle keyframe = (%t, %t), want (true, false)", allowed, enteredIdle) + } + if allowed, enteredIdle := gate.Allow(false, false); allowed || enteredIdle { + t.Fatalf("later idle delta = (%t, %t), want (false, false)", allowed, enteredIdle) + } + if allowed, enteredIdle := gate.Allow(true, false); !allowed || enteredIdle { + t.Fatalf("resumed delta = (%t, %t), want (true, false)", allowed, enteredIdle) + } + if allowed, enteredIdle := gate.Allow(false, true); !allowed || !enteredIdle { + t.Fatalf("next idle keyframe = (%t, %t), want (true, true)", allowed, enteredIdle) + } +} + func TestWriteWatchdogTracksActiveWrite(t *testing.T) { now := time.Unix(10, 0) watchdog := WriteWatchdog{} diff --git a/machinery/src/cloud/livemoq_enabled.go b/machinery/src/cloud/livemoq_enabled.go index cdf78c0..f9da5b8 100644 --- a/machinery/src/cloud/livemoq_enabled.go +++ b/machinery/src/cloud/livemoq_enabled.go @@ -4,6 +4,7 @@ package cloud import ( "context" + "errors" "fmt" "net/url" "os" @@ -128,8 +129,9 @@ func boundedMoQDuration(name string, fallback, minimum, maximum time.Duration) t // 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. +// other path. Each tier uploads every frame while watched and keyframes only +// while idle, keeping the relay cache current without continuously sending the +// full unused stream. func StartLiveStreamMoQ( ctx context.Context, configuration *models.Configuration, @@ -280,24 +282,34 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { if err != nil { return fmt.Errorf("create broadcast: %w", err) } - defer broadcast.Finish() + var finishBroadcastOnce sync.Once + var finishBroadcastErr error + finishBroadcast := func() error { + finishBroadcastOnce.Do(func() { + finishBroadcastErr = broadcast.Finish() + }) + return finishBroadcastErr + } + defer finishBroadcast() stream, err := broadcast.PublishMedia("avc3", nil) if err != nil { return fmt.Errorf("create H.264 media stream: %w", err) } var finishStreamOnce sync.Once - finishStream := func() { + var finishStreamErr error + finishStream := func() error { finishStreamOnce.Do(func() { - _ = stream.Finish() + finishStreamErr = stream.Finish() }) + return finishStreamErr } defer finishStream() - // 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. + // Publish the complete stream while watched and only keyframes while idle. + // `publishing` starts true so the track becomes discoverable on the relay + // before the first subscriber; after that, the subscriber watcher selects the + // full-rate or keyframe-refresh mode. publishing := &atomic.Bool{} publishing.Store(true) subscriberWatchDone := make(chan struct{}) @@ -308,8 +320,7 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { writeWatchdog := &livemoq.WriteWatchdog{} writeWatchDone := make(chan struct{}) closePublisher := func() error { - finishStream() - return client.Close() + return closeLiveMoQPublisher(finishStream, finishBroadcast, client.Close) } go func() { defer close(writeWatchDone) @@ -324,10 +335,10 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { cursor := config.queue.Latest() gate := livemoq.FrameGate{} + audienceGate := livemoq.AudienceGate{} deduplicator := livemoq.KeyframeDeduplicator{} var lastSlowWriteWarning time.Time var lastDuplicateKeyframeWarning time.Time - idle := false for { select { case err := <-sessionClosed: @@ -347,18 +358,15 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { } 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 - } + if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") { continue } - idle = false - if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") { + allowedForAudience, enteredIdle := audienceGate.Allow(publishing.Load(), packet.IsKeyFrame) + if enteredIdle { + gate.Reset() + deduplicator.Reset() + } + if !allowedForAudience { continue } allowed, event := gate.Allow(packet.IsKeyFrame, packet.CurrentTime, time.Now(), config.packetAgeLimit()) @@ -435,14 +443,24 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { } } -func watchLiveStreamMoQWrites(ctx context.Context, closeClient func() error, watchdog *livemoq.WriteWatchdog, config liveMoQConfig) { +func closeLiveMoQPublisher(finishStream, finishBroadcast, closeClient func() error) error { + streamErr := finishStream() + broadcastErr := finishBroadcast() + clientErr := closeClient() + return errors.Join(streamErr, broadcastErr, clientErr) +} + +func watchLiveStreamMoQWrites(ctx context.Context, closePublisher func() error, watchdog *livemoq.WriteWatchdog, config liveMoQConfig) { ticker := time.NewTicker(moQWriteWatchInterval) defer ticker.Stop() for { select { case <-ctx.Done(): - closeClient() + if err := closePublisher(); err != nil { + config.logEntry("publisher_close_failed").WithError(err). + Debug("MoQ publisher reported an error while closing") + } return case now := <-ticker.C: writeDuration, active := watchdog.Elapsed(now) @@ -457,7 +475,10 @@ func watchLiveStreamMoQWrites(ctx context.Context, closeClient func() error, wat if config.communication != nil { config.communication.RecordMoQWriteTimeout() } - closeClient() + if err := closePublisher(); err != nil { + config.logEntry("publisher_close_failed").WithError(err). + Warn("MoQ publisher reported an error while closing after a write timeout") + } return } } @@ -486,6 +507,6 @@ func watchLiveStreamMoQSubscribers(ctx context.Context, stream *moq.MediaProduce return } publishing.Store(false) - config.logEntry("subscribers_idle").Info("MoQ broadcast idle; waiting for subscribers") + config.logEntry("subscribers_idle").Info("MoQ broadcast idle; refreshing keyframes only") } } diff --git a/machinery/src/cloud/livemoq_enabled_test.go b/machinery/src/cloud/livemoq_enabled_test.go index adeebc7..067370c 100644 --- a/machinery/src/cloud/livemoq_enabled_test.go +++ b/machinery/src/cloud/livemoq_enabled_test.go @@ -4,6 +4,8 @@ package cloud import ( "context" + "errors" + "reflect" "sync/atomic" "testing" "time" @@ -13,6 +15,37 @@ import ( "github.com/moq-dev/moq-go/moq" ) +func TestCloseLiveMoQPublisherFinishesBeforeClosingClient(t *testing.T) { + var order []string + streamErr := errors.New("stream finish failed") + broadcastErr := errors.New("broadcast finish failed") + clientErr := errors.New("client close failed") + + err := closeLiveMoQPublisher( + func() error { + order = append(order, "stream") + return streamErr + }, + func() error { + order = append(order, "broadcast") + return broadcastErr + }, + func() error { + order = append(order, "client") + return clientErr + }, + ) + + if want := []string{"stream", "broadcast", "client"}; !reflect.DeepEqual(order, want) { + t.Fatalf("shutdown order = %v, want %v", order, want) + } + for _, want := range []error{streamErr, broadcastErr, clientErr} { + if !errors.Is(err, want) { + t.Errorf("shutdown error %v does not include %v", err, want) + } + } +} + func TestWatchLiveStreamMoQWritesClosesClientWhenContextEnds(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) closed := make(chan struct{}) diff --git a/machinery/verify-moq-devcontainer.sh b/machinery/verify-moq-devcontainer.sh index 0c58308..b0d4dd6 100644 --- a/machinery/verify-moq-devcontainer.sh +++ b/machinery/verify-moq-devcontainer.sh @@ -17,7 +17,7 @@ binary="${TMPDIR:-/tmp}/agent-moq" trap 'rm -f "$binary"' EXIT echo "==> Linking the MoQ Agent" -GOWORK=off go build -tags moq -o "$binary" ./main.go +GOWORK=off go build -tags moq -o "$binary" . echo "==> Running the linked binary" "$binary" -action version