moq-net: bound incomplete messages with receive credit

doc/bin/relay/config.md

@@ -61,8 +61,8 @@ idle_timeout = "10s"                 # Drop a connection after this long with no
 keep_alive = "3s"                    # Ping interval; "0s" disables it. iroh still pings at least every 5s. Default.
 gso = true                           # UDP segmentation offload. iroh cannot turn it off.
 mtu_discovery = false                # Path MTU discovery. Default.
-receive_window = 67108864            # Whole-connection window, in bytes. Default (64 MiB).
-stream_receive_window = 8388608      # Per-stream window. Omit for the backend default.
+receive_window = 134217728           # Whole-connection window, in bytes. Default (128 MiB).
+stream_receive_window = 8388608      # Per-stream window. Omit to follow receive_window.
 send_window = 33554432               # Unacknowledged outgoing data. Omit for the backend default.
 qlog = "/var/log/moq/qlog"           # Existing directory. Needs the `qlog` build feature.
 ```
@@ -72,11 +72,17 @@ sessions, and so its [cluster routes](/bin/relay/cluster#failure-detection).
 QUIC uses the smaller of the two endpoints' values, so this also bounds the
 other end. Keep `keep_alive` under a third of it.
 
-Raise the receive windows when a fat, long path idles below the link rate. The
-64 MiB `receive_window` default carries several Gbps at a 100 ms RTT and bounds
-how much unread data one peer can make the relay buffer. Keep
-`stream_receive_window` well under it so one slow group cannot starve the
-connection.
+Raise `receive_window` when a fat, long path idles below the link rate. The
+128 MiB default carries several Gbps at a 100 ms RTT and bounds how much unread
+data and incomplete messages one peer can make the relay buffer. A message holds
+its share until it is complete, so the window must fit the largest one, just over
+64 MiB, and the relay refuses to start with less. When incomplete messages fill
+it, the relay cancels the least urgent stream and carries on. iroh returns credit
+as it reads, so there the window bounds only unread data, and incomplete messages
+are bounded by `max_streams` streams of up to the largest message each.
+
+Stream credit returns as data is read, so a `stream_receive_window` under
+`receive_window` only adds round trips to a large frame.
 
 ## \[session] and \[peer]
 

quest/m1/quic/README.md

@@ -63,7 +63,7 @@ cache limits remain separate; bitrate pacing composes with this memory cap.
 ## Required
 
 - [Regression net](/quest/m1/quic/coverage/README.md) - benchmarks, fuzzing, characterization tests, and a seeded simulation, so later QUIC changes see every behavior or cost change
-- [Bound message assembly](/quest/m1/quic/receive-memory.md) - complete headers and frames return credit, and receiver priority cancellation restores progress
+- [Prove receive memory](/quest/m1/quic/receive-memory-coverage.md) - a JavaScript sender exhausts both native receivers and they recover; nightly measures the cost
 - [A reset group keeps its header](/quest/m1/quic/group-reset.md) - a lite
   publisher resets a group with `RESET_STREAM_AT`, so its subscriber drops the
   tail grace

quest/m1/quic/peer-limits.md

@@ -35,9 +35,10 @@ its own.
 - Refuse a `peer` value below the client default rather than silently
   shrinking.
 
-The [receive-memory plan](/quest/m1/quic/receive-memory.md) requires client
-and peer memory allowances to fit the largest supported encoded unit,
-including framing. Preserve that validation and retained-credit accounting
+Message assembly holds connection credit until each unit is complete, so
+client and peer receive windows must fit the largest supported encoded unit,
+`moq_net::transport::MIN_RECEIVE_WINDOW`; the relay refuses a smaller
+`[quic]` window at config load. Preserve that validation and retained-credit accounting
 when applying runtime limits; wider stream counts must not create a second
 unbounded assembly pool.
 

quest/m1/quic/receive-memory-coverage.md

@@ -0,0 +1,39 @@
+# [M] Prove receive memory with JavaScript senders and benchmarks
+
+## Goal
+
+A JavaScript sender drives both native receivers into receive-credit
+exhaustion and they recover, and nightly measures what bounded receive memory
+costs. This is coverage for the message-assembly work in #5735, with no new
+transport behavior.
+
+## Plan
+
+Split out of #5735, which lands native credit holding, pressure recovery, and
+the minimum receive window with Rust-only end-to-end tests over tokio and
+io_uring.
+
+A JavaScript sender over the stock receivers cannot stall them. The minimum
+window fits the largest unit, and a strict-priority sender finishes one
+stream before the next, so every frame completes. The test therefore needs a
+receiver whose connection window is far below the minimum. Neither stock
+config allows that, so the test assembles a WebTransport (HTTP/3) server
+from each raw driver: tokio's `quic::driver` with `quic::web::Server`, and
+an io_uring endpoint in web mode. Keep it test-only; do not add a public way
+to configure a window below `MIN_RECEIVE_WINDOW`.
+
+The sender is a Bun script using `@moq/web-transport`, pinned to the test
+server's certificate hash with WebSocket disabled. It mirrors the Rust tests
+in `rs/moq-tokio/tests/integration/receive_credit.rs`. A low-priority track sends a frame
+larger than the window. A high-priority frame then has to complete, and a
+later low-priority group has to arrive after the low stream is stopped. Run
+it from Rust the way `rs/moq-tokio/tests/integration/max_age.rs` drives its Bun client,
+and wire it into the interop workflow, since normal CI has no Bun.
+
+The benchmark sweeps concurrent streams and frame sizes, and compares a
+finite stream window with one that follows the connection window. Report
+retained bytes (held plus unread credit), throughput, and frame completion
+latency. Sweep streams and frame sizes on separate axes, so a cost that grows
+with the stream count shows up as a slope. Wire it into `sh/gh/nightly.sh`
+next to `benchmark_held_reads`. Unit tests stay on mocked time; the benchmark
+is the only wall-clock part.

quest/m2/rate-quic.md

@@ -18,8 +18,8 @@ backpressure the peer cannot ignore, rather than metering and closing.
   so a modified client cannot exceed the cap. An honest one goes
   flow-control limited (its congestion controller does not back off) and
   queues until [the grant](/quest/m2/rate-grant.md) clamps its encoder.
-  Keep rate pacing separate from the connection memory allowance in
-  [message assembly](/quest/m1/quic/receive-memory.md). The memory allowance
+  Keep rate pacing separate from the connection memory allowance that
+  message assembly holds credit against. The memory allowance
   must fit the largest supported encoded unit even when it takes many rate
   intervals to receive. Limit each grant by both rate allowance and available
   memory backing; do not wait for a whole-message credit release to grant
@@ -90,4 +90,3 @@ that a one-second rate window can also be the complete assembly allowance.
 ## Required
 
 - [Peer limits](/quest/m1/quic/peer-limits.md) - adds the runtime `set_limits(Limits)` seam and shrink-as-debt behavior this extends
-- [Bound message assembly](/quest/m1/quic/receive-memory.md) - separate retained memory from paced outstanding credit

quest/m2/unbounded-groups/segmented-frames.md

@@ -22,8 +22,9 @@ See the [epic](/quest/m2/unbounded-groups/README.md) for the decisions.
 - Choose the segment size from benchmarks, and keep it a storage detail
   rather than a public knob. Coalescing the chunks the transport delivers is
   fine.
-- Update [Receive memory](/quest/m1/quic/receive-memory.md) so a frame's
-  payload releases credit per segment. The frame header stays a whole unit.
+- Change `moq-net`'s `coding::Reader`, which holds connection credit until a
+  whole frame is handed off (#5735), so a frame's payload releases credit per
+  segment. The frame header stays a whole unit.
 - Measure `group.rs`, `frame.rs` (`frame_recv`), and `memory.rs` before and
   after. Add a large-frame case swept over frame size.
 

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

@@ -107,6 +107,9 @@ impl<S: backend::Session<SendStream: 'static, RecvStream: 'static> + Clone + 'st
 	fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll<Self::Error> {
 		backend::Session::poll_closed(&mut self.0, cx).map(Error)
 	}
+	fn poll_credit_exhausted(&mut self, _: &mut Context<'_>) -> Poll<()> {
+		Poll::Pending
+	}
 	fn stats(&self) -> impl moq::Stats {
 		Stats(backend::Session::stats(&self.0))
 	}
@@ -141,20 +144,39 @@ impl<S: backend::SendStream + 'static> poll::SendStream for SendStream<S> {
 }
 /// An incoming WebTransport stream adapted to moq-net.
 pub struct RecvStream<S>(S);
+// The WebTransport backend returns connection credit as it reads, so there is none to hold.
 impl<S: backend::RecvStream + 'static> poll::RecvStream for RecvStream<S> {
 	type Error = Error<S::Error>;
-	fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll<Result<Option<usize>, Self::Error>> {
-		backend::RecvStream::poll_read(&mut self.0, cx, dst).map_err(Error)
+	type Credit = moq::Unheld;
+	type Stopper = ();
+	fn stopper(&self) {}
+	fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll<Result<Option<moq::Unheld>, Self::Error>> {
+		backend::RecvStream::poll_read(&mut self.0, cx, dst)
+			.map_err(Error)
+			.map_ok(|n| n.map(|n| moq::Unheld(n as u64)))
 	}
 	fn poll_read_buf<B: BufMut>(
 		&mut self,
 		cx: &mut Context<'_>,
 		buf: &mut B,
-	) -> Poll<Result<Option<usize>, Self::Error>> {
-		backend::RecvStream::poll_read_buf(&mut self.0, cx, buf).map_err(Error)
+	) -> Poll<Result<Option<moq::Unheld>, Self::Error>> {
+		backend::RecvStream::poll_read_buf(&mut self.0, cx, buf)
+			.map_err(Error)
+			.map_ok(|n| n.map(|n| moq::Unheld(n as u64)))
 	}
-	fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll<Result<Option<Bytes>, Self::Error>> {
-		backend::RecvStream::poll_read_chunk(&mut self.0, cx, max).map_err(Error)
+	fn poll_read_chunk(
+		&mut self,
+		cx: &mut Context<'_>,
+		max: usize,
+	) -> Poll<Result<Option<(Bytes, moq::Unheld)>, Self::Error>> {
+		backend::RecvStream::poll_read_chunk(&mut self.0, cx, max)
+			.map_err(Error)
+			.map_ok(|chunk| {
+				chunk.map(|b| {
+					let n = b.len() as u64;
+					(b, moq::Unheld(n))
+				})
+			})
 	}
 	fn stop(&mut self, code: u32) {
 		backend::RecvStream::stop(&mut self.0, code)
@@ -309,7 +331,7 @@ mod tests {
 		let Poll::Ready(Ok(Some(chunk))) = poll::RecvStream::poll_read_chunk(&mut recv, &mut cx, 2) else {
 			panic!("read failed")
 		};
-		assert_eq!(chunk, b"hi"[..]);
+		assert_eq!(chunk.0, b"hi"[..]);
 		poll::RecvStream::stop(&mut recv, 13);
 		assert_eq!(state.get(), 13);
 		let Poll::Ready(Err(error)) = poll::SendStream::poll_closed(&mut send, &mut cx) else {

rs/moq-net/src/client.rs

@@ -657,6 +657,10 @@ mod tests {
 			Poll::Ready(FakeError)
 		}
 
+		fn poll_credit_exhausted(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<()> {
+			std::task::Poll::Pending
+		}
+
 		fn stats(&self) -> impl crate::transport::Stats {
 			FakeStats {
 				send_rate: *self.state.send_rate.lock().unwrap(),
@@ -717,8 +721,15 @@ mod tests {
 
 	impl crate::transport::poll::RecvStream for FakeRecvStream {
 		type Error = FakeError;
+		type Credit = crate::transport::Unheld;
+		type Stopper = ();
+		fn stopper(&self) {}
 
-		fn poll_read(&mut self, _cx: &mut Context<'_>, dst: &mut [u8]) -> Poll<Result<Option<usize>, Self::Error>> {
+		fn poll_read(
+			&mut self,
+			_cx: &mut Context<'_>,
+			dst: &mut [u8],
+		) -> Poll<Result<Option<Self::Credit>, Self::Error>> {
 			if self.data.is_empty() {
 				return Poll::Ready(Ok(None));
 			}
@@ -727,7 +738,7 @@ mod tests {
 			for slot in dst.iter_mut().take(size) {
 				*slot = self.data.pop_front().unwrap();
 			}
-			Poll::Ready(Ok(Some(size)))
+			Poll::Ready(Ok(Some(crate::transport::Unheld(size as u64))))
 		}
 
 		fn stop(&mut self, _code: u32) {}
@@ -1056,6 +1067,10 @@ mod tests {
 			self.inner.poll_closed(cx)
 		}
 
+		fn poll_credit_exhausted(&mut self, _cx: &mut std::task::Context<'_>) -> std::task::Poll<()> {
+			std::task::Poll::Pending
+		}
+
 		fn stats(&self) -> impl crate::transport::Stats {
 			self.inner.stats()
 		}
@@ -1092,8 +1107,15 @@ mod tests {
 
 	impl crate::transport::poll::RecvStream for LocalRecv {
 		type Error = FakeError;
+		type Credit = crate::transport::Unheld;
+		type Stopper = ();
+		fn stopper(&self) {}
 
-		fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll<Result<Option<usize>, Self::Error>> {
+		fn poll_read(
+			&mut self,
+			cx: &mut Context<'_>,
+			dst: &mut [u8],
+		) -> Poll<Result<Option<Self::Credit>, Self::Error>> {
 			self.inner.poll_read(cx, dst)
 		}