feat: optimize live stream publishing and enhance audience gate logic

README.md

@@ -574,11 +574,14 @@ its own broadcast and switching quality is simply a resubscribe:
 | `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.
+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:
 

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

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

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

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{})

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