diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index a5d46f11b4..2ec97237f0 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -6326,6 +6326,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"); } } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index b49b33b995..2f29e882dd 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -6365,6 +6365,49 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + /// An aborted edge hands the edge to a group between it and the successor, and neither + /// the abort nor that group's first frame touches the track, so the read must still wake. + #[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_delay(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:?}"); + } + /// 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]