diff --git a/js/net/src/cache.test.ts b/js/net/src/cache.test.ts index df22fc88c58..deb89674746 100644 --- a/js/net/src/cache.test.ts +++ b/js/net/src/cache.test.ts @@ -581,6 +581,20 @@ test("a kept local skip expires once the edge passes it or the track closes", as }); }); +// A final drop ends a group still arriving, but leaves one that finished here to every reader, as a +// group arriving after its drop replaces it. Mirrors Rust `a_final_drop_leaves_a_finished_group`. +test("a final drop leaves a finished group", async () => { + const producer = pooled(new cache.Pool({ capacity: 1 << 30 })); + hooks.receiveTrack(producer); + producer.createGroup(0n).finish(); + const open = producer.createGroup(1n); + hooks.dropTrackGroups(producer, 0n, 2n, { error: new StreamError(StreamCode.Cancel) }); + + expect(dropsHeld(producer)).toEqual([1n]); + expect(open.closed.peek()).toBeInstanceOf(Error); + expect(await cached(producer)).toContain(0n); +}); + // A received stale run the pool forgets is accounted for, as an evicted group is: no reader waits // on it from the live feed, so a reader that wants it asks a fill. Mirrors Rust // `a_reader_asks_a_fill_for_a_forgotten_run_it_judged`. diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 364fd2d5bcc..814634bcedc 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -1511,7 +1511,10 @@ test("lite draft-07: a final drop replaces an expired stream", async () => { expect("start" in (await decodeSubscribeResponse(sub.client.reader, Version.DRAFT_07))).toBe(true); const reset = sub.stall(); - sub.serve(1n); + // Still arriving, as a relay's copy from upstream: the upstream's final drop ends it. One + // held whole would stay, since a group stream wins over its drop. + const arriving = sub.track.createGroup(1n); + arriving.writeFrame(textFrame("hello", Timestamp.fromMillis(1))); expect(await sub.servedHeader()).toBe(1n); await flush(); await clock.advance(Duration.millis(budget * 10)); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index cbf920d49f0..350898f37e0 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -1482,33 +1482,29 @@ export class Producer { // Record `[start, end)` given up as `hole` says, returning the runs recorded. A skip is final // and the most authoritative verdict, so a drop never replaces one. An upstream drop (`open`) // also ends the groups still arriving in it with its error, so their readers settle now - // rather than once the stream's reset lands; one that finished here stays whole. A skip names - // nothing produced. + // rather than once the stream's reset lands. A skip names nothing produced. #drop(start: bigint, end: bigint, hole: Hole, open = false): Run[] { if (start >= end) return []; const skipped = hole.skip ? [] : this.#state.drops.overlapping(start, end).filter((run) => run.skip); let runs = runGaps(skipped, start, end); - // A received stale verdict leaves a group that arrived and was not aborted: a lite-07 - // publisher may serve a group after its stale drop, and the group's stream may land first. - // The group wins, still arriving or not. - if (open && hole.stale !== undefined) { + const arriving: GroupProducer[] = []; + if (open && !hole.skip) { for (const [sequence, entry] of this.#cached) { - if (sequence >= start && sequence < end && !(entry.group.closed.peek() instanceof Error)) { - runs = removeRun(runs, sequence, sequence + 1n); - } + const closed = entry.group.closed.peek(); + if (sequence < start || sequence >= end || closed instanceof Error) continue; + // A lite-07 publisher may serve a group after its stale drop, and the group's stream + // may land first. A final drop ends a group still arriving, but one that finished here + // stays whole, as a re-sent copy the subscriber refused leaves it: naming it given up + // would hide it from every reader. + if (hole.stale === undefined && closed === undefined) arriving.push(entry.group); + else runs = removeRun(runs, sequence, sequence + 1n); } } // A fill no longer owes what the track gave up for good. A stale verdict settles only // the readers it covers, so a fill asked by any other is still owed. if (hole.stale === undefined) this.#fills.dropped(start, end); for (const run of runs) this.#setDrops(run.start, run.end, hole); - if (open && !hole.skip && hole.stale === undefined) { - for (const [sequence, entry] of this.#cached) { - if (sequence >= start && sequence < end && entry.group.closed.peek() === undefined) { - entry.group.abort(hole.error); - } - } - } + for (const group of arriving) group.abort(hole.error); // A pending tail owes nothing the publisher gave up. A stale verdict is not receipt: new // terms may supersede it, leaving those groups unknown again, as in Rust. if (hole.stale === undefined) this.#arrive(start, end); diff --git a/quest/m0/nightly/lite06-join-hang.md b/quest/m0/nightly/lite06-join-hang.md index b6fce85fc9b..020c2859426 100644 --- a/quest/m0/nightly/lite06-join-hang.md +++ b/quest/m0/nightly/lite06-join-hang.md @@ -1,36 +1,19 @@ -# [M] A lite-06 viewer joining a relay reads its first group again +# [XS] A scheduled Nightly's relay session job finishes inside its hour ## Goal -`relay_session_join_broadcasts` and `relay_session_join_stale` finish their -moq-lite-06 cells, as they did before -[moq-dev/moq#8476](https://github.com/moq-dev/moq/pull/8476), so `Nightly (Relay session benchmarks)` (`just gh nightly rust-relay`) finishes -inside its hour with every cell. +inside its hour with every cell, including the moq-lite-06 cells of +`relay_session_join_broadcasts` and `relay_session_join_stale`. ## Plan -Found 2026-11-12 while timing the relay session group for -[moq-dev/moq#8555](https://github.com/moq-dev/moq/pull/8555). The first -moq-lite-06 cell of `relay_session_join_broadcasts` -(`relays=1/publishers=1/broadcasts=1/viewers=16`) and of -`relay_session_join_stale` never leaves its warm-up: the bench thread parks in -the cell's `join` loop (`rs/moq-net/benches/session/mod.rs`) at 0% CPU, -waiting on a join, watch, or the first group read that never completes. The -moq-transport and moq-lite-07 cells of the same groups finish in seconds, and -`relay_session_join_relays` finishes on every version. +The lite-06 join hang that kept those cells in their warm-up is fixed: a final +drop no longer hides a group the relay holds whole. Locally, every +`relay_session_join_*` cell finishes, the first lite-06 one in about 0.2 ms per +join. Confirm the first scheduled Nightly after the fix lands; never dispatch +Nightly. -`git bisect` over `c916855d16..75f00f92c8` (the 11-11 Nightly's commit, whose -lite-06 join cells finished in 0.2 s, to main on 2026-11-12) names -f3edb0b8d9 ([moq-dev/moq#8476](https://github.com/moq-dev/moq/pull/8476), "a -failed fetch below the end no longer holds it open") as the first hanging -commit. Reproduce with -`cargo bench -p moq-relay --features test-support --bench session -- --quick 'relay_session_join_broadcasts/moq-lite-06/'`. +## Waiting -Find which await hangs, reproduce it as a moq-net integration test of a -lite-06 viewer joining a relay that holds history, and fix the cause. A -viewer that never receives a group it subscribed to is a product bug, not -only a bench one, so do not skip the cells. - -Then confirm a scheduled Nightly's relay session job finishes inside its -hour; never dispatch Nightly. +- moq-dev/moq: the first scheduled Nightly on `main` after the fix lands finishes its `Relay session benchmarks` job inside the hour; check with `gh run list --workflow nightly.yml --event schedule`, then delete this quest diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ddd992be98b..1e972644253 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -4447,8 +4447,10 @@ impl Producer { } /// Record `groups` given up as `hole` describes, returning the runs recorded: a drop - /// leaves a skipped sequence skipped, and a stale verdict leaves a group that arrived and - /// was not aborted, since a group stream wins over its drop. + /// leaves a skipped sequence skipped, and a group a copy still serves whole, since a group + /// stream wins over its drop: a stale drop leaves it still arriving or not, and a final one + /// ends the live copy still arriving but leaves a group finished here or served by a + /// fetched head. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] pub(crate) fn give_up(&self, groups: ops::Range, hole: Hole) -> RangeSet { let error = match &hole { @@ -4467,38 +4469,38 @@ impl Producer { } let (runs, open): (_, Vec) = match self.modify() { Ok(mut state) => { - // Only the live feed's copy: a drop judges the subscription, so a fetched head - // or fetch-only backfill may still complete. - let open: Vec<_> = state - .lookup - .range(groups.clone()) - .filter(|(_, slot)| slot.visible && !slot.group.is_aborted()) - .map(|(&sequence, slot)| (sequence, slot.group.clone())) - .collect(); - let recorded = match hole.stale_budget() { + let stale = hole.stale_budget().is_some(); + let mut left = RangeSet::new(); + left.insert(groups.clone()); + let mut open = Vec::new(); + for (&sequence, slot) in state.lookup.range(groups) { + if !slot.visible { + continue; + } + // A final drop ends only the live feed's copy still arriving: a drop judges + // the subscription, so a fetched head or fetch-only backfill may still complete. + let arriving = !slot.group.is_aborted() && !slot.group.is_settled(); + if !stale && arriving { + open.push(slot.group.clone()); + } // A lite-07 publisher may serve a group after its stale drop, and the group's - // stream may land first: the group wins, still arriving or not. - Some(_) => { - let mut left = RangeSet::new(); - left.insert(groups); - for (sequence, _) in &open { - if let Some(end) = sequence.checked_add(1) { - left.remove(*sequence..end); - } - } - let mut runs = RangeSet::new(); - for run in left.iter() { - runs.extend(state.drop_groups(run.clone(), hole.clone()).iter().cloned()); - } - (runs, Vec::new()) + // stream may land first. A group a copy still serves whole stays, whichever + // copy serves it, as a re-sent copy the subscriber refused leaves it: naming + // it given up would hide it from every reader. + let ended = !stale && arriving && std::ptr::eq(slot.serving(), &slot.group); + if !ended + && slot.serves_end() + && let Some(end) = sequence.checked_add(1) + { + left.remove(sequence..end); } - None => ( - state.drop_groups(groups, hole), - open.into_iter().map(|(_, group)| group).collect(), - ), - }; + } + let mut runs = RangeSet::new(); + for run in left.iter() { + runs.extend(state.drop_groups(run.clone(), hole.clone()).iter().cloned()); + } state.charge_debt(); - recorded + (runs, open) } Err(_) => return RangeSet::new(), }; @@ -10927,11 +10929,8 @@ mod test { indexed(); producer.drop_groups(15..25, StreamError::Cancel); indexed(); - producer - .create_group(group::Info { sequence: 17 }) - .unwrap() - .finish() - .unwrap(); + // Still arriving, so the next drop ends it and covers it again. + let _recreated = producer.create_group(group::Info { sequence: 17 }).unwrap(); indexed(); producer.drop_groups(14..26, StreamError::Old); indexed(); @@ -10941,6 +10940,65 @@ mod test { ); } + /// A final drop ends a group still arriving, but leaves one that finished here to every + /// reader, as a group arriving after its drop replaces it. + #[test] + fn a_final_drop_leaves_a_finished_group() { + let producer = track_producer("test", Info::default()); + producer + .create_group(group::Info { sequence: 0 }) + .unwrap() + .finish() + .unwrap(); + let open = producer.create_group(group::Info { sequence: 1 }).unwrap(); + producer.drop_groups(0..2, StreamError::Cancel); + + let drops: Vec<_> = producer.state.read().drops.iter().map(|(run, _)| run.clone()).collect(); + assert_eq!(drops, vec![1..2]); + assert!(open.is_aborted(), "the group still arriving ends"); + let mut consumer = producer.subscribe(None); + assert_eq!(consumer.assert_group().sequence, 0); + } + + /// A drop leaves a group a fetched head serves whole, whatever its live copy is doing: a + /// final drop still ends a live copy still arriving, and the head serves in its place. + #[kio::test] + #[cfg(feature = "async")] + async fn a_drop_leaves_a_group_a_fetched_head_serves() { + let producer = track_producer("test", Info::timed()); + let dynamic = producer.requests(); + let consumer = producer.consume(); + let mut lives = Vec::new(); + for sequence in 0..2 { + let mut live = producer.create_group(group::Info { sequence }).unwrap(); + live.start_at(1).unwrap(); + live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta")) + .unwrap(); + let fetch = consumer.fetch_group(sequence, None); + let request = dynamic.requested_fill_group().await.unwrap(); + let mut head = request.accept(Info::timed()).unwrap(); + for payload in [&b"snapshot"[..], b"delta"] { + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(payload)) + .unwrap(); + } + head.finish().unwrap(); + fetch.await.unwrap(); + lives.push(live); + } + lives.pop().unwrap().abort(Error::Cancel).unwrap(); + + producer.drop_groups(0..1, StreamError::App(5)); + producer.give_up(1..2, Hole::stale(Duration::from_millis(500))); + + assert!(producer.state.read().drops.iter().next().is_none()); + assert!(lives[0].is_aborted(), "the live copy still arriving ends"); + let mut sub = producer.subscribe(None); + assert!(sub.poll_drop(&kio::Waiter::noop()).is_pending()); + for sequence in 0..2 { + assert_eq!(sub.assert_group().sequence, sequence); + } + } + /// Let `duration` pass on the cache pool `producer`'s groups charge into. fn elapse(producer: &Producer, duration: Duration) { producer.broadcast.pool.step(duration); diff --git a/rs/moq-net/tests/integration/history_join.rs b/rs/moq-net/tests/integration/history_join.rs new file mode 100644 index 00000000000..442dbf5d63e --- /dev/null +++ b/rs/moq-net/tests/integration/history_join.rs @@ -0,0 +1,67 @@ +#![cfg(feature = "async")] + +//! Viewers joining a relay that already holds a track's history, each on its own session. + +use std::time::Duration; + +use moq_net::{Timestamp, Version, cache, origin, track}; + +use crate::support::harness::{MockConnectOptions, connect_mock, spawn}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +fn origin(hop: u64) -> origin::Producer { + let mut config = origin::Config::new(std::num::NonZeroU64::new(hop).unwrap()); + config.pool = cache::Pool::unbounded(); + let (origin, driver) = origin::Producer::new(config); + spawn(driver); + origin.with_proxy(true) +} + +/// Every viewer that joins reads the group the relay holds. Each join subscribes upstream +/// again once the last viewer left, so the publisher serves the held group again; the relay +/// refuses that copy, and a lite publisher before lite-07 then drops the group for good. +/// That drop must not hide the group the relay holds whole. +#[kio::test] +async fn each_joining_viewer_reads_the_held_group() { + for version in ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip", "moq-transport-22"] { + let version: Version = version.parse().unwrap(); + let publisher = origin(1); + let relay = origin(2); + let mut options = MockConnectOptions::new(version); + options.client_publish = Some(publisher.consume()); + options.server_subscribe = Some(relay.clone()); + let _upstream = connect_mock(options).await; + + let broadcast = publisher.route("live", origin::RouteOptions::default()).unwrap(); + let info = track::Info::default() + .with_timescale(moq_net::Timescale::MILLI) + .with_max_age(None); + let source = broadcast.broadcast().create_track("video", info).unwrap(); + let mut group = source.append_group().unwrap(); + group + .write_frame(Timestamp::from_millis(0).unwrap(), &b"x"[..]) + .unwrap(); + group.finish().unwrap(); + + for join in 0..3 { + let viewer = origin(3 + join); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(viewer.clone()); + let downstream = connect_mock(options).await; + kio::clock::timeout(TIMEOUT, async { + let remote = viewer.consume().follow("live", None).unwrap().available().await; + let remote = remote.unwrap().expect("the broadcast resolves"); + let mut subscriber = remote.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = subscriber.recv_group().await.unwrap().expect("a group"); + assert_eq!(group.sequence, 0, "{version}: viewer {join}"); + let frame = group.read_frame().await.unwrap().expect("a frame"); + assert_eq!(&frame.payload[..], b"x", "{version}: viewer {join}"); + }) + .await + .unwrap_or_else(|_| panic!("{version}: viewer {join} never read the held group")); + drop(downstream); + } + } +} diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index 64c6820deb5..dad1e413018 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -38,6 +38,7 @@ mod flights; mod goaway; mod group_order; mod history_groups; +mod history_join; mod idle_cancel_refetch; mod idle_session; mod implementation; diff --git a/rs/moq-net/verify/subscribe/SubscribeAccounting.tla b/rs/moq-net/verify/subscribe/SubscribeAccounting.tla index 72e1d93b5e8..f57caf80bd2 100644 --- a/rs/moq-net/verify/subscribe/SubscribeAccounting.tla +++ b/rs/moq-net/verify/subscribe/SubscribeAccounting.tla @@ -342,7 +342,9 @@ TailFold(g) == \* `apply_response` for SUBSCRIBE_DROP. A stale drop under a replaced update is no \* verdict. Otherwise the tail accounts for the group; a stale drop records it in the \* copy unless its stream finished under this update or the copy holds it -\* (`give_up` keeps an open group), and a final drop records it and ends its readers. +\* (`give_up` keeps an open group), and a final drop records it and ends its readers +\* unless the copy holds it whole (`give_up` keeps a group a finished copy serves, +\* the live one or a fetched head). \* A stale drop of a group inside the span a fold forgot that the ledger already \* accounted (`Tail::overtaken`) may be one a finished stream overtook, so the next \* update owes it nothing again. @@ -365,7 +367,7 @@ OnDrop(m) == /\ UNCHANGED <> ELSE /\ acct' = acct \cup {g} /\ sstale' = sstale \ {g} - /\ rec' = [rec EXCEPT ![g] = IF @ = "skip" THEN "skip" ELSE "final"] + /\ rec' = IF rec[g] = "skip" \/ cached[g] = "whole" THEN rec ELSE [rec EXCEPT ![g] = "final"] /\ cached' = [cached EXCEPT ![g] = IF @ = "open" THEN "aborted" ELSE @] /\ UNCHANGED violation