diff --git a/quest/m1/release-perf/README.md b/quest/m1/release-perf/README.md
index 7a9ad5acda3..8651e966305 100644
--- a/quest/m1/release-perf/README.md
+++ b/quest/m1/release-perf/README.md
@@ -1,4 +1,4 @@
-# Main is no slower than release
+# [M] Main is no slower than release
## Goal
@@ -68,6 +68,33 @@ noise.
optimal; a representation that hides that reclamation would widen scope for
no measured benefit. Sparse arrivals cost O(log gaps) amortized, and at 4096
gaps flush, read, and expiry instructions fell 80 to 95%.
+- The mux exports' and Annex B import's fixed cost per broadcast is accepted.
+ It publishes required output: a per-track timeline and an immediately
+ completed catalog record. Their cold DEFLATE state is most of it (81% of a
+ one-frame import's instructions are its zero-fill), and the encoders already
+ build that state lazily on first use. It is paid once per broadcast: the
+ import ratio falls from 3.97x at one frame to 1.49x at 64. TS export's
+ growth to 2.09x at `4x64` is in the export reap that
+ [moq-dev/moq#6281](https://github.com/moq-dev/moq/pull/6281) changes.
+- The model's remaining per-group and per-frame cost is accepted, after
+ [moq-dev/moq#6294](https://github.com/moq-dev/moq/pull/6294) cut the
+ accidental part (a second track upgrade per first frame, a 128-bit division
+ per buffered datagram on join, and draining empty waiter lists on every
+ write). What remains pays for features: the arrival stamp each frame reads
+ from the clock ([#6249](https://github.com/moq-dev/moq/pull/6249)), largest
+ notifications ([#5725](https://github.com/moq-dev/moq/pull/5725)), the
+ rewind check and starts index each first frame fills
+ ([#5803](https://github.com/moq-dev/moq/pull/5803),
+ [#6074](https://github.com/moq-dev/moq/pull/6074)), indexed expiry wakes,
+ the bounded wall-clock expiry scan per append, and a subscriber's ranges and
+ datagram max delay ([#5195](https://github.com/moq-dev/moq/pull/5195)).
+ None of it grows with the sweeps. `track_stamp_groups/sparse` stays highest
+ at 4.4x, because release's first frame skipped nearly all of that work.
+- Session delivery is accepted at 1.28x to 1.77x in instructions. It is the
+ per-group and per-frame cost above on every hop, measured through the
+ flow-controlled mock transport
+ ([#5530](https://github.com/moq-dev/moq/pull/5530)) that release's harness
+ lacked.
### Release as a baseline
@@ -89,44 +116,52 @@ usually pays for a feature.
| `track_gc/idle/16384` | 11.6 ns | 73.7 ns | 6.3x | accepted (above) |
| `origin/resume_overlap/sequence/1r_64g_32c` | 93 us | 426 us | 4.6x | fixed in [moq-dev/moq#6292](https://github.com/moq-dev/moq/pull/6292) |
| `session_delivery_group_cost/moq-transport-22/.../viewers=64/.../fetch` | 1.19 ms | 3.86 ms | 3.2x | accepted (above) |
-| `group_write_frames/single/8192` | 424 us | 486 us | 1.17x | [Re-check](/quest/m1/release-perf/recheck.md) |
+| `group_write_frames/single/8192` | 424 us | 486 us | 1.17x | accepted (above) |
Main is much faster on subscriber churn, joins, range demand, and idle
sessions (by 100x to 3000x). The relay load test used 21% less CPU on video
and 12% less on fanout, with the same delivery.
-### Screen only (one run, unconfirmed)
-
-- moq-mux: `container-export/ts/4x1` 4.7x and `mkv/4x1` 3.9x (median 1.41x
- over 27 cases); `annexb-import-{bytes,frames}` median 2.1x (up to 3.6x).
-- moq-net track: `track_stamp_groups` median 2.7x (4.7x at `sparse/4096`),
- `track_sparse_read/timed/4096` 4.3x, `track_datagram_subscriber_join`
- median 2.3x, `track_parallel_write` 1.6x, `track_fanout_frame` 1.42x,
- `track_empty_run` median 1.36x.
-- moq-net origin and session: `origin/announce_duplicate/30d_1s` 4.3x,
- `session_delivery_relays` median 1.56x, `paced` 1.45x, `scale` 1.43x.
-- moq-stats `aggregate` at live=16, departed=4096: 2.1x. The other rows were
- 1.0 to 1.3x, so this one is probably noise.
-- kio `contended_notify` and `contended_log` at 16 threads: 4 to 5x. These
- are contention benches on a busy host, so they are likely noise.
-- JS `net/bench/frames.ts` on bun: lite rows 1.2 to 1.3x, IETF rows 1.3 to
- 2.2x. Fixed in Chromium:
- [#6296](https://github.com/moq-dev/moq/pull/6296) removed a retained-history
- scan, and [#6479](https://github.com/moq-dev/moq/pull/6479) stopped
- releasing a finished stream's reader lock, which built a `TypeError` per
- group stream. Every row then sits within the A/A spread of release.
- The group write/read microbench stays 3 to 12% above release against an
- A/A of 3 to 8% on the same host. Decided 2026-10-23 (maintainer): browser
- parity accepted on this evidence; the final re-run above checks it again.
+### Screen only, re-measured
+
+Re-measured on 2026-10-22 with callgrind instructions per operation, release
+rebuilt under main's dev shell with fat LTO. Ratios are main over release;
+the moq-net and kio rows include the fixes above.
+
+| Bench | Screen | Instructions | Owner |
+|---|---|---|---|
+| `container-export/ts/4x1`, `mkv/4x1` | 4.7x, 3.9x | 1.70x, 1.59x | accepted (above) |
+| `annexb-import`, 1 and 64 frames | median 2.1x | 3.97x, 1.49x | accepted (above) |
+| `track_stamp_groups/append/64`, `sparse/64` | median 2.7x | 2.21x, 4.43x | accepted (above) |
+| `track_parallel_write/1` | 1.6x | 1.99x | accepted (above) |
+| `track_fanout_frame/1` | 1.42x | 1.86x | accepted (above) |
+| `track_empty_run/{fresh/1,stale/8}/512` | median 1.36x | 1.53x | accepted (above) |
+| `track_datagram_subscriber_join/0d/1`, `64d/64` | median 2.3x | 2.12x, 3.55x | accepted (above) |
+| `track_sparse_read/timed/4096` | 4.3x | | [moq-dev/moq#6287](https://github.com/moq-dev/moq/pull/6287) |
+| `origin/announce_duplicate`, `1d_1s` to `240d_240s` | 4.3x | 1.06x to 1.16x | noise |
+| `session_delivery_{relays,paced,scale}` | 1.43x to 1.56x | 1.28x to 1.77x | accepted (above) |
+| moq-stats `aggregate`, live=16, departed=4096 | 2.1x | 1.05x | noise |
+| kio `contended_{notify,log}`, 16 threads | 4x to 5x | wall 1.04x to 1.15x | noise |
+| JS `net/bench/frames.ts` on bun | 1.2x to 2.2x | | accepted (below) |
+
+The kio contention benches spin across threads, so instructions don't measure
+them; the row is the least loaded of three alternating wall-time rounds, and
+later rounds swung both ways.
+
+JS `net/bench/frames.ts` on bun had lite rows at 1.2 to 1.3x and IETF rows at
+1.3 to 2.2x. Fixed in Chromium:
+[#6296](https://github.com/moq-dev/moq/pull/6296) removed a retained-history
+scan, and [#6479](https://github.com/moq-dev/moq/pull/6479) stopped
+releasing a finished stream's reader lock, which built a `TypeError` per
+group stream. Every row then sits within the A/A spread of release.
+The group write/read microbench stays 3 to 12% above release against an
+A/A of 3 to 8% on the same host. Decided 2026-10-23 (maintainer): browser
+parity accepted on this evidence; the final re-run above checks it again.
The raw report lived in a scratch directory that has since been removed.
`just bench ` now pairs full Criterion IDs; point `MOQ_BENCH_ARTIFACTS`
outside the worktree to keep the next one.
-## Required
-
-- [Re-check](/quest/m1/release-perf/recheck.md) - the screen-only moves are re-measured with instruction counts, and the ones that hold are fixed or accepted
-
## Related
- [Performance audit](/quest/m2/audit-perf.md) - compares nightly to nightly; this epic compares main to release
diff --git a/quest/m1/release-perf/recheck.md b/quest/m1/release-perf/recheck.md
deleted file mode 100644
index 96366e428ba..00000000000
--- a/quest/m1/release-perf/recheck.md
+++ /dev/null
@@ -1,68 +0,0 @@
-# [M] Re-check
-
-## Goal
-
-Every screen-only move in the [epic](/quest/m1/release-perf/README.md) is
-re-measured with instruction counts. Each one that holds is fixed in this
-quest's PR, one commit per fix, or accepted with a reason recorded in the
-epic.
-
-## Plan
-
-Use the instruction-count method from the [performance audit](/quest/m2/audit-perf.md).
-Wall time on a shared host can't tell these apart. Build each bench at
-`origin/release` and at main, and compare instructions per iteration. A
-change that needs a maintainer decision, or is too large to review alongside
-the rest, becomes a follow-up through `/quest-plan`.
-
-Leads from reading code on 2026-10-21, strongest first:
-
-1. moq-mux `container-export/{ts,mkv}/4x1` (4.7x, 3.9x). The worst ratio is at
- one frame per track, so look for a fixed cost per export or per track. The
- reap cost from [#5942](https://github.com/moq-dev/moq/pull/5942) that the
- audit found grows with buffered groups instead. Suspects: every export now
- goes through `jitter::Buffer`
- ([#4645](https://github.com/moq-dev/moq/pull/4645),
- [#5309](https://github.com/moq-dev/moq/pull/5309)), and `container/source.rs`
- start-up ([#5947](https://github.com/moq-dev/moq/pull/5947),
- [#5854](https://github.com/moq-dev/moq/pull/5854)). If the jitter path
- dominates, skip it when the delay is zero.
-2. `annexb-import` (median 2.1x). Check `group_*` and `append_group` first. If
- they regressed by the same factor, the cost is per group in moq-net
- ([#5803](https://github.com/moq-dev/moq/pull/5803) rewind check,
- [#5725](https://github.com/moq-dev/moq/pull/5725) largest notifications,
- [#6020](https://github.com/moq-dev/moq/pull/6020)), not in the importer.
-3. `track_stamp_groups` (median 2.7x). The first frame on a timed track now
- takes the track lock and `bounds`, `starts`, and `sequence` mutexes
- ([#5803](https://github.com/moq-dev/moq/pull/5803),
- [#6074](https://github.com/moq-dev/moq/pull/6074)). Pass the checked start
- into `first_frame`, merge `bounds` and `starts` under one lock, and skip
- `advance_largest`'s write and wake when nothing changed.
-4. `track_datagram_subscriber_join` (median 2.3x). `Cursor::new` scans the
- datagram ring twice ([#5195](https://github.com/moq-dev/moq/pull/5195)).
- Track the newest datagram timestamp on push. The window's construction cost
- from [#5929](https://github.com/moq-dev/moq/pull/5929) was cut in
- [#6112](https://github.com/moq-dev/moq/pull/6112); an inline single-range
- window ([#6241](https://github.com/moq-dev/moq/pull/6241)) regressed join
- time, so the rest stays.
-5. `track_parallel_write`, `track_fanout_frame`, `track_empty_run`,
- `group_write_frames` (1.2 to 1.6x). This is constant per-group and
- per-frame cost from several features. Resolve `now` once per poll in
- `Drift`, and skip no-op `advance_largest` writes. Accept the rest.
-6. `origin/announce_duplicate/30d_1s` (4.3x). Compare against `1d_1s`, to
- tell a constant cost per change from a cost per duplicate. Leads: an
- epochless winner switch is now a `Restart`
- ([#5087](https://github.com/moq-dev/moq/pull/5087),
- [#5443](https://github.com/moq-dev/moq/pull/5443),
- [#4942](https://github.com/moq-dev/moq/pull/4942),
- [#5144](https://github.com/moq-dev/moq/pull/5144)).
-7. `session_delivery_{relays,paced,scale}` (1.4 to 1.6x). Expected to be the
- per-group cost above plus the flow-controlled mock transport
- ([#5530](https://github.com/moq-dev/moq/pull/5530)). Re-measure after the
- fixes above, and accept what remains.
-8. moq-stats `aggregate` at live=16, departed=4096, and kio
- `contended_{notify,log}` at 16 threads: confirm they are noise. If the
- aggregate row holds, suspect [#4874](https://github.com/moq-dev/moq/pull/4874)
- and [#4904](https://github.com/moq-dev/moq/pull/4904).
-
-Add a bench sweep wherever a fix removes a slope, so the slope stays visible.
diff --git a/rs/kio/benches/channel.rs b/rs/kio/benches/channel.rs
index 562710ab557..98996a7ad00 100644
--- a/rs/kio/benches/channel.rs
+++ b/rs/kio/benches/channel.rs
@@ -127,5 +127,30 @@ fn bench_notify(c: &mut Criterion) {
group.finish();
}
-criterion_group!(benches, bench_create_parallel, bench_notify);
+/// Measure writes nobody waits on after a burst of readers parked and left, swept over
+/// the burst size. A write must not keep paying for waiters that are already gone.
+fn bench_write_abandoned(c: &mut Criterion) {
+ let mut group = c.benchmark_group("channel_write_abandoned");
+ for abandoned in [0, 128, 4_096] {
+ group.bench_with_input(BenchmarkId::from_parameter(abandoned), &abandoned, |b, &abandoned| {
+ let producer = Producer::new(0u64);
+ let consumer = producer.consume();
+ let waiters: Vec<_> = (0..abandoned).map(|_| Waiter::new(Waker::noop().clone())).collect();
+ for waiter in &waiters {
+ assert!(consumer.poll(waiter, |_| Poll::<()>::Pending).is_pending());
+ }
+ drop(waiters);
+
+ b.iter(|| {
+ let Ok(mut state) = producer.write() else {
+ panic!("benchmark channel is open");
+ };
+ *state += 1;
+ });
+ });
+ }
+ group.finish();
+}
+
+criterion_group!(benches, bench_create_parallel, bench_notify, bench_write_abandoned);
criterion_main!(benches);
diff --git a/rs/kio/src/producer.rs b/rs/kio/src/producer.rs
index 2e8d9ed2f50..155192a2caf 100644
--- a/rs/kio/src/producer.rs
+++ b/rs/kio/src/producer.rs
@@ -411,6 +411,13 @@ impl Drop for Mut<'_, T> {
return;
}
+ if !state.closed && state.waiters_value.is_empty() {
+ // A write nobody is parked on wakes nobody, so skip moving the list out. Drop
+ // any dead entries in place, or every later write would rescan them.
+ state.waiters_value.clear();
+ return;
+ }
+
// Drain wakers while holding lock, then wake after releasing.
// A modification that also closed the channel (e.g. `close()`) must
// wake the closed and consumer-count waiters too, since they resolve
diff --git a/rs/kio/src/waiter.rs b/rs/kio/src/waiter.rs
index 28e0ddcc0e7..0dc9c3a8b45 100644
--- a/rs/kio/src/waiter.rs
+++ b/rs/kio/src/waiter.rs
@@ -308,6 +308,12 @@ impl WaiterList {
self.entries.iter().all(|entry| entry.strong_count() == 0)
}
+ /// Drop every entry without waking it.
+ pub(crate) fn clear(&mut self) {
+ self.drained();
+ self.entries.clear();
+ }
+
/// Wake all live waiters, draining the list.
pub fn wake(&mut self) {
self.drained();
diff --git a/rs/moq-mux/benches/containers.rs b/rs/moq-mux/benches/containers.rs
index 3943a8fb48d..fa067578ca8 100644
--- a/rs/moq-mux/benches/containers.rs
+++ b/rs/moq-mux/benches/containers.rs
@@ -183,6 +183,13 @@ impl Annexb {
}
}
+// Collect only timed import work, excluding the output validation in setup.
+#[inline(never)]
+fn profiled_import(mut import: Annexb, payload: &Bytes, frames: usize) -> Annexb {
+ import.decode(payload, frames);
+ import
+}
+
/// `fragments` moofs whose `tracks` trafs each split their samples into `runs` one-sample runs,
/// interleaved in the mdat run by run, after the init of an exported `tracks`-track broadcast.
fn interleaved(init: &[u8], tracks: usize, runs: usize, fragments: usize, payload: &Bytes) -> Bytes {
@@ -360,9 +367,8 @@ fn containers(c: &mut Criterion) {
} else {
b.iter_batched(
Annexb::new,
- |mut import| {
- import.decode(std::hint::black_box(&payload), frames);
- import
+ |import| {
+ profiled_import(import, std::hint::black_box(&payload), frames)
},
BatchSize::PerIteration,
);
diff --git a/rs/moq-mux/benches/export.rs b/rs/moq-mux/benches/export.rs
index da97d3bb42d..6815de0c639 100644
--- a/rs/moq-mux/benches/export.rs
+++ b/rs/moq-mux/benches/export.rs
@@ -75,30 +75,56 @@ async fn prepare(tracks: usize, frames: usize) -> (moq_mux::Source, moq_mux::cat
(source, stream, fixture)
}
-async fn export(format: &str, source: moq_mux::Source, stream: moq_mux::catalog::Consumer) -> usize {
+/// Opened during setup so the timed drain has the same boundary across API changes.
+enum Export {
+ Flv(Box>),
+ Mkv(Box>),
+ Ts(Box>),
+}
+
+fn open(format: &str, source: moq_mux::Source, stream: moq_mux::catalog::Consumer) -> Export {
+ match format {
+ "flv" => Export::Flv(Box::new(
+ moq_mux::container::flv::Export::new(source, stream).with_multitrack(true),
+ )),
+ "ts" => Export::Ts(Box::new(moq_mux::container::ts::Export::new(source, stream))),
+ _ => Export::Mkv(Box::new(
+ moq_mux::container::mkv::Export::new(source, stream).with_fragment_duration(Duration::ZERO),
+ )),
+ }
+}
+
+async fn export(export: Export) -> usize {
let mut chunks = 0;
- if format == "flv" {
- let mut export = moq_mux::container::flv::Export::new(source, stream).with_multitrack(true);
- while let Some(chunk) = export.next().await.unwrap() {
- std::hint::black_box(chunk);
- chunks += 1;
+ match export {
+ Export::Flv(mut export) => {
+ while let Some(chunk) = export.next().await.unwrap() {
+ std::hint::black_box(chunk);
+ chunks += 1;
+ }
}
- } else if format == "ts" {
- let mut export = moq_mux::container::ts::Export::new(source, stream);
- while let Some(chunk) = export.next().await.unwrap() {
- std::hint::black_box(chunk);
- chunks += 1;
+ Export::Ts(mut export) => {
+ while let Some(chunk) = export.next().await.unwrap() {
+ std::hint::black_box(chunk);
+ chunks += 1;
+ }
}
- } else {
- let mut export = moq_mux::container::mkv::Export::new(source, stream).with_fragment_duration(Duration::ZERO);
- while let Some(chunk) = export.next().await.unwrap() {
- std::hint::black_box(chunk);
- chunks += 1;
+ Export::Mkv(mut export) => {
+ while let Some(chunk) = export.next().await.unwrap() {
+ std::hint::black_box(chunk);
+ chunks += 1;
+ }
}
}
chunks
}
+// A stable profiling boundary excludes publishing and opening from instruction counts.
+#[inline(never)]
+fn profiled_export(runtime: &tokio::runtime::Runtime, opened: Export) -> usize {
+ runtime.block_on(export(opened))
+}
+
fn bench_export(c: &mut Criterion) {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
@@ -112,11 +138,14 @@ fn bench_export(c: &mut Criterion) {
for frames in [1, 16, 64] {
group.throughput(Throughput::Elements((tracks * frames) as u64));
group.bench_function(BenchmarkId::new(format, format!("{tracks}x{frames}")), |b| {
- // Publishing the broadcast is setup; only exporting it is measured.
+ // Publishing and opening are setup; only draining the export is measured.
b.iter_batched(
- || runtime.block_on(prepare(tracks, frames)),
- |(source, stream, fixture)| {
- let chunks = runtime.block_on(export(format, source, stream));
+ || {
+ let (source, stream, fixture) = runtime.block_on(prepare(tracks, frames));
+ (open(format, source, stream), fixture)
+ },
+ |(opened, fixture)| {
+ let chunks = profiled_export(&runtime, opened);
if format == "ts" {
assert!(chunks > 0);
} else {
diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs
index 1d59be2aef0..e0fc9c3f251 100644
--- a/rs/moq-net/src/model/group.rs
+++ b/rs/moq-net/src/model/group.rs
@@ -1216,8 +1216,8 @@ impl Producer {
return Err(Error::FrameTooLarge);
}
- let track = self.first_frame_track();
- let track = track.as_ref().map(|track| track.read());
+ let producer = self.first_frame_track();
+ let track = producer.as_ref().map(|track| track.read());
let neighbors = track
.as_ref()
.and_then(|track| track.neighbors(self.handle.info.sequence));
@@ -1265,7 +1265,7 @@ impl Producer {
track.first_frame(self.handle.info.sequence, bound);
}
drop(track);
- self.handle.start();
+ self.handle.start(producer);
// With the group lock released (lock order is track then group), settle
// eviction debt if enough has been written since the track last paid.
@@ -1299,8 +1299,8 @@ impl Producer {
let count = frames.len();
let bytes: u64 = frames.filled().iter().map(|frame| frame.payload.len() as u64).sum();
- let track = self.first_frame_track();
- let track = track.as_ref().map(|track| track.read());
+ let producer = self.first_frame_track();
+ let track = producer.as_ref().map(|track| track.read());
let neighbors = track
.as_ref()
.and_then(|track| track.neighbors(self.handle.info.sequence));
@@ -1371,7 +1371,7 @@ impl Producer {
}
drop(track);
if count > 0 {
- self.handle.start();
+ self.handle.start(producer);
}
self.handle.cache.settle(now);
@@ -1451,8 +1451,8 @@ impl Producer {
let scratch = self.handle.track.compress && header.inflated.is_none();
let inflated = header.inflated.unwrap_or(frame.size);
- let track = self.first_frame_track();
- let track = track.as_ref().map(|track| track.read());
+ let producer = self.first_frame_track();
+ let track = producer.as_ref().map(|track| track.read());
let neighbors = track
.as_ref()
.and_then(|track| track.neighbors(self.handle.info.sequence));
@@ -1510,7 +1510,7 @@ impl Producer {
track.first_frame(self.handle.info.sequence, bound);
}
drop(track);
- self.handle.start();
+ self.handle.start(producer);
// With the group lock released (lock order is track then group), settle
// eviction debt if enough has been written since the track last paid.
@@ -1626,7 +1626,7 @@ impl Producer {
}
drop(state);
// An empty group counts as content once it ends.
- self.handle.start();
+ self.handle.start(None);
// A finished group can no longer raise the track's largest, which may end its stream.
self.handle.cache.finish_largest(self.sequence);
Ok(())
@@ -1814,7 +1814,7 @@ impl Handle {
}
let now = state.charge.add(payload + metadata + growth);
drop(state);
- self.start();
+ self.start(None);
self.cache.settle(now);
}
@@ -1980,11 +1980,14 @@ impl Handle {
/// The first frame opened or the group finished: tell the track once, so it re-judges
/// the reads that skipped this group while it was frameless. Called with the group
/// lock released, since the track's is taken first.
- fn start(&self) {
+ ///
+ /// `track` is the track a first frame already reached, so stamping a group upgrades
+ /// its link once.
+ fn start(&self, track: Option>) {
if self.alive.started.swap(true, Ordering::Relaxed) {
return;
}
- let Some(track) = self.cache.track() else {
+ let Some(track) = track.or_else(|| self.cache.track()) else {
return;
};
let answers = track.read().reveal(self);
diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs
index 33cf960b66e..40d9b6271ac 100644
--- a/rs/moq-net/src/model/track.rs
+++ b/rs/moq-net/src/model/track.rs
@@ -5673,15 +5673,16 @@ impl Cursor {
// subsequent delivery stays in arrival order, including late datagrams.
let newest = state.datagrams.iter().filter_map(|datagram| datagram.timestamp).max();
let start = newest.map_or(0, |newest| {
+ // Every datagram carries the track's scale: convert the cutoff into it once,
+ // rounded up, rather than each timestamp into a `Duration` (a 128-bit division).
let cutoff = Duration::from(newest).saturating_sub(preferences.max_delay);
+ let scale = newest.scale().as_u64() as u128;
+ let cutoff = (cutoff.as_nanos() * scale).div_ceil(1_000_000_000) as u64;
+ let cutoff = Timestamp::new(cutoff, newest.scale()).expect("at most the newest");
state
.datagrams
.iter()
- .position(|datagram| {
- datagram
- .timestamp
- .is_none_or(|timestamp| Duration::from(timestamp) >= cutoff)
- })
+ .position(|datagram| datagram.timestamp.is_none_or(|timestamp| timestamp >= cutoff))
.expect("the newest datagram is within max delay")
});
let filter = Filter {
@@ -8619,6 +8620,24 @@ mod test {
assert_eq!(recv_datagram(&mut subscriber).sequence, 6);
}
+ #[test]
+ #[cfg(feature = "async")]
+ fn datagram_replay_cutoff_between_track_units() {
+ // A 90 kHz unit doesn't divide the nanosecond cutoff, which lands inside unit 89_995.
+ let info = Info::default().with_timescale(Timescale::new(90_000).unwrap());
+ let mut producer = track_producer("video", info);
+ for units in [89_995, 89_996, 90_000] {
+ producer
+ .append_datagram(Timestamp::from_scale(units, 90_000).unwrap(), b"x".as_slice())
+ .unwrap();
+ }
+ let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_micros(50)));
+ for sequence in [1, 2] {
+ assert_eq!(recv_datagram(&mut subscriber).sequence, sequence);
+ }
+ assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending());
+ }
+
#[test]
#[cfg(feature = "async")]
fn datagram_replay_starts_after_evicted_entries_with_zero_delay() {