From 2ef14a4238cb601bad84a46506b05606dedd29f5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 22:59:13 -0800 Subject: [PATCH 1/9] chore(quest): claim quest/m1/bench-test-mode.md Quest-Claim: 268386aae45b142d From df14216eaf8a8022d0a930e4caf512b17e40424e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:26:43 -0800 Subject: [PATCH 2/9] fix(net): retain unread benchmark batches through delivery Co-Authored-By: GPT-6.1 Sol --- quest/m1/bench-test-mode.md | 62 ++++++++++++------------------- rs/moq-net/benches/memory.rs | 29 ++++++++++++--- rs/moq-net/benches/origin.rs | 1 + rs/moq-net/benches/session/mod.rs | 30 +++++++-------- 4 files changed, 61 insertions(+), 61 deletions(-) diff --git a/quest/m1/bench-test-mode.md b/quest/m1/bench-test-mode.md index 0c0952cfcc9..96b10236c6f 100644 --- a/quest/m1/bench-test-mode.md +++ b/quest/m1/bench-test-mode.md @@ -11,48 +11,32 @@ such break. ## Plan -Planned 2026-11-04 with the maintainer. While iterating -[moq-dev/moq#7729](https://github.com/moq-dev/moq/pull/7729) (authored tracks -default to zero max age, merged 2026-11-04), its agent ran the Rust benches in -test mode: `group` panicked under the new default and was fixed there, while -`origin`, `session` and `track` in `rs/moq-net/benches/` panicked on main -before the change. - -Nightly already runs these targets measured: `sh/gh/nightly.sh`'s `rust-bench` -step runs `criterion net --no-fail-fast -p moq-net --features fuzz --bench '*'`. -Scheduled run -[41656613559](https://github.com/moq-dev/moq/actions/runs/41656613559) -(2026-11-03, before #7729) failed it on four targets: - -- `origin` (`origin.rs:574` and `:576`, `winner changed`) in - `origin/reprice_duplicate/1d_1s_cost2`. -- `session` (`session/mod.rs:371` and `:492`, `Evicted`) in - `session_delivery_burst/moq-transport-22/...`. -- `memory` (`memory.rs:194`, `Evicted`) in `memory/streams=16/size=1048576`. -- `track` (`track.rs:1718` and `:1730`) in `track_range_widening/1r/64`, fixed - by [moq-dev/moq#7840](https://github.com/moq-dev/moq/pull/7840) (merged - 2026-11-04). - -Start by re-running every moq-net bench target on main, in test mode and in a -short measured run, since #7729's zero max-age default may have broken more. -Fix each at its cause: a bench whose expectation predates a deliberate -behaviour change is updated; a bench exposing a product bug gets the product -fix. `origin` and `session` need `fuzz`, and Cargo skips a bench whose -`required-features` are off without a word, so a test-mode run without it -passes while they're broken. - -Where CI catches the next break is the maintainer's call; ask in the PR. A -test-mode run in Nightly adds nothing, since the measured run covers the same -targets. Today the PR gate builds benches with their features but leaves them -unrun (`rs/justfile` above the `test` recipe; `.config/nextest.toml`'s -`default-filter = "not kind(bench)"`). Running them there, which would have -caught these before merge, reverses the -[Nightly epic](/quest/m0/nightly/README.md)'s 2026-10-24 decision that the -gate keeps skipping benches and a green Nightly catches them within a day. -Measure the test-mode run's cost before asking. +The implementation PR fixes the benchmark fixtures without changing the product: + +- `origin/reprice_duplicate` uses covering discovery, since the announced prefix + is above each cursor's allowed paths. +- Memory and session delivery retain unread groups without byte-pressure + shedding. Cache expiry between rounds bounds history outside measured work. + The shared session fixture also covers the production relay benchmark. +- The full test-mode and short measured sweeps verify every moq-net target. + +Nightly already runs the measured targets in `sh/gh/nightly.sh`'s `rust-bench` +step. The PR gate builds benches with their required features, but excludes +running them through `.config/nextest.toml`'s `not kind(bench)` default filter. +No CI placement change is included pending the maintainer's decision below. Public API: none. Wire: none. +## Waiting + +- Maintainer: choose where CI catches the next benchmark failure. Options: + retain measured Nightly coverage, or run moq-net benches in test mode in the + PR gate as well. Recommend retaining Nightly coverage, consistent with the + Nightly epic's existing policy; the PR records the full test-mode cost. + This reverses a recorded maintainer policy if changed, so an agent cannot + choose it unattended. Check the decision is recorded in the PR, then apply + any chosen gate change and finish this quest. + ## Related - [Nightly passes on main](/quest/m0/nightly/README.md) - this quest owns the moq-net bench panics in Nightly's "Rust bench targets" job diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index d6bea4918ee..4815e0d775a 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -100,6 +100,8 @@ fn now() -> moq_net::time::Instant { /// One publisher and one subscriber session over an in-memory pair. struct Link { + pools: Vec, + cache_now: moq_net::time::Instant, tracks: Vec, subscribers: Vec, payload: Bytes, @@ -110,12 +112,13 @@ struct Link { impl Link { async fn new(streams: usize, size: usize) -> Self { + let mut pools = Vec::new(); let version: Version = VERSION.parse().unwrap(); - let origin = |hop: u64| { + let mut origin = |hop: u64| { let mut config = origin::Config::new(Hop::new(hop).unwrap()); - // Hold one round's groups, with room for the handover, and no more. - config.pool = - cache::Pool::new(cache::Config::default().with_capacity((4 * streams * FRAMES * size) as u64)); + // Expire history between rounds without shedding unread frames under byte pressure. + config.pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY)); + pools.push(config.pool.clone()); let (producer, driver) = origin::Producer::new(config); tokio::spawn(moq_net::time::run(driver)); producer @@ -165,6 +168,8 @@ impl Link { } let mut link = Self { + pools, + cache_now: now(), tracks, subscribers, payload: Bytes::from(vec![0; size]), @@ -174,6 +179,7 @@ impl Link { }; // Warm every stream so the timed rounds skip first-group setup. for _ in 0..4 { + link.tick(); link.round().await; } link @@ -200,13 +206,24 @@ impl Link { bytes } + // Bound retained history to two rounds plus the protected latest group. + fn tick(&mut self) { + self.cache_now += cache::DEFAULT_EXPIRY; + for pool in &self.pools { + pool.poll(self.cache_now, &kio::Waiter::noop()); + } + } + async fn measure(&mut self, iters: u64) -> Duration { let expected = self.tracks.len() * FRAMES * self.payload.len(); - let start = Instant::now(); + let mut elapsed = Duration::ZERO; for _ in 0..iters { + self.tick(); + let start = Instant::now(); assert_eq!(self.round().await, expected); + elapsed += start.elapsed(); } - start.elapsed() + elapsed } } diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 3ad61f8c0f8..24f3ccc154a 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -564,6 +564,7 @@ fn bench_reprice_duplicate(c: &mut Criterion) { ) .unwrap() .with_hidden(true) + .covering() .announced() }) .collect(); diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index 61744ecb853..b411622d13d 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -123,9 +123,8 @@ const DASH_VERSIONS: [&str; 2] = ["moq-lite-06", "moq-transport-22"]; /// A stats frame: a small JSON snapshot. const DASH_FRAME: usize = 128; -/// Each endpoint gets its own bounded pool, as a relay configures one per -/// process: without a byte target, every hop keeps each group for the whole -/// expiry window and memory grows with the run length instead of the shape. +/// Initial budgets for join and churn fixtures. Delivery rooms use expiry +/// between rounds instead, so cache pressure never sheds unread frames. const RELAY_CACHE: u64 = 16 * 1024 * 1024; const CLIENT_CACHE: u64 = 1024 * 1024; @@ -571,7 +570,9 @@ fn make_origin(id: u64, capacity: u64) -> (origin::Producer, cache::Pool) { let pool = cache::Pool::new(if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY) } else { - cache::Config::default().with_capacity(capacity) + cache::Config::default() + .with_capacity(capacity) + .with_expiry(cache::DEFAULT_EXPIRY) }); config.pool = pool.clone(); let (producer, driver) = origin::Producer::new(config); @@ -876,9 +877,8 @@ impl Room { room.writing.resize_with(room.watched.len(), || None); room.reading.resize_with(shape.viewers * shape.watch, || None); - // First measure the complete batch without eviction. The delivery benchmark must - // fit its working set; cache-pressure loss is a different workload. - let capacities: Vec<_> = room.pools().map(cache::Pool::capacity).collect(); + // Delivery must retain every unread group. Bound history by expiry between + // rounds instead of byte pressure, which may shed an in-flight group's head. for pool in room.pools() { pool.resize(None); } @@ -895,11 +895,6 @@ impl Room { room.tick(); room.round().await; } - for (pool, capacity) in room.pools().zip(capacities) { - // Keep the old and current batch resident during the handover, including - // bookkeeping rather than estimating a group from payload bytes alone. - pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 2))); - } // Exercise turnover as well as the empty cache, including Criterion's one-pass // nightly smoke mode. for _ in 0..4 { @@ -942,11 +937,14 @@ impl Room { self.cluster.pools.iter().chain(viewers) } - /// Unpaced rounds otherwise collapse many complete batches into one cache tick, - /// where yesterday's read outranks today's unread write. Sample a new recency - /// epoch per round without sleeping or changing protocol time. + /// Expire old batches outside timing. Activity is dated on the following poll, + /// so history holds two rounds plus each track's protected latest group. fn tick(&mut self) { - self.cache_now += Duration::from_secs(1); + self.cache_now += if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { + Duration::from_secs(1) + } else { + cache::DEFAULT_EXPIRY + }; for pool in self.pools() { pool.poll(self.cache_now, &kio::Waiter::noop()); } From 29af2a67982b872da120089c4a639de25f7c9721 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:37:17 -0800 Subject: [PATCH 3/9] fix(bench): scope expiry policy to delivery fixtures Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/benches/session/mod.rs | 40 +++++++++++++++++-------------- rs/moq-relay/benches/session.rs | 4 ++-- 2 files changed, 24 insertions(+), 20 deletions(-) diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index b411622d13d..edb997c2dc0 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -34,9 +34,10 @@ //! one datagram per track per round across publisher and viewer axes, then larger payloads //! and cadences, on the lite versions (moq-transport carries no datagram tracks). Throughput //! counts delivered datagrams, each checked for its sequence and payload, so a lost one fails. -//! `SESSION_EXPIRY_ONLY=1` uses the standalone 30s-expiry pool and reports groups -//! written plus charged cache bytes every 10s during long profiling runs. Each round -//! advances the cache clock by a second, so the expiry window spans 30 rounds. +//! Delivery history expires between rounds, outside timing, without byte-pressure loss. +//! `SESSION_EXPIRY_ONLY=1` advances the cache clock by a second instead of the 30s +//! expiry window, retaining 30 rounds for profiling. It reports groups written and +//! charged cache bytes every 10s during long runs. //! //! `session_join_*` times a new viewer connecting, resolving a broadcast, and //! receiving its latest group, swept over what the relays already announce. @@ -164,6 +165,8 @@ struct Shape { compress: bool, /// Read each group via a fresh FETCH rather than a live subscription. fetch: bool, + /// Expire delivery history between rounds instead of applying byte pressure. + expiry: bool, /// Deliver each round as `burst` datagrams of `frame` bytes per watched broadcast's /// datagram track instead of groups. datagrams: bool, @@ -189,6 +192,7 @@ impl Shape { fetch: false, compress: false, datagrams: false, + expiry: false, }; /// What each viewer subscribes with: a burst is a backlog, which the default @@ -516,7 +520,7 @@ pub(crate) trait Backend { type Relay: Clone; const PREFIX: &'static str; - fn relay(id: u64, capacity: u64) -> (Self::Relay, cache::Pool); + fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool); fn origin(relay: &Self::Relay) -> &origin::Producer; async fn connect(options: MockConnectOptions, relay: &Self::Relay, hidden: bool) -> Pair; } @@ -528,7 +532,7 @@ impl Backend for Model { type Relay = origin::Producer; const PREFIX: &'static str = ""; - fn relay(id: u64, capacity: u64) -> (Self::Relay, cache::Pool) { + fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool) { make_origin(id, capacity) } @@ -565,15 +569,15 @@ impl Drop for Pair { } } -fn make_origin(id: u64, capacity: u64) -> (origin::Producer, cache::Pool) { +fn make_origin(id: u64, capacity: Option) -> (origin::Producer, cache::Pool) { let mut config = origin::Config::new(Hop::new(id).unwrap()); - let pool = cache::Pool::new(if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { - cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY) - } else { - cache::Config::default() - .with_capacity(capacity) - .with_expiry(cache::DEFAULT_EXPIRY) - }); + let pool = cache::Pool::new( + if capacity.is_none() || std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { + cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY) + } else { + cache::Config::default().with_capacity(capacity) + }, + ); config.pool = pool.clone(); let (producer, driver) = origin::Producer::new(config); harness::spawn(driver); @@ -583,6 +587,7 @@ fn make_origin(id: u64, capacity: u64) -> (origin::Producer, cache::Pool) { /// Relays and publishers, before any viewer joins. Holds every handle that /// keeps a session or broadcast alive. struct Cluster { + expiry: bool, version: Version, window: usize, credit: Option, @@ -604,6 +609,7 @@ struct Cluster { impl Cluster { async fn new(version: &str, shape: Shape) -> Self { let mut this = Self { + expiry: shape.expiry, version: version.parse().unwrap(), window: shape.window, credit: shape.credit, @@ -621,7 +627,7 @@ impl Cluster { for _ in 0..shape.relays { this.next_hop += 1; - let (relay, pool) = B::relay(this.next_hop, RELAY_CACHE); + let (relay, pool) = B::relay(this.next_hop, (!shape.expiry).then_some(RELAY_CACHE)); this.relays.push(relay); this.pools.push(pool); } @@ -679,7 +685,7 @@ impl Cluster { fn origin(&mut self, capacity: u64) -> (origin::Producer, cache::Pool) { self.next_hop += 1; - make_origin(self.next_hop, capacity) + make_origin(self.next_hop, (!self.expiry).then_some(capacity)) } /// Connect a viewer session to `relay`. @@ -809,6 +815,7 @@ impl Room { ); assert!(!shape.paced || shape.burst == 1, "a paced round writes one frame"); assert!(!shape.fetch || !shape.paced, "fetch rounds read finished groups"); + let shape = Shape { expiry: true, ..shape }; let mut cluster = Cluster::::new(version, shape).await; let mut viewers = Vec::new(); @@ -879,9 +886,6 @@ impl Room { // Delivery must retain every unread group. Bound history by expiry between // rounds instead of byte pressure, which may shed an in-flight group's head. - for pool in room.pools() { - pool.resize(None); - } let blocked_before: usize = room .viewers .iter() diff --git a/rs/moq-relay/benches/session.rs b/rs/moq-relay/benches/session.rs index 730adc32264..f91d0c76b89 100644 --- a/rs/moq-relay/benches/session.rs +++ b/rs/moq-relay/benches/session.rs @@ -32,10 +32,10 @@ impl Backend for Relay { type Relay = Node; const PREFIX: &'static str = "relay_"; - fn relay(id: u64, capacity: u64) -> (Node, cache::Pool) { + fn relay(id: u64, capacity: Option) -> (Node, cache::Pool) { let mut config = moq_relay::cache::Config::default(); if std::env::var_os("SESSION_EXPIRY_ONLY").is_none() { - config.capacity = Some(capacity.to_string()); + config.capacity = capacity.map(|capacity| capacity.to_string()); } let cache = config.init().unwrap(); let pool = cache.pool.clone(); From fb55ec87e5a4670c43ab635aae2a2f73debdd40c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:41:10 -0800 Subject: [PATCH 4/9] fix(bench): retain byte budgets alongside history expiry Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/benches/memory.rs | 8 +++++-- rs/moq-net/benches/session/mod.rs | 39 +++++++++++++++++-------------- rs/moq-relay/benches/session.rs | 4 ++-- 3 files changed, 30 insertions(+), 21 deletions(-) diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index 4815e0d775a..f96ccf71f98 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -116,8 +116,12 @@ impl Link { let version: Version = VERSION.parse().unwrap(); let mut origin = |hop: u64| { let mut config = origin::Config::new(Hop::new(hop).unwrap()); - // Expire history between rounds without shedding unread frames under byte pressure. - config.pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY)); + // Hold the working batch and handover; expire read history between rounds. + config.pool = cache::Pool::new( + cache::Config::default() + .with_capacity((4 * streams * FRAMES * size) as u64) + .with_expiry(cache::DEFAULT_EXPIRY), + ); pools.push(config.pool.clone()); let (producer, driver) = origin::Producer::new(config); tokio::spawn(moq_net::time::run(driver)); diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index edb997c2dc0..d413bfcb8be 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -34,7 +34,7 @@ //! one datagram per track per round across publisher and viewer axes, then larger payloads //! and cadences, on the lite versions (moq-transport carries no datagram tracks). Throughput //! counts delivered datagrams, each checked for its sequence and payload, so a lost one fails. -//! Delivery history expires between rounds, outside timing, without byte-pressure loss. +//! Delivery history expires between rounds, outside timing, while byte budgets stay enabled. //! `SESSION_EXPIRY_ONLY=1` advances the cache clock by a second instead of the 30s //! expiry window, retaining 30 rounds for profiling. It reports groups written and //! charged cache bytes every 10s during long runs. @@ -124,8 +124,7 @@ const DASH_VERSIONS: [&str; 2] = ["moq-lite-06", "moq-transport-22"]; /// A stats frame: a small JSON snapshot. const DASH_FRAME: usize = 128; -/// Initial budgets for join and churn fixtures. Delivery rooms use expiry -/// between rounds instead, so cache pressure never sheds unread frames. +/// Initial byte budgets. Delivery rooms also expire read history between rounds. const RELAY_CACHE: u64 = 16 * 1024 * 1024; const CLIENT_CACHE: u64 = 1024 * 1024; @@ -165,7 +164,7 @@ struct Shape { compress: bool, /// Read each group via a fresh FETCH rather than a live subscription. fetch: bool, - /// Expire delivery history between rounds instead of applying byte pressure. + /// Expire delivery history between rounds before applying byte pressure. expiry: bool, /// Deliver each round as `burst` datagrams of `frame` bytes per watched broadcast's /// datagram track instead of groups. @@ -520,7 +519,7 @@ pub(crate) trait Backend { type Relay: Clone; const PREFIX: &'static str; - fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool); + fn relay(id: u64, capacity: u64, expiry: bool) -> (Self::Relay, cache::Pool); fn origin(relay: &Self::Relay) -> &origin::Producer; async fn connect(options: MockConnectOptions, relay: &Self::Relay, hidden: bool) -> Pair; } @@ -532,8 +531,8 @@ impl Backend for Model { type Relay = origin::Producer; const PREFIX: &'static str = ""; - fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool) { - make_origin(id, capacity) + fn relay(id: u64, capacity: u64, expiry: bool) -> (Self::Relay, cache::Pool) { + make_origin(id, capacity, expiry) } fn origin(relay: &Self::Relay) -> &origin::Producer { @@ -569,14 +568,13 @@ impl Drop for Pair { } } -fn make_origin(id: u64, capacity: Option) -> (origin::Producer, cache::Pool) { +fn make_origin(id: u64, capacity: u64, expiry: bool) -> (origin::Producer, cache::Pool) { let mut config = origin::Config::new(Hop::new(id).unwrap()); + let profiling = std::env::var_os("SESSION_EXPIRY_ONLY").is_some(); let pool = cache::Pool::new( - if capacity.is_none() || std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { - cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY) - } else { - cache::Config::default().with_capacity(capacity) - }, + cache::Config::default() + .with_capacity((!profiling).then_some(capacity)) + .with_expiry((expiry || profiling).then_some(cache::DEFAULT_EXPIRY)), ); config.pool = pool.clone(); let (producer, driver) = origin::Producer::new(config); @@ -627,7 +625,7 @@ impl Cluster { for _ in 0..shape.relays { this.next_hop += 1; - let (relay, pool) = B::relay(this.next_hop, (!shape.expiry).then_some(RELAY_CACHE)); + let (relay, pool) = B::relay(this.next_hop, RELAY_CACHE, shape.expiry); this.relays.push(relay); this.pools.push(pool); } @@ -685,7 +683,7 @@ impl Cluster { fn origin(&mut self, capacity: u64) -> (origin::Producer, cache::Pool) { self.next_hop += 1; - make_origin(self.next_hop, (!self.expiry).then_some(capacity)) + make_origin(self.next_hop, capacity, self.expiry) } /// Connect a viewer session to `relay`. @@ -884,8 +882,12 @@ impl Room { room.writing.resize_with(room.watched.len(), || None); room.reading.resize_with(shape.viewers * shape.watch, || None); - // Delivery must retain every unread group. Bound history by expiry between - // rounds instead of byte pressure, which may shed an in-flight group's head. + // Size the byte budgets from a full batch, including bookkeeping. Expiry + // between rounds reclaims read history before it competes with unread data. + let capacities: Vec<_> = room.pools().map(cache::Pool::capacity).collect(); + for pool in room.pools() { + pool.resize(None); + } let blocked_before: usize = room .viewers .iter() @@ -899,6 +901,9 @@ impl Room { room.tick(); room.round().await; } + for (pool, capacity) in room.pools().zip(capacities) { + pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 2))); + } // Exercise turnover as well as the empty cache, including Criterion's one-pass // nightly smoke mode. for _ in 0..4 { diff --git a/rs/moq-relay/benches/session.rs b/rs/moq-relay/benches/session.rs index f91d0c76b89..817b78cfc84 100644 --- a/rs/moq-relay/benches/session.rs +++ b/rs/moq-relay/benches/session.rs @@ -32,10 +32,10 @@ impl Backend for Relay { type Relay = Node; const PREFIX: &'static str = "relay_"; - fn relay(id: u64, capacity: Option) -> (Node, cache::Pool) { + fn relay(id: u64, capacity: u64, _expiry: bool) -> (Node, cache::Pool) { let mut config = moq_relay::cache::Config::default(); if std::env::var_os("SESSION_EXPIRY_ONLY").is_none() { - config.capacity = capacity.map(|capacity| capacity.to_string()); + config.capacity = Some(capacity.to_string()); } let cache = config.init().unwrap(); let pool = cache.pool.clone(); From e635b1809d5418210cde37e05b163e475819e371 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:42:30 -0800 Subject: [PATCH 5/9] fix(bench): budget for expiry dating and batch handover Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/benches/session/mod.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index d413bfcb8be..8cdcc85dc45 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -902,7 +902,9 @@ impl Room { room.round().await; } for (pool, capacity) in room.pools().zip(capacities) { - pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 2))); + // Expiry dates activity on the next poll: retain those two batches + // plus the incoming batch until the following cleanup reclaims history. + pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 3))); } // Exercise turnover as well as the empty cache, including Criterion's one-pass // nightly smoke mode. From 4f8ee97c9d63c94ba3978ccc31b7b24541419925 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:48:45 -0800 Subject: [PATCH 6/9] fix(bench): count only delivery allocations in memory rounds Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/benches/memory.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index f96ccf71f98..00ec41c3313 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -220,12 +220,19 @@ impl Link { async fn measure(&mut self, iters: u64) -> Duration { let expected = self.tracks.len() * FRAMES * self.payload.len(); + let counting = std::env::var_os("MEMORY_ALLOCS").is_some(); let mut elapsed = Duration::ZERO; for _ in 0..iters { self.tick(); + if counting { + COUNTING.store(true, Ordering::Relaxed); + } let start = Instant::now(); assert_eq!(self.round().await, expected); elapsed += start.elapsed(); + if counting { + COUNTING.store(false, Ordering::Relaxed); + } } elapsed } @@ -240,9 +247,7 @@ fn allocations() { let mut link = rt.block_on(Link::new(streams, size)); ALLOCATIONS.store(0, Ordering::Relaxed); ALLOCATED.store(0, Ordering::Relaxed); - COUNTING.store(true, Ordering::Relaxed); rt.block_on(link.measure(ROUNDS as u64)); - COUNTING.store(false, Ordering::Relaxed); let frames = (ROUNDS * streams * FRAMES) as f64; let allocations = ALLOCATIONS.load(Ordering::Relaxed) as f64 / frames; From d7456b85a10951f07fbd8e644bfd27d02e6a69f9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:54:08 -0800 Subject: [PATCH 7/9] fix(bench): include protected edges in cache working sets Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/benches/memory.rs | 16 +++++++++------- rs/moq-net/benches/session/mod.rs | 9 +++++---- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index 00ec41c3313..c0a091baab0 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -116,12 +116,7 @@ impl Link { let version: Version = VERSION.parse().unwrap(); let mut origin = |hop: u64| { let mut config = origin::Config::new(Hop::new(hop).unwrap()); - // Hold the working batch and handover; expire read history between rounds. - config.pool = cache::Pool::new( - cache::Config::default() - .with_capacity((4 * streams * FRAMES * size) as u64) - .with_expiry(cache::DEFAULT_EXPIRY), - ); + config.pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY)); pools.push(config.pool.clone()); let (producer, driver) = origin::Producer::new(config); tokio::spawn(moq_net::time::run(driver)); @@ -181,6 +176,13 @@ impl Link { _sessions: sessions, _origins: (publisher, subscriber), }; + // Size from a complete batch, including group and frame bookkeeping. + link.tick(); + link.round().await; + for pool in &link.pools { + // Dating, the expiry boundary, a demoted live edge, and the incoming batch. + pool.resize(pool.used() * 4); + } // Warm every stream so the timed rounds skip first-group setup. for _ in 0..4 { link.tick(); @@ -210,7 +212,7 @@ impl Link { bytes } - // Bound retained history to two rounds plus the protected latest group. + // Expire read history between rounds without advancing protocol time. fn tick(&mut self) { self.cache_now += cache::DEFAULT_EXPIRY; for pool in &self.pools { diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index 8cdcc85dc45..e2047ba704c 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -902,9 +902,10 @@ impl Room { room.round().await; } for (pool, capacity) in room.pools().zip(capacities) { - // Expiry dates activity on the next poll: retain those two batches - // plus the incoming batch until the following cleanup reclaims history. - pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 3))); + // Expiry dates activity after demotion and reclaims strictly past its + // window. Allow the dating batch, the window boundary, the demoted + // live edge, and the incoming batch, including their bookkeeping. + pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 4))); } // Exercise turnover as well as the empty cache, including Criterion's one-pass // nightly smoke mode. @@ -949,7 +950,7 @@ impl Room { } /// Expire old batches outside timing. Activity is dated on the following poll, - /// so history holds two rounds plus each track's protected latest group. + /// and the exact expiry boundary survives until the following poll. fn tick(&mut self) { self.cache_now += if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { Duration::from_secs(1) From b367efeb46531d4d57574866136cc2370f63c67b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 00:01:26 -0800 Subject: [PATCH 8/9] refactor(bench): bound delivery history with expiry alone Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/benches/memory.rs | 7 ------ rs/moq-net/benches/session/mod.rs | 41 +++++++++++-------------------- rs/moq-relay/benches/session.rs | 4 +-- 3 files changed, 17 insertions(+), 35 deletions(-) diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index c0a091baab0..d1c3bab0612 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -176,13 +176,6 @@ impl Link { _sessions: sessions, _origins: (publisher, subscriber), }; - // Size from a complete batch, including group and frame bookkeeping. - link.tick(); - link.round().await; - for pool in &link.pools { - // Dating, the expiry boundary, a demoted live edge, and the incoming batch. - pool.resize(pool.used() * 4); - } // Warm every stream so the timed rounds skip first-group setup. for _ in 0..4 { link.tick(); diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index e2047ba704c..6faca5cbded 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -34,7 +34,8 @@ //! one datagram per track per round across publisher and viewer axes, then larger payloads //! and cadences, on the lite versions (moq-transport carries no datagram tracks). Throughput //! counts delivered datagrams, each checked for its sequence and payload, so a lost one fails. -//! Delivery history expires between rounds, outside timing, while byte budgets stay enabled. +//! Delivery uses expiry-only pools: history expires between rounds outside timing, +//! so byte pressure cannot shed unread frames during a batch. //! `SESSION_EXPIRY_ONLY=1` advances the cache clock by a second instead of the 30s //! expiry window, retaining 30 rounds for profiling. It reports groups written and //! charged cache bytes every 10s during long runs. @@ -124,7 +125,7 @@ const DASH_VERSIONS: [&str; 2] = ["moq-lite-06", "moq-transport-22"]; /// A stats frame: a small JSON snapshot. const DASH_FRAME: usize = 128; -/// Initial byte budgets. Delivery rooms also expire read history between rounds. +/// Initial budgets for join and churn fixtures. Delivery rooms use expiry-only pools. const RELAY_CACHE: u64 = 16 * 1024 * 1024; const CLIENT_CACHE: u64 = 1024 * 1024; @@ -164,7 +165,7 @@ struct Shape { compress: bool, /// Read each group via a fresh FETCH rather than a live subscription. fetch: bool, - /// Expire delivery history between rounds before applying byte pressure. + /// Expire delivery history between rounds instead of applying byte pressure. expiry: bool, /// Deliver each round as `burst` datagrams of `frame` bytes per watched broadcast's /// datagram track instead of groups. @@ -519,7 +520,7 @@ pub(crate) trait Backend { type Relay: Clone; const PREFIX: &'static str; - fn relay(id: u64, capacity: u64, expiry: bool) -> (Self::Relay, cache::Pool); + fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool); fn origin(relay: &Self::Relay) -> &origin::Producer; async fn connect(options: MockConnectOptions, relay: &Self::Relay, hidden: bool) -> Pair; } @@ -531,8 +532,8 @@ impl Backend for Model { type Relay = origin::Producer; const PREFIX: &'static str = ""; - fn relay(id: u64, capacity: u64, expiry: bool) -> (Self::Relay, cache::Pool) { - make_origin(id, capacity, expiry) + fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool) { + make_origin(id, capacity) } fn origin(relay: &Self::Relay) -> &origin::Producer { @@ -568,14 +569,14 @@ impl Drop for Pair { } } -fn make_origin(id: u64, capacity: u64, expiry: bool) -> (origin::Producer, cache::Pool) { +fn make_origin(id: u64, capacity: Option) -> (origin::Producer, cache::Pool) { let mut config = origin::Config::new(Hop::new(id).unwrap()); let profiling = std::env::var_os("SESSION_EXPIRY_ONLY").is_some(); - let pool = cache::Pool::new( - cache::Config::default() - .with_capacity((!profiling).then_some(capacity)) - .with_expiry((expiry || profiling).then_some(cache::DEFAULT_EXPIRY)), - ); + let pool = cache::Pool::new(if capacity.is_none() || profiling { + cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY) + } else { + cache::Config::default().with_capacity(capacity) + }); config.pool = pool.clone(); let (producer, driver) = origin::Producer::new(config); harness::spawn(driver); @@ -625,7 +626,7 @@ impl Cluster { for _ in 0..shape.relays { this.next_hop += 1; - let (relay, pool) = B::relay(this.next_hop, RELAY_CACHE, shape.expiry); + let (relay, pool) = B::relay(this.next_hop, (!shape.expiry).then_some(RELAY_CACHE)); this.relays.push(relay); this.pools.push(pool); } @@ -683,7 +684,7 @@ impl Cluster { fn origin(&mut self, capacity: u64) -> (origin::Producer, cache::Pool) { self.next_hop += 1; - make_origin(self.next_hop, capacity, self.expiry) + make_origin(self.next_hop, (!self.expiry).then_some(capacity)) } /// Connect a viewer session to `relay`. @@ -882,12 +883,6 @@ impl Room { room.writing.resize_with(room.watched.len(), || None); room.reading.resize_with(shape.viewers * shape.watch, || None); - // Size the byte budgets from a full batch, including bookkeeping. Expiry - // between rounds reclaims read history before it competes with unread data. - let capacities: Vec<_> = room.pools().map(cache::Pool::capacity).collect(); - for pool in room.pools() { - pool.resize(None); - } let blocked_before: usize = room .viewers .iter() @@ -901,12 +896,6 @@ impl Room { room.tick(); room.round().await; } - for (pool, capacity) in room.pools().zip(capacities) { - // Expiry dates activity after demotion and reclaims strictly past its - // window. Allow the dating batch, the window boundary, the demoted - // live edge, and the incoming batch, including their bookkeeping. - pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 4))); - } // Exercise turnover as well as the empty cache, including Criterion's one-pass // nightly smoke mode. for _ in 0..4 { diff --git a/rs/moq-relay/benches/session.rs b/rs/moq-relay/benches/session.rs index 817b78cfc84..f91d0c76b89 100644 --- a/rs/moq-relay/benches/session.rs +++ b/rs/moq-relay/benches/session.rs @@ -32,10 +32,10 @@ impl Backend for Relay { type Relay = Node; const PREFIX: &'static str = "relay_"; - fn relay(id: u64, capacity: u64, _expiry: bool) -> (Node, cache::Pool) { + fn relay(id: u64, capacity: Option) -> (Node, cache::Pool) { let mut config = moq_relay::cache::Config::default(); if std::env::var_os("SESSION_EXPIRY_ONLY").is_none() { - config.capacity = Some(capacity.to_string()); + config.capacity = capacity.map(|capacity| capacity.to_string()); } let cache = config.init().unwrap(); let pool = cache.pool.clone(); From 89de2ac78d360ad0cb5c874f8531ceb6a56424eb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 00:10:21 -0800 Subject: [PATCH 9/9] refactor(bench): make delivery cache policy a fixture choice Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/benches/memory.rs | 2 +- rs/moq-net/benches/session/mod.rs | 24 ++++++++++++------------ 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index d1c3bab0612..28cdb674deb 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -205,7 +205,7 @@ impl Link { bytes } - // Expire read history between rounds without advancing protocol time. + // Advance the cache/model clock between rounds to expire read history. fn tick(&mut self) { self.cache_now += cache::DEFAULT_EXPIRY; for pool in &self.pools { diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index 6faca5cbded..e8f7ef93802 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -165,8 +165,6 @@ struct Shape { compress: bool, /// Read each group via a fresh FETCH rather than a live subscription. fetch: bool, - /// Expire delivery history between rounds instead of applying byte pressure. - expiry: bool, /// Deliver each round as `burst` datagrams of `frame` bytes per watched broadcast's /// datagram track instead of groups. datagrams: bool, @@ -192,7 +190,6 @@ impl Shape { fetch: false, compress: false, datagrams: false, - expiry: false, }; /// What each viewer subscribes with: a burst is a backlog, which the default @@ -606,9 +603,9 @@ struct Cluster { } impl Cluster { - async fn new(version: &str, shape: Shape) -> Self { + async fn new(version: &str, shape: Shape, expiry: bool) -> Self { let mut this = Self { - expiry: shape.expiry, + expiry, version: version.parse().unwrap(), window: shape.window, credit: shape.credit, @@ -626,7 +623,7 @@ impl Cluster { for _ in 0..shape.relays { this.next_hop += 1; - let (relay, pool) = B::relay(this.next_hop, (!shape.expiry).then_some(RELAY_CACHE)); + let (relay, pool) = B::relay(this.next_hop, (!expiry).then_some(RELAY_CACHE)); this.relays.push(relay); this.pools.push(pool); } @@ -814,8 +811,7 @@ impl Room { ); assert!(!shape.paced || shape.burst == 1, "a paced round writes one frame"); assert!(!shape.fetch || !shape.paced, "fetch rounds read finished groups"); - let shape = Shape { expiry: true, ..shape }; - let mut cluster = Cluster::::new(version, shape).await; + let mut cluster = Cluster::::new(version, shape, true).await; let mut viewers = Vec::new(); let mut watched = vec![false; shape.total()]; @@ -938,8 +934,9 @@ impl Room { self.cluster.pools.iter().chain(viewers) } - /// Expire old batches outside timing. Activity is dated on the following poll, - /// and the exact expiry boundary survives until the following poll. + /// Advance the cache/model clock outside timing to expire old batches. + /// Each round completes its reads before this step; pending read deadlines + /// share this clock too. fn tick(&mut self) { self.cache_now += if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { Duration::from_secs(1) @@ -1311,6 +1308,7 @@ pub(crate) fn range_updates(c: &mut Criterion) { publishers: 1, ..Shape::BASE }, + false, ) .await; let mut viewers = Vec::new(); @@ -1403,7 +1401,7 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator::new(version, *shape).await; + let cluster = Cluster::::new(version, *shape, false).await; let payload = Bytes::from(vec![0; shape.frame]); for track in &cluster.tracks { for sequence in 0..shape.history { @@ -1499,6 +1497,7 @@ impl DashRoom { viewers: 0, ..Shape::BASE }, + false, ) .await; @@ -1662,6 +1661,7 @@ impl ChurnRoom { viewers: 0, ..Shape::BASE }, + false, ) .await; let published = &cluster._broadcasts[0]; @@ -1876,7 +1876,7 @@ pub(crate) fn withdrawal(c: &mut Criterion) { publishers: 0, ..Shape::BASE }; - let cluster = Cluster::::new("moq-lite-06", shape).await; + let cluster = Cluster::::new("moq-lite-06", shape, false).await; let mut announced = B::origin(cluster.relays.last().unwrap()).consume().announced(); let live: Vec<_> = (0..broadcasts) .map(|i| {