feat(net)!: pace a batch fill's handler by its fastest reader

dart/moq/MIGRATE.md

@@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### `Fill.wantedChanged()` waits for a reader to pick up the group served before
+
+`await fill.wantedChanged()` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wantedChanged()` until it returns null, accepting one group per call, rather than looping over `wanted()`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take concurrently, since each now waits on its own readers.
+
 ### Socket sessions
 
 package:moq/socket.dart now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change.

dart/moq_ffi/lib/src/moq.dart

@@ -24711,10 +24711,10 @@ void _checkApiChecksums() {
   if (uniffi_moq_ffi_checksum_method_moqfill_unwanted() != 13637) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqfill_wanted() != 65141) {
+  if (uniffi_moq_ffi_checksum_method_moqfill_wanted() != 16564) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqfill_wanted_changed() != 42560) {
+  if (uniffi_moq_ffi_checksum_method_moqfill_wanted_changed() != 7000) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqgroupproducer_abort() != 59787) {

go/wrapper/MIGRATE.md

@@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### `Fill.WantedChanged` waits for a reader to pick up the group served before
+
+`Fill.WantedChanged(ctx)` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `WantedChanged` until it returns nil, accepting one group per call, rather than looping over `Wanted`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take in their own goroutines, since each now waits on its own readers.
+
 ### Socket sessions
 
 moq.dev/moq/socket now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change.

go/wrapper/publish.go

@@ -317,11 +317,13 @@ func (d *TrackRequests) Cancel() {
 
 // Fill is a run of groups readers miss, for a TrackRequests to serve.
 //
-// Accept each group you have and DropGroups the rest; Wanted names what is still
-// owed, in the fill's order. Releasing the fill names whatever is left as dropped
+// Accept each group you have and DropGroups the rest, one group at a time as
+// WantedChanged returns them; Wanted names what is still owed, in the fill's
+// order. Releasing the fill names whatever is left as dropped
 // while a reader still wants it; once nobody does, it is withdrawn, so stop
 // serving it once Wanted is nil. Stop serving a single group when Unwanted
-// returns for it, and move on to Wanted.
+// returns for it, and move on to Wanted. Serve the fills you take
+// concurrently: each waits on its own readers.
 type Fill struct {
 	inner *ffi.MoqFill
 }
@@ -358,12 +360,16 @@ func (f *Fill) Order() Order {
 }
 
 // Wanted is the next group still owed in the fill's order, or nil once every
-// group is answered or no reader waits on the fill.
+// group is answered or no reader waits on the fill. It does not wait for the
+// readers; WantedChanged does.
 func (f *Fill) Wanted() *uint64 {
 	return f.inner.Wanted()
 }
 
-// WantedChanged blocks until Wanted changes and returns it. The first call returns at once.
+// WantedChanged blocks until Wanted changed and every group this fill accepted
+// reached a reader, which picks it up when it next reads, and returns it. Nil,
+// once the fill is done, never waits; the first call returns at once unless an
+// accepted group has yet to reach a reader.
 func (f *Fill) WantedChanged(ctx context.Context) (*uint64, error) {
 	return f.inner.WantedChanged(ctx)
 }

kt/moq/MIGRATE.md

@@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### `Fill.wantedChanged()` waits for a reader to pick up the group served before
+
+`fill.wantedChanged()` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wantedChanged()` until it returns null, accepting one group per call, rather than looping over `wanted()`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take in their own coroutines, since each now waits on its own readers.
+
 ### Socket sessions
 
 dev.moq.socket now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. Like broadcasts, endpoint `end()` forces shutdown while AutoCloseable `close()` releases the handle.

py/moq-rs/MIGRATE.md

@@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### `Fill.wanted_changed()` waits for a reader to pick up the group served before
+
+`await fill.wanted_changed()` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wanted_changed()` until it returns `None`, accepting one group per call, rather than looping over `wanted()`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take in their own tasks, since each now waits on its own readers.
+
 ### Socket sessions
 
 moq.socket now exposes TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change.

py/moq-rs/moq/publish.py

@@ -208,11 +208,13 @@ def abort(self, error_code: int) -> None:
 class Fill:
     """A run of groups readers miss, for a :class:`TrackRequests` to serve.
 
-    Call :meth:`accept` for each group you have and :meth:`drop_groups` for the rest;
-    :meth:`wanted` names what is still owed, in the fill's order. Dropping the fill names
+    Call :meth:`accept` for each group you have and :meth:`drop_groups` for the rest, one
+    group at a time as :meth:`wanted_changed` yields them; :meth:`wanted` names what is still
+    owed, in the fill's order. Dropping the fill names
     whatever is left as dropped while a reader still wants it; once nobody does, it is
     withdrawn, so stop serving it once :meth:`wanted` is ``None``. Stop serving a single group
-    when :meth:`unwanted` returns for it, and move on to :meth:`wanted`.
+    when :meth:`unwanted` returns for it, and move on to :meth:`wanted`. Serve the fills you
+    take concurrently: each waits on its own readers.
     """
 
     def __init__(self, inner: MoqFill) -> None:
@@ -251,12 +253,18 @@ def order(self) -> MoqOrder:
     def wanted(self) -> int | None:
         """The next group still owed in the fill's order.
 
-        ``None`` once every group is answered or no reader waits on the fill.
+        ``None`` once every group is answered or no reader waits on the fill. It does not wait
+        for the readers; :meth:`wanted_changed` does.
         """
         return self._inner.wanted()
 
     async def wanted_changed(self) -> int | None:
-        """Wait for ``wanted`` to change and return it. The first call returns at once."""
+        """The next group to serve, once every group this fill accepted reached a reader.
+
+        Returns ``wanted`` once it changed and a reader picked up what was served, as it next
+        reads. ``None``, once the fill is done, never waits; the first call returns at once
+        unless an accepted group has yet to reach a reader.
+        """
         return await self._inner.wanted_changed()
 
     def accept(self, sequence: int) -> GroupProducer:

quest/m2/unbounded-groups/README.md

@@ -33,7 +33,9 @@ flights:
   prefix is transferred once per flight. Carrying the inflater's state from the
   cached copy, and keeping refresh or `Evicted` for compressed groups, were
   rejected.
-- Paced batch fills are [their own quest](/quest/m2/unbounded-groups/paced-batch-fills.md).
+- Paced batch fills were their own quest, landed in
+  [moq-dev/moq#8017](https://github.com/moq-dev/moq/pull/8017): `Fill::wanted_changed`
+  yields the next group only once a reader picked up the one before.
 - No cap on concurrent flights per group: memory is bounded by each flight's
   tail and pool debt, and the cost of more flights is upstream transfer.
   Revisit with relay evidence.
@@ -178,7 +180,6 @@ the model layer (maintainer, 2026-10-30).
 
 ## Required
 
-- [Paced batch fills](/quest/m2/unbounded-groups/paced-batch-fills.md) - fills that write a batch of groups stop evicting their own earlier groups before the reader takes them
 - [Byte holes](/quest/m2/unbounded-groups/byte-holes.md) - segments inside a frame shed like front frames and flights start at `(frame, byte)` on the current lite wip; the frame cap goes
 - [JS byte holes](/quest/m2/unbounded-groups/js-byte-holes.md) - `@moq/net` mirrors the byte-offset wire and continues a frame cut off mid-way
 - [Compressed frames have no size cap and inflate as a bounded stream](/quest/m2/unbounded-groups/compressed-stream.md) - A compressed frame of any size publishes and is received in Rust and `@moq/net`, like a plain one

rs/moq-archive/src/consumer/track.rs

@@ -154,17 +154,13 @@ impl Live {
 	}
 }
 
-/// Serve `fill` from the archive, a group at a time in the fill's order, until every group is
-/// answered or nobody waits on it.
-pub(super) async fn serve<S: ObjectStore>(shared: Arc<Shared<S>>, index: Arc<Mutex<Index>>, fill: Fill) {
+/// Serve `fill` from the archive, a group at a time in the fill's order as its readers pick
+/// them up, until every group is answered or nobody waits on it.
+pub(super) async fn serve<S: ObjectStore>(shared: Arc<Shared<S>>, index: Arc<Mutex<Index>>, mut fill: Fill) {
 	if let Some(pts) = fill.time() {
 		return look_up(&shared, &index, &fill, pts).await;
 	}
-	let mut last = None;
-	while let Some(sequence) = fill.wanted()
-		&& last != Some(sequence)
-	{
-		last = Some(sequence);
+	while let Some(sequence) = fill.wanted_changed().await {
 		serve_group(&shared, &index, &fill, sequence).await;
 	}
 }

rs/moq-c/src/publish.rs

@@ -957,14 +957,11 @@ impl Publish {
 		}
 	}
 
-	/// Hand `fill` to C a group at a time, each once the one before it is answered or nobody
-	/// reads it any more, until every group is answered or nobody waits on the fill.
+	/// Hand `fill` to C a group at a time, each once the one before it reached a reader or
+	/// nobody reads it any more, until every group is answered or nobody waits on the fill.
 	async fn serve_fill(callback: OnStatus, fill: Arc<moq_net::track::Fill>) {
-		let mut last = None;
-		while let Some(sequence) = fill.wanted()
-			&& last != Some(sequence)
-		{
-			last = Some(sequence);
+		let mut changes = (*fill).clone();
+		while let Some(sequence) = changes.wanted_changed().await {
 			let (pending, answered) = oneshot::channel();
 			let request = GroupRequest {
 				fill: fill.clone(),

rs/moq-ffi/MIGRATE.md

@@ -5,6 +5,10 @@ These are the UniFFI names; each language wrapper has its own `MIGRATE.md` in it
 
 ## Unreleased
 
+### `MoqFill::wanted_changed` waits for a reader to pick up the group served before
+
+`MoqFill::wanted_changed` now returns the next group only once every group the fill accepted reached a reader, which picks it up when it next reads, so a handler writes a batch one group at a time at its fastest reader's pace. Serve a fill by looping on `wanted_changed` until it returns null, accepting one group per call, rather than looping over `wanted`, which still returns at once: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read. Serve the fills you take concurrently, since each now waits on its own readers.
+
 ### Socket sessions
 
 MoqSocketEndpoint, MoqSocketIncomingTcp, MoqSocketIncomingTrack, MoqSocketTcpStream, and MoqSocketTrackStream now expose TCP group halves and reliably subscribed track halves beside origin. Regenerate bindings and rebuild the matching native library together. Existing callers need no migration. Socket endpoints require lite-07 or moq-transport on every hop; there is no wire change. Dropping an unfinished group producer now resets it with `Unroutable` instead of leaking Rust's programming-error `Dropped`; finished groups still end cleanly.

rs/moq-ffi/src/producer.rs

@@ -478,10 +478,12 @@ impl MoqTrackRequests {
 /// A run of groups readers miss, mirroring [`moq_net::track::Fill`].
 ///
 /// Call [`accept`](Self::accept) for each group you have and [`drop_groups`](Self::drop_groups)
-/// for the rest; [`wanted`](Self::wanted) names what is still owed, in the fill's order. Releasing
+/// for the rest, one group at a time as [`wanted_changed`](Self::wanted_changed) yields them;
+/// [`wanted`](Self::wanted) names what is still owed, in the fill's order. Releasing
 /// it names whatever is left as dropped while a reader still wants it; once nobody does, it is
 /// withdrawn, so stop serving it once [`wanted`](Self::wanted) is null. Stop serving a single
-/// group when [`unwanted`](Self::unwanted) resolves for it, and move on to the next.
+/// group when [`unwanted`](Self::unwanted) resolves for it, and move on to the next. Serve the
+/// fills you take concurrently: each waits on its own readers.
 #[derive(uniffi::Object)]
 pub struct MoqFill {
 	inner: moq_net::track::Fill,
@@ -530,12 +532,15 @@ impl MoqFill {
 	}
 
 	/// The next group still owed in the fill's order, or null once every group is answered or
-	/// no reader waits on the fill.
+	/// no reader waits on the fill. It does not wait for the readers; `wanted_changed` does.
 	pub fn wanted(&self) -> Option<u64> {
 		self.inner.wanted()
 	}
 
-	/// Wait for `wanted` to change and return the new value. The first call returns at once.
+	/// The next group to serve: `wanted` once it changed and every group this fill accepted
+	/// reached a reader, which picks it up when it next reads. Null, once the fill is done,
+	/// never waits; the first call returns at once unless an accepted group has yet to reach a
+	/// reader. Serve one group per call, so the fill writes no faster than its readers read.
 	pub async fn wanted_changed(&self) -> Result<Option<u64>, MoqError> {
 		// Polled in place: a cancelled call left unfreed must not advance the
 		// watcher and swallow the change the next call waits for.

rs/moq-net/MIGRATE.md

@@ -4,6 +4,13 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### `track::Fill::wanted_changed` waits for a reader to pick up the group served before
+
+`Fill::wanted_changed` (and `poll_wanted_changed`) now yields the next group only once every group the fill accepted reached a reader, which picks it up as it asks for its next group, so a handler writes a batch one group at a time at its fastest reader's pace.
+Serve a fill with `while let Some(sequence) = fill.wanted_changed().await` rather than looping over `Fill::wanted`, which still names the next group owed without waiting: a batch written ahead of its readers lets a bounded cache evict its earlier groups before they are read.
+Serve the fills you take concurrently: each waits on its own readers, so serving them in turn holds every fill behind the readers of the one before.
+Don't block that loop on `group::Producer::ready` while writing a group: a reader reading in another order than the fill's picks the group up but reads none of it until its older groups arrive.
+
 ### Snapshots split connected sessions into idle and active
 
 `stats::Snapshot::activity()` returns each tier's `stats::Activity`: connected sessions with

rs/moq-net/src/lite/subscriber.rs

@@ -11113,6 +11113,8 @@ struct FillServeRun {
 	fill: track::Fill,
 	info: FetchInfo,
 	run: Option<(u64, FetchServeRun)>,
+	/// The group the last run answered.
+	last: Option<u64>,
 }
 
 impl Drop for FillServeRun {
@@ -11132,6 +11134,7 @@ impl FillServeRun {
 			fill,
 			info,
 			run: None,
+			last: None,
 		}
 	}
 }
@@ -11141,25 +11144,24 @@ impl kio::Task for FillServeRun {
 
 	fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> {
 		loop {
-			let last = match &mut self.run {
-				Some((sequence, run)) => {
-					ready!(run.poll(waiter));
-					let sequence = *sequence;
-					self.run = None;
-					Some(sequence)
-				}
-				None => None,
-			};
+			if let Some((sequence, run)) = &mut self.run {
+				ready!(run.poll(waiter));
+				self.last = Some(*sequence);
+				self.run = None;
+			}
 			// Withdrawn: nobody waits on what is left, so it records nothing.
 			if self.fill.poll_withdrawn(waiter).is_ready() {
 				return Poll::Ready(());
 			}
+			// The next fetch waits for a reader to pick up the group the last one brought, so a
+			// fill wider than the cache never evicts its own groups before they are read.
+			ready!(self.fill.poll_paced(waiter));
 			let Some(sequence) = self.fill.wanted() else {
 				return Poll::Ready(());
 			};
 			// Every run answers its group, so the same one coming back is a bug; ending the
 			// fill names it dropped rather than fetching it forever.
-			if last == Some(sequence) {
+			if self.last == Some(sequence) {
 				debug_assert!(false, "fill group {sequence} left unanswered");
 				return Poll::Ready(());
 			}