feat(moq-net): cache compressed frame payloads, decode at delivery

Compressible tracks (TrackInfo.compress) are Deflate-coded on the lite-05+
wire. The model previously inflated every frame on ingress and re-deflated it
per egress connection, so a relay burned CPU decompressing and recompressing
identical bytes for each downstream peer, and its in-RAM byte count tracked the
decompressed size rather than what it actually stores and sends.

Cache the bytes verbatim instead. The lite subscriber stores frames in the
codec they arrived in, recorded on the new Frame::compression field. This is
distinct from the compress egress hint: an origin marks a track compress but
writes plaintext, so its cached frames stay None. Egress then:

- forwards verbatim when the cached codec matches the wire codec (the common
  relay to modern-peer path: no inflate or deflate), and
- recodes only when they differ: an origin compressing its plaintext, or a
  compression-incapable peer (old lite draft, IETF moq-transport, or an
  in-process consumer) that needs plaintext.

FrameConsumer::read_all decodes against Frame::compression, so application
consumers (the moq-json catalog reader, etc.) and the IETF / old-lite egress
paths keep getting plaintext unchanged. read_chunk streams the stored bytes
verbatim for the relay passthrough. The wire format is unchanged.

The model's natural byte count now equals the compressed/wire count, which is
what usage billing should meter. The old-client-decompress path undercounts
egress (counts compressed, sends decompressed), an accepted tradeoff over
overcounting.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: kixelated

rs/hang/src/container/frame.rs

@@ -62,6 +62,7 @@ impl Frame {
 		let net_frame = moq_net::Frame {
 			size,
 			timestamp: net_timestamp,
+			compression: moq_net::Compression::None,
 		};
 		let mut chunked = group.create_frame(net_frame)?;
 		chunked.write(header.freeze())?;

rs/moq-mux/src/container/fmp4/import.rs

@@ -613,6 +613,7 @@ impl Import {
 			let mut frame = g.create_frame(moq_net::Frame {
 				size: fragment_bytes.len() as u64,
 				timestamp: Some(timestamp),
+				compression: moq_net::Compression::None,
 			})?;
 			frame.write(fragment_bytes)?;
 			frame.finish()?;

rs/moq-native/tests/broadcast.rs

@@ -140,6 +140,7 @@ async fn lite05_timestamp_roundtrip(scheme: &str) {
 		let frame = moq_native::moq_net::Frame {
 			size: payload.len() as u64,
 			timestamp: Some(Timestamp::new(us, Timescale::MICRO).unwrap()),
+			compression: moq_native::moq_net::Compression::None,
 		};
 		let mut writer = group.create_frame(frame).expect("failed to create frame");
 		writer
@@ -257,6 +258,7 @@ async fn lite05_fetch_roundtrip(scheme: &str) {
 		let frame = moq_native::moq_net::Frame {
 			size: payload.len() as u64,
 			timestamp: Some(Timestamp::new(us, Timescale::MICRO).unwrap()),
+			compression: moq_native::moq_net::Compression::None,
 		};
 		let mut writer = group.create_frame(frame).expect("failed to create frame");
 		writer
@@ -353,6 +355,126 @@ async fn broadcast_moq_lite_05_fetch_webtransport() {
 	lite05_fetch_roundtrip("https").await;
 }
 
+/// A modern (lite-05) subscriber caches a `compress` track's frames in their
+/// Deflate-compressed form: the model holds the packed bytes (so its byte count
+/// matches the wire/compressed size and exposes the codec), and only inflates
+/// them when the payload is actually read. This is the relay-billing invariant —
+/// the cache counts compressed bytes — and the decode-at-delivery contract.
+async fn lite05_compress_caches_compressed(scheme: &str) {
+	use moq_net::{Compression, Timescale, Timestamp};
+
+	let pub_origin = Origin::random().produce();
+	let mut broadcast = pub_origin.create_broadcast("test").expect("failed to create broadcast");
+	let mut track = broadcast
+		.create_track(
+			"meta",
+			moq_net::TrackInfo::default()
+				.with_timescale(Timescale::MICRO)
+				.with_compress(true),
+		)
+		.expect("failed to create track");
+
+	// Highly compressible payload so the cached (compressed) length is
+	// unmistakably smaller than the decoded payload. Written verbatim, the way an
+	// origin produces frames (the publisher compresses on egress).
+	let payload = bytes::Bytes::from(vec![b'a'; 4096]);
+	let mut group = track.append_group().expect("failed to append group");
+	let mut writer = group
+		.create_frame(moq_net::Frame {
+			size: payload.len() as u64,
+			timestamp: Some(Timestamp::new(1_000, Timescale::MICRO).unwrap()),
+			compression: Compression::None,
+		})
+		.expect("failed to create frame");
+	writer.write(payload.clone()).expect("failed to write frame");
+	writer.finish().expect("failed to finish frame");
+	group.finish().expect("failed to finish group");
+
+	let mut server_config = moq_native::ServerConfig::default();
+	server_config.bind = Some("[::]:0".to_string());
+	server_config.tls.generate = vec!["localhost".into()];
+	server_config.version = vec!["moq-lite-05-wip".parse().unwrap()];
+	let mut server = server_config.init().expect("failed to init server");
+	let addr = server.local_addr().expect("failed to get local addr");
+
+	let sub_origin = Origin::random().produce();
+	let mut announcements = sub_origin.consume().announced();
+
+	let mut client_config = moq_native::ClientConfig::default();
+	client_config.tls.disable_verify = Some(true);
+	client_config.version = vec!["moq-lite-05-wip".parse().unwrap()];
+	let client = client_config.init().expect("failed to init client");
+	let url: url::Url = format!("{scheme}://localhost:{}", addr.port()).parse().unwrap();
+
+	let server_handle = tokio::spawn(async move {
+		let request = server.accept().await.expect("no incoming connection");
+		let session = request.with_publisher(&pub_origin).ok().await?;
+		let _broadcast = broadcast;
+		let _track = track;
+		let _ = session.closed().await;
+		Ok::<_, anyhow::Error>(())
+	});
+
+	let client = client.with_subscriber(sub_origin);
+	let session = tokio::time::timeout(TIMEOUT, client.connect(url))
+		.await
+		.expect("client connect timed out")
+		.expect("client connect failed");
+
+	let (path, bc) = tokio::time::timeout(TIMEOUT, announcements.next())
+		.await
+		.expect("announce timed out")
+		.expect("origin closed");
+	assert_eq!(path.as_str(), "test");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
+
+	let mut track_sub = tokio::time::timeout(TIMEOUT, async {
+		bc.track("meta").unwrap().subscribe(None).unwrap().await
+	})
+	.await
+	.expect("subscribe timed out")
+	.expect("subscribe failed");
+	assert!(track_sub.info().compress, "track should be compress-hinted");
+
+	let mut group_sub = tokio::time::timeout(TIMEOUT, track_sub.next_group())
+		.await
+		.expect("next_group timed out")
+		.expect("next_group failed")
+		.expect("expected a group");
+
+	let mut frame_sub = tokio::time::timeout(TIMEOUT, group_sub.next_frame())
+		.await
+		.expect("next_frame timed out")
+		.expect("next_frame failed")
+		.expect("expected a frame");
+
+	// The cache holds the Deflate-compressed bytes, not the plaintext: the codec
+	// is recorded and the stored size is the compressed length.
+	assert_eq!(frame_sub.compression, Compression::Deflate);
+	assert!(
+		(frame_sub.size as usize) < payload.len(),
+		"cached size {} should be the compressed length, smaller than {}",
+		frame_sub.size,
+		payload.len()
+	);
+
+	// Reading inflates the cached bytes back to the original payload.
+	let decoded = frame_sub.read_all().await.expect("failed to read frame");
+	assert_eq!(decoded, payload);
+
+	drop(session);
+	server_handle
+		.await
+		.expect("server task panicked")
+		.expect("server task failed");
+}
+
+#[tracing_test::traced_test]
+#[tokio::test]
+async fn broadcast_moq_lite_05_compress_caches_compressed_webtransport() {
+	lite05_compress_caches_compressed("https").await;
+}
+
 /// A fetch must be served while a live subscription is active on the same track.
 /// The relay subscribes starting at the latest group, so an older group isn't
 /// cached and the fetch has to issue a wire FETCH concurrently with the
@@ -365,6 +487,7 @@ async fn lite05_fetch_during_subscribe(scheme: &str) {
 		moq_net::Frame {
 			size: payload.len() as u64,
 			timestamp: Some(Timestamp::new(us, Timescale::MICRO).unwrap()),
+			compression: moq_net::Compression::None,
 		}
 	}
 

rs/moq-net/src/ietf/publisher.rs

@@ -331,6 +331,27 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 				stream.encode(&0u64).await?;
 			}
 
+			// IETF carries no per-frame compression, so a cached compressed frame
+			// (e.g. relayed from a lite-05 origin) must be inflated for this peer.
+			if !frame.compression.is_none() {
+				let mut payload = tokio::select! {
+					biased;
+					_ = stream.closed() => return Err(Error::Cancel),
+					payload = frame.read_all() => payload?,
+				};
+				let n = payload.len() as u64;
+				stream.encode(&n).await?;
+				track_stats.frame();
+				if n == 0 {
+					// Have to write the object status too.
+					stream.encode(&0u8).await?;
+				} else {
+					stream.write_all(&mut payload).await?;
+					track_stats.bytes(n);
+				}
+				continue;
+			}
+
 			// Write the size of the frame.
 			stream.encode(&frame.size).await?;
 			track_stats.frame();

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

@@ -3,8 +3,9 @@ use std::collections::{HashMap, hash_map::Entry};
 use std::sync::Arc;
 
 use crate::{
-	BroadcastDynamic, BroadcastInfo, Error, Frame, FrameProducer, Group, GroupProducer, MAX_FRAME_SIZE, OriginProducer,
-	OriginPublish, Path, PathOwned, StatsHandle, SubscriberStats, SubscriberTrack, TrackProducer, TrackRequest,
+	BroadcastDynamic, BroadcastInfo, Compression, Error, Frame, FrameProducer, Group, GroupProducer, MAX_FRAME_SIZE,
+	OriginProducer, OriginPublish, Path, PathOwned, StatsHandle, SubscriberStats, SubscriberTrack, TrackProducer,
+	TrackRequest,
 	coding::{Reader, Stream},
 	ietf::{self, Control, FilterType, GroupOrder, RequestId},
 	model::BroadcastProducer,
@@ -817,6 +818,7 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 					let mut frame = producer.create_frame(Frame {
 						size: 0,
 						timestamp: None,
+						compression: Compression::None,
 					})?;
 					track_stats.frame();
 					frame.finish()?;
@@ -829,7 +831,11 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 				if size > MAX_FRAME_SIZE {
 					return Err(Error::FrameTooLarge);
 				}
-				let mut frame = producer.create_frame(Frame { size, timestamp: None })?;
+				let mut frame = producer.create_frame(Frame {
+					size,
+					timestamp: None,
+					compression: Compression::None,
+				})?;
 				track_stats.frame();
 
 				if let Err(err) = self.run_frame(stream, frame.clone(), &track_stats).await {

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

@@ -788,25 +788,25 @@ async fn write_fetch_frame<W: web_transport_trait::SendStream>(
 ) -> Result<(), Error> {
 	encode_frame_timing(writer, frame, timescale, prev_ts).await?;
 
-	match compression {
-		Compression::None => {
-			writer.encode(&frame.size).await?;
-			track_stats.frame();
-			while let Some(mut chunk) = frame.read_chunk().await? {
-				let n = chunk.len() as u64;
-				writer.write_all(&mut chunk).await?;
-				track_stats.bytes(n);
-			}
-		}
-		compression => {
-			let payload = frame.read_all().await?;
-			let mut chunk = bytes::Bytes::from(compression.compress(&payload));
+	if frame.compression == compression {
+		// Cached codec matches the wire: forward the bytes verbatim.
+		writer.encode(&frame.size).await?;
+		track_stats.frame();
+		while let Some(mut chunk) = frame.read_chunk().await? {
 			let n = chunk.len() as u64;
-			writer.encode(&n).await?;
-			track_stats.frame();
 			writer.write_all(&mut chunk).await?;
 			track_stats.bytes(n);
 		}
+	} else {
+		// Recode for this peer: `read_all` decodes the cached bytes, then we
+		// re-encode for the wire (exactly one side is a real op; the other is None).
+		let payload = frame.read_all().await?;
+		let mut chunk = bytes::Bytes::from(compression.compress(&payload));
+		let n = chunk.len() as u64;
+		writer.encode(&n).await?;
+		track_stats.frame();
+		writer.write_all(&mut chunk).await?;
+		track_stats.bytes(n);
 	}
 
 	Ok(())
@@ -960,10 +960,12 @@ impl<S: web_transport_trait::Session> Subscription<S> {
 		Ok(())
 	}
 
-	/// Send one frame. Uncompressed frames stream chunk-by-chunk so we never
-	/// buffer the whole payload; a compressed frame must buffer to feed the
-	/// codec, and its wire size becomes the compressed length (the subscriber
-	/// inflates it from the track's codec, known from TRACK_INFO on lite-05+).
+	/// Send one frame. When the cached codec already matches the wire codec the
+	/// bytes stream through verbatim, never buffering the whole payload — this is
+	/// the hot path, and it's where a modern peer pulling an already-compressed
+	/// track avoids a needless inflate/deflate. Otherwise the payload is decoded
+	/// and re-encoded for this peer (an origin compressing its plaintext on egress,
+	/// or a compression-incapable peer that needs plain bytes), which buffers it.
 	async fn serve_frame(
 		&mut self,
 		stream: &mut Writer<S::SendStream, Version>,
@@ -973,22 +975,22 @@ impl<S: web_transport_trait::Session> Subscription<S> {
 	) -> Result<(), Error> {
 		encode_frame_timing(stream, &frame, self.timescale, prev_ts).await?;
 
-		match self.compression {
-			Compression::None => {
-				stream.encode(&frame.size).await?;
-				self.track_stats.frame();
+		if frame.compression == self.compression {
+			stream.encode(&frame.size).await?;
+			self.track_stats.frame();
 
-				while let Some(chunk) = self.read_chunk(stream, priority, &mut frame).await? {
-					self.write_chunk(stream, priority, chunk).await?;
-				}
-			}
-			compression => {
-				let payload = self.read_all(stream, priority, &mut frame).await?;
-				let chunk = bytes::Bytes::from(compression.compress(&payload));
-				stream.encode(&(chunk.len() as u64)).await?;
-				self.track_stats.frame();
+			while let Some(chunk) = self.read_chunk(stream, priority, &mut frame).await? {
 				self.write_chunk(stream, priority, chunk).await?;
 			}
+		} else {
+			// `read_all` decodes the cached bytes to the logical payload; we then
+			// re-encode in the wire codec. Codecs differ, so exactly one of decode
+			// and encode is a real operation (the other side is `None`).
+			let payload = self.read_all(stream, priority, &mut frame).await?;
+			let chunk = bytes::Bytes::from(self.compression.compress(&payload));
+			stream.encode(&(chunk.len() as u64)).await?;
+			self.track_stats.frame();
+			self.write_chunk(stream, priority, chunk).await?;
 		}
 
 		Ok(())

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

@@ -646,34 +646,24 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 				return Err(Error::FrameTooLarge);
 			}
 
-			match compression {
-				Compression::None => {
-					let mut frame = group.create_frame(Frame { size, timestamp })?;
-					track_stats.frame();
-
-					if let Err(err) = self.run_frame(stream, &mut frame, &track_stats).await {
-						let _ = frame.abort(err.clone());
-						return Err(err);
-					}
+			// Cache the bytes exactly as they arrive, tagged with the negotiated
+			// codec. For a compressed track the cache holds the packed bytes (no
+			// inflate on ingress): a relay forwards them verbatim to a modern peer
+			// and only the model decodes for one that can't speak the codec. `size`
+			// is the on-wire (stored) length in either case.
+			let mut frame = group.create_frame(Frame {
+				size,
+				timestamp,
+				compression,
+			})?;
+			track_stats.frame();
 
-					frame.finish()?;
-				}
-				compression => {
-					// `size` is the compressed length; pull it off the wire, then
-					// inflate. The frame the consumer sees carries the original size.
-					let packed = stream.read_exact(size as usize).await?;
-					track_stats.frame();
-					track_stats.bytes(size);
-
-					let payload = compression.decompress(&packed)?;
-					let mut frame = group.create_frame(Frame {
-						size: payload.len() as u64,
-						timestamp,
-					})?;
-					frame.write(bytes::Bytes::from(payload))?;
-					frame.finish()?;
-				}
+			if let Err(err) = self.run_frame(stream, &mut frame, &track_stats).await {
+				let _ = frame.abort(err.clone());
+				return Err(err);
 			}
+
+			frame.finish()?;
 		}
 
 		Ok(())

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

@@ -58,6 +58,11 @@ impl Compression {
 		}
 	}
 
+	/// True for [`Compression::None`] (frames stored/sent verbatim).
+	pub fn is_none(&self) -> bool {
+		matches!(self, Self::None)
+	}
+
 	/// The varint code used on the wire.
 	pub fn to_code(self) -> u64 {
 		match self {