From 03c7a93ce9340dd59ad4851a8ff6243cbc6ea3d8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 15:44:45 -0700 Subject: [PATCH] feat(stats): linger an empty group broadcast before unannouncing it (backport #4871) 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`. Cherry-pick of #4871 (a52ee7ea2) onto `release`; the announce test helper uses `release`'s announce update API. 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 841d8bb13b..8282c145b4 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -220,6 +220,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 ea370e84ef..6fec5048a7 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(first_sequence(SystemTime::now()))), }; @@ -219,6 +243,7 @@ struct Task { prefix: PathOwned, node: Option, depth: usize, + linger: Duration, interval: Duration, sequence: Arc, } @@ -244,7 +269,7 @@ impl Task { } drain.collect(); - drain.publish(); + drain.publish(Instant::now()); } } fn node(&self) -> Option<&str> { @@ -343,14 +368,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. @@ -758,6 +783,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. @@ -830,9 +857,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 { @@ -1290,6 +1329,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) @@ -1489,6 +1540,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(update)) = events.next().now_or_never() { + out.push((update.prefix.as_str().to_string(), update.kind.is_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")); @@ -1890,6 +2049,7 @@ mod tests { node: None, sequence: Arc::new(AtomicU64::new(0)), depth, + linger: DEFAULT_LINGER, interval: Duration::from_secs(1), }) .expect("drain"); @@ -1897,13 +2057,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