From 442519d5aeaad00ed36374213e35fc2ec0ac76f3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 18:15:30 -0700 Subject: [PATCH 1/2] fix(net): move open IETF group streams on a priority update A subscription REQUEST_UPDATE changed the subscriber priority for groups opened afterwards, but each group stream captured its priority once, so a long or flow-control-blocked stream kept the old one after the update was acknowledged. TrackServe now owns the priority in a kio channel that every group stream follows while open, like lite's per-group track priority. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 106 ++++++++++++++++++++++++++----- 1 file changed, 91 insertions(+), 15 deletions(-) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 0c8d4a4b28..b6210d98a3 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -942,12 +942,10 @@ where }; return Some(result.and(Err(Error::Unsupported))); } - if let Some(priority) = update.priority { - let mut subscription = serve.track.subscription(); - subscription.priority = super::priority::from_wire(priority); - if let Err(err) = serve.track.update(subscription) { - return Some(Err(err)); - } + if let Some(priority) = update.priority + && let Err(err) = serve.set_priority(super::priority::from_wire(priority)) + { + return Some(Err(err)); } if answer && let Err(err) = async { @@ -2659,6 +2657,8 @@ struct TrackServe { range: ServeRange, timescale: Option, children: kio::Tasks>, + /// The subscriber priority, which every group stream follows while it is open. + priority: kio::Producer, /// The track finished: the in-flight group machines drain, then FIN. draining: bool, /// Group streams opened, shared with the group machines. @@ -2695,6 +2695,7 @@ impl TrackServe { // never write TIMESCALE, whatever the caller passed in, so datagrams and group // objects on those drafts go out unstamped. let timescale = timescale.filter(|_| ietf::Properties::sends_timescale(version)); + let priority = kio::Producer::new(track.subscription().priority); Self { datagrams, @@ -2705,6 +2706,7 @@ impl TrackServe { range, timescale, children: kio::Tasks::new(), + priority, draining: false, opened: Default::default(), end: None, @@ -2712,6 +2714,17 @@ impl TrackServe { } } + /// Move the subscriber priority, for the group streams already open as well as later ones. + fn set_priority(&mut self, priority: u8) -> Result<(), Error> { + let mut subscription = self.track.subscription(); + subscription.priority = priority; + self.track.update(subscription)?; + if let Ok(mut current) = self.priority.write() { + *current = priority; + } + Ok(()) + } + /// Group streams opened so far. fn opened(&self) -> u64 { self.opened.load(Ordering::Relaxed) @@ -2815,7 +2828,7 @@ impl TrackServe { GroupServe::new( self.session.clone(), msg, - self.track.subscription().priority, + self.priority.consume(), group, self.timescale, self.version, @@ -2904,7 +2917,10 @@ struct GroupServe { /// The subscription's count of opened streams, bumped once this one opens. opened: Arc, msg: ietf::GroupHeader, - priority: u8, + /// The subscription's priority, followed while the stream is open. + priority: kio::Consumer, + /// The priority last handed to the stream. + applied: u8, group: group::Consumer, timescale: Option, version: Version, @@ -2952,7 +2968,7 @@ impl GroupServe { fn new( session: S, msg: ietf::GroupHeader, - priority: u8, + priority: kio::Consumer, mut group: group::Consumer, timescale: Option, version: Version, @@ -2961,10 +2977,12 @@ impl GroupServe { group.skip_to(slice.skip); group.end_at(slice.until.map_or(Bound::Unbounded, Bound::Excluded)); let object_delta = group.index(); + let applied = *priority.read(); Self { session, opened: Default::default(), msg, + applied, priority, group, timescale, @@ -2983,6 +3001,18 @@ impl GroupServe { fn poll_serve(&mut self, waiter: &kio::Waiter) -> Poll> { let mut cx = waiter.context(); + // A REQUEST_UPDATE can move the subscriber priority while this stream is open. + let applied = self.applied; + let mut moved = match self.priority.poll(waiter, |priority| match **priority == applied { + true => Poll::Pending, + false => Poll::Ready(**priority), + }) { + Poll::Ready(Ok(priority)) => { + self.applied = priority; + Some(priority) + } + _ => None, + }; loop { match &mut self.state { GroupState::Open => { @@ -2999,7 +3029,8 @@ impl GroupServe { }; self.opened.fetch_add(1, Ordering::Relaxed); let mut stream = stream; - stream.set_priority(self.priority.into()); + self.applied = *self.priority.read(); + stream.set_priority(self.applied.into()); let mut writer = Writer::new(stream, self.version); if let Err(err) = writer.buffer(&self.msg) { @@ -3021,6 +3052,9 @@ impl GroupServe { batch, batch_pos, } => { + if let Some(priority) = moved.take() { + writer.set_priority(priority); + } // The peer closing first cancels the group. if writer.poll_closed(&mut cx).is_ready() { self.state = GroupState::Done; @@ -3139,6 +3173,9 @@ impl GroupServe { } } GroupState::Closed { writer } => { + if let Some(priority) = moved.take() { + writer.set_priority(priority); + } // Wait until everything is acknowledged by the peer so we can still // cancel the stream. poll_close releases the stream on completion so // the Drop fallback cannot reset the acknowledged stream. @@ -3266,7 +3303,7 @@ mod group_priority_test { let mut serve = GroupServe::new( session, msg, - 200, + kio::Producer::new(200).consume(), consumer, Some(Timescale::default()), Version::Draft14, @@ -3564,7 +3601,7 @@ mod group_priority_test { publisher_priority: 0, flags: Default::default(), }, - 0, + kio::Producer::new(0).consume(), group, Some(Timescale::default()), Version::Draft19, @@ -3607,7 +3644,7 @@ mod group_priority_test { publisher_priority: 0, flags: Default::default(), }, - 0, + kio::Producer::new(0).consume(), group, Some(Timescale::default()), Version::Draft19, @@ -3655,7 +3692,7 @@ mod group_priority_test { publisher_priority: 0, flags: Default::default(), }, - 0, + kio::Producer::new(0).consume(), group, Some(Timescale::default()), Version::Draft19, @@ -3914,6 +3951,45 @@ mod serve_tests { assert_eq!(occurrences(&session.log, &[0x07, 0, 1, 0]), 2); } + /// A priority update moves the group streams already in flight too, not only the + /// ones opened after it. + #[moq_net_sim::test] + async fn subscription_update_reprioritizes_open_group_streams() { + let version = Version::Draft19; + let h = serve(version); + let session = ScriptedSession::new(Vec::new()); + let mut stream = Stream::open(&mut session.clone(), version).await.unwrap(); + let mut serving = TrackServe::new( + h.session.clone(), + h.track.subscribe(None), + RequestId(0), + version, + ServeRange::default(), + None, + ); + let mut finished = false; + let mut run = std::pin::pin!( + h.publisher + .run_subscription(&mut stream, &mut serving, &mut finished, async {}) + ); + + // An unfinished group keeps its stream open. + let mut group = h.track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(timestamp(), b"open".as_slice()).unwrap(); + assert!(futures::poll!(run.as_mut()).is_pending()); + let before = h.log.priorities(); + assert_eq!(before.len(), 1, "{before:?}"); + + // REQUEST_UPDATE with SUBSCRIBER_PRIORITY 10. + session.push(&[0x02, 0, 4, 2, 1, 0x20, 10]); + for _ in 0..4 { + assert!(futures::poll!(run.as_mut()).is_pending()); + } + assert_eq!(h.track.subscription().unwrap().priority, 245); + let after = h.log.priorities(); + assert_eq!(after.last(), Some(&245), "{after:?}"); + } + #[moq_net_sim::test] async fn unsupported_subscription_update_fails_the_subscription() { let version = Version::Draft19; @@ -6161,7 +6237,7 @@ mod serve_tests { let mut serve = GroupServe::new( session, header(), - 0, + kio::Producer::new(0).consume(), consumer, Some(Timescale::default()), Version::Draft20, From b704fe84fdfa129965d7f79e9c5fe9397c2a8ba6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 05:43:13 -0700 Subject: [PATCH 2/2] test(net): a stream awaiting its FIN ack follows priority updates Also note why the priority write cannot fail and which states apply a moved priority. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 46 ++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index b6210d98a3..1f834098b0 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -2719,6 +2719,7 @@ impl TrackServe { let mut subscription = self.track.subscription(); subscription.priority = priority; self.track.update(subscription)?; + // Only a closed channel refuses the write, and this is the producer, which never closes it. if let Ok(mut current) = self.priority.write() { *current = priority; } @@ -3002,6 +3003,8 @@ impl GroupServe { fn poll_serve(&mut self, waiter: &kio::Waiter) -> Poll> { let mut cx = waiter.context(); // A REQUEST_UPDATE can move the subscriber priority while this stream is open. + // Serve and Closed, the only states holding a writer, apply the change; Open + // reads the current value once it gets its stream. let applied = self.applied; let mut moved = match self.priority.poll(waiter, |priority| match **priority == applied { true => Poll::Pending, @@ -3667,6 +3670,49 @@ mod group_priority_test { assert_eq!(log.resets(), vec![crate::StreamError::Cancel.to_code()]); } + /// A stream parked on its FIN acknowledgement still follows each priority update. + #[moq_net_sim::test] + async fn unacknowledged_fin_follows_priority_updates() { + let session = SinkSession::new(Default::default()).with_unacked_fin(); + let log = session.log.clone(); + let track = track::Producer::new(std::sync::Arc::new(crate::broadcast::Info::default()), "test", None); + let mut group = track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(crate::Timestamp::ZERO, b"done".as_slice()).unwrap(); + let consumer = group.consume(); + group.finish().unwrap(); + + let priority = kio::Producer::new(200); + let mut serve = GroupServe::new( + session, + ietf::GroupHeader { + track_alias: 0, + group_id: 0, + sub_group_id: 0, + publisher_priority: 0, + flags: Default::default(), + }, + priority.consume(), + consumer, + Some(Timescale::default()), + Version::Draft19, + GroupSlice::default(), + ); + let mut serving = std::pin::pin!(kio::wait(|waiter| serve.poll_serve(waiter))); + assert!( + futures::poll!(serving.as_mut()).is_pending(), + "the FIN is unacknowledged" + ); + assert_eq!(log.priorities(), vec![200]); + + *priority.write().ok().unwrap() = 100; + assert!(futures::poll!(serving.as_mut()).is_pending()); + assert_eq!(log.priorities(), vec![200, 100]); + + *priority.write().ok().unwrap() = 50; + assert!(futures::poll!(serving.as_mut()).is_pending()); + assert_eq!(log.priorities(), vec![200, 100, 50]); + } + /// The final payload remains guarded after its frame has advanced the group cursor. #[moq_net_sim::test] async fn blocked_final_transport_chunk_expires_with_the_group() {