Enhance MoQ streaming: implement quality tier broadcasting and subscriber management

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,6 +5,7 @@ import (
 	"strings"
 
 	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
+	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
 var annexBStartCode = []byte{0x00, 0x00, 0x00, 0x01}
@@ -51,12 +52,22 @@ func NormalizeH264AccessUnit(payload []byte) ([]byte, error) {
 	return h264.AnnexBMarshal(normalized)
 }
 
-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) {
@@ -65,12 +67,15 @@ 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)
+	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"); got != "devices/camera-1/live.hang" {
+	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/recovery.go

@@ -17,6 +17,14 @@ type FrameGate struct {
 	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 {

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"
@@ -33,8 +34,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
@@ -50,39 +66,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) {
@@ -138,25 +162,46 @@ 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()
 	gate := livemoq.FrameGate{}
 	var lastSlowWriteWarning 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()
+				idle = true
+			}
+			continue
+		}
+		idle = false
 		if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") {
 			continue
 		}
 		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")
+			log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; " + config.label() + " broadcast is live")
 		case livemoq.FrameGateEventLagging:
-			log.Log.Warning("cloud.publishLiveStreamMoQ(): stream is lagging; dropping packets until a recent keyframe")
+			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 live stream at a recent keyframe")
+			log.Log.Info("cloud.publishLiveStreamMoQ(): caught up with the " + config.label() + " live stream at a recent keyframe")
 		}
 		if !allowed {
 			continue
@@ -183,10 +228,37 @@ func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
 				}
 			}
 			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,
+				"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")
+	}
+}