From 0eb19cd15b7a585f840f849d0ab65d734fe746c2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 08:02:02 -0800 Subject: [PATCH 1/3] fix(net): drain subscriptions awaiting track info Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/src/lite/subscriber.rs | 75 +++++++++++++--- rs/moq-net/src/model/front.rs | 59 ++++++++++--- rs/moq-net/src/model/origin.rs | 139 +++++++++++++++++++++--------- 3 files changed, 207 insertions(+), 66 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index bb5a71c72e3..3adab426076 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4791,6 +4791,54 @@ mod tests { } } + /// An in-flight TRACK survives withdrawal, like its pipelined SUBSCRIBE, but + /// never accepts a replacement instance's info. + #[kio::test] + async fn an_in_flight_track_info_keeps_its_instance() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + for replace in [false, true] { + let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()); + let subscriber = pipelining(crate::driver::Transport::new(session.clone()), version); + let (_waiting, mut run) = subscribing(&subscriber, Subscription::default()); + let route = kio::Producer::new(()); + run.serve.route = Some(route.consume()); + let mut selection = crate::model::TestSelection::new(); + run.serve.validity = selection.validity(); + let mut fetch = TrackInfoFetch::new(&run.serve); + let waiter = kio::Waiter::noop(); + assert!(fetch.poll_fetch(&run.serve, true, &waiter).is_pending()); + drop(route); + selection.withdraw(); + if replace { + selection.replace(); + } else { + assert!( + fetch.poll_fetch(&run.serve, true, &waiter).is_pending(), + "{version}: the request already went out" + ); + } + + let mut info = Vec::new(); + lite::TrackInfo { + refresh: None, + priority: 0, + max_age: None, + timescale: Some(Timescale::MILLI), + compress: false, + } + .encode(&mut crate::coding::Encoder::new(&mut info, version.into()), version) + .unwrap(); + session.push(&info); + let result = fetch.poll_fetch(&run.serve, true, &waiter); + if replace { + assert!(matches!(result, Poll::Ready(Err(Error::Unroutable))), "{version}"); + } else { + assert!(matches!(result, Poll::Ready(Ok(_))), "{version}"); + } + } + } + } + /// Accepting the TRACK_INFO sends no SUBSCRIBE_UPDATE, even when the publisher keeps /// groups for less than the subscriber waits: the publisher applies its own max age. #[kio::test] @@ -8369,7 +8417,10 @@ impl kio::Task for TrackServeRun { } // TRACK opens first: its stream fails fast without credit, where a fetch's // waits for it, so fetches opened ahead could starve their own prerequisite. - let fetched = info.poll_fetch(&self.serve, waiter); + let subscribed = pipelined + .as_ref() + .is_some_and(|subscribing| matches!(subscribing.state, PipelinedState::Open(_))); + let fetched = info.poll_fetch(&self.serve, subscribed, waiter); let failed = match pipelined { Some(subscribing) => subscribing.poll(waiter), None => Poll::Pending, @@ -8536,15 +8587,19 @@ impl TrackInfoFetch { fn poll_fetch( &mut self, serve: &TrackServe, + subscribed: bool, waiter: &kio::Waiter, ) -> Poll> { #[every_poll] - self.validity.check(waiter)?; - // A withdrawn route fails an info still owed. One already delivered stays usable, like - // one read before the withdrawal: a SUBSCRIBE sent alongside may already be served. - #[every_poll] - if !matches!(self.state, TrackInfoState::Read(_)) && withdrawn(serve, waiter) { - return Poll::Ready(Err(Error::Unroutable)); + if subscribed && matches!(self.state, TrackInfoState::Read(..) | TrackInfoState::Properties(..)) { + // The request is on the wire: withdrawal leaves it to its answer, just like + // the SUBSCRIBE sent alongside it. A replacement still invalidates that answer. + self.validity.check_unanswered(waiter)?; + } else { + self.validity.check(waiter)?; + if withdrawn(serve, waiter) { + return Poll::Ready(Err(Error::Unroutable)); + } } let mut stream: Stream = @@ -8562,14 +8617,10 @@ impl TrackInfoFetch { if let Poll::Ready(err) = self.closed.poll_closed(waiter) { return Poll::Ready(Err(crate::lite::error::from_transport(err))); } - let read = match serve.subscriber.peer_setup.poll_ietf(waiter) { + match serve.subscriber.peer_setup.poll_ietf(waiter) { Poll::Ready(ietf) => stream.reader.poll_decode::(waiter).map_ok(|info| (ietf, info)), Poll::Pending => Poll::Pending, - }; - if read.is_pending() && withdrawn(serve, waiter) { - return Poll::Ready(Err(Error::Unroutable)); } - read })?; // Publisher Max Age rides on the wire, so the local retention diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index 797d9742adc..566134aaa63 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -122,8 +122,8 @@ pub(super) enum Action { /// Request the path through `route`; feed the outcome back as /// [`Event::Resolved`]. Request { route: u64 }, - /// Drop the driver's handle on a source. - Detach { source: u64 }, + /// Drop the driver's source handle, keeping its pending track queries when `drain` is true. + Detach { source: u64, drain: bool }, /// Resolve the requesters parked on the front: it has a source. Resolve, /// Ask `source` for its copy of `track`; feed the answer back as @@ -394,7 +394,7 @@ impl Front { let Some(upstream) = self.upstream.filter(|upstream| upstream.route == route) else { // A source for a request we no longer want: let it go. if let Ok(source) = result { - actions.push(Action::Detach { source }); + actions.push(Action::Detach { source, drain: false }); } return; }; @@ -422,7 +422,10 @@ impl Front { fn attach(&mut self, source: u64, candidate: Candidate, actions: &mut Vec) { if let Some((old, _)) = self.serving.take() { - actions.push(Action::Detach { source: old }); + actions.push(Action::Detach { + source: old, + drain: false, + }); } // Every copy is from an older source, including one a track still drains from a // source the previous one replaced. @@ -458,7 +461,7 @@ impl Front { self.excluded.insert(candidate.route); self.serving = None; self.serving_closing = false; - actions.push(Action::Detach { source }); + actions.push(Action::Detach { source, drain: true }); self.drain_copies(); actions.push(Action::Reselect); } @@ -856,7 +859,10 @@ mod tests { result: Ok(200), }), &[ - Action::Detach { source: 100 }, + Action::Detach { + source: 100, + drain: false, + }, Action::Query { track: name("video"), source: 200, @@ -886,7 +892,10 @@ mod tests { route: 2, result: Ok(200), }), - &[Action::Detach { source: 200 }], + &[Action::Detach { + source: 200, + drain: false, + }], ); } @@ -895,7 +904,13 @@ mod tests { let mut front = serving(remote(1), 100); assert_actions( front.step(Event::SourceClosed { source: 100 }), - &[Action::Detach { source: 100 }, Action::Reselect], + &[ + Action::Detach { + source: 100, + drain: true, + }, + Action::Reselect, + ], ); assert_eq!(front.excluded_routes(), &HashSet::from([1])); assert_actions( @@ -1349,7 +1364,13 @@ mod tests { front.step(Event::Forgotten { track: name("video") }); assert_actions( front.step(Event::SourceClosed { source: 100 }), - &[Action::Detach { source: 100 }, Action::Reselect], + &[ + Action::Detach { + source: 100, + drain: true, + }, + Action::Reselect, + ], ); assert_actions(front.step(Event::Unheld), &[]); assert_actions( @@ -1424,7 +1445,10 @@ mod tests { route: 2, result: Ok(200), }), - &[Action::Detach { source: 100 }], + &[Action::Detach { + source: 100, + drain: false, + }], ); let t1 = t0 + LINGER; front.step(Event::Unwanted { @@ -1573,7 +1597,13 @@ mod tests { let mut front = serving(local(1), 100); assert_actions( front.step(Event::SourceClosed { source: 100 }), - &[Action::Detach { source: 100 }, Action::Reselect], + &[ + Action::Detach { + source: 100, + drain: true, + }, + Action::Reselect, + ], ); assert_actions( front.step(Event::Selected { @@ -1759,7 +1789,7 @@ mod tests { assert!(splices <= 1); // A Detach names a source the front no longer serves from. for action in &actions { - if let Action::Detach { source } = action { + if let Action::Detach { source, .. } = action { assert_ne!(next.serving.map(|(s, _)| s), Some(*source)); } } @@ -1807,7 +1837,10 @@ mod tests { result: Ok(300), }), &[ - Action::Detach { source: 200 }, + Action::Detach { + source: 200, + drain: false, + }, Action::Query { track: name("video"), source: 300, diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index c56d3bb5fb5..62a7cf24e3f 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3378,8 +3378,9 @@ impl TrackIo { /// Serve the track from `asked`'s copy, accepting it with that copy's info first. fn splice(&mut self, asked: Asked) { self.ready.wake_by_ref(); - if let Some(request) = self.request.take() { - let info = asked.info.clone().unwrap_or_default(); + if let Some(info) = asked.info.clone() + && let Some(request) = self.request.take() + { self.accepted = Some(request.accept(info)); } self.routes.splice(asked.copy.clone(), asked.staged); @@ -3436,10 +3437,14 @@ struct FrontListing { struct InFlight { weak: track::TrackWeak, routes: super::resume::Producer, + /// Why the front ended, used if its pending copy cannot answer. + end: Error, + /// Still owed when the front ended before the copy's info arrived. + request: Option, // Kept for its readers: the logical track, the query holding its copy, and the copy's // subscription, which follows their demand as it did while the front served it. accepted: Option, - _copy: Option, + query: Option, held: Option>, } @@ -3453,10 +3458,12 @@ fn follow_demand( routes: &super::resume::Producer, waiter: &kio::Waiter, ) { - let (Some(hold), Some(accepted)) = (held.as_mut(), accepted) else { + let Some(hold) = held.as_mut() else { return; }; - while let Poll::Ready(Ok(_)) = accepted.poll_subscription_changed(waiter) {} + if let Some(accepted) = accepted { + while let Poll::Ready(Ok(_)) = accepted.poll_subscription_changed(waiter) {} + } match weak.subscription() { Some((subscription, subscribers)) => { let _ = hold.update(routes.subscription(subscription, subscribers)); @@ -3470,6 +3477,9 @@ impl InFlight { /// as a cancel rather than as a producer dropped unfinished. fn release(self) { self.routes.release(); + if let Some(request) = self.request { + request.reject(Error::Cancel); + } if let Some(accepted) = self.accepted { let _ = accepted.abort(Error::Cancel); } @@ -3489,9 +3499,37 @@ async fn run_front(task: FrontTask, origin: TasksWeak) { // Dropped without a release, so a reader on its way keeps the copy; it stays held // only while the logical track's state does, for an origin nobody owns. if origin.poll_orphaned(waiter).is_ready() { + for io in &mut in_flight { + if let Some(request) = io.request.take() { + request.reject(io.end.clone()); + } + } return Poll::Ready(()); } for mut io in std::mem::take(&mut in_flight) { + if io.request.is_some() { + match io + .query + .as_ref() + .expect("a pending request holds its query") + .poll_ok(waiter) + { + Poll::Ready(Ok(info)) => { + io.accepted = Some(io.request.take().expect("pending above").accept(info)); + } + Poll::Ready(Err(err)) => { + // No metadata ever arrived: a lost route cannot answer the request. + let err = match super::resume::route_failed(&err) { + true => io.end.clone(), + false => err, + }; + io.request.take().expect("pending above").reject(err.clone()); + io.routes.end(Err(err)); + continue; + } + Poll::Pending => {} + } + } follow_demand(&mut io.held, io.accepted.as_mut(), &io.weak, &io.routes, waiter); match io.weak.poll_unwanted(waiter) { Poll::Pending => in_flight.push(io), @@ -3871,17 +3909,14 @@ async fn serve_front(task: FrontTask) -> Vec { asked_renewals: serve.renewals.load(Ordering::Acquire), }); } - Action::Detach { source } => { - // Readers keep reading the source's copy until a replacement is - // spliced, so a route being beaten keeps serving until then; a - // route that died ends its copy, which says so. + Action::Detach { source, drain } => { sources.remove(&source); instances.remove(&source); for io in tracks.values_mut() { - if io.query.as_ref().is_some_and(|asked| asked.source == source) { + if !drain && io.query.as_ref().is_some_and(|asked| asked.source == source) { io.query = None; } - if io.staged.as_ref().is_some_and(|asked| asked.source == source) { + if !drain && io.staged.as_ref().is_some_and(|asked| asked.source == source) { io.staged = None; } // Any source leaving is rare: drop every parked shortcut rather @@ -4059,6 +4094,7 @@ async fn serve_front(task: FrontTask) -> Vec { let TrackIo { weak, routes, + request, accepted, copy, held, @@ -4067,8 +4103,10 @@ async fn serve_front(task: FrontTask) -> Vec { in_flight.push(InFlight { weak, routes, + end: err.clone(), + request, accepted, - _copy: copy.map(|(_, _, query)| query), + query: copy.map(|(_, _, query)| query), held, }); } @@ -9875,38 +9913,57 @@ mod tests { assert!(!again.is_clone(&resolved)); } - /// A subscriber still waiting on the source's track info is in flight too: - /// unannouncing leaves it on the copy it asked for, which the source can - /// still answer and finish. - #[kio::test] - #[cfg(feature = "async")] - async fn unannounce_keeps_a_track_awaiting_its_info() { - let producer = origin(1).produce(); - let consumer = producer.consume(); - - let broadcast = producer.publish("room/alice", Route::default()).unwrap(); - let mut dynamic = broadcast.requests(); - let resolved = consumer.request("room/alice", None).await.expect("resolves"); - let track = resolved.track("video").unwrap(); - let subscribing = kio::clock::sim::spawn(async move { track.subscribe(None).await }); - let request = kio::clock::timeout(Duration::from_secs(1), dynamic.requested_track()) - .await - .expect("the front asked the source") - .expect("request"); + /// Ending a publication leaves a subscriber waiting on its track info in flight: + /// the source can still answer and finish the copy it already handed out. + #[kio::test] + #[cfg(feature = "async")] + async fn ending_a_publication_keeps_a_track_awaiting_its_info() { + for close in [false, true] { + for replacement in [false, true] { + let producer = origin(1).produce(); + let consumer = producer.consume(); + + let broadcast = producer + .publish("room/alice", Route::default().with_epoch(crate::Epoch::mint())) + .unwrap(); + let mut dynamic = broadcast.requests(); + let resolved = consumer.request("room/alice", None).await.expect("resolves"); + let track = resolved.track("video").unwrap(); + let subscribing = kio::clock::sim::spawn(async move { track.subscribe(None).await }); + let request = kio::clock::timeout(Duration::from_secs(1), dynamic.requested_track()) + .await + .expect("the front asked the source") + .expect("request"); + let _replacement = replacement.then(|| { + producer + .requests_route( + "room/alice", + Route::default().with_epoch(crate::Epoch::mint()).with_hops(hops(&[10])), + None, + ) + .unwrap() + }); - broadcast.unannounce(); - settle(|| resolved.is_closed()).await; + if close { + broadcast.close(); + } else { + broadcast.unannounce(); + } + settle(|| resolved.is_closed()).await; - let source = request.accept(crate::track::Info::timed()); - let mut group = source.append_group().unwrap(); - group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap(); - group.finish().unwrap(); - source.finish().unwrap(); + let source = request.accept(crate::track::Info::timed()); + let mut group = source.append_group().unwrap(); + group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap(); + group.finish().unwrap(); + source.finish().unwrap(); - let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction"); - let mut group = subscription.recv_group().await.unwrap().expect("the source's group"); - assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"late"); - assert!(matches!(subscription.recv_group().await, Ok(None)), "ends cleanly"); + let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction"); + assert_eq!(subscription.info(), source.info(), "the source's actual track info"); + let mut group = subscription.recv_group().await.unwrap().expect("the source's group"); + assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"late"); + assert!(matches!(subscription.recv_group().await, Ok(None)), "ends cleanly"); + } + } } /// A front resolved through a served route, as a relay's upstream session serves From 81b09b3a42e750457ccc289b7448a8bc6b021e9b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 08:13:12 -0800 Subject: [PATCH 2/3] fix(net): retain draining query state and follow pending demand Co-Authored-By: GPT-6.1 Sol --- rs/moq-net/src/model/front.rs | 43 ++++++++++++++++++++++++++++------ rs/moq-net/src/model/origin.rs | 17 ++++++++++++-- 2 files changed, 51 insertions(+), 9 deletions(-) diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index 566134aaa63..85ddfa4cc8a 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -429,7 +429,7 @@ impl Front { } // Every copy is from an older source, including one a track still drains from a // source the previous one replaced. - self.drain_copies(); + self.drain_copies(false); self.serving = Some((source, candidate)); self.serving_closing = false; if !self.resolved { @@ -462,21 +462,22 @@ impl Front { self.serving = None; self.serving_closing = false; actions.push(Action::Detach { source, drain: true }); - self.drain_copies(); + self.drain_copies(true); actions.push(Action::Reselect); } /// The serving source is going: a spliced copy keeps feeding its track (readers read it /// until a replacement's is spliced, or it runs out), and one still being asked for is - /// dropped. - fn drain_copies(&mut self) { + /// kept while that source can still answer, until a replacement attaches. + fn drain_copies(&mut self, keep_queries: bool) { for track in self.tracks.values_mut() { match track.state { TrackState::Spliced { source } => { track.draining = Some(source); track.state = TrackState::Idle; } - TrackState::Querying { .. } => track.state = TrackState::Idle, + TrackState::Querying { .. } if !keep_queries => track.state = TrackState::Idle, + TrackState::Querying { .. } => {} TrackState::Idle | TrackState::Parked { .. } => {} } } @@ -761,8 +762,8 @@ mod tests { assert_eq!(front.next_deadline(), scanned.first().map(|(at, _)| *at)); } - /// A front serving `source` through `candidate`, with `video` spliced and read. - fn serving(candidate: Candidate, source: u64) -> Front { + /// A front querying `source` through `candidate` for the read track `video`. + fn querying(candidate: Candidate, source: u64) -> Front { let mut front = Front::new(LINGER); assert_actions( front.step(Event::Selected { @@ -796,6 +797,12 @@ mod tests { Action::Arm { at: None }, ], ); + front + } + + /// A front serving `source` through `candidate`, with `video` spliced and read. + fn serving(candidate: Candidate, source: u64) -> Front { + let mut front = querying(candidate, source); assert_actions( front.step(Event::TrackInfo { track: name("video"), @@ -872,6 +879,28 @@ mod tests { assert_eq!(front.serving, Some((200, remote(2)))); } + #[test] + fn a_closing_source_answers_while_another_route_is_resolving() { + let mut front = querying(remote(1), 100); + front.step(Event::SourceClosed { source: 100 }); + front.step(Event::Selected { + best: Some(remote(2)), + serving_closing: false, + }); + assert_actions( + front.step(Event::TrackInfo { + track: name("video"), + source: 100, + closing: true, + result: Ok(info()), + }), + &[Action::Splice { + track: name("video"), + source: 100, + }], + ); + } + #[test] fn a_stale_resolution_is_let_go() { let mut front = serving(remote(1), 100); diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 62a7cf24e3f..c791c7c9d6e 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3507,6 +3507,9 @@ async fn run_front(task: FrontTask, origin: TasksWeak) { return Poll::Ready(()); } for mut io in std::mem::take(&mut in_flight) { + if let Some(request) = &mut io.request { + while let Poll::Ready(Some(_)) = request.poll_subscription_changed(waiter) {} + } if io.request.is_some() { match io .query @@ -9929,7 +9932,8 @@ mod tests { let mut dynamic = broadcast.requests(); let resolved = consumer.request("room/alice", None).await.expect("resolves"); let track = resolved.track("video").unwrap(); - let subscribing = kio::clock::sim::spawn(async move { track.subscribe(None).await }); + let mut subscribing = track.subscribe(crate::track::Subscription::default().with_priority(3)); + assert!((&mut subscribing).now_or_never().is_none()); let request = kio::clock::timeout(Duration::from_secs(1), dynamic.requested_track()) .await .expect("the front asked the source") @@ -9950,6 +9954,15 @@ mod tests { broadcast.unannounce(); } settle(|| resolved.is_closed()).await; + subscribing + .update(crate::track::Subscription::default().with_priority(7)) + .unwrap(); + settle(|| { + request + .subscription() + .is_some_and(|subscription| subscription.priority == 7) + }) + .await; let source = request.accept(crate::track::Info::timed()); let mut group = source.append_group().unwrap(); @@ -9957,7 +9970,7 @@ mod tests { group.finish().unwrap(); source.finish().unwrap(); - let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction"); + let mut subscription = subscribing.await.expect("subscribe survives the retraction"); assert_eq!(subscription.info(), source.info(), "the source's actual track info"); let mut group = subscription.recv_group().await.unwrap().expect("the source's group"); assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"late"); From a72875d375a59961f2761bbb4e7439b0b83dff0c Mon Sep 17 00:00:00 2001 From: quest fixer Date: Tue, 3 Nov 2026 08:04:03 -0800 Subject: [PATCH 3/3] test(net): follow base route instance API rename --- rs/moq-net/src/model/origin.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 3a03cc3c4d3..f87ab6dfa06 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -9699,7 +9699,10 @@ mod tests { let consumer = producer.consume(); let broadcast = producer - .publish("room/alice", Route::default().with_epoch(crate::Epoch::mint())) + .publish( + "room/alice", + Route::default().with_instance(crate::Instance::random(), 0), + ) .unwrap(); let mut dynamic = broadcast.requests(); let resolved = consumer.request("room/alice", None).await.expect("resolves"); @@ -9714,7 +9717,9 @@ mod tests { producer .requests_route( "room/alice", - Route::default().with_epoch(crate::Epoch::mint()).with_hops(hops(&[10])), + Route::default() + .with_instance(crate::Instance::random(), 0) + .with_hops(hops(&[10])), None, ) .unwrap()