From f3f57154630e7b51739cc695d9188e3476755cc4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 15:31:54 -0700 Subject: [PATCH 1/3] quest: claim quest/m0/stats-linger Co-Authored-By: Claude Opus 5.5 From 51c88e73ee85a8d7a7a515b0b5caf56352a417d1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 15:44:45 -0700 Subject: [PATCH 2/3] feat(stats): linger an empty group broadcast before unannouncing it A grouped stats broadcast (`//node/`) now stays announced for `Config::linger` (default 5 minutes) after its group's last traffic and session entries leave, so viewer churn no longer unannounces and re-announces it across the mesh. While it lingers empty its tracks read `{}`; a returning entry re-arms the linger. The relay exposes it as `stats.linger` / `--stats-linger` / `MOQ_STATS_LINGER`. Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/config.md | 1 + doc/concept/stats.md | 8 +- rs/moq-relay/src/config.rs | 12 ++- rs/moq-relay/src/settings.rs | 3 + rs/moq-relay/src/stats.rs | 28 +++++- rs/moq-stats/src/produce.rs | 180 +++++++++++++++++++++++++++++++++-- 6 files changed, 217 insertions(+), 15 deletions(-) diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 0524a10e71..be3b7bd702 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -224,6 +224,7 @@ prefix = ".stats" # Broadcasts appear under /node/, + + #[usage(env = "MOQ_STATS_LINGER", cli("--stats-linger"))] + linger: Option, } #[derive(usage::Config)] diff --git a/rs/moq-relay/src/stats.rs b/rs/moq-relay/src/stats.rs index 6fb8bfbb9a..12fa708423 100644 --- a/rs/moq-relay/src/stats.rs +++ b/rs/moq-relay/src/stats.rs @@ -77,6 +77,19 @@ pub struct Config { setting = "stats.depth" )] pub depth: usize, + + /// How long a group's stats broadcast stays announced after the group's + /// last session and traffic leave, e.g. "5m" or "30s". Defaults to 5 + /// minutes. A group that returns within it keeps its broadcast, so viewer + /// churn doesn't unannounce and re-announce it across the mesh. Only applies + /// at `depth` 1 or more. See [`moq_stats::produce::Config::linger`]. + #[usage(skip)] + #[serde(with = "crate::duration::serde_option")] + pub linger: Option, + + #[usage(long = "stats-linger", env = "MOQ_STATS_LINGER", setting = "stats.linger")] + #[serde(default, rename = "__cli_linger", skip_serializing_if = "Option::is_none")] + pub(crate) linger_arg: Option, } impl Default for Config { @@ -87,11 +100,18 @@ impl Default for Config { interval: 1, node: None, depth: 0, + linger: None, + linger_arg: None, } } } impl Config { + /// The linger after command-line overrides; `None` keeps the producer's default. + pub(crate) fn linger(&self) -> Option { + self.linger_arg.map(crate::duration::Duration::into_std).or(self.linger) + } + /// Build a [`moq_stats::Producer`] from this config, publishing on `origin`. /// /// Returns a no-op producer when [`Self::enabled`] is false, so the relay can @@ -107,13 +127,17 @@ impl Config { let interval = Duration::from_secs(self.interval.max(1)); let node = self.node.clone().map(PathOwned::from); let depth = self.depth; - tracing::info!(prefix, interval_secs = interval.as_secs(), node = ?node, depth, "stats publishing enabled"); - let config = moq_stats::produce::Config::new() + let linger = self.linger(); + tracing::info!(prefix, interval_secs = interval.as_secs(), node = ?node, depth, ?linger, "stats publishing enabled"); + let mut config = moq_stats::produce::Config::new() .with_origin(origin) .with_prefix(prefix) .with_interval(interval) .with_node(node) .with_depth(depth); + if let Some(linger) = linger { + config = config.with_linger(linger); + } moq_stats::Producer::new(config) } } diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index 07c0987e3b..d54c8f36a5 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -11,6 +11,7 @@ use moq_net::stats::{Presence, Registry, Report, Role, Tier, Traffic}; use moq_net::{Path, PathOwned, broadcast, kio, origin, track}; use serde::Serialize; use web_async::spawn; +use web_async::time::Instant; use crate::{COMPRESSED_SUFFIX, sessions_track, traffic_track}; @@ -46,9 +47,18 @@ pub struct Config { /// Default `0` publishes one `/node/` broadcast carrying every /// path. `1` publishes one broadcast per first segment at /// `//node/`, and larger values include more leading - /// segments. Group broadcasts are announced while their group has live traffic; - /// at depth `0`, the single broadcast stays announced for the producer's life. + /// segments. Group broadcasts are announced while their group has live traffic, + /// plus the [`Self::linger`]; at depth `0`, the single broadcast stays + /// announced for the producer's life. pub depth: usize, + /// How long a group broadcast stays announced after its group's last entry + /// leaves. Default 5 minutes. + /// + /// A group that returns within the linger keeps its broadcast, so a viewer + /// leaving and another arriving causes no unannounce and re-announce across + /// the mesh. While it lingers empty, every track reads `{}`. Zero + /// unannounces on the first drain the group is empty. Unused at depth `0`. + pub linger: Duration, } impl Config { @@ -62,6 +72,7 @@ impl Config { node: None, interval: Duration::from_secs(1), depth: 0, + linger: DEFAULT_LINGER, } } @@ -95,8 +106,19 @@ impl Config { self.depth = depth; self } + + /// Set how long an empty group broadcast stays announced (default 5 + /// minutes). See [`Self::linger`]. + pub fn with_linger(mut self, linger: Duration) -> Self { + self.linger = linger; + self + } } +/// The default [`Config::linger`]. Sized from a live fleet, where an idle +/// group's viewers returned 46 to 112 seconds after the last one left. +const DEFAULT_LINGER: Duration = Duration::from_secs(300); + impl Default for Config { fn default() -> Self { Self::new() @@ -156,6 +178,7 @@ impl Producer { node, interval, depth, + linger, } = config; // An empty path after normalization is indistinguishable from "no node // set"; collapse it so downstream code only sees a single representation. @@ -180,6 +203,7 @@ impl Producer { prefix, node, depth, + linger, interval, sequence: Arc::new(AtomicU64::new(0)), }; @@ -206,6 +230,7 @@ struct Task { prefix: PathOwned, node: Option, depth: usize, + linger: Duration, interval: Duration, sequence: Arc, } @@ -231,7 +256,7 @@ impl Task { } drain.collect(); - drain.publish(); + drain.publish(Instant::now()); } } fn node(&self) -> Option<&str> { @@ -330,14 +355,14 @@ impl Drain { } /// Write every group's pending frames, serve consumer track requests, and - /// unpublish groups with nothing left to report. - fn publish(&mut self) { + /// unpublish groups that have lingered empty. + fn publish(&mut self, now: Instant) { // At depth 0 the single broadcast stays for the producer's life; a - // group broadcast lives while its group has entries. - let depth = self.task.depth; + // group broadcast lives while its group has entries, plus the linger. + let Task { depth, linger, .. } = self.task; for (_, group) in self .groups - .extract_if(|_, group| depth > 0 && group.traffic_rows.is_empty() && group.session_rows.is_empty()) + .extract_if(|_, group| depth > 0 && group.lingered(now, linger)) { // Deliberate unpublish: finish (tracks included) rather than drop, // so there is no dropped-without-finish warning. @@ -745,6 +770,8 @@ struct GroupPublisher { /// This drain's entries for the group, as indices into the report. traffic_rows: Vec, session_rows: Vec, + /// When the group's last entry left; `None` while it has entries. + empty_since: Option, } /// The plain track names one tier's entries land on. @@ -817,9 +844,21 @@ impl GroupPublisher { names: HashMap::new(), traffic_rows: Vec::new(), session_rows: Vec::new(), + empty_since: None, }) } + /// Whether this drain left the group empty for at least `linger`. An entry + /// returning within the linger re-arms it. + fn lingered(&mut self, now: Instant, linger: Duration) -> bool { + if !self.traffic_rows.is_empty() || !self.session_rows.is_empty() { + self.empty_since = None; + return false; + } + let since = *self.empty_since.get_or_insert(now); + now.saturating_duration_since(since) >= linger + } + /// Run this drain's rows through change detection into the pending frames. fn collect(&mut self, report: &Report, tick: u64) { let Self { @@ -1270,6 +1309,18 @@ mod tests { serde_json::from_slice(&frame.payload).expect("json parse") } + async fn read_session_frame_last( + broadcast: &moq_net::broadcast::Consumer, + name: &str, + ) -> BTreeMap { + let mut track = subscribe(broadcast, name).await; + let mut last = next_frame(&mut track).await; + while let Some(frame) = try_next_frame(&mut track) { + last = frame; + } + serde_json::from_slice(&last.payload).expect("json parse") + } + async fn subscribe(broadcast: &moq_net::broadcast::Consumer, name: &str) -> track::Ordered { broadcast .track(name) @@ -1469,6 +1520,114 @@ mod tests { ); } + /// A depth-1 producer with `linger`, its origin, and a hidden announce + /// listener started before anything publishes. + fn grouped_producer(linger: Duration) -> (Producer, origin::Producer, announce::Consumer) { + let origin = produce_origin(); + let events = origin.consume().with_hidden(true).announced(); + let producer = Producer::new( + Config::new() + .with_origin(origin.clone()) + .with_node(PathOwned::from("sjc")) + .with_depth(1) + .with_linger(linger), + ); + (producer, origin, events) + } + + /// The announce events buffered so far, as `(path, active)`. + fn take_events(events: &mut announce::Consumer) -> Vec<(String, bool)> { + use futures::FutureExt; + let mut out = Vec::new(); + while let Some(Some((route, active))) = next_update(events).now_or_never() { + out.push((route.prefix.as_str().to_string(), active)); + } + out + } + + const ACME: &str = ".stats/acme/node/sjc"; + const FEED: &str = ".stats/feed/node/sjc"; + + #[tokio::test(start_paused = true)] + async fn empty_group_lingers_then_unannounces() { + let (producer, origin, mut events) = grouped_producer(Duration::from_secs(10)); + let registry = producer.registry(); + + // `feed` records traffic on `acme/live` (group `acme`) and holds a + // session rooted at `feed` (group `feed`). + let first = feed(registry, Tier::default(), "acme/live", true, 1, 100).await; + drive_tick().await; + let mut started = take_events(&mut events); + started.sort(); + assert_eq!(started, [(ACME.to_string(), true), (FEED.to_string(), true)]); + let acme = origin.consume().request_broadcast(ACME).await.expect("resolve"); + let sessions = origin.consume().request_broadcast(FEED).await.expect("resolve"); + assert_eq!(read_last_frame(&acme, "publisher.json").await["acme/live"].bytes, 100); + + // The viewer leaves: both groups empty but stay announced, reading zero. + drop(first); + for _ in 0..3 { + drive_tick().await; + } + assert_eq!(take_events(&mut events), [], "an empty group lingers"); + let frame = read_last_frame(&acme, "publisher.json").await; + assert!(frame.is_empty(), "no live counters while lingering, got {frame:?}"); + let frame = read_session_frame_last(&sessions, "sessions.json").await; + assert!(frame.is_empty(), "no presence while lingering, got {frame:?}"); + + // Another viewer arrives within the linger: same broadcast, no announce. + // The path restarted, so a reader diffing frames counts 100 + 50, the + // same as it would across an unannounce and re-announce. + let second = feed(registry, Tier::default(), "acme/live", true, 1, 50).await; + drive_tick().await; + assert_eq!( + take_events(&mut events), + [], + "a return within the linger re-announces nothing" + ); + assert_eq!(read_last_frame(&acme, "publisher.json").await["acme/live"].bytes, 50); + + // The return re-armed the linger: it unannounces only once it elapses + // again with the group still empty. + drop(second); + for _ in 0..8 { + drive_tick().await; + } + assert_eq!(take_events(&mut events), [], "the return re-armed the linger"); + for _ in 0..4 { + drive_tick().await; + } + let mut ended = take_events(&mut events); + ended.sort(); + assert_eq!(ended, [(ACME.to_string(), false), (FEED.to_string(), false)]); + + // A return after the linger is a new broadcast counted from zero. + let _third = feed(registry, Tier::default(), "acme/live", true, 1, 25).await; + drive_tick().await; + let mut restarted = take_events(&mut events); + restarted.sort(); + assert_eq!(restarted, [(ACME.to_string(), true), (FEED.to_string(), true)]); + let acme = origin.consume().request_broadcast(ACME).await.expect("resolve"); + assert_eq!(read_last_frame(&acme, "publisher.json").await["acme/live"].bytes, 25); + } + + #[tokio::test(start_paused = true)] + async fn zero_linger_unannounces_once_empty() { + let (producer, _origin, mut events) = grouped_producer(Duration::ZERO); + let first = feed(producer.registry(), Tier::default(), "acme/live", true, 1, 100).await; + drive_tick().await; + assert_eq!(take_events(&mut events).len(), 2); + + // The closing readout drain still carries the entry; the next is empty. + drop(first); + drive_tick().await; + assert_eq!(take_events(&mut events), []); + drive_tick().await; + let mut ended = take_events(&mut events); + ended.sort(); + assert_eq!(ended, [(ACME.to_string(), false), (FEED.to_string(), false)]); + } + #[tokio::test(start_paused = true)] async fn session_track_surfaces_by_root() { let (producer, origin) = test_producer(Some("sjc")); @@ -1870,6 +2029,7 @@ mod tests { node: None, sequence: Arc::new(AtomicU64::new(0)), depth, + linger: DEFAULT_LINGER, interval: Duration::from_secs(1), }) .expect("drain"); @@ -1877,13 +2037,13 @@ mod tests { // Warm up: create the groups and tracks and grow every buffer. for _ in 0..3 { drain.collect(); - drain.publish(); + drain.publish(Instant::now()); } let before = counting::allocs(); drain.collect(); let allocs = counting::allocs() - before; - drain.publish(); + drain.publish(Instant::now()); let pending: usize = drain .groups From c8a40601e8ae3eef15a9c08f6ce3649d24f72552 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 15:45:45 -0700 Subject: [PATCH 3/3] quest: trim stats linger to its release backport Co-Authored-By: Claude Opus 5.5 --- quest/m0/README.md | 2 +- quest/m0/broadcast-epoch/stats-split.md | 13 +++-- quest/m0/stats-linger.md | 65 ++++++------------------- 3 files changed, 22 insertions(+), 58 deletions(-) diff --git a/quest/m0/README.md b/quest/m0/README.md index 680f4c71e5..afd703f4c9 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -59,7 +59,7 @@ a published `@moq/watch` break. - [noq reassembly cap](/quest/m0/noq-reassembly-cap.md) - iroh's upstream noq carries quinn's stream reassembly cap, once n0 releases it - [qmux credit](/quest/m0/qmux-credit.md) - qmux returns connection credit for dropped and stopped streams and delivers its close frame, on both lines - [Shared fronts](/quest/m0/shared-fronts.md) - viewer sessions share a front, so fronts scale with peers, not viewers, and viewer churn no longer leaks fronts -- [Stats linger](/quest/m0/stats-linger.md) - a grouped stats broadcast stays announced for a linger after its last session, so viewer churn stops re-announcing it across the mesh, backported to `release` +- [Stats linger](/quest/m0/stats-linger.md) - landed on `main`; the `release` backport remains, so moq.pro's grouped stats broadcasts stop re-announcing on viewer churn - [Wildcard](/quest/m0/wildcard/README.md) - a relay resolves subscriptions against advertised prefixes, a service claims the prefix it could serve and refuses the rest instead of enumerating broadcasts, and the browser player treats a covering claim as availability - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - every first-party publisher that can restart mints a fresh `@` epoch, viewers follow the newest live one, and bare names still resolve on every version - [Audio jitter target](/quest/m0/audio-jitter-target/README.md) - the audio playout target is a measured estimate of arrival timing in both languages, not a round-trip guess diff --git a/quest/m0/broadcast-epoch/stats-split.md b/quest/m0/broadcast-epoch/stats-split.md index d133da227f..cbf944fae1 100644 --- a/quest/m0/broadcast-epoch/stats-split.md +++ b/quest/m0/broadcast-epoch/stats-split.md @@ -25,13 +25,11 @@ Decided 2026-10-05 (planned from moq-dev/moq.pro#2202): can't be published as-is: keep the totals per group, folding an entry into its group's total when it is pruned, or two projects sharing a tier would merge. -- **Idle groups** (decided 2026-10-05). Today a group broadcast disappears - the moment it has no entries (`rs/moq-stats/src/produce.rs`). Instead it - stays announced for the stats linger (`quest/m0/stats-linger.md`, an m0 - quest planned in [#4843](https://github.com/moq-dev/moq/pull/4843); list it - under Related once it lands, not Required, so it does not gate the release) - after its last entry ends, so a group that returns within the linger - continues its totals. A zero linger is valid: a returning group then always +- **Idle groups** (decided 2026-10-05). A group broadcast already stays + announced for the stats linger (`produce::Config::linger`, landed in + [#4871](https://github.com/moq-dev/moq/pull/4871)) after its last entry + ends. Today a path that left drops out of frames while the group lingers; + with totals, a group that returns within the linger continues its totals. A zero linger is valid: a returning group then always takes a new epoch. After the linger it unannounces and drops its totals; a return announces under a new [epoch](/quest/m0/broadcast-epoch/stats-epoch.md) counted from zero. Totals are cumulative and a lingering group's frames all @@ -90,6 +88,7 @@ across nodes. ## Related +- [Stats linger](/quest/m0/stats-linger.md) - the idle-group window these totals continue across - [Bounded stats aggregate](/quest/m0/broadcast-epoch/stats-aggregate-bound.md) - retired nodes fold into a bounded total; per-epoch totals feed it - [Media stats](/quest/m1/stats/README.md) - publisher and viewer media stats diff --git a/quest/m0/stats-linger.md b/quest/m0/stats-linger.md index 7e1255a20b..85e70862f8 100644 --- a/quest/m0/stats-linger.md +++ b/quest/m0/stats-linger.md @@ -1,4 +1,4 @@ -# [S] Stats broadcasts linger +# [XS] Stats broadcasts linger ## Goal @@ -10,55 +10,20 @@ them, are unchanged. ## Plan -Decided 2026-10-05 in moq.pro's quest audit: approved as recommended. The -earlier plan was in the closed -[#4052](https://github.com/moq-dev/moq/pull/4052) beside tree-routed -announces; the linger is independent of that and lands alone. It sits in m0 -as a standalone quest, not a release gate, because moq.pro's m0 stats-linger -quest waits on a `release` commit carrying it (decided 2026-10-05). - -Data from moq.pro's live fleet: on 2026-09-29 one customer's node stats -broadcasts ended on 31 nodes at once every 2 to 4 minutes and returned 46 to -112 s later, and the fleet served about 148,000 `.stats` subscriptions against -7 media ones in that hour. Size the linger from those gaps, not the one minute -first proposed, and make it configurable. - -- `moq-stats`'s producer unpublishes a group broadcast on the drain where its - group has no traffic or session rows (`publish` in - `rs/moq-stats/src/produce.rs`). Keep it, and the epoch and group sequence it - publishes under ([stats epochs](/quest/m0/broadcast-epoch/stats-epoch.md)), - until the linger elapses with the group still empty; a row returning - within the linger re-arms it. While it lingers empty, its live gauges and - session presence read zero, so a reader sees no stale live counters. -- Under one epoch totals never go backwards or count twice. On `main`, #4846 - (stats split, idle groups, decided 2026-10-05) settles it: a group - returning within the linger keeps its epoch and its totals continue, and - the per-path maps are gone. The `release` backport keeps today's per-path - map, which drops a path's entry and totals when it leaves (`flush` in - `produce.rs`); there, either the path drops out of frames while the group - lingers so its return reads as a restart, or the producer carries its last - totals forward. Pick while building the backport and record why. -- Once the linger elapses, the group broadcast unannounces. A later return - re-announces it under a new epoch, with its group sequence and totals - counted from zero (decided 2026-10-05). This reverses stats epochs' one-epoch-per-producer - rule; [#4846](https://github.com/moq-dev/moq/pull/4846) writes that reversal - into the epoch line, so this quest only follows it. -- Time decisions use `max(wall, pts)`; tests mock time. -- Depth 0 already lives for the producer's life and is unchanged. - -moq.pro tracks the `release` branch, so once this lands on `main` it is -backported to `release` as an additive PR. The backport is rewritten against -`release`'s `produce.rs`, which has no epochs: a returning group starts a new -broadcast from the wall-clock group seed of #4810 instead. - -Test: a session closes and reopens within the linger with no unannounce, -and no path's total goes backwards or counts twice across the return; while -the group lingers empty, live gauges and presence are zero; the group unannounces after the linger, and -a return after that announces a new epoch counted from zero; the reported -totals match a run without the linger. - -Public API: a linger knob on the stats producer config, surfaced in the relay -as `stats.linger` / `--stats-linger` beside `--stats-enabled`. Wire: none. +Landed on `main` in [#4871](https://github.com/moq-dev/moq/pull/4871): +`moq_stats::produce::Config::linger` (default 5 minutes), surfaced in the +relay as `stats.linger` / `--stats-linger`. While a group lingers empty its +tracks read `{}`, and a path that left drops out of frames, so its return +reads as a restart. + +Remaining: backport #4871 to `release` as an additive cherry-pick PR, since +moq.pro tracks `release` and its m0 stats-linger quest waits on a `release` +commit carrying it (decided 2026-10-05). `release`'s `produce.rs` differs +from `main`'s only by the wall-clock group seed of #4810, which keeps group +numbers increasing across a return. Both branches still have the per-path +maps, so the backport keeps #4871's choice: the path drops out of frames while the +group lingers, rather than the producer carrying its last totals forward. +Delete this quest once the backport merges. ## Related