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 95b3019084..0000000000 --- a/quest/m0/broadcast-epoch/stats-aggregate-bound.md +++ /dev/null @@ -1,47 +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 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 08441a5579..00cb36db17 100644 --- a/quest/m0/broadcast-epoch/stats-split.md +++ b/quest/m0/broadcast-epoch/stats-split.md @@ -88,7 +88,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/rs/moq-stats/Cargo.toml b/rs/moq-stats/Cargo.toml index cc6e27e3f8..cad9db9b78 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..e6ab87594a --- /dev/null +++ b/rs/moq-stats/benches/aggregate.rs @@ -0,0 +1,140 @@ +//! 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; + } + + 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; + } + // 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 { + 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..50001c3ffa 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,15 @@ 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 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 +/// sessions immediately. pub struct Consumer { origin: origin::Consumer, config: Config, @@ -215,8 +234,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 +247,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 { @@ -249,6 +272,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. @@ -261,8 +298,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 +324,8 @@ impl Merged { json }, nodes: HashMap::new(), + grace: config.grace, + retired: BTreeMap::new(), } } @@ -308,12 +352,24 @@ 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); + !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. + if self.nodes.capacity() > 4 * self.nodes.len().max(16) { + self.nodes.shrink_to_fit(); } if changed { @@ -348,6 +404,10 @@ 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); } @@ -358,15 +418,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 +441,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 +870,172 @@ 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. 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; + + 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