feat(quic)!: schedule datagrams in their send group

doc/concept/moq-lite.md

@@ -197,7 +197,7 @@ Each subscription carries the knobs that decide behavior under congestion:
 
 | Knob | Effect |
 | --- | --- |
-| **Priority** (0..255) | Higher-priority tracks get bandwidth first. Audio above video, base layer above enhancement. From a native Rust publisher over QUIC or WebTransport (iroh excepted), subscriptions at the same priority split stream bandwidth evenly by bytes, however many groups each has queued. Datagrams are not part of that split yet. |
+| **Priority** (0..255) | Higher-priority tracks get bandwidth first. Audio above video, base layer above enhancement. From a native Rust publisher over QUIC or WebTransport (iroh excepted), subscriptions at the same priority split bandwidth evenly by bytes, streams and datagrams alike, however many groups each has queued. A browser publisher and other transports without send groups are not covered. |
 | **Order** | Which group to send first when several are pending. Newest first for live (the default), oldest first for catch-up. Needs moq-lite 07 or moq-transport; older moq-lite always sends newest first. |
 | **Max delay** | How long a non-latest group stays worth delivering before it is skipped. Zero means "live edge only", and raising it is also what asks for history. |
 
@@ -255,6 +255,12 @@ datagram: unreliable, unordered, under about 1200 bytes, and never
 retransmitted. It suits real-time audio and sensor data. There is no stream
 fallback, so a datagram that doesn't fit isn't delivered that way.
 
+A datagram shares its subscription's priority and group order with streams,
+so a low-priority datagram waits behind higher-priority groups instead of
+jumping the queue. From a native Rust publisher over QUIC or WebTransport
+(iroh excepted), a full send buffer drops the lowest-priority subscription's
+oldest datagram.
+
 Nothing caches a datagram. A subscription gets the ones inside its group
 range, and a new one may get the few still in the publisher's short send
 buffer, but `FETCH` never returns one. A publisher that knows the sequence was

drafts/draft-lcurley-moq-lite.md

@@ -596,6 +596,7 @@ The subscriber's `Group Order` determines which Group within the Track is transm
 
 A publisher SHOULD attempt to transmit streams based on these rules.
 This depends on the QUIC implementation and it may not be possible to get fine-grained control.
+A publisher SHOULD order a [datagram](#datagrams) among the subscription's streams as if it were its Group's stream, ahead of that stream when both exist.
 
 ### Priority
 The `Subscriber Priority` is scoped to the connection and MAY change over the life of the subscription via SUBSCRIBE_UPDATE.
@@ -725,7 +726,7 @@ A publisher sends a datagram to a subscription only when its Group Sequence is i
 A datagram is never cached or retransmitted, and FETCH never returns one.
 A publisher MAY send a new subscription the datagrams still in a short send buffer, so a subscriber cannot assume a datagram was sent after it subscribed.
 A publisher that knows a FETCH names a group delivered only as a datagram resets the stream with NOT_FETCHABLE; otherwise it answers as for a group that does not exist, with NOT_FOUND.
-A publisher SHOULD only send a datagram if the congestion controller can transmit it immediately.
+A publisher SHOULD drop a datagram rather than block when the transport has no room for it.
 There is no separate subscription for datagram delivery: datagrams are routed to existing subscriptions via the Subscribe ID, and a subscriber receiving the same group via both a stream and a datagram MUST deduplicate by group sequence.
 
 Each datagram body has the following encoding (note: there is no message length prefix; the QUIC datagram boundary delimits the payload):
@@ -1586,6 +1587,7 @@ The `Message Length` describes the payload size on the wire.
 - SUBSCRIBE_OK `Group` names where delivery starts when the subscription resolves, which need not be the first group sent. It is not a new floor. A subscriber does not wait for a group between the requested floor and it, but a publisher still delivers one that arrives later within Subscriber Max Age.
 - Corrected the SUBSCRIBE note that offset `Group Start` by 1. Only `Group End` and `Frame End` are offset so 0 can mean absent. `Group Start` is an absolute sequence, and 0 is group 0.
 - A subscription's range bounds datagrams like groups, and FETCH never returns a datagram.
+- A publisher SHOULD order a datagram among the subscription's streams as if it were its Group's stream, ahead of that stream when both exist. It SHOULD drop a datagram rather than block when the transport has no room for it, where it previously sent one only if the congestion controller could transmit it immediately.
 - Assigned 0x3A NOT_FETCHABLE in the stream error table: a FETCH for a group delivered only as a datagram.
 - The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset.
 - A subscriber MAY cancel by FIN on its Subscribe Stream, and the publisher then FINs without SUBSCRIBE_END.

rs/moq-ffi/src/transport/adapter.rs

@@ -80,8 +80,14 @@ impl<S: backend::Session<SendStream: 'static, RecvStream: 'static> + Clone + 'st
 	fn send_group(&mut self, _: i64) -> Option<poll::Ungrouped> {
 		None
 	}
-	fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll<Result<(), Self::Error>> {
-		backend::Session::poll_send_datagram(&mut self.0, cx, payload).map_err(Error)
+	/// The backend cannot rank a datagram against a stream, so the group and order are ignored,
+	/// and a datagram it has no room for is dropped.
+	fn send_datagram(&mut self, payload: Bytes, _: Option<&poll::Ungrouped>, _: i64) -> Result<(), Self::Error> {
+		let mut cx = Context::from_waker(std::task::Waker::noop());
+		match backend::Session::poll_send_datagram(&mut self.0, &mut cx, &payload) {
+			Poll::Ready(res) => res.map_err(Error),
+			Poll::Pending => Ok(()),
+		}
 	}
 	fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll<Result<Bytes, Self::Error>> {
 		backend::Session::poll_recv_datagram(&mut self.0, cx).map_err(Error)

rs/moq-net/benches/schedule.rs

@@ -14,6 +14,8 @@
 //!
 //! `fanout` sweeps subscriptions against groups in flight per subscription, timing one frame,
 //! so a scheduling cost that grows with the table rather than the touched path is a slope.
+//! `datagram` sweeps send groups, domains, and queued datagrams per group against one
+//! datagram sent and one evicted, for the same reason.
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
@@ -38,7 +40,7 @@ mod track {
 mod schedule;
 
 use priority::{Priority, PriorityHandle, PriorityQueue};
-use schedule::{Schedule, Slot};
+use schedule::{Next, Schedule, Slot};
 
 /// The most a frame carries, one packet.
 const PACKET: usize = 1200;
@@ -188,7 +190,7 @@ struct Sim {
 	kind: Kind,
 	tracks: Vec<Track>,
 	capacity: usize,
-	schedule: Schedule<u64>,
+	schedule: Schedule<u64, ()>,
 	groups: Vec<usize>,
 	queue: PriorityQueue,
 	moved: Arc<Mutex<Vec<u64>>>,
@@ -324,7 +326,7 @@ impl Sim {
 
 		let mut budget = self.capacity;
 		while budget > 0 {
-			let Some(id) = self.schedule.pop() else {
+			let Some(Next::Stream(id)) = self.schedule.pop() else {
 				break;
 			};
 			let stream = self.streams.get_mut(&id).unwrap();
@@ -420,7 +422,7 @@ fn bench_workloads(c: &mut Criterion) {
 }
 
 /// A saturated schedule: `subscriptions` groups at one priority, each with `streams` queued.
-fn saturated(subscriptions: usize, streams: usize) -> (Schedule<u64>, Vec<Slot>) {
+fn saturated(subscriptions: usize, streams: usize) -> (Schedule<u64, ()>, Vec<Slot>) {
 	let mut schedule = Schedule::new();
 	let mut slots = Vec::with_capacity(subscriptions * streams);
 	for _ in 0..subscriptions {
@@ -449,7 +451,9 @@ fn bench_fanout(c: &mut Criterion) {
 				&(),
 				|b, ()| {
 					b.iter(|| {
-						let id = schedule.pop().unwrap();
+						let Some(Next::Stream(id)) = schedule.pop() else {
+							unreachable!("only streams are queued");
+						};
 						let slot = &mut slots[id as usize];
 						schedule.sent(id, slot, PACKET, Some((id % streams as u64) as i64), true);
 					});
@@ -485,5 +489,75 @@ fn bench_fanout(c: &mut Criterion) {
 	group.finish();
 }
 
-criterion_group!(benches, bench_workloads, bench_fanout);
+/// A full datagram budget: `groups` send groups at one priority, each in its own domain when
+/// `domains`, with `datagrams` queued in each, one packet apiece.
+fn datagrams(groups: usize, datagrams: usize, domains: bool) -> (Schedule<u64, ()>, Vec<usize>) {
+	let mut schedule = Schedule::new();
+	let ids: Vec<_> = (0..groups)
+		.map(|_| {
+			let domain = domains.then(|| schedule.open_domain());
+			schedule.open_group(0, domain)
+		})
+		.collect();
+	let budget = groups * datagrams * PACKET;
+	for seq in 0..datagrams {
+		for &group in &ids {
+			assert!(schedule.push_datagram(Some(group), seq as i64, (), PACKET, budget));
+		}
+	}
+	(schedule, ids)
+}
+
+fn bench_datagram(c: &mut Criterion) {
+	for (name, domains) in [("groups", false), ("domains", true)] {
+		// A datagram into a full budget, which evicts the oldest of the busiest domain's
+		// lowest-priority group.
+		let mut group = c.benchmark_group(format!("schedule_datagram_evict_{name}"));
+		for groups in [1, 16, 256, 4_096] {
+			for queued in [1, 16, 256] {
+				if groups * queued > 100_000 {
+					continue;
+				}
+				let (mut schedule, ids) = datagrams(groups, queued, domains);
+				let budget = schedule.datagram_bytes();
+				let mut seq = queued as i64;
+				let mut next = 0;
+				group.bench_with_input(BenchmarkId::new(format!("{name}_{groups}"), queued), &(), |b, ()| {
+					b.iter(|| {
+						// Round robin, so every group keeps its share of the budget.
+						schedule.push_datagram(Some(ids[next]), seq, (), PACKET, budget);
+						next = (next + 1) % ids.len();
+						seq += 1;
+					});
+				});
+			}
+		}
+		group.finish();
+	}
+
+	// A datagram sent and another queued in its place, through every round robin.
+	let mut group = c.benchmark_group("schedule_datagram_send");
+	for groups in [1, 16, 256, 4_096] {
+		for queued in [1, 16, 256] {
+			if groups * queued > 100_000 {
+				continue;
+			}
+			let (mut schedule, _) = datagrams(groups, queued, false);
+			let mut seq = queued as i64;
+			group.bench_with_input(BenchmarkId::new(format!("groups_{groups}"), queued), &(), |b, ()| {
+				b.iter(|| {
+					let Some(Next::Datagram(at)) = schedule.pop() else {
+						unreachable!("only datagrams are queued");
+					};
+					schedule.sent_datagram(at);
+					schedule.push_datagram(at, seq, (), PACKET, usize::MAX);
+					seq += 1;
+				});
+			});
+		}
+	}
+	group.finish();
+}
+
+criterion_group!(benches, bench_workloads, bench_fanout, bench_datagram);
 criterion_main!(benches);

rs/moq-net/src/client.rs

@@ -624,8 +624,13 @@ mod tests {
 			None
 		}
 
-		fn poll_send_datagram(&mut self, _cx: &mut Context<'_>, _payload: &[u8]) -> Poll<Result<(), Self::Error>> {
-			Poll::Ready(Ok(()))
+		fn send_datagram(
+			&mut self,
+			_: Bytes,
+			_: Option<&crate::transport::poll::Ungrouped>,
+			_: i64,
+		) -> Result<(), Self::Error> {
+			Ok(())
 		}
 
 		fn poll_recv_datagram(&mut self, _cx: &mut Context<'_>) -> Poll<Result<Bytes, Self::Error>> {
@@ -1019,8 +1024,13 @@ mod tests {
 			None
 		}
 
-		fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll<Result<(), Self::Error>> {
-			self.inner.poll_send_datagram(cx, payload)
+		fn send_datagram(
+			&mut self,
+			payload: Bytes,
+			group: Option<&crate::transport::poll::Ungrouped>,
+			order: i64,
+		) -> Result<(), Self::Error> {
+			self.inner.send_datagram(payload, group, order)
 		}
 
 		fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll<Result<Bytes, Self::Error>> {

rs/moq-net/src/ietf/adapter.rs

@@ -1176,10 +1176,15 @@ impl<S: crate::transport::poll::Session> crate::transport::poll::Session for Con
 		self.inner.send_group(priority)
 	}
 
-	fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll<Result<(), Self::Error>> {
+	fn send_datagram(
+		&mut self,
+		payload: Bytes,
+		group: Option<&<Self::SendStream as crate::transport::poll::SendStream>::Group>,
+		order: i64,
+	) -> Result<(), Self::Error> {
 		self.inner
-			.poll_send_datagram(cx, payload)
-			.map(|r| r.map_err(|_| crate::Error::Closed))
+			.send_datagram(payload, group, order)
+			.map_err(|_| crate::Error::Closed)
 	}
 
 	fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll<Result<Bytes, Self::Error>> {

rs/moq-net/src/ietf/publisher.rs

@@ -3937,7 +3937,10 @@ impl<S: crate::transport::poll::Session> TrackServe<S> {
 				);
 				continue;
 			}
-			let _ = self.session.send_datagram(&body);
+			// Ranked like a stream carrying the group from its start, so it merges into the
+			// subscription's group order and wins a tie with that group's stream.
+			let order = crate::priority::in_group(sequence, 0, self.ranking.order);
+			let _ = self.session.send_datagram(body, self.ranking.group.as_ref(), order);
 		}
 		Ok(())
 	}
@@ -5123,6 +5126,49 @@ mod group_priority_test {
 		assert_eq!(log.orders().len(), 2, "the streams keep their place in the group");
 	}
 
+	/// An OBJECT_DATAGRAM goes in its subscription's send group, ranked like the stream
+	/// carrying its group from the start, so the transport merges it into the group order.
+	#[test]
+	fn datagrams_rank_within_their_subscription() {
+		let log = crate::lite::test_transport::Log::default();
+		let session = SinkSession::new(log.clone())
+			.with_groups()
+			.with_datagrams(1200)
+			.with_unacked_fin();
+		let mut track = track::Producer::new(
+			std::sync::Arc::new(crate::broadcast::Info::default()),
+			"test",
+			crate::track::Info::timed(),
+		);
+		let subscriber = track.subscribe(
+			track::Subscription::default()
+				.with_priority(3)
+				.with_max_delay(std::time::Duration::from_secs(3600)),
+		);
+		let mut serve = TrackServe::new(
+			session,
+			subscriber,
+			RequestId(0),
+			Version::Draft14,
+			ServeRange::default(),
+			None,
+		);
+		let mut group = track.create_group(group::Info { sequence: 0 }).unwrap();
+		group
+			.write_frame(crate::Timestamp::from_millis(0).unwrap(), b"x".as_slice())
+			.unwrap();
+		group.finish().unwrap();
+		let sequence = track
+			.append_datagram(crate::Timestamp::from_millis(1).unwrap(), b"d".as_slice())
+			.unwrap();
+
+		let waiter = kio::Waiter::noop();
+		assert!(serve.poll(&waiter).is_pending());
+		let order = crate::priority::in_group(sequence, 0, Order::Descending);
+		assert_eq!(log.datagram_ranks(), [(Some(0), order)]);
+		assert_eq!(log.orders(), [(0, 1)], "the stream for group 0 ranks below");
+	}
+
 	/// A group with more objects ready than one poll's budget drains over several polls.
 	#[test]
 	fn a_long_group_drains_within_the_budget() {