fix(moq-net): read frames in place, without minting a per-poll FrameConsumer

rs/kio/src/producer.rs

@@ -111,6 +111,34 @@ impl<T> Producer<T> {
 		Poll::Pending
 	}
 
+	/// Poll-based **read-only** access with waker registration.
+	///
+	/// Like [`Self::poll`] but hands `f` a [`Ref`] instead of a [`Mut`], so it
+	/// never flags the state modified and never wakes consumers. Use it to wait on
+	/// a read condition (e.g. a `fin` flag) from the producer side without creating
+	/// a [`Consumer`] — creating/dropping a consumer churns the consumer count and
+	/// wakes the value waiters, which would spin a polling reader.
+	pub fn poll_ref<F, R>(&self, waiter: &Waiter, mut f: F) -> Poll<Result<R, Ref<'_, T>>>
+	where
+		F: FnMut(&Ref<'_, T>) -> Poll<R>,
+	{
+		let state = self.state.lock();
+		let state = Ref { state };
+
+		if let Poll::Ready(res) = f(&state) {
+			return Poll::Ready(Ok(res));
+		}
+
+		if state.state.closed {
+			return Poll::Ready(Err(state));
+		}
+
+		let mut state = state.state;
+		waiter.register(&mut state.waiters);
+
+		Poll::Pending
+	}
+
 	/// Wait for the closure to return [`Poll::Ready`], re-polling on each state change.
 	///
 	/// Returns `Ok(R)` when the closure returns [`Poll::Ready`], or `Err(Ref)` with

rs/moq-net/src/model/frame.rs

@@ -226,6 +226,35 @@ impl FrameProducer {
 		}
 	}
 
+	/// Poll for the frame's full payload, resolving once it's finished.
+	///
+	/// Reads from the producer side (via `kio::Producer::poll_ref`) so a polling
+	/// reader doesn't mint a transient [`FrameConsumer`] per poll: that would churn
+	/// the consumer count and wake the value waiters, spinning the poll. Always
+	/// returns the whole payload (offset 0), so it's safe to call from any number
+	/// of readers in parallel.
+	pub(crate) fn poll_read_all(&self, waiter: &kio::Waiter) -> Poll<Result<Bytes>> {
+		let res = ready!(self.state.poll_ref(waiter, |state| {
+			if state.fin {
+				Poll::Ready(Ok(()))
+			} else if let Some(err) = &state.abort {
+				Poll::Ready(Err(err.clone()))
+			} else {
+				Poll::Pending
+			}
+		}));
+
+		match res {
+			Ok(Ok(())) => {
+				// `fin` implies written == capacity (the producer fills the whole buffer).
+				let written = self.buf.written(Ordering::Acquire);
+				Poll::Ready(Ok(Bytes::from_owner(self.buf.clone()).slice(0..written)))
+			}
+			Ok(Err(err)) => Poll::Ready(Err(err)),
+			Err(state) => Poll::Ready(Err(state.abort.clone().unwrap_or(Error::Dropped))),
+		}
+	}
+
 	/// Block until there are no active consumers.
 	pub async fn unused(&self) -> Result<()> {
 		self.state

rs/moq-net/src/model/group.rs

@@ -104,6 +104,26 @@ impl GroupState {
 		}
 	}
 
+	/// Poll for the full payload of the frame at `index`, reading it in place.
+	///
+	/// Unlike [`Self::poll_get_frame`] this never mints a [`FrameConsumer`] (which
+	/// would churn the frame's consumer count and wake its waiters every poll); it
+	/// reads the cached [`FrameProducer`] directly. `waiter` is registered on the
+	/// frame's state so the reader wakes when it finishes.
+	fn poll_frame_read_all(&self, index: usize, waiter: &kio::Waiter) -> Poll<Result<Option<Bytes>>> {
+		if index < self.offset {
+			return Poll::Ready(Err(Error::CacheFull));
+		}
+		match self.frames.get(index - self.offset) {
+			Some(frame) => Poll::Ready(Ok(Some(ready!(frame.poll_read_all(waiter))?))),
+			None if self.fin => Poll::Ready(Ok(None)),
+			None => match &self.abort {
+				Some(err) => Poll::Ready(Err(err.clone())),
+				None => Poll::Pending,
+			},
+		}
+	}
+
 	fn poll_finished(&self) -> Poll<Result<u64>> {
 		if self.fin {
 			Poll::Ready(Ok((self.offset + self.frames.len()) as u64))
@@ -347,13 +367,12 @@ impl GroupConsumer {
 
 	/// Read the next frame's data all at once, without blocking.
 	pub fn poll_read_frame(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Bytes>>> {
-		let Some(mut frame) = ready!(self.poll(waiter, |state| state.poll_get_frame(self.index))?) else {
+		let index = self.index;
+		let Some(data) = ready!(self.poll(waiter, |state| state.poll_frame_read_all(index, waiter))?) else {
 			return Poll::Ready(Ok(None));
 		};
 
-		let data = ready!(frame.poll_read_all(waiter))?;
 		self.index += 1;
-
 		Poll::Ready(Ok(Some(data)))
 	}
 
@@ -364,14 +383,15 @@ impl GroupConsumer {
 
 	/// Read all of the chunks of the next frame, without blocking.
 	pub fn poll_read_frame_chunks(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Vec<Bytes>>>> {
-		let Some(mut frame) = ready!(self.poll(waiter, |state| state.poll_get_frame(self.index))?) else {
+		let index = self.index;
+		let Some(data) = ready!(self.poll(waiter, |state| state.poll_frame_read_all(index, waiter))?) else {
 			return Poll::Ready(Ok(None));
 		};
 
-		let data = ready!(frame.poll_read_all_chunks(waiter))?;
 		self.index += 1;
-
-		Poll::Ready(Ok(Some(data)))
+		// In-place reads return the whole frame as one slice; keep the chunked API
+		// shape (empty payload -> no chunks).
+		Poll::Ready(Ok(Some(if data.is_empty() { Vec::new() } else { vec![data] })))
 	}
 
 	/// Read all of the chunks of the next frame.