From 7cd4a790bc23784292236458b2f1a3dfd92929e8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:14:49 -0700 Subject: [PATCH 1/2] quest(broadcast-epoch): claim gst From 93a405f1f715a6809965fcf3400e5cb0f6f0f20d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:19:39 -0700 Subject: [PATCH 2/2] feat(moq-gst): moqsink publishes each run under a fresh epoch Session::start mints a publisher epoch and announces the broadcast with it, so a restarted pipeline takes over from a lingering old route instead of resuming viewers into the old run's group sequence. Reconnects within a run keep the epoch. The epoch is logged at info. Completes quest/m0/broadcast-epoch/gst.md; the per-generation half moves into the #3115 quest. Co-Authored-By: Claude Opus 5.5 --- doc/bin/gstreamer.md | 4 +++ quest/m0/broadcast-epoch/README.md | 1 - quest/m0/broadcast-epoch/gst.md | 24 -------------- ...generated-media-timelines-to-wall-clock.md | 5 ++- ...ublication-has-no-generation-so-a-flush.md | 2 ++ rs/moq-gst/src/sink/session.rs | 31 +++++++++++++++++-- 6 files changed, 36 insertions(+), 31 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/gst.md diff --git a/doc/bin/gstreamer.md b/doc/bin/gstreamer.md index e51a5fc504..abcc8449af 100644 --- a/doc/bin/gstreamer.md +++ b/doc/bin/gstreamer.md @@ -56,6 +56,10 @@ Each `sink_%u` request pad is one track. Pad properties: `track` names it `estimated-send-rate`, `estimated-recv-rate`, `connection-stats`, and `sessions`. The sink reconnects for as long as the pipeline runs and only reports `failed` on an answer redialing can't change, such as a rejected token. +Each run from `READY` publishes under a fresh +[publisher epoch](/concept/moq-lite#publisher-epochs), so viewers switch to a +restarted pipeline at once instead of stalling on the old one; reconnects +within a run keep it. `connection-stats` is null while disconnected. While connected it is a `GstStructure` named `moq-connection-stats` containing the transport metrics diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index e8d4b4cd75..b6bc106787 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -72,7 +72,6 @@ timeout, and killing the newest epoch falls back to a still-live older one. - [Gateways](/quest/m0/broadcast-epoch/gateways.md) - RTMP, SRT, and WHIP ingest mint an epoch per incoming connection, so an encoder reconnect is a clean takeover - [TS restart](/quest/m0/broadcast-epoch/ts-restart.md) - a signalled backward TS discontinuity finishes the broadcast and continues the same input under a fresh epoch - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one -- [GStreamer](/quest/m0/broadcast-epoch/gst.md) - moqsink publishes each run under a fresh epoch - [Remove `--hop`](/quest/m0/broadcast-epoch/hop-removal.md) - `moq` takes an optional `--epoch` instead of `--hop`, a plain publisher declares a random Hop ID, and the per-session hop stamp is gone - [Stats epochs](/quest/m0/broadcast-epoch/stats-epoch.md) - moq-stats publishes each group announcement under its own epoch, so neither a restarted node nor a returning idle group stalls its viewers - [Stats totals and prefix tracks](/quest/m0/broadcast-epoch/stats-split.md) - the same release retires the per-path stats maps for totals and on-demand prefix tracks (decided 2026-10-05) diff --git a/quest/m0/broadcast-epoch/gst.md b/quest/m0/broadcast-epoch/gst.md deleted file mode 100644 index a647f7f0f6..0000000000 --- a/quest/m0/broadcast-epoch/gst.md +++ /dev/null @@ -1,24 +0,0 @@ -# [S] GStreamer publishes under epochs - -## Goal - -`moqsink` publishes each run under a fresh epoch, so a restarted pipeline is -a clean takeover for viewers. - -## Plan - -`moqsink` mints an epoch per session and announces it. When -[#3115](/quest/m2/3115-moqsink-the-publication-has-no-generation-so-a-flush.md) -lands, each of its publication generations is a new epoch. Update -`doc/bin/gstreamer.md` if it shows paths. - -Decided in the 2026-10-05 audit: the OBS half moved to -[OBS publishes under epochs](/quest/m1/obs-epoch.md), so the release gate no -longer waits on the C++ line. `moqsink` is Rust on moq-net and needs nothing -from it. moq-c announces through `broadcast::Producer::announce`, which takes the -route's epoch as given, so the OBS quest has to pass one. - - -## Related - -- [OBS publishes under epochs](/quest/m1/obs-epoch.md) - the OBS plugin's half, after the C++ line diff --git a/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md b/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md index 662337a9e6..0ccdbfda3b 100644 --- a/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md +++ b/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md @@ -19,8 +19,8 @@ reference clock is not UTC. Every pad uses the same epoch rather than sampling its own. Do not define a GStreamer-specific catalog shape. Decided in the 2026-09-30 audit: a restart is a new broadcast epoch, not a -forward re-anchor on the old clock (per remove-live and -[GStreamer](/quest/m0/broadcast-epoch/gst.md)), and +forward re-anchor on the old clock (per remove-live; `moqsink` mints a +publisher epoch per run), and [#3115](/quest/m2/3115-moqsink-the-publication-has-no-generation-so-a-flush.md) handles the sink side. @@ -33,5 +33,4 @@ delayed first buffers, and multiple pads sharing one epoch. ## Related -- [GStreamer](/quest/m0/broadcast-epoch/gst.md) - a restarted pipeline publishes a new epoch - [Shared import clock](/quest/m1/shared-clock.md) - when a catalog clock becomes final and how a `with_clock` catalog offsets every importer; set this epoch through that API, either `with_clock` or the first anchor, consistent with its rules diff --git a/quest/m2/3115-moqsink-the-publication-has-no-generation-so-a-flush.md b/quest/m2/3115-moqsink-the-publication-has-no-generation-so-a-flush.md index 2c06d09427..66f48cc13f 100644 --- a/quest/m2/3115-moqsink-the-publication-has-no-generation-so-a-flush.md +++ b/quest/m2/3115-moqsink-the-publication-has-no-generation-so-a-flush.md @@ -24,6 +24,8 @@ globally (a fresh `CompletionHandle`, broadcast, catalog, and producers); every other pad joins that generation on its next buffer rather than each pad opening its own, so aggregate EOS membership is one set per generation. +- Each generation is a new broadcast at the same path, so it announces a + freshly minted publisher epoch, as `Session::start` does per run. - Separate "the producers were finalized" from "the EOS message was posted"; the per-pad lifecycle from #2998 already distinguishes them for pads, and the element-level latch (`eos_delivered`) needs the same split so a diff --git a/rs/moq-gst/src/sink/session.rs b/rs/moq-gst/src/sink/session.rs index 8b53fd1516..5cfe0edaaa 100644 --- a/rs/moq-gst/src/sink/session.rs +++ b/rs/moq-gst/src/sink/session.rs @@ -282,7 +282,11 @@ impl Session { let _rt = RUNTIME.enter(); let origin = moq_tokio::origin::spawn(); - let mut broadcast = origin.publish(&settings.broadcast, moq_net::origin::Route::default())?; + // Each run is a new publisher instance: a restarted pipeline takes over from one whose route still + // lingers, rather than resuming viewers into the old run's group sequence. Reconnects keep it. + let epoch = moq_net::Epoch::mint(); + gst::info!(CAT, "publishing {} under epoch {epoch}", settings.broadcast); + let mut broadcast = origin.publish(&settings.broadcast, moq_net::origin::Route::default().with_epoch(epoch))?; let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default())?; let status = Arc::new(Status::default()); @@ -530,7 +534,7 @@ mod tests { assert_eq!(structure.get::("ended"), Ok(2)); } - fn started() -> (Session, moq_tokio::Connection) { + fn started_with_broadcast() -> (Session, moq_net::broadcast::Producer) { gst::init().unwrap(); let settings = ResolvedSettings { url: "https://127.0.0.1:1".parse().unwrap(), @@ -539,12 +543,33 @@ mod tests { quic_idle_timeout: None, quic_keep_alive: None, }; - let (session, registration, _, _) = Session::start(settings, glib::WeakRef::new()).unwrap(); + let (session, registration, broadcast, _) = Session::start(settings, glib::WeakRef::new()).unwrap(); registration.mark_registered(); + (session, broadcast) + } + + fn started() -> (Session, moq_tokio::Connection) { + let (session, _) = started_with_broadcast(); let connection = session.connection.clone(); (session, connection) } + fn announced_epoch(broadcast: &moq_net::broadcast::Producer) -> moq_net::Epoch { + broadcast.route().expect("announced").epoch.expect("an epoch") + } + + // Without an epoch, a restart whose old route lingers resumes viewers into the old run's groups. + #[test] + fn each_run_announces_a_newer_epoch() { + let (first, first_broadcast) = started_with_broadcast(); + let first_epoch = announced_epoch(&first_broadcast); + first.stop(); + + let (second, second_broadcast) = started_with_broadcast(); + assert!(announced_epoch(&second_broadcast) > first_epoch); + second.stop(); + } + fn is_closed(connection: &moq_tokio::Connection) -> bool { connection.poll_closed(&moq_net::kio::Waiter::noop()).is_ready() }