fix(quic): release stream table and control queue capacity at rest

quest/m1/README.md

@@ -203,7 +203,6 @@ per-frame CPU growth since #5926.
 - [A returning route serves past the cap its old copy was given](/quest/m1/returning-route-cap.md) - when a resumed reader's route returns, the reused serving copy asks with the reader's current resolution, not the capped resolution it was left with when it was replaced
 - [The resume model passes its long run again](/quest/m1/resume-model-7542.md) - fix #7542's regression that drops groups a long-budget reader can still get
 - [A resume more than one group behind never hangs](/quest/m1/ietf-resume-behind.md) - a resumed read gets every group or a drop on every wire version
-- [moq-quic's idle connections keep no burst-sized state](/quest/m1/quic-readable-dedupe.md) - readable events are deduplicated and a quiet connection keeps no per-peak-stream bytes
 - [Stats name the route instance's epoch with a type](/quest/m1/stats-epoch-type.md) - `GroupFrame::epoch` gets a dedicated type and tests stop mislabelling the instance
 
 ## Related

rs/moq-quic/README.md

@@ -97,7 +97,7 @@ Changes on top of the upstream commit, besides the renames:
 - **Namespaced streams** (`send`, `recv`, `stream`): quinn's `SendStream`, `RecvStream`, and their error zoo become `send::Stream` and `recv::Stream` with one `send::Error` (streams and datagrams) and one `recv::Error` (reads and leases), adding `Blocked`, and `Closed` for an ID the stream no longer has state for. `send::Stream::write` takes `&mut impl Buf`, splitting a `Bytes` source off without a copy, in place of `write` and `write_chunks`, and `Datagrams::send` takes `impl Buf` the same way, so a chained header costs no copy. An outgoing stream's `send::End` (finished and acknowledged, stopped, or reset) stays queryable until `send::Stream::release`, so a driver needs no table of its own for how a stream ended. A `send::Offset` comes only from `offset()` and is tagged to its stream, so `reset_at` and `acked` cannot be handed one past the data written. A port touching a public stream method has to be redone against these names.
 - **Connection limits** (`Limits`, `Connection::set_limits`, and the drivers' `set_limits`): one validated call changes a live connection's stream limit and windows, refusing values a peer would reject and changing nothing then. moq-tokio and moq-uring re-export the type, so both runtimes check limits the same way. Not offered upstream: it only bundles quinn's existing setters.
 - **Indexed send buffer** (`SendBuffer::get`): each retained write keeps its stream offset, and a STREAM frame's data comes from one binary search, instead of quinn walking every write from the oldest unacknowledged one for each piece, which costs a triangular number of visits when many small writes are in flight. Not offered upstream yet.
-- **No receive-state recycling** (`StreamsState::stream_recv_freed`): a completed stream's `Recv` is dropped, where quinn keeps it in `free_recv` with its chunk list at peak capacity for the next stream. Recycled state stayed allocated per peak concurrent stream for the connection's lifetime: about 550 B per stream, and about 113 KiB for one that once buffered a 2,048-packet backlog. Reusing it saved only 0.1 to 0.3% of a burst's instructions. `benches/receive_bursts.rs` measures both. A cherry-pick touching `StreamRecv`, `free_recv`, or `reinit` has nothing to apply to. Not offered upstream yet.
+- **No receive-state recycling** (`StreamsState::stream_recv_freed`): a completed stream's `Recv` is dropped, where quinn keeps it in `free_recv` with its chunk list at peak capacity for the next stream. Recycled state stayed allocated per peak concurrent stream for the connection's lifetime: about 550 B per stream, and about 113 KiB for one that once buffered a 2,048-packet backlog. Reusing it saved only 0.1 to 0.3% of a burst's instructions. `benches/receive_bursts.rs` measures both. Stream ID hash tables and waiter tables contract geometrically as streams go away and release every slot when empty; drained control queues and obsolete credit updates are dropped when nothing remains to send. A surviving stream cannot pin a burst's hash-table capacity. A cherry-pick touching `StreamRecv`, `free_recv`, or `reinit` has nothing to apply to. Not offered upstream yet.
 - **Bandwidth probing** (`Connection::set_probe_target`, `connection/bandwidth_probe.rs`): an application-limited sender below its target fills the headroom BBR3 opens, in Startup and in one ProbeBW\_UP per probe wait, with duplicates of STREAM data still in flight, paced to the target, so the controller's samples measure the path rather than the application. That inflight bounds an attempt like a window once it sends its first duplicate; media alone past it, as beside another flow's standing queue, leaves the sender application-limited. `Controller::probes` and `Metrics::probe_inflight` tell the sender when and how far it may fill; CUBIC never does. A duplicate is never retransmitted, a reset or stopped stream's data is never resent, and a loss retransmits only the ranges no copy delivered (`SendBuffer::retransmit`). An attempt that raises nothing for two round trips, or ends short of the target, sits out 3 s and then 4 s before retrying the same target; reaching the target or a new one retries at once. `PathStats::sent_bandwidth_probe_bytes` counts the redundant bytes. A cherry-pick touching `poll_transmit`'s space loop or `populate_packet` has to keep the probe after the ordinary frames. Not offered upstream.
 - **Exact datagram overhead** (`Datagram::max_payload`): `Datagrams::max_size` subtracts the frame type and the length's actual varint size from the path budget and the peer's `max_datagram_frame_size`, instead of `Datagram::SIZE_BOUND`'s worst-case 8-byte length. RFC 9221 section 3 counts the frame type and length in the limit, so a peer advertising 1200 bytes (quicly does) accepts a 1197-byte payload, where reserving 8 bytes allows only 1191. quinn has the same behaviour; not offered upstream yet. A limit too small for even an empty frame, including RFC 9221's 0 for no support, makes `max_size` return `None` instead of accepting an empty payload.
 - **Waits** (`connection/waits.rs`): every wait is a `poll_*` taking the caller's `&kio::Waiter`, so a driver keeps no waiter tables of its own: `recv::Chunks::poll_next`, `recv::Stream::poll_lease` and `poll_received_reset`, `send::Stream::poll_write`, `poll_end`, and `poll_acked`, `Streams::poll_accept` and `poll_open`, `Datagrams::poll_recv`, and `Connection::poll_handshake_data`, `poll_connected`, `poll_handshake_confirmed`, and `poll_closed`, over QUIC and QMux alike, with `Connection::error` for why it was lost. quinn's `StreamEvent` and the `Event` variants a handle waited on are gone (`Event` keeps `PathUpdated`, and QMux has no events): the stream engine wakes each list where quinn queued the matching event, and wakes connection-blocked writers as soon as the window has room (`StreamsState::unblock`), so a port touching one of quinn's event sites has to wake the matching list instead. A lost connection wakes every waiter, and each wait drains what arrived (accepted streams, readable bytes, datagrams) before it fails, a stream wait with `ConnectionLost`. Not offered upstream: quinn keeps its wakers in the async layer.

rs/moq-quic/benches/receive_bursts.rs

@@ -13,7 +13,8 @@
 //! `server_` columns are what dropping the server's connections frees: when new, when quiet after
 //! one burst, and when quiet after the resume and a repeat burst. `kept_b` is what quiet holds over
 //! new, per stream of the burst. The CPU columns time the server; the allocation counts include the
-//! clients', which a receive change leaves alone. `teardown_b` is what the case leaks.
+//! clients'. `quiet_delta_b` and `repeat_delta_b` keep the exact difference from cold, so table
+//! rounding cannot hide a small retained allocation. `teardown_b` is what the case leaks.
 //!
 //! Criterion then times the server's side of a burst on a connection that already freed the
 //! previous one's streams, so the cost of allocating fresh receive state for each stream shows.
@@ -267,12 +268,12 @@ fn server_heap(mut pair: Pair) -> isize {
 
 fn table() {
 	println!(
-		"mode conns streams packets peak_kib server_cold_kib server_quiet_kib kept_b resume_allocs resume_cpu_us repeat_allocs repeat_cpu_us server_repeat_kib teardown_b rss_kib"
+		"mode conns streams packets peak_kib server_cold_kib server_quiet_kib kept_b resume_allocs resume_cpu_us repeat_allocs repeat_cpu_us server_repeat_kib quiet_delta_b repeat_delta_b teardown_b rss_kib"
 	);
 	let kib = |bytes: isize| bytes / 1024;
 	for mode in Mode::ALL {
 		for conns in [1, 4] {
-			for streams in [16, 256, 1024] {
+			for streams in [1, 16, 256, 1024] {
 				for packets in [1, 16, 256, 2048] {
 					if conns * streams * packets > BURST_PACKETS {
 						continue;
@@ -299,14 +300,16 @@ fn table() {
 
 					let kept = (quiet - cold) / (conns * streams) as isize;
 					println!(
-						"{} {conns} {streams} {packets} {} {} {} {kept} {resume_allocs} {} {repeat_allocs} {} {} {teardown} {}",
+						"{} {conns} {streams} {packets} {} {} {} {kept} {resume_allocs} {} {repeat_allocs} {} {} {} {} {teardown} {}",
 						mode.name(),
 						kib(peak),
 						kib(cold),
 						kib(quiet),
 						resume_cpu.as_micros(),
 						repeat_cpu.as_micros(),
 						kib(repeat),
+						quiet - cold,
+						repeat - cold,
 						rss_kib(),
 					);
 				}
@@ -319,8 +322,8 @@ fn table() {
 fn bench_bursts(c: &mut Criterion<ThreadCpu>) {
 	let mut group = c.benchmark_group("receive_bursts");
 	for mode in Mode::ALL {
-		for streams in [16, 256] {
-			for packets in [16, 256] {
+		for streams in [1, 16, 256] {
+			for packets in [1, 16, 256] {
 				let mut pair = None;
 				group.throughput(Throughput::Bytes((streams * packets * PACKET) as u64));
 				group.bench_function(

rs/moq-quic/src/connection/mod.rs

@@ -1269,7 +1269,7 @@ impl Connection {
 	}
 
 	/// Indicate what types of frames are ready to send for the given space
-	fn space_can_send(&self, space_id: SpaceId, frame_space_1rtt: usize) -> SendableFrames {
+	fn space_can_send(&mut self, space_id: SpaceId, frame_space_1rtt: usize) -> SendableFrames {
 		if self.spaces[space_id].crypto.is_none() {
 			// No keys available for this space
 			return SendableFrames::empty();

rs/moq-quic/src/connection/spaces.rs

@@ -168,9 +168,15 @@ impl PacketSpace {
 		x
 	}
 
-	pub(super) fn can_send(&self, streams: &StreamsState) -> SendableFrames {
+	pub(super) fn can_send(&mut self, streams: &StreamsState) -> SendableFrames {
 		let acks = self.pending_acks.can_send();
-		let other = !self.pending.is_empty(streams) || self.ping_pending || self.immediate_ack_pending;
+		let pending = !self.pending.is_empty(streams);
+		if !pending {
+			// Drop both drained queue capacity and obsolete per-stream credit updates. Merely
+			// testing them as unsendable would retain their slots until more traffic arrives.
+			self.pending = Retransmits::default();
+		}
+		let other = pending || self.ping_pending || self.immediate_ack_pending;
 
 		SendableFrames { acks, other }
 	}

rs/moq-quic/src/connection/streams/send.rs

@@ -460,6 +460,9 @@ impl<'a> Stream<'a> {
 	/// kept for the connection's life.
 	pub fn release(self) {
 		if self.state.ends.remove(&self.id).is_some() {
+			if self.state.ends.capacity() > self.state.ends.len().saturating_mul(4) {
+				self.state.ends.shrink_to_fit();
+			}
 			return;
 		}
 		let max_data = self.state.max_send_data(self.id);

rs/moq-quic/src/connection/streams/state.rs

@@ -672,6 +672,9 @@ impl StreamsState {
 			buf.write_var(limit);
 			stats.stream_data_blocked += 1;
 		}
+		if pending.is_empty(self) {
+			*pending = Retransmits::default();
+		}
 	}
 
 	/// Write STREAM and DATAGRAM frames in schedule order until the packet is full.
@@ -1284,6 +1287,21 @@ impl StreamsState {
 		if half == StreamHalf::Send {
 			self.connection_blocked.swap_remove(&id);
 			self.send_streams -= 1;
+			if self.send.capacity() > self.send.len().saturating_mul(4) {
+				self.send.shrink_to_fit();
+			}
+			if self.connection_blocked.capacity() > self.connection_blocked.len().saturating_mul(4) {
+				self.connection_blocked.shrink_to_fit();
+			}
+		} else {
+			// Contract geometrically, so a surviving stream cannot pin a burst's empty slots and
+			// removing a burst costs linear work overall. Empty tables release every slot.
+			if self.recv.capacity() > self.recv.len().saturating_mul(4) {
+				self.recv.shrink_to_fit();
+			}
+			if self.unlooked.capacity() > self.unlooked.len().saturating_mul(4) {
+				self.unlooked.shrink_to_fit();
+			}
 		}
 		self.waits.freed(id);
 	}
@@ -1350,6 +1368,158 @@ mod tests {
 		)
 	}
 
+	/// Notifications coalesce while a reader is parked, and freed streams leave no table storage.
+	#[test]
+	fn receive_bursts_release_stream_storage() {
+		for dir in [Dir::Uni, Dir::Bi] {
+			for streams in [1, 16, 128] {
+				for packets in [1, 16, 256] {
+					let mut server = make(Side::Server);
+					let mut pending = Retransmits::default();
+					for round in 0..3 {
+						let readers: Vec<_> = (0..streams).map(|_| Counter::new()).collect();
+						let ids: Vec<_> = (0..streams)
+							.map(|i| StreamId::new(Side::Client, dir, round * streams + i))
+							.collect();
+						for (&id, reader) in ids.iter().zip(&readers) {
+							data(&mut server, id, 0, 0, false);
+							server.waits.register(id, Wait::Read, reader);
+							assert_eq!(Streams::new(&mut server, &ConnState::Established).accept(dir), Some(id));
+						}
+						for packet in 0..packets {
+							for &id in &ids {
+								data(&mut server, id, packet, 1, false);
+							}
+						}
+						assert!(readers.iter().all(|reader| reader.wakes() == 1));
+						assert_eq!(server.waits.stream_entries(), 0);
+						assert_eq!(server.unlooked.len(), streams as usize);
+						for id in ids {
+							data(&mut server, id, packets, 0, true);
+							// Read, cancel, and observe a reset on successive bursts of one connection.
+							if round == 0 {
+								let mut stream = recv(&mut server, &mut pending, id);
+								let mut chunks = stream.read().unwrap();
+								let mut bytes = 0;
+								while let Some(chunk) = chunks.next(usize::MAX).unwrap() {
+									bytes += chunk.bytes.len();
+								}
+								assert_eq!(bytes as u64, packets);
+								let _ = chunks.finalize();
+							} else if round == 1 {
+								recv(&mut server, &mut pending, id).stop(0u32.into()).unwrap();
+							} else {
+								let _ = server.received_reset(reset_at(id, 7, packets as u32, None)).unwrap();
+								assert_eq!(
+									recv(&mut server, &mut pending, id).received_reset(),
+									Ok(Some(7u32.into()))
+								);
+							}
+							assert!(server.recv.capacity() <= server.recv.len() * 4);
+							assert!(server.unlooked.capacity() <= server.unlooked.len() * 4);
+							if dir == Dir::Bi {
+								let mut stream = send::Stream {
+									id,
+									state: &mut server,
+									pending: &mut pending,
+									conn_state: &ConnState::Established,
+								};
+								stream.finish().unwrap();
+								stream.release();
+								server.received_ack_of(frame::StreamMeta {
+									id,
+									offsets: 0..0,
+									fin: true,
+								});
+								assert!(server.send.capacity() <= server.send.len() * 4);
+							}
+						}
+						assert_eq!(server.recv.capacity(), 0, "receive slots outlived their streams");
+						assert_eq!(server.send.capacity(), 0, "send slots outlived their streams");
+						assert_eq!(server.unlooked.capacity(), 0, "unread index outlived its streams");
+					}
+				}
+			}
+		}
+	}
+
+	#[test]
+	fn control_bursts_release_stream_storage() {
+		for streams in [1, 16, 128] {
+			let mut server = make(Side::Server);
+			server.set_params(&TransportParameters {
+				initial_max_streams_uni: (streams as u32).into(),
+				initial_max_data: 1024u32.into(),
+				..TransportParameters::default()
+			});
+			server.initial_stream_receive_window = 1;
+			server.stream_receive_window = 1;
+			let mut pending = Retransmits::default();
+			let mut sent = Vec::new();
+			for index in 0..streams {
+				let id = Streams::new(&mut server, &ConnState::Established)
+					.open(Dir::Uni, Default::default())
+					.unwrap();
+				let mut stream = send::Stream {
+					id,
+					state: &mut server,
+					pending: &mut pending,
+					conn_state: &ConnState::Established,
+				};
+				assert_eq!(stream.write(&mut &b"x"[..]), Err(send::Error::Blocked));
+				stream.reset(0u32.into()).unwrap();
+				sent.push(id);
+
+				let id = StreamId::new(Side::Client, Dir::Uni, index);
+				data(&mut server, id, 0, 1, false);
+				let mut stream = recv(&mut server, &mut pending, id);
+				let mut chunks = stream.read().unwrap();
+				assert_eq!(chunks.next(1).unwrap().unwrap().bytes.len(), 1);
+				assert!(matches!(chunks.next(1), Err(recv::Error::Blocked)));
+				let _ = chunks.finalize();
+				assert!(pending.max_stream_data.contains(&id));
+				recv(&mut server, &mut pending, id).stop(0u32.into()).unwrap();
+				data(&mut server, id, 1, 0, true);
+			}
+			// Control frames retain their obligations until sent, but no capacity after draining.
+			let mut stats = FrameStats::default();
+			while !pending.is_empty(&server) {
+				let mut written = ThinRetransmits::default();
+				server.write_control_frames(&mut Vec::new(), &mut pending, &mut written, &mut stats, 128);
+				if let Some(written) = written.get() {
+					for &(id, size) in &written.reset_stream {
+						server.reset_acked(id, size);
+					}
+				}
+			}
+			// QMux also serializes an idle batch, which discards obsolete credit updates.
+			server.write_control_frames(
+				&mut Vec::new(),
+				&mut pending,
+				&mut ThinRetransmits::default(),
+				&mut stats,
+				128,
+			);
+			for id in sent {
+				send::Stream {
+					id,
+					state: &mut server,
+					pending: &mut pending,
+					conn_state: &ConnState::Established,
+				}
+				.release();
+				assert!(server.ends.capacity() <= server.ends.len() * 4);
+			}
+			assert_eq!(stats.reset_stream, streams);
+			assert_eq!(stats.stop_sending, streams);
+			assert_eq!(server.ends.capacity(), 0);
+			assert_eq!(pending.reset_stream.capacity(), 0);
+			assert_eq!(pending.stop_sending.capacity(), 0);
+			assert_eq!(pending.max_stream_data.capacity(), 0);
+			assert_eq!(pending.stream_data_blocked.capacity(), 0);
+		}
+	}
+
 	#[test]
 	fn connection_blocked_reset_churn_is_bounded() {
 		let mut server = make(Side::Server);