diff --git a/README.md b/README.md index 748d626..98b6f8b 100644 --- a/README.md +++ b/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//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//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. The `/anon` relay route is intended for interoperability testing. Production deployments must set `AGENT_LIVE_MOQ_URL` to a short-lived, device-scoped diff --git a/machinery/src/cloud/livemoq/annexb.go b/machinery/src/cloud/livemoq/annexb.go index 9d9eb3d..7f4e4f5 100644 --- a/machinery/src/cloud/livemoq/annexb.go +++ b/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. diff --git a/machinery/src/cloud/livemoq/annexb_test.go b/machinery/src/cloud/livemoq/annexb_test.go index 1f37609..a361d6f 100644 --- a/machinery/src/cloud/livemoq/annexb_test.go +++ b/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) { diff --git a/machinery/src/cloud/livemoq/recovery.go b/machinery/src/cloud/livemoq/recovery.go index 68cbf78..a9afcd6 100644 --- a/machinery/src/cloud/livemoq/recovery.go +++ b/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 { diff --git a/machinery/src/cloud/livemoq_enabled.go b/machinery/src/cloud/livemoq_enabled.go index c970da2..92e3f6d 100644 --- a/machinery/src/cloud/livemoq_enabled.go +++ b/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") + } +}