From 8e684e251a075396b385a8b31187e8f7c90c6118 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 10:41:23 -0700 Subject: [PATCH 1/8] fix(net): wake parked reads when stamped successors abort (#4710) Co-authored-by: GPT-6 (cherry picked from commit 7e6fd81d959c5c7d8496cb4a5d1f164cd736a47c) Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/benches/track.rs | 43 +++++++++++++++++- rs/moq-net/src/model/resume.rs | 81 ++++++++++++++++++++++++++++++++++ rs/moq-net/src/model/track.rs | 50 +++++++++++++++++++-- 3 files changed, 170 insertions(+), 4 deletions(-) diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 14f7661d3b..637157a266 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -320,6 +320,46 @@ fn bench_subscriber_join(c: &mut Criterion) { group.finish(); } +/// Parked reads re-judge their reach, including the successor's abort waiter. +/// Sweep readers and cached groups so added per-reader scans show up as a slope. +fn bench_parked_read(c: &mut Criterion) { + let mut group = c.benchmark_group("track_parked_read"); + for cached in [8, 64, 512] { + for readers in [1, 8, 64] { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", None).unwrap(); + let mut head = track.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); + for _ in 1..cached { + let mut next = track.append_group().unwrap(); + next.write_frame(Timestamp::ZERO, Bytes::from_static(b"next")).unwrap(); + next.finish().unwrap(); + } + let waiters: Vec<_> = (0..readers).map(|_| kio::Waiter::noop()).collect(); + let mut held: Vec<_> = waiters + .iter() + .map(|waiter| { + let mut sub = track.subscribe(track::Subscription::default().with_max_age(Duration::from_secs(1))); + let Poll::Ready(Ok(Some(mut head))) = sub.poll_recv_group(waiter) else { + panic!("head is cached"); + }; + assert!(matches!(head.poll_read_frame(waiter), Poll::Ready(Ok(Some(_))))); + head + }) + .collect(); + group.throughput(Throughput::Elements(readers as u64)); + group.bench_function(BenchmarkId::new(format!("cached_{cached}"), readers), |b| { + b.iter(|| { + for (head, waiter) in held.iter_mut().zip(&waiters) { + assert!(black_box(head.poll_read_frame(waiter)).is_pending()); + } + }); + }); + } + } + group.finish(); +} + criterion_group!( benches, bench_fanout, @@ -327,6 +367,7 @@ criterion_group!( bench_aborted_scan, bench_subscriber_churn, bench_subscriber_churn_after_peak, - bench_subscriber_join + bench_subscriber_join, + bench_parked_read ); criterion_main!(benches); diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index b3445d889b..b5b352ca9a 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -3721,6 +3721,87 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + #[tokio::test] + async fn aborted_stamped_successor_wakes_a_parked_read() { + tokio::time::pause(); + let (a, a_read) = track_pair("a"); + let (mut b, b_read) = track_pair("b"); + let mut head = a.create_group(0u64.into()).unwrap(); + head.write_frame(Timestamp::ZERO, b"a0".to_vec()).unwrap(); + let mut successor = b.create_group(1u64.into()).unwrap(); + successor + .write_frame(Duration::from_secs(30).try_into().unwrap(), b"b1".to_vec()) + .unwrap(); + write_group_at(&mut b, 2, "b2", Duration::from_secs(1)); + write_group_at(&mut b, 3, "edge", Duration::from_secs(20)); + let mut producer = Producer::new(); + producer.switch(a_read, None).unwrap(); + producer.switch(b_read, Position::group(1)).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); + assert_eq!(read(&mut reading), b"a0"); + + let (counter, waker) = CountWaker::new(); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + let before = counter.count(); + successor.abort(Error::Cancel).unwrap(); + assert!( + counter.count() > before, + "the stamped successor's abort lost its wakeup" + ); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + + /// Cache eviction removes an unstamped successor from its track, so the next + /// segment's first group becomes the bound for the read already parked on A. + #[tokio::test] + async fn evicted_unstamped_successor_re_resolves_across_segments() { + tokio::time::pause(); + let (a, a_read) = track_pair("a"); + let pool = crate::cache::Pool::new(crate::cache::Config::default().with_expiry(Duration::from_secs(1))); + let info = broadcast::Info { + pool: pool.clone(), + ..Default::default() + }; + let b = track::Producer::new(Arc::new(info), "b", None); + let (mut c, c_read) = track_pair("c"); + let mut head = a.create_group(0u64.into()).unwrap(); + head.write_frame(Timestamp::ZERO, b"a0".to_vec()).unwrap(); + let successor = b.create_group(1u64.into()).unwrap(); + // The live edge is protected from eviction and lies beyond B's segment. + let _protected = b.create_group(4u64.into()).unwrap(); + write_group_at(&mut c, 2, "c2", Duration::from_secs(1)); + write_group_at(&mut c, 3, "edge", Duration::from_secs(20)); + let mut producer = Producer::new(); + producer.switch(a_read, None).unwrap(); + producer.switch(b.consume(), Position::group(1)).unwrap(); + producer.switch(c_read, Position::group(2)).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the unstamped successor leaves reach unbounded"); + assert_eq!(read(&mut reading), b"a0"); + + let (counter, waker) = CountWaker::new(); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + let before = counter.count(); + crate::model::clock::advance(Duration::from_secs(2)); + pool.gc(crate::model::clock::now()); + assert!(successor.is_aborted(), "cache GC evicted the successor"); + assert!( + b.consume().peek_group(1).is_none(), + "the slot was removed from the cache" + ); + assert!(counter.count() > before, "the successor's eviction lost its wakeup"); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + #[tokio::test] async fn pruned_segment_boundary_is_judged_against_later_segments() { let (mut a, a_read) = track_pair("a"); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 4b258985b9..e6040327d2 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2636,10 +2636,16 @@ impl Consumer { ConsumerKind::Plain(state) => { let mut start = None; let _ = state.poll(waiter, |state| { - start = state.first_servable(from, cap).map(|group| { + while let Some(group) = state.first_servable(from, cap) { + // A stamped group can abort without changing its track. Register + // before checking it again, so an abort either wakes or re-resolves. + let _ = group.poll_closed(waiter); let _ = group.poll_timestamp(waiter); - group.timestamp() - }); + if !group.is_aborted() { + start = Some(group.timestamp()); + break; + } + } Poll::<()>::Pending }); start @@ -3407,6 +3413,11 @@ impl group::Expiry for GroupExpiry { let _ = self.state.poll(waiter, |state| { let budget = clamp_max_age(max_age, state.max_age_bound()); loop { + // An abort can change reach even after the successor is stamped. + // Register before judging, so a racing abort is observed or wakes us. + if let Some(group) = state.first_servable(self.sequence.saturating_add(1), cap) { + let _ = group.poll_closed(waiter); + } let edge = state.drift_edge(cap, outer, successor); expired = state.is_stale(self.sequence, &edge, budget); if expired { @@ -6039,6 +6050,39 @@ mod test { assert!(matches!(result, Ok(None)), "the held group ends: {result:?}"); } + /// A rewound successor can keep a drained group within budget until its abort + /// moves the reach to the next cached group, without changing the track itself. + #[tokio::test] + async fn aborted_stamped_successor_wakes_a_parked_read() { + tokio::time::pause(); + let mut producer = track_producer("test", None); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap(); + let mut successor = producer.append_group().unwrap(); + successor + .write_frame(Timestamp::from_millis(30_000).unwrap(), b"next".as_slice()) + .unwrap(); + append_at(&mut producer, 1000); + append_at(&mut producer, 20_000); + let mut sub = producer.subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); + assert!(reading.read_frame().await.unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + successor.abort(Error::Cancel).unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the stamped successor's abort lost its wakeup" + ); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// A first timestamp on a *newer* group can convict a held one, so the held reader has /// to be woken by it. The conviction needs a group beyond the held one's successor: /// a group is bounded by where its successor begins, so the successor itself never From 16847352a5604f82d70e632558646f5178f16bde Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 21:12:31 -0700 Subject: [PATCH 2/8] fix(net): a parked group's expiry no longer walks the backlog (#4892) (cherry picked from commit d70edf644d069ab5d4257dcde62bf46a687798d9) Adapted for release: its drift_edge still takes the splice anchors (outer, successor), so the re-selected successor group is bound as `next` to avoid shadowing the anchor successor timestamp. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 138 ++++++++++++++++++++++++++++++---- 1 file changed, 123 insertions(+), 15 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index e6040327d2..9b5be711d4 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3415,9 +3415,15 @@ impl group::Expiry for GroupExpiry { loop { // An abort can change reach even after the successor is stamped. // Register before judging, so a racing abort is observed or wakes us. - if let Some(group) = state.first_servable(self.sequence.saturating_add(1), cap) { - let _ = group.poll_closed(waiter); - } + // An abort only closes the group, never touching the track, so one that + // landed before registering must re-select or the replacement goes unwatched. + let next = loop { + let next = state.first_servable(self.sequence.saturating_add(1), cap); + match next { + Some(group) if group.poll_closed(waiter).is_ready() && group.is_aborted() => continue, + next => break next, + } + }; let edge = state.drift_edge(cap, outer, successor); expired = state.is_stale(self.sequence, &edge, budget); if expired { @@ -3427,8 +3433,11 @@ impl group::Expiry for GroupExpiry { // A first timestamp can change the verdict without mutating the track: // on a group past the edge (a new edge), or on the candidate's // unstamped immediate successor (a reach where there was none). - // Register on the candidate and every unstamped servable group above - // it. If one raced this scan, resolve the edge again before Pending. + // Register on the candidate, its successor, and every servable group + // past the edge. If one raced this scan, resolve the edge again before + // Pending. Groups between the successor and the edge can move neither, + // and walking them would cost each of N parked serves the whole + // backlog, O(N^2) per track change. let mut timestamp_raced = false; if let Some(slot) = state.lookup.get(&self.sequence) { let group = &slot.group; @@ -3439,17 +3448,18 @@ impl group::Expiry for GroupExpiry { timestamp_raced = true; } } - for (_, slot) in state + let past = edge + .presentation + .map_or(self.sequence, |live| live.sequence.max(self.sequence)); + let beyond = state .lookup - .range((std::ops::Bound::Excluded(self.sequence), std::ops::Bound::Unbounded)) - { - let group = &slot.group; - if !super::subscription::before_end(group.sequence, cap) { - break; - } - if slot.visible - && !group.is_aborted() - && group.timestamp().is_none() + .range((std::ops::Bound::Excluded(past), std::ops::Bound::Unbounded)) + .map(|(_, slot)| slot) + .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) + .filter(|slot| slot.visible && !slot.group.is_aborted()) + .map(|slot| &slot.group); + for group in next.into_iter().chain(beyond) { + if group.timestamp().is_none() && group.poll_timestamp(waiter).is_ready() && group.timestamp().is_some() { @@ -6120,6 +6130,104 @@ mod test { assert!(matches!(result, Ok(None)), "the held group ends: {result:?}"); } + /// A parked read registers only where a first timestamp can change its verdict: its + /// immediate successor and the groups past the edge. Walking every group between them + /// costs each parked publisher stream the whole backlog, enough for a 2s audio backlog + /// at 400 groups/s to pin a publisher's runtime and starve its connection. + #[test] + fn a_parked_read_ignores_first_frames_between_its_successor_and_the_edge() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10))); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + append_at(&mut producer, 1000); // the successor bounds the head's reach + let mut between = producer.append_group().unwrap(); + append_at(&mut producer, 3000); // the edge + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0); + assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(held.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + + between + .write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert!( + !woken.load(Ordering::SeqCst), + "a first frame below the edge moves neither the reach nor the edge" + ); + + // Control: a first frame past the edge is still observed. + let mut beyond = producer.append_group().unwrap(); + assert!(next.as_mut().poll(&mut cx).is_pending()); + woken.store(false, Ordering::SeqCst); + beyond + .write_frame(Timestamp::from_millis(4000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert!(woken.load(Ordering::SeqCst), "a new edge wakes the parked read"); + } + + /// An aborted successor hands the reach to the next group, which sits below the edge + /// and so is watched only because the read re-selects its successor. + #[test] + fn a_parked_read_watches_the_replacement_for_an_aborted_successor() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500))); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + let successor = producer.append_group().unwrap(); + let mut replacement = producer.append_group().unwrap(); + append_at(&mut producer, 20_000); // the edge + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0, "an unstamped successor leaves the reach unbounded"); + assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(held.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + + successor.abort(Error::Cancel).unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the successor's abort wakes the parked read" + ); + woken.store(false, Ordering::SeqCst); + assert!( + next.as_mut().poll(&mut cx).is_pending(), + "the unstamped replacement leaves the reach unbounded" + ); + + replacement + .write_frame(Timestamp::from_millis(1000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the replacement's first frame bounds the reach" + ); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// The ordinary live case, at the default real-time budget: 2s GOPs produced one at /// a time and read as they arrive. The budget must take the live edge without /// shortening the group the reader is already on, so every frame of every group is From 7a378e2e435e05b8195c9f583780115ac8aa54ea Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 12:53:30 -0700 Subject: [PATCH 3/8] fix(net): a capped IETF stream never claims END_OF_GROUP (#5077) Backport to release. The Rust regression test decodes the header with release's GroupHeader::decode form and runs under tokio::test, since release lacks coding::decode_buf and moq_net_sim. Quest edits dropped. (cherry picked from commit 9680ca9e412d9ecefef7aec4ce8a7102bfca7bf9) Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.test.ts | 18 ++++++++++++--- js/net/src/ietf/publisher.ts | 4 +++- rs/moq-net/src/ietf/publisher.rs | 37 +++++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 4 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index b7a4a0889e..9f94be6bdc 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -1492,6 +1492,8 @@ interface ServedGroup { sequence: number; /** Whether the header claimed the stream starts at the group's first object. */ firstObject: boolean; + /** Whether the header claimed the stream's FIN ends the group (END_OF_GROUP). */ + endOfGroup: boolean; /** Each object's absolute id (reconstructed from its delta) and payload. */ objects: { id: number; payload: string }[]; } @@ -1623,7 +1625,12 @@ async function readGroup(stream: ReadableStream): Promise): Promise /** * An absolute filter names the objects it wants, so the boundary groups are trimmed to it * and the groups outside it are never opened. The first object written carries its absolute - * id, or the subscriber would read a silently renumbered group. + * id, or the subscriber would read a silently renumbered group, and a capped tail does not + * claim END_OF_GROUP, or the subscriber would think the group ended at the cap. */ test("draft-20: an absolute filter trims the range it serves", async () => { const fx = fixture(); @@ -1715,6 +1723,7 @@ test("draft-20: an absolute filter trims the range it serves", async () => { sequence: 1, // The head was trimmed, so the stream does not start at the group's first object. firstObject: false, + endOfGroup: true, objects: [ { id: 1, payload: "1.1" }, { id: 2, payload: "1.2" }, @@ -1726,6 +1735,8 @@ test("draft-20: an absolute filter trims the range it serves", async () => { expect(await readGroup(second)).toEqual({ sequence: 2, firstObject: true, + // The filter ends at object 0 of 3, so the stream stops before the group does. + endOfGroup: false, objects: [{ id: 0, payload: "2.0" }], }); @@ -1787,6 +1798,7 @@ test("draft-20: a fill serves the current group's head on a fetch stream", async expect(await readGroup(live)).toEqual({ sequence: 0, firstObject: false, + endOfGroup: true, objects: [{ id: 2, payload: "0.2" }], }); } finally { @@ -1863,7 +1875,7 @@ test("draft-20: a backwards range within one group serves nothing and ends the s try { const served = await nextUni(fx.uni); if (!served) throw new Error("the group stream never opened"); - expect(await readGroup(served)).toEqual({ sequence: 0, firstObject: false, objects: [] }); + expect(await readGroup(served)).toEqual({ sequence: 0, firstObject: false, endOfGroup: false, objects: [] }); } finally { fx.close(); client.close(); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index ae3ff38850..5d39644a22 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -591,7 +591,9 @@ export class Publisher { hasExtensions: stamped, hasSubgroup: false, hasSubgroupObject: false, - hasEnd: true, + // A stream capped by the range stops before the group may end, so its FIN + // cannot claim END_OF_GROUP. + hasEnd: slice.until === undefined, hasPriority: true, // Only honest when the stream really starts at the group's first object; // a trimmed head starts partway through. diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index d42749c8ea..cd218539e1 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -2315,6 +2315,9 @@ impl TrackServe { flags: ietf::GroupFlags { has_extensions: self.timescale.is_some(), first_object: slice.skip == 0, + // A stream capped by the range stops before the group may end, so its + // FIN cannot claim END_OF_GROUP. + has_end: slice.until.is_none(), ..Default::default() }, }; @@ -4394,6 +4397,40 @@ mod serve_tests { "the cap excludes cc" ); } + + /// A stream cut short by the range's end Location does not claim END_OF_GROUP, since + /// the group goes on past it. A range ending on a whole group still does. A cap at or past + /// the group's real end clears the bit too, since the header is written before we know. + #[tokio::test] + async fn capped_group_does_not_claim_its_end() { + async fn serve(object: Option) -> ietf::GroupHeader { + let log = crate::lite::test_transport::Log::default(); + let session = SinkSession::new(log.clone()); + + let track = track::Producer::new(std::sync::Arc::new(crate::broadcast::Info::default()), "test", None); + let subscriber = track.subscribe(None); + + let mut group = track.append_group().unwrap(); + for payload in [b"aa", b"bb", b"cc"] { + group.write_frame(crate::Timestamp::ZERO, payload.as_slice()).unwrap(); + } + group.finish().unwrap(); + track.finish().unwrap(); + + let range = ServeRange { + start: Some(Location { group: 0, object: 0 }), + end: Some(EndLocation { group: 0, object }), + }; + let mut serve = TrackServe::new(session, subscriber, RequestId(0), Version::Draft20, range, None); + kio::wait(|waiter| serve.poll(waiter)).await.unwrap(); + + let mut buf = bytes::Bytes::from(log.writes.lock().unwrap().clone()); + ietf::GroupHeader::decode(&mut buf, Version::Draft20).expect("a group header") + } + + assert!(!serve(Some(0)).await.flags.has_end, "capped at object 0 of 3"); + assert!(serve(None).await.flags.has_end, "the whole group"); + } } #[cfg(test)] From d2aa3865f688fc9e5ed3d55495fddbbd08b22e83 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 02:14:42 -0700 Subject: [PATCH 4/8] fix(hls): retain init decoded from completed segments (#4840) Co-authored-by: GPT-6 (cherry picked from commit cccf5001bfc9de5fb95dd981a5196be3e87ecadf) Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 132 +++++++++++++++++++++++++++++ rs/moq-hls/src/export/rendition.rs | 16 +++- 2 files changed, 147 insertions(+), 1 deletion(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 058eca36c6..7623b81267 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -945,6 +945,76 @@ mod tests { (catalog, registration, media) } + /// A completed segment already decoded its inline codec metadata. Finishing the + /// catalog can index a trailing group whose media producer is still live. + #[tokio::test(start_paused = true)] + async fn a_transmuxed_segment_keeps_its_init_with_an_unfinished_tail() { + let origin = produce_origin(); + let mut broadcast = origin.create_broadcast("live").expect("publish allowed"); + let (mut catalog, mut registration, _media) = publish_vp8(&mut broadcast); + registration + .set(hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8)) + .unwrap(); + broadcast.announce(moq_net::origin::Route::default()).unwrap(); + settle().await; + let source = moq_mux::Source::new(origin.consume(), "live"); + let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap(); + broadcaster.ready().await; + let rendition = broadcaster.rendition(Kind::Video, "video0").expect("rendition"); + let mut cursor = rendition.segments(); + let segment = cursor.next().await.unwrap().expect("completed segment"); + assert!(!segment.media.is_empty()); + + catalog.finish().unwrap(); + while !rendition.snapshot().finished { + tokio::task::yield_now().await; + } + let init = tokio::time::timeout(Duration::from_secs(1), rendition.init()) + .await + .expect("init must not wait for the unfinished trailing group") + .unwrap() + .expect("transmuxed segment has an init"); + assert_eq!(&init[4..8], b"ftyp"); + } + + /// An init bootstrap already blocked on the unfinished tail must not hold back a later + /// request once a completed segment has cached the init. + #[tokio::test(start_paused = true)] + async fn a_blocked_init_bootstrap_does_not_hide_a_cached_init() { + let origin = produce_origin(); + let mut broadcast = origin.create_broadcast("live").expect("publish allowed"); + let (mut catalog, mut registration, _media) = publish_vp8(&mut broadcast); + registration + .set(hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8)) + .unwrap(); + broadcast.announce(moq_net::origin::Route::default()).unwrap(); + settle().await; + let source = moq_mux::Source::new(origin.consume(), "live"); + let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap(); + broadcaster.ready().await; + let rendition = broadcaster.rendition(Kind::Video, "video0").expect("rendition"); + + catalog.finish().unwrap(); + while !rendition.snapshot().finished { + tokio::task::yield_now().await; + } + let mut blocked = std::pin::pin!(rendition.init()); + assert!( + tokio::time::timeout(Duration::from_millis(10), blocked.as_mut()) + .await + .is_err(), + "the bootstrap reads the unfinished trailing group" + ); + + let segment = rendition.segments().next().await.unwrap().expect("completed segment"); + assert!(!segment.media.is_empty()); + tokio::time::timeout(Duration::from_secs(1), rendition.init()) + .await + .expect("init must not wait behind the blocked bootstrap") + .unwrap() + .expect("transmuxed segment has an init"); + } + // A reconfigure that changes the init bytes changes the init URL, and the old URL stops // resolving, so a cache can never hand a player the previous init under the new one. #[tokio::test] @@ -2068,6 +2138,68 @@ mod tests { drop((new_media, _new_track, new_server, catalog, live)); } + // A restart can land while a segment request waits for its media broadcast. The init + // decoded from the old run's groups must not be cached for the new run. + #[tokio::test] + async fn a_segment_straddling_a_restart_keeps_its_init_out_of_the_new_run() { + tokio::time::pause(); + // A VP8 keyframe whose 320x240 geometry is only known inline. + const KEYFRAME: &[u8] = &[0x10, 0x00, 0x00, 0x9d, 0x01, 0x2a, 0x40, 0x01, 0xf0, 0x00]; + + let origin = produce_origin(); + let mut live = origin.create_broadcast("live").expect("publish allowed"); + live.announce(Default::default()).expect("publish allowed"); + let catalog = moq_mux::catalog::Producer::new(&mut live, moq_mux::catalog::Config::default()).unwrap(); + let recorder = catalog.enroll_test("video0").unwrap(); + + let mut media = moq_net::broadcast::Info::new().produce(); + let _track = write_routed_media(&mut media, KEYFRAME, recorder, 0); + let server = origin.dynamic("media", sibling_route(10)).unwrap(); + settle().await; + + let source = moq_mux::Source::new(origin.consume(), "live"); + let upstream = Upstream { + broadcast: source.broadcast().await.unwrap(), + source, + }; + let mut config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.broadcast = Some(moq_net::path::Relative::new("media").to_owned()); + + let mut snapshot = moq_mux::catalog::hang::Catalog::default(); + snapshot.video.renditions.insert("video0".to_string(), config); + let archive = hang::catalog::Archive::new(hang::timeline::DEFAULT_NAME); + snapshot.archive = Some(archive.clone()); + let renditions = renditions::Producer::new(Config::default().window); + renditions.sync(&upstream, &snapshot); + let fanout = renditions.fanout(); + let watcher = tokio::spawn(watch_timeline(upstream.broadcast.clone(), archive, fanout.clone())); + let rendition = renditions.get(Kind::Video, "video0").unwrap(); + tokio::time::timeout(Duration::from_secs(5), rendition.playable()) + .await + .expect("the timeline reaches the rendition"); + + // The sibling request is unanswered, so the segment waits for its media broadcast. + let mut segment = std::pin::pin!(rendition.segment(0)); + assert!( + tokio::time::timeout(Duration::from_millis(10), segment.as_mut()) + .await + .is_err() + ); + fanout.set_generation(Some("run-1".into())); + fanout.set_generation(Some("run-2".into())); + + accept_sibling(&server, &media).await; + segment + .await + .unwrap() + .expect("the old run's rows were captured before the restart"); + assert!( + rendition.init().await.unwrap().is_none(), + "the new run has no media of its own yet" + ); + watcher.abort(); + } + // A replacement already present when the export starts is the sibling the request // resolves to: it is served, not treated as a stale epoch. #[tokio::test] diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index af468b35cc..729d593556 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -695,6 +695,11 @@ impl Rendition { async fn load_init(&self) -> Result>> { let binding = self.media.sync(&self.live); + // A transmuxed segment can cache the init while another bootstrap holds `building`, + // blocked on an unfinished group. + if let Some(init) = self.run().init { + return Ok(Some(init)); + } let _building = self.building.lock().await; let run = self.run(); if let Some(init) = run.init { @@ -738,6 +743,9 @@ impl Rendition { /// cache. pub async fn segment(&self, segment: u64) -> Result> { let binding = self.media.sync(&self.live); + // Read the run before the rows (see `restart`), so an init decoded from them is never + // cached for a run that started while they were being fetched. + let run = self.run(); let Some(ranges) = self.live.segment_ranges(segment) else { return Ok(None); }; @@ -796,7 +804,13 @@ impl Rendition { if frames.is_empty() { return Ok(None); } - Ok(Some(muxer.fragment(segment as u32, &frames)?)) + let fragment = muxer.fragment(segment as u32, &frames)?; + // Inline codec metadata was already decoded from these completed groups. + // Reuse it instead of fetching a newer, possibly unfinished timeline tail. + if let Some(bytes) = muxer.init()? { + self.cache_init(run.epoch, bytes); + } + Ok(Some(fragment)) } } From 1aec3e6fe510d5935b14573f2e35248c1a8a4487 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 17:17:13 -0700 Subject: [PATCH 5/8] test(net): pin a cap at or past a group's end clearing END_OF_GROUP Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index cd218539e1..a3bf47e7b0 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -4429,6 +4429,8 @@ mod serve_tests { } assert!(!serve(Some(0)).await.flags.has_end, "capped at object 0 of 3"); + assert!(!serve(Some(2)).await.flags.has_end, "capped at the last object"); + assert!(!serve(Some(3)).await.flags.has_end, "capped past the last object"); assert!(serve(None).await.flags.has_end, "the whole group"); } } From 1f67f5c0c3b0af0eb33da344cac4df783500cee0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 17:20:24 -0700 Subject: [PATCH 6/8] fix(net): a parked read watches its local edge abort The narrowed expiry scan skips groups between the successor and the local edge. An aborted edge hands the edge to one of those groups, and neither the abort nor that group's first frame touches the track, so the parked read slept through a verdict change the old backlog walk delivered. Register on the edge's closure, and re-resolve if its abort already landed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 57 +++++++++++++++++++++++++++++++++-- 1 file changed, 55 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 9b5be711d4..75b56fc902 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3425,6 +3425,15 @@ impl group::Expiry for GroupExpiry { } }; let edge = state.drift_edge(cap, outer, successor); + // Likewise the edge: its abort hands the edge to a lower group, which the + // scan below skips. + if let Some(live) = edge.presentation + && let Some(slot) = state.lookup.get(&live.sequence) + && slot.group.poll_closed(waiter).is_ready() + && slot.group.is_aborted() + { + continue; + } expired = state.is_stale(self.sequence, &edge, budget); if expired { break; @@ -3435,8 +3444,8 @@ impl group::Expiry for GroupExpiry { // unstamped immediate successor (a reach where there was none). // Register on the candidate, its successor, and every servable group // past the edge. If one raced this scan, resolve the edge again before - // Pending. Groups between the successor and the edge can move neither, - // and walking them would cost each of N parked serves the whole + // Pending. Groups between the successor and the edge can move neither + // while the edge stands (its abort is watched above), and walking them would cost each of N parked serves the whole // backlog, O(N^2) per track change. let mut timestamp_raced = false; if let Some(slot) = state.lookup.get(&self.sequence) { @@ -6228,6 +6237,50 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + /// An aborted edge hands the edge back to a lower group, which sits below the old edge + /// and so is watched only because the read observes the edge's abort. Neither the + /// abort nor that group's first frame touches the track. + #[test] + fn a_parked_read_watches_its_edge_abort() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(5))); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + append_at(&mut producer, 1000); // the successor bounds the head's reach + let mut between = producer.append_group().unwrap(); + let mut edge = producer.append_group().unwrap(); + edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + edge.finish().unwrap(); + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0); + assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(held.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + + edge.abort(Error::Cancel).unwrap(); + between + .write_frame(Timestamp::from_millis(10_000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the edge's abort, then a new edge below it, lost the wakeup" + ); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// The ordinary live case, at the default real-time budget: 2s GOPs produced one at /// a time and read as they arrive. The budget must take the live edge without /// shortening the group the reader is already on, so every frame of every group is From 5187fedeb747daba495c10941c8356493f5d024c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 18:02:26 -0700 Subject: [PATCH 7/8] chore(net): reflow the parked-read scan comment Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 75b56fc902..597459d529 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3445,8 +3445,9 @@ impl group::Expiry for GroupExpiry { // Register on the candidate, its successor, and every servable group // past the edge. If one raced this scan, resolve the edge again before // Pending. Groups between the successor and the edge can move neither - // while the edge stands (its abort is watched above), and walking them would cost each of N parked serves the whole - // backlog, O(N^2) per track change. + // while the edge stands (its abort is watched above), and walking them + // would cost each of N parked serves the whole backlog, O(N^2) per + // track change. let mut timestamp_raced = false; if let Some(slot) = state.lookup.get(&self.sequence) { let group = &slot.group; From 560d9622a0b6a33405fa321db858d4f92f587a2d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 20:00:48 -0700 Subject: [PATCH 8/8] fix(deps): bump moq-noq to 1.3.5 on release Carries moq-dev/noq#33, the release/1.3 backport of moq-dev/noq#32: reset the PTO backoff when Initial or Handshake keys are discarded (RFC 9002 A.11). Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 16 ++++++++-------- Cargo.toml | 6 +++--- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c6ff75f065..f3d3c5c930 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4536,9 +4536,9 @@ dependencies = [ [[package]] name = "moq-noq" -version = "1.3.4" +version = "1.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90fc18fcbed942196f9124d7987848cbd5dd4a7407b1e353d1b6a2bfa57d0ddf" +checksum = "d95f06ae483784bc30afc67a14d886d2eae88e8e0bbc03819b01ae270eb298cc" dependencies = [ "bytes", "cfg_aliases", @@ -4558,9 +4558,9 @@ dependencies = [ [[package]] name = "moq-noq-proto" -version = "1.3.4" +version = "1.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd9ff00e72d7158264ff794f7da0955859bda424bb0154a5fc7d60aa084a5897" +checksum = "df1804ba442ea8613d96b770302e48c54570e425d36b298f1b4f733ff1c57740" dependencies = [ "aes-gcm", "aws-lc-rs", @@ -4589,9 +4589,9 @@ dependencies = [ [[package]] name = "moq-noq-udp" -version = "1.3.4" +version = "1.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0094cf49d3c3a79e9f732edb263477ba70a750baf321f53502cace5ad2cdc120" +checksum = "7e82e79f6ebef4f07d578eb2c054d084a1c14b7716a7172af3d364cae6a3c31d" dependencies = [ "cfg_aliases", "libc", @@ -9848,9 +9848,9 @@ dependencies = [ [[package]] name = "web-transport-moq" -version = "1.3.4" +version = "1.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8051685a029b2b5257b2f0ca51eb385495b437437786c7442f00c36178de90f" +checksum = "b896208c8b083650b384c7b152ec05831a32df5029c85d88dc6fa11fd5f3627b" dependencies = [ "bytes", "futures", diff --git a/Cargo.toml b/Cargo.toml index 884d703af2..c73760ec6c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -151,8 +151,8 @@ moq-mux = { version = "0.10.11", path = "rs/moq-mux" } moq-net = { version = "0.3.10", path = "rs/moq-net" } # The MoQ fork of noq (moq-dev/noq). iroh keeps upstream noq, so a build with the # iroh feature carries both stacks. -moq-noq-proto = { version = "1.3.4", default-features = false } -moq-noq-udp = "1.3.4" +moq-noq-proto = { version = "1.3.5", default-features = false } +moq-noq-udp = "1.3.5" # NVENC bindings, forked from ViliamVadocz/nvidia-video-codec-sdk to dlopen the # driver at runtime. Compiles on any platform (macOS included) but only actually # used by moq-video on Linux. @@ -236,7 +236,7 @@ web-transport-iroh = "0.7" # default-features off so the QUIC crypto provider is chosen by moq-tokio's # aws-lc-rs / ring features. # The WebTransport adapter released with moq-noq, from the same repository. -web-transport-moq = { version = "1.3.4", default-features = false } +web-transport-moq = { version = "1.3.5", default-features = false } web-transport-proto = "0.6" web-transport-trait = "0.4" web-transport-wasm = "0.6"