fix(net): reuse compressed group encoders per track

quest/m1/fuzz-findings/README.md

@@ -1,4 +1,4 @@
-# Media fuzz findings
+# [S] Media fuzz findings
 
 ## Goal
 
@@ -16,9 +16,9 @@ maintainer). Its short local runs found the bugs the children fix; the
 minimized inputs are in [its PR comment](https://github.com/moq-dev/moq/pull/7313#issuecomment-6520535160).
 
 Each child commits its input under `regressions/` with its fix, since an
-input committed before the fix would fail the merge gate. After the last
-child, run every target nightly for a few sessions; a new finding becomes a
-child here.
+input committed before the fix would fail the merge gate. Every known finding
+is now fixed: run every target nightly for a few sessions; a new finding
+becomes a child here.
 
 For the crates outside the `kixelated` and `moq-dev` orgs, moq-mux fixes
 the bug on its own side and the bug is reported upstream (decided
@@ -33,7 +33,3 @@ maintainer's approval.
 webm-iterable's block parsing with moq-mux's own block-header reader, which
 fixed the MKV block-header panic; its input is
 `rs/moq-mux/fuzz/regressions/mkv/short-block-header`.
-
-## Required
-
-- [A catalog update reuses its encoder](/quest/m1/fuzz-findings/catalog-encoder.md) - republishing a compressed catalog or HLS group stops allocating a fresh DEFLATE encoder (about 200 KB) each time

rs/moq-net/benches/group.rs

@@ -30,6 +30,9 @@
 //! reader, as a paced ingest runs, swept over readers: each frame read reports progress,
 //! and the fastest reader's read wakes the parked writer.
 //!
+//! `group_compressed_updates` republishes single-frame compressed groups, swept over
+//! tracks and overlapping groups, to expose encoder allocation on the update path.
+//!
 //! Run with `cargo bench -p moq-net`.
 
 #![expect(
@@ -503,6 +506,56 @@ fn bench_inflate(c: &mut Criterion) {
 	g.finish();
 }
 
+/// Repeated single-frame compressed updates on independent tracks. Groups overlap
+/// within each update so only the one spare per track can be reused.
+fn bench_compressed_updates(c: &mut Criterion) {
+	const UPDATES: usize = 32;
+	let payload = Bytes::from(format!("{{\"status\":\"{}\"}}", "ready".repeat(100)));
+	let mut g = c.benchmark_group("group_compressed_updates");
+	for tracks in [1usize, 16] {
+		for concurrent in [1usize, 4] {
+			g.throughput(Throughput::Elements((UPDATES * tracks * concurrent) as u64));
+			g.bench_function(BenchmarkId::new(format!("tracks{tracks}"), concurrent), |b| {
+				b.iter_batched(
+					|| {
+						let broadcast = broadcast::Info::default().produce();
+						let producers: Vec<_> = (0..tracks)
+							.map(|i| {
+								let track = broadcast
+									.create_track(format!("bench{i}"), track::Info::default().with_compress(true))
+									.unwrap();
+								let mut group = track.append_group().unwrap();
+								group.write_frame(None, payload.clone()).unwrap();
+								group.finish().unwrap();
+								track
+							})
+							.collect();
+						(broadcast, producers, Vec::with_capacity(concurrent))
+					},
+					|(broadcast, producers, mut groups)| {
+						for _ in 0..UPDATES {
+							for track in &producers {
+								for _ in 0..concurrent {
+									let mut group = track.append_group().unwrap();
+									group.write_frame(None, payload.clone()).unwrap();
+									groups.push(group);
+								}
+								for group in &mut groups {
+									group.finish().unwrap();
+								}
+								groups.clear();
+							}
+						}
+						(broadcast, producers, groups)
+					},
+					BatchSize::SmallInput,
+				);
+			});
+		}
+	}
+	g.finish();
+}
+
 /// Worker threads the contended readers are spread over.
 const THREADS: [usize; 3] = [1, 4, 16];
 
@@ -952,6 +1005,7 @@ criterion_group!(
 	bench_stamp,
 	bench_relay_frames,
 	bench_inflate,
+	bench_compressed_updates,
 	bench_contended,
 	bench_streaming,
 	bench_shed,

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

@@ -47,7 +47,7 @@ use std::task::{Poll, ready};
 use std::time::Duration;
 
 use super::track::{self, Position, TrackState};
-use super::{aborts, expiry, group};
+use super::{aborts, expiry, flate, group};
 
 /// Fixed bookkeeping charged per cached group on top of its frame payload bytes.
 ///
@@ -719,6 +719,10 @@ pub(super) struct Track {
 	// What the frames this track's groups are still receiving may allocate up front,
 	// shared with every track minted from the same origin handle.
 	frames: super::frame::Budget,
+
+	// One finished local group's encoder, kept until another group needs it or the
+	// account drops. Raw codec only: an Extra here would keep this account alive.
+	encoder: Mutex<Option<flate::Encoder>>,
 }
 
 /// The greatest location a track has written or its source reported, followed by the track
@@ -825,6 +829,7 @@ impl Track {
 			aborts: OnceLock::new(),
 			largest: Default::default(),
 			declared: AtomicBool::new(false),
+			encoder: Mutex::default(),
 		});
 		if let Some(key) = track.pool.register(&track) {
 			let _ = track.sweep.set(key);
@@ -1018,6 +1023,15 @@ impl Track {
 
 impl Drop for Track {
 	fn drop(&mut self) {
+		if self
+			.encoder
+			.get_mut()
+			.unwrap_or_else(PoisonError::into_inner)
+			.take()
+			.is_some()
+		{
+			self.pool.sub(flate::ENCODER_HEAP);
+		}
 		if let Some(key) = self.sweep.get() {
 			self.pool.unregister(*key);
 		}
@@ -1288,6 +1302,34 @@ pub(super) struct Extra {
 }
 
 impl Extra {
+	/// Take this track's spare encoder with its charge, or bill a new one.
+	pub(super) fn take_encoder(&self) -> flate::Encoder {
+		if let Some(track) = &self.track {
+			let encoder = track.encoder.lock().unwrap_or_else(PoisonError::into_inner).take();
+			if let Some(mut encoder) = encoder {
+				// Moving existing heap is not a write and must not accrue eviction debt.
+				self.bytes.fetch_add(flate::ENCODER_HEAP, Ordering::Relaxed);
+				encoder.reset();
+				return encoder;
+			}
+		}
+		self.add(flate::ENCODER_HEAP);
+		flate::Encoder::new()
+	}
+
+	/// Park a finished encoder and its charge, dropping it if the track has a spare.
+	pub(super) fn park_encoder(&self, encoder: flate::Encoder) {
+		if let Some(track) = &self.track {
+			let mut spare = track.encoder.lock().unwrap_or_else(PoisonError::into_inner);
+			if spare.is_none() {
+				self.bytes.fetch_sub(flate::ENCODER_HEAP, Ordering::Relaxed);
+				*spare = Some(encoder);
+				return;
+			}
+		}
+		self.sub(flate::ENCODER_HEAP);
+	}
+
 	/// Bill `n` more bytes.
 	pub(super) fn add(&self, n: u64) {
 		if let Some(track) = &self.track {
@@ -1350,6 +1392,42 @@ mod test {
 		Pool::new(config)
 	}
 
+	/// Reuse transfers existing heap, concurrent users allocate separately, and only
+	/// one finished encoder stays parked until the account's final owner drops.
+	#[test]
+	fn encoder_spare_transfers_its_charge_without_write_debt() {
+		let pool = Pool::unbounded();
+		let charge = charge(&pool);
+		let first = charge.extra();
+		let second = charge.extra();
+		let account = charge.track.as_ref().unwrap();
+		account.take_written();
+		let encoder = first.take_encoder();
+		let concurrent = second.take_encoder();
+		assert_eq!(pool.used(), ENTRY_OVERHEAD + 2 * flate::ENCODER_HEAP);
+		assert_eq!(account.take_written(), 2 * flate::ENCODER_HEAP);
+
+		first.park_encoder(encoder);
+		assert_eq!(first.size(), 0);
+		assert_eq!(pool.used(), ENTRY_OVERHEAD + 2 * flate::ENCODER_HEAP);
+		let encoder = first.take_encoder();
+		assert_eq!(first.size(), flate::ENCODER_HEAP);
+		assert_eq!(account.take_written(), 0, "reuse must not inflate write debt");
+		first.park_encoder(encoder);
+		second.park_encoder(concurrent);
+		assert_eq!(second.size(), 0);
+		assert_eq!(pool.used(), ENTRY_OVERHEAD + flate::ENCODER_HEAP, "only one spare");
+
+		// A group and its inflated copy may outlive the track. They still own its
+		// account, so the parked encoder stays billed until that last owner drops.
+		drop(charge);
+		assert_eq!(pool.used(), flate::ENCODER_HEAP);
+		drop(first);
+		assert_eq!(pool.used(), flate::ENCODER_HEAP);
+		drop(second);
+		assert_eq!(pool.used(), 0);
+	}
+
 	struct Flag(std::sync::atomic::AtomicBool);
 
 	impl std::task::Wake for Flag {

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

@@ -26,9 +26,10 @@ const CHUNK: usize = 8 * 1024;
 /// The DEFLATE window, 2^15 bytes for raw DEFLATE at the default `windowBits`.
 const WINDOW: u64 = 1 << 15;
 
-/// Heap an [`Encoder`] holds, as zlib sizes it at the default `memLevel` of 8: the doubled
-/// window, its `prev` chain and hash `head` tables, and the pending buffer, 64 KiB each.
-pub(super) const ENCODER_HEAP: u64 = 8 * WINDOW;
+/// Heap an [`Encoder`] holds with zlib-rs at the default `memLevel` of 8: 64 KiB each
+/// for the doubled window, `prev` chain and pending buffer, a 128 KiB hash table and
+/// a separate 48 KiB symbol buffer, plus the codec state and alignment rounded to 16 KiB.
+pub(super) const ENCODER_HEAP: u64 = 12 * WINDOW;
 
 /// Heap a [`Decoder`] holds: the window, plus its inflate state rounded up to 8 KiB.
 pub(super) const DECODER_HEAP: u64 = WINDOW + 8 * 1024;
@@ -44,6 +45,11 @@ impl Encoder {
 		Self(Box::new(Compress::new(flate2::Compression::new(LEVEL), false)))
 	}
 
+	/// Start another group with a cold window, retaining the codec's allocations.
+	pub(super) fn reset(&mut self) {
+		self.0.reset();
+	}
+
 	/// Compress the next frame, returning its slice minus the sync-flush marker. Empty in
 	/// yields empty out, leaving the window untouched.
 	pub(super) fn frame(&mut self, payload: &[u8]) -> Bytes {