From cf33a499f3ada4a61783a870a6ed007d28455dfe Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 15:40:15 -0700 Subject: [PATCH 1/4] quest: claim stats-aggregate-bound Co-Authored-By: Claude Opus 5.5 From f0c93d1b5ad29c02e008c08dd60f5eaa712d039b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 16:09:41 -0700 Subject: [PATCH 2/4] fix(moq-stats): fold departed nodes into a retired total after a grace The aggregator kept every sticky node it had ever seen, so a long-lived aggregator grew with every distinct node path (scale-outs, renames, and soon every epoch). An unannounced node now waits `aggregate::Config::grace` (default 10 minutes) for its path to return, then its last counters fold into one retired total per key and the entry is dropped. Merged totals stay exact and monotonic. Adds an aggregate benchmark swept over live and departed node counts. Co-Authored-By: Claude Opus 5.5 --- quest/m0/broadcast-epoch/README.md | 1 - .../broadcast-epoch/stats-aggregate-bound.md | 48 ---- quest/m0/broadcast-epoch/stats-epoch.md | 8 +- quest/m0/broadcast-epoch/stats-split.md | 2 - quest/m0/stats-linger.md | 1 - rs/moq-stats/Cargo.toml | 4 + rs/moq-stats/benches/aggregate.rs | 138 ++++++++++ rs/moq-stats/src/aggregate.rs | 248 +++++++++++++++++- 8 files changed, 378 insertions(+), 72 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/stats-aggregate-bound.md create mode 100644 rs/moq-stats/benches/aggregate.rs diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 7f968003ee..e9a7db07f3 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -88,6 +88,5 @@ This README owns: - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and inherit the default - [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 -- [Bounded stats aggregate](/quest/m0/broadcast-epoch/stats-aggregate-bound.md) - the stats aggregator folds departed nodes into a retired total, so epoch churn stops growing its memory - [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 per-broadcast tracks](/quest/m0/broadcast-epoch/stats-split.md) - the same release retires the per-path stats maps for totals and per-broadcast tracks (decided 2026-10-05) diff --git a/quest/m0/broadcast-epoch/stats-aggregate-bound.md b/quest/m0/broadcast-epoch/stats-aggregate-bound.md deleted file mode 100644 index 6930f722dc..0000000000 --- a/quest/m0/broadcast-epoch/stats-aggregate-bound.md +++ /dev/null @@ -1,48 +0,0 @@ -# [S] Bounded stats aggregate - -## Goal - -The stats aggregator's memory is bounded by its keys plus recent churn, not -by every node it has ever seen, while merged traffic totals stay monotonic. - -Today `aggregate::Merged` keeps a sticky traffic node with `Reader::Ended` -forever (the `STICKY` unannounce branch in `rs/moq-stats/src/aggregate.rs`). -A returning path reuses its entry, but every distinct path ever announced (a -scale-out, a renamed node) stays in a long-lived aggregator's `nodes`. This is -a bug on main regardless of [stats epochs](/quest/m0/broadcast-epoch/stats-epoch.md), which -make it worse by giving every restart a distinct path. - -## Plan - -Decided (2026-10-02): grace-window fold. - -- An Ended sticky entry re-arms if its path returns within a grace window, so a - reconnect with intact counters is not new traffic. -- After the window, its last counters fold into one retired total per key and - the entry is dropped. The merged total is live nodes plus the retired total. -- Open: a path that returns after the window with intact counters would count - its pre-departure traffic twice. Decide whether that bounded error is - acceptable (rare once the window exceeds reconnect times, and with epochs - only a same-instance return after a long outage), or whether some bounded - per-path baseline reconciles it. Test exact totals across that return either - way, not just monotonicity. -- With an epoch per group announcement, a stats group returning from idle is - always a new path, so the grace re-arm covers only reconnects of a still - announced path, and the double-count above needs a same-epoch return. -- Rejected: folding on depart double-counts a node that reconnects with its - counters intact. TTL eviction without a fold makes merged totals regress. -- Timers follow the repo rule: time decisions use `max(wall, pts)`. Tests mock - time. -- A regression test churns N distinct nodes and asserts `nodes` stays bounded - while the merged total never regresses. -- Add or extend a benchmark swept over node count, per the fan-out guidance in - `AGENTS.md`. - -Public API: none expected beyond a possible grace-window knob on -`aggregate::Config`. Wire: none. - -## Related - -- [Stats epochs](/quest/m0/broadcast-epoch/stats-epoch.md) - every restarted node gets a new name, so retired entries churn faster -- [Binary delta stats](/quest/m2/stats-delta.md) - also sweeps the aggregate over node count -- [Stats linger](/quest/m0/stats-linger.md) - a group's node broadcast outlives a short gap instead of churning an aggregate entry diff --git a/quest/m0/broadcast-epoch/stats-epoch.md b/quest/m0/broadcast-epoch/stats-epoch.md index 38300cc73b..85f21240f1 100644 --- a/quest/m0/broadcast-epoch/stats-epoch.md +++ b/quest/m0/broadcast-epoch/stats-epoch.md @@ -28,8 +28,8 @@ a broadcast name or the group numbers a relay cached under it. - Decide what an unset node becomes once the epoch follows it (#4739 used `local`). - The aggregator treats a new epoch as a new node: its counters add to the - retired total from the [bounded aggregate](/quest/m0/broadcast-epoch/stats-aggregate-bound.md) - instead of regressing the merged total. + merged total instead of regressing it, and the old epoch folds into the + retired total once `aggregate::Config::grace` elapses. Test it. - Expose each group broadcast's epoch (a per-group accessor replacing #4739's `Producer::epoch()`) so a consumer can tell epochs apart. MoQ Pro's VOD `storage.json` mints its own epoch rather than sharing a @@ -47,7 +47,3 @@ a broadcast name or the group numbers a relay cached under it. Public API: path shape change for every stats consumer. Wire: stats broadcast names gain a trailing epoch segment. - -## Required - -- [Bounded stats aggregate](/quest/m0/broadcast-epoch/stats-aggregate-bound.md) - retired nodes fold into a bounded total, which epochs churn diff --git a/quest/m0/broadcast-epoch/stats-split.md b/quest/m0/broadcast-epoch/stats-split.md index d133da227f..a2be5e8bac 100644 --- a/quest/m0/broadcast-epoch/stats-split.md +++ b/quest/m0/broadcast-epoch/stats-split.md @@ -90,7 +90,5 @@ across nodes. ## Related -- [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 stay hang tracks, separate from the relay's stats diff --git a/quest/m0/stats-linger.md b/quest/m0/stats-linger.md index 7e1255a20b..fa3b15f564 100644 --- a/quest/m0/stats-linger.md +++ b/quest/m0/stats-linger.md @@ -63,5 +63,4 @@ as `stats.linger` / `--stats-linger` beside `--stats-enabled`. Wire: none. ## Related - [Stats epochs](/quest/m0/broadcast-epoch/stats-epoch.md) - the name a lingering broadcast keeps -- [Bounded stats aggregate](/quest/m0/broadcast-epoch/stats-aggregate-bound.md) - the aggregator's grace window for departed nodes - [moq.pro: stats linger](https://github.com/moq-dev/moq.pro/blob/main/quest/m0/stats-linger.md) - the fleet adoption and its measurements diff --git a/rs/moq-stats/Cargo.toml b/rs/moq-stats/Cargo.toml index e03c1ca484..1c4f577e42 100644 --- a/rs/moq-stats/Cargo.toml +++ b/rs/moq-stats/Cargo.toml @@ -28,3 +28,7 @@ tokio = { workspace = true, features = ["macros", "rt", "test-util", "time"] } [[bench]] name = "decode" harness = false + +[[bench]] +name = "aggregate" +harness = false diff --git a/rs/moq-stats/benches/aggregate.rs b/rs/moq-stats/benches/aggregate.rs new file mode 100644 index 0000000000..ab4c9a6e04 --- /dev/null +++ b/rs/moq-stats/benches/aggregate.rs @@ -0,0 +1,138 @@ +//! Cost of one merged traffic frame from [`moq_stats::aggregate`], swept over live nodes and +//! nodes that have come and gone, with and without the grace that folds departed nodes away. +//! +//! Each node reports the same broadcast paths, like a restarted node under a new name. A bounded +//! aggregate's cost follows the live nodes; an unbounded one (`grace` never elapsing) also pays for +//! every departed node on each frame. Time is paused, so the churn ages past the grace instantly. +//! +//! Run `cargo bench -p moq-stats --bench aggregate` and compare the table across revisions. + +use std::time::{Duration, Instant}; + +use moq_net::origin; +use moq_stats::{Role, Tier, Traffic, TrafficFrame, aggregate, traffic_track}; + +/// Broadcast paths each node reports. +const KEYS: usize = 16; + +/// Frames measured per configuration, after one warm-up frame. +const FRAMES: u32 = 64; + +/// The grace configured when folding; the churn ages past it before measuring. +const GRACE: Duration = Duration::from_secs(1); + +/// A hand-published node stats broadcast with a default-tier publisher traffic track. +struct Node { + _source: moq_net::broadcast::Producer, + traffic: moq_json::snapshot::Producer, + frame: TrafficFrame, +} + +impl Node { + fn new(origin: &origin::Producer, name: &str) -> Self { + let source = origin + .create_broadcast(format!(".stats/acme/node/{name}").as_str()) + .unwrap(); + source.announce(origin::Route::default()).unwrap(); + let track = source + .create_track(traffic_track(&Tier::default(), Role::Publisher, false), None) + .unwrap(); + let config = moq_json::snapshot::Config::default().with_delta_ratio(0); + let frame = (0..KEYS) + .map(|key| (format!("acme/room-{key:02}"), Traffic::default())) + .collect(); + Self { + _source: source, + traffic: moq_json::snapshot::Producer::new(track, config), + frame, + } + } + + /// Bump every path's byte counter and publish the whole snapshot. + fn publish(&mut self, bytes: u64) { + for traffic in self.frame.values_mut() { + traffic.bytes += bytes; + } + self.traffic.update(&self.frame).unwrap(); + } +} + +/// Read merged frames until the first path reaches `want` bytes. +async fn read_until(traffic: &mut aggregate::TrafficConsumer, want: u64) { + loop { + let frame = traffic.next().await.unwrap().unwrap(); + if frame.values().next().is_some_and(|t| t.bytes >= want) { + return; + } + } +} + +/// Mean wall time per merged frame with `live` nodes publishing after `departed` nodes left. +async fn measure(live: usize, departed: usize, grace: Duration) -> Duration { + let (origin, driver) = origin::Producer::new(origin::Config::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let agg = aggregate::Consumer::new( + origin.consume(), + aggregate::Config::new().with_depth(1).with_grace(grace), + ); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + + // Every node contributes 1 byte per path, so the total counts nodes seen. + let mut total = 0; + for index in 0..departed { + let mut node = Node::new(&origin, &format!("gone-{index}")); + node.publish(1); + total += 1; + read_until(&mut traffic, total).await; + } + tokio::time::advance(GRACE * 2).await; + + let mut nodes: Vec = (0..live) + .map(|index| Node::new(&origin, &format!("live-{index}"))) + .collect(); + for node in &mut nodes { + node.publish(1); + total += 1; + read_until(&mut traffic, total).await; + } + + let mut elapsed = Duration::ZERO; + for frame in 0..=FRAMES { + let node = &mut nodes[frame as usize % live]; + let start = Instant::now(); + node.publish(1); + total += 1; + read_until(&mut traffic, total).await; + // The first frame folds the departed nodes; it is the warm-up. + if frame > 0 { + elapsed += start.elapsed(); + } + } + + drop(nodes); + driver.abort(); + elapsed / FRAMES +} + +fn main() { + // Nextest lists all targets as potential test binaries. + if std::env::args().any(|arg| arg == "--list") { + return; + } + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .start_paused(true) + .build() + .unwrap(); + + println!("live departed grace us/frame"); + for live in [1, 16, 128] { + for departed in [0, 256, 4096] { + for (label, grace) in [("1s", GRACE), ("never", Duration::MAX)] { + let elapsed = runtime.block_on(measure(live, departed, grace)); + let micros = elapsed.as_secs_f64() * 1e6; + println!("{live:>4} {departed:>8} {label:>5} {micros:>11.1}"); + } + } + } +} diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 9203ee6612..0ed7533079 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -9,11 +9,13 @@ use std::collections::hash_map::Entry; use std::collections::{BTreeMap, HashMap}; use std::task::Poll; +use std::time::Duration; use moq_net::kio::{self, Pending, Waiter}; use moq_net::stats::{Presence, Role, Tier, Traffic}; use moq_net::track::Subscribing; use moq_net::{PathOwned, origin}; +use web_async::time::Instant; use crate::{Result, SessionsFrame, TrafficFrame, parse_node_path, sessions_track, traffic_track}; @@ -38,6 +40,11 @@ pub struct Config { /// Same data for a fraction of the bytes, but requires a producer that /// publishes them. Defaults to `false`. pub compression: bool, + /// How long a departed node's traffic waits for its path to return before + /// folding into the retired total (default 10 minutes). A return within it + /// resumes the node's own counters; a return after it with counters intact + /// counts its earlier traffic twice, so size this above reconnect times. + pub grace: Duration, } impl Config { @@ -64,6 +71,12 @@ impl Config { self.compression = compression; self } + + /// Override how long a departed node waits to return (default 10 minutes). + pub fn with_grace(mut self, grace: Duration) -> Self { + self.grace = grace; + self + } } impl Default for Config { @@ -72,6 +85,8 @@ impl Default for Config { prefix: PathOwned::from(".stats"), depth: 0, compression: false, + // Well above reconnect times; a longer grace only costs memory. + grace: Duration::from_secs(600), } } } @@ -83,11 +98,14 @@ impl Default for Config { /// announce cursor and subscribes to that track on every node broadcast in the /// group, summing the cumulative counters per key. Traffic is sticky: a node /// dropping out (its broadcast unannounces or its reader ends) keeps its last -/// contribution, so a relay that returns with its boot-lifetime counters -/// intact never looks like new traffic. Only a genuine per-node counter -/// regression (a restarted relay) regresses the merged counter, the same reset -/// contract a single node's own restart follows. Presence is not sticky: a -/// departed node stops counting sessions immediately. +/// contribution, so a relay that returns within [`Config::grace`] with its +/// boot-lifetime counters intact never looks like new traffic. After the grace, +/// a departed node's contribution folds into one retired total and the node is +/// forgotten, so memory follows live nodes, not every node ever seen. A node +/// returning within the grace with a lower counter (a restarted relay) +/// regresses the merged counter, the same reset contract a single node's own +/// restart follows. Presence is not sticky: a departed node stops counting +/// sessions immediately. pub struct Consumer { origin: origin::Consumer, config: Config, @@ -215,8 +233,8 @@ enum Reader { /// and one lives per node in a map. Active(Box>>), /// The subscription failed or the track ended; the node no longer reads. It - /// lingers until it unannounces or reannounces, still contributing its last - /// frame when [`Mergeable::STICKY`]. + /// lingers until it reannounces or, once unannounced, its grace elapses, + /// still contributing its last frame when [`Mergeable::STICKY`]. Ended, } @@ -228,6 +246,10 @@ struct Node { /// after the reader ends requests again. path: PathOwned, last: Option>, + /// When the path unannounced, starting its grace; cleared by a reannounce. + /// Only an unannounced node folds, so a still-announced node whose reader + /// failed keeps its own counters. + departed: Option, } impl Node { @@ -261,8 +283,13 @@ struct Merged { /// Track name subscribed on each node broadcast. name: String, config: moq_json::snapshot::consumer::Config, - /// One entry per live node broadcast, keyed by absolute announced path. + /// One entry per live or recently departed node broadcast, keyed by + /// absolute announced path. nodes: HashMap>, + /// How long a departed node waits before folding into `retired`. + grace: Duration, + /// The summed last frames of every node whose grace elapsed. + retired: BTreeMap, } impl Merged { @@ -282,6 +309,8 @@ impl Merged { json }, nodes: HashMap::new(), + grace: config.grace, + retired: BTreeMap::new(), } } @@ -308,12 +337,32 @@ impl Merged { } } - // Advance each node's reader, collapsing any backlog to its latest frame. + // Advance each node's reader, collapsing any backlog to its latest frame, + // then fold every departed node whose grace elapsed. A fold leaves the + // merged view unchanged, and only an announce adds a node, which polls + // here, so no timer is needed to keep the map bounded. + let now = Instant::now(); let config = &self.config; let name = self.name.as_str(); let origin = &self.origin; - for node in self.nodes.values_mut() { + let grace = self.grace; + let retired = &mut self.retired; + self.nodes.retain(|_, node| { changed |= advance(node, origin, config, name, waiter); + // A grace too long to represent never folds. + let Some(departed) = node.departed else { return true }; + if departed.checked_add(grace).is_none_or(|at| at > now) { + return true; + } + for (key, value) in node.last.take().into_iter().flatten() { + V::merge(retired.entry(key).or_default(), value); + } + false + }); + // Release a churn peak's buckets: walking a map costs its capacity, so a + // map that once held thousands of nodes would slow every frame. + if self.nodes.capacity() > 4 * self.nodes.len().max(16) { + self.nodes.shrink_to_fit(); } if changed { @@ -348,6 +397,7 @@ impl Merged { match self.nodes.entry(absolute) { Entry::Occupied(mut entry) => { let node = entry.get_mut(); + node.departed = None; if matches!(node.reader, Reader::Ended) { node.reader = resolve(&self.origin, &node.path); } @@ -358,15 +408,20 @@ impl Merged { reader: resolve(&self.origin, &path), path, last: None, + departed: None, }); false } } } else if V::STICKY { // Unannounce: keep the cumulative totals, retire the live gauges, and - // stop reading. The entry stays so a reannounce re-arms it. + // stop reading. The entry stays so a reannounce within the grace + // re-arms it. match self.nodes.get_mut(&absolute) { - Some(node) => node.depart(), + Some(node) => { + node.departed = Some(Instant::now()); + node.depart() + } None => false, } } else { @@ -376,9 +431,9 @@ impl Merged { } } - /// Sum every node's last frame, per key. + /// Sum the retired total and every node's last frame, per key. fn merged(&self) -> BTreeMap { - let mut acc: BTreeMap = BTreeMap::new(); + let mut acc = self.retired.clone(); for node in self.nodes.values() { if let Some(last) = &node.last { for (key, value) in last { @@ -805,6 +860,171 @@ mod tests { ); } + /// Let the origin deliver pending announces, then poll `traffic` once so it + /// stamps departures and folds expired nodes. Returns a frame if one was + /// ready. + async fn settle(traffic: &mut TrafficConsumer) -> Option { + use futures::FutureExt; + for _ in 0..8 { + tokio::task::yield_now().await; + } + traffic + .next() + .now_or_never() + .map(|frame| frame.expect("read").expect("frame")) + } + + /// Node A (100) and node B (40) merged to 140, then node A departs and the + /// aggregator notices. Returns A's frame so a test can bring it back with its + /// counters intact, plus the still-live node B. + async fn depart_node_a(origin: &origin::Producer, traffic: &mut TrafficConsumer) -> (TrafficFrame, NodeBroadcast) { + let mut node_a = NodeBroadcast::new(origin, "acme", "a"); + let mut node_b = NodeBroadcast::new(origin, "acme", "b"); + node_a.publish("acme/room", 100); + node_b.publish("acme/room", 40); + read_until_bytes(traffic, "acme/room", 140).await; + + let frame = node_a.frame.clone(); + drop(node_a); + settle(traffic).await; + assert_eq!(traffic.inner.nodes.len(), 2, "a departed node lingers for its grace"); + (frame, node_b) + } + + const GRACE: Duration = Duration::from_secs(60); + + #[tokio::test(start_paused = true)] + async fn return_within_grace_resumes_counters() { + // A node returning within the grace with its counters intact replaces + // its own kept contribution: no traffic counts twice. + let origin = produce_origin(); + let agg = Consumer::new(origin.consume(), Config::new().with_depth(1).with_grace(GRACE)); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + let (frame, _node_b) = depart_node_a(&origin, &mut traffic).await; + + tokio::time::advance(GRACE - Duration::from_secs(1)).await; + let mut node_a = NodeBroadcast::new(&origin, "acme", "a"); + node_a.frame = frame; + node_a.publish("acme/room", 20); + + let frame = read_monotonic_until(&mut traffic, "acme/room", 140, 160).await; + assert_eq!(frame.get("acme/room").expect("entry").bytes, 160, "100 + 20 + 40"); + assert!(traffic.inner.retired.is_empty(), "nothing folded"); + } + + #[tokio::test(start_paused = true)] + async fn return_after_grace_counts_twice() { + // After the grace, a departed node folds into the retired total and its + // entry is dropped. If the same path then returns with its counters + // intact, its pre-departure traffic counts twice: the accepted, bounded + // cost of forgetting it. The total still never regresses. + let origin = produce_origin(); + let agg = Consumer::new(origin.consume(), Config::new().with_depth(1).with_grace(GRACE)); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + let (frame, _node_b) = depart_node_a(&origin, &mut traffic).await; + + tokio::time::advance(GRACE).await; + assert!( + settle(&mut traffic).await.is_none(), + "a fold leaves the merged view alone" + ); + assert_eq!(traffic.inner.nodes.len(), 1, "the departed node folded"); + assert_eq!(traffic.inner.retired.get("acme/room").map(|t| t.bytes), Some(100)); + + let mut node_a = NodeBroadcast::new(&origin, "acme", "a"); + node_a.frame = frame; + node_a.publish("acme/room", 20); + + let frame = read_monotonic_until(&mut traffic, "acme/room", 140, 260).await; + assert_eq!( + frame.get("acme/room").expect("entry").bytes, + 260, + "retired 100 + returned 120 + 40" + ); + } + + #[tokio::test(start_paused = true)] + async fn restart_after_grace_adds_to_the_total() { + // A node returning after the grace with fresh counters (it restarted) + // adds to the retired total instead of regressing it. + let origin = produce_origin(); + let agg = Consumer::new(origin.consume(), Config::new().with_depth(1).with_grace(GRACE)); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + let (_frame, _node_b) = depart_node_a(&origin, &mut traffic).await; + + tokio::time::advance(GRACE).await; + settle(&mut traffic).await; + + let mut node_a = NodeBroadcast::new(&origin, "acme", "a"); + node_a.publish("acme/room", 30); + + let frame = read_monotonic_until(&mut traffic, "acme/room", 140, 170).await; + assert_eq!(frame.get("acme/room").expect("entry").bytes, 170, "100 + 30 + 40"); + } + + #[tokio::test(start_paused = true)] + async fn failed_reader_does_not_fold_while_announced() { + // A node whose reader failed under a still-announced broadcast keeps its + // own entry past the grace: only an unannounced node is forgotten. + let origin = produce_origin(); + let mut node_a = NodeBroadcast::new(&origin, "acme", "a"); + node_a.publish("acme/room", 100); + + let agg = Consumer::new(origin.consume(), Config::new().with_depth(1).with_grace(GRACE)); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + read_until_bytes(&mut traffic, "acme/room", 100).await; + + node_a.fail_traffic(); + settle(&mut traffic).await; + tokio::time::advance(GRACE * 2).await; + settle(&mut traffic).await; + assert_eq!(traffic.inner.nodes.len(), 1, "still announced, so not folded"); + assert!(traffic.inner.retired.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn churn_keeps_nodes_bounded() { + // Distinct nodes come and go, as epochs make every restart do. The + // node map holds only the live node plus those within their grace, while + // the merged total stays exact and never regresses. + const CHURN: u64 = 64; + const STEP: Duration = Duration::from_secs(15); + + let origin = produce_origin(); + let agg = Consumer::new(origin.consume(), Config::new().with_depth(1).with_grace(GRACE)); + let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); + + for i in 0..CHURN { + let mut node = NodeBroadcast::new(&origin, "acme", &format!("n{i}")); + node.publish("acme/room", 10); + let want = 10 * (i + 1); + let frame = read_monotonic_until(&mut traffic, "acme/room", 10 * i, want).await; + assert_eq!(frame.get("acme/room").expect("entry").bytes, want); + + drop(node); + settle(&mut traffic).await; + tokio::time::advance(STEP).await; + + // Departures within the last grace, plus one that came due since the + // last poll. + let bound = (GRACE.as_secs() / STEP.as_secs() + 1) as usize; + assert!( + traffic.inner.nodes.len() <= bound, + "{} nodes after {} departures", + traffic.inner.nodes.len(), + i + 1 + ); + } + + tokio::time::advance(GRACE).await; + settle(&mut traffic).await; + assert!(traffic.inner.nodes.is_empty(), "every departed node folded"); + assert_eq!( + traffic.inner.retired.get("acme/room").map(|t| t.bytes), + Some(10 * CHURN) + ); + } + #[tokio::test(start_paused = true)] async fn unannounce_retires_live_gauges() { // A departed node's totals stay in the merged view, but its live gauges From 4f8b93970b84fbe0cbcb4723b04d8397b2ad7314 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 17:28:32 -0700 Subject: [PATCH 3/4] fix(moq-stats): fold an expired node when its path returns before a sweep A reannounce cleared the departure stamp before the poll's sweep, so a path returning as the first wake after the grace escaped the fold: a restarted node regressed the merged total. Co-Authored-By: Claude Opus 5.5 --- rs/moq-stats/src/aggregate.rs | 32 +++++++++++++++++++++----------- 1 file changed, 21 insertions(+), 11 deletions(-) diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 0ed7533079..5cc3caa7df 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -271,6 +271,20 @@ impl Node { } changed } + + /// Fold the last frame into `retired` if the node departed at least `grace` + /// before `now`, leaving the merged view unchanged. Returns whether it + /// folded. A grace too long to represent never folds. + fn fold_expired(&mut self, retired: &mut BTreeMap, grace: Duration, now: Instant) -> bool { + let Some(departed) = self.departed else { return false }; + if departed.checked_add(grace).is_none_or(|at| at > now) { + return false; + } + for (key, value) in self.last.take().into_iter().flatten() { + V::merge(retired.entry(key).or_default(), value); + } + true + } } /// Watches a group's node announces and folds one track across all of them. @@ -349,15 +363,7 @@ impl Merged { let retired = &mut self.retired; self.nodes.retain(|_, node| { changed |= advance(node, origin, config, name, waiter); - // A grace too long to represent never folds. - let Some(departed) = node.departed else { return true }; - if departed.checked_add(grace).is_none_or(|at| at > now) { - return true; - } - for (key, value) in node.last.take().into_iter().flatten() { - V::merge(retired.entry(key).or_default(), value); - } - false + !node.fold_expired(retired, grace, now) }); // Release a churn peak's buckets: walking a map costs its capacity, so a // map that once held thousands of nodes would slow every frame. @@ -397,6 +403,9 @@ impl Merged { match self.nodes.entry(absolute) { Entry::Occupied(mut entry) => { let node = entry.get_mut(); + // A return after the grace starts over, even when this announce + // is the first poll since the grace elapsed. + node.fold_expired(&mut self.retired, self.grace, Instant::now()); node.departed = None; if matches!(node.reader, Reader::Ended) { node.reader = resolve(&self.origin, &node.path); @@ -946,14 +955,15 @@ mod tests { #[tokio::test(start_paused = true)] async fn restart_after_grace_adds_to_the_total() { // A node returning after the grace with fresh counters (it restarted) - // adds to the retired total instead of regressing it. + // adds to the retired total instead of regressing it. No poll runs + // between the grace elapsing and the return, so the return itself must + // fold the expired entry. let origin = produce_origin(); let agg = Consumer::new(origin.consume(), Config::new().with_depth(1).with_grace(GRACE)); let mut traffic = agg.traffic(&Tier::default(), Role::Publisher); let (_frame, _node_b) = depart_node_a(&origin, &mut traffic).await; tokio::time::advance(GRACE).await; - settle(&mut traffic).await; let mut node_a = NodeBroadcast::new(&origin, "acme", "a"); node_a.publish("acme/room", 30); From 1c7c625006f7b305845c0dcec79f934e9e075bcd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 19:32:04 -0700 Subject: [PATCH 4/4] fix(moq-stats): fold every departed node before the aggregate bench measures Also scope the Consumer memory claim to live nodes plus reported keys. Co-Authored-By: Claude Opus 5.5 --- rs/moq-stats/benches/aggregate.rs | 4 +++- rs/moq-stats/src/aggregate.rs | 3 ++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/rs/moq-stats/benches/aggregate.rs b/rs/moq-stats/benches/aggregate.rs index ab4c9a6e04..e6ab87594a 100644 --- a/rs/moq-stats/benches/aggregate.rs +++ b/rs/moq-stats/benches/aggregate.rs @@ -85,7 +85,6 @@ async fn measure(live: usize, departed: usize, grace: Duration) -> Duration { total += 1; read_until(&mut traffic, total).await; } - tokio::time::advance(GRACE * 2).await; let mut nodes: Vec = (0..live) .map(|index| Node::new(&origin, &format!("live-{index}"))) @@ -95,6 +94,9 @@ async fn measure(live: usize, departed: usize, grace: Duration) -> Duration { total += 1; read_until(&mut traffic, total).await; } + // Advance only after a read has seen the last departure, so every departed + // node is past its grace. + tokio::time::advance(GRACE * 2).await; let mut elapsed = Duration::ZERO; for frame in 0..=FRAMES { diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 5cc3caa7df..50001c3ffa 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -101,7 +101,8 @@ impl Default for Config { /// contribution, so a relay that returns within [`Config::grace`] with its /// boot-lifetime counters intact never looks like new traffic. After the grace, /// a departed node's contribution folds into one retired total and the node is -/// forgotten, so memory follows live nodes, not every node ever seen. A node +/// forgotten, so memory follows live nodes and the keys ever reported, not +/// every node ever seen. A node /// returning within the grace with a lower counter (a restarted relay) /// regresses the merged counter, the same reset contract a single node's own /// restart follows. Presence is not sticky: a departed node stops counting