feat(moq-net): make payload compression per-hop, per-frame (drafts#39)

Implements the moq-lite half of moq-dev/drafts#39: the publisher only signals
*whether* a track is worth compressing, while the algorithm is negotiated per hop
and recorded per frame. A single frame can opt out — e.g. a small JSON merge-patch
delta that DEFLATE would only enlarge — instead of bloating it.

Wire (moq-lite-05-wip):
- TRACK_INFO `Publisher Compression` becomes a boolean hint (>1 reserved, decodes
  as true so it stays additive). It names no algorithm.
- New SETUP `Compression` parameter (id 0x3): each endpoint advertises the
  algorithms it can decompress, packed varints, none(0) omitted. moq-net advertises
  [deflate]. Per hop, per direction; never forwarded by a relay.
- New per-frame `Compression` field in FRAME, present iff the track is hinted,
  after the timestamp delta and before the length. Names the codec actually used
  (none/deflate); it doubles as the group's end-of-frames sentinel when no
  timestamp precedes it.

Behavior:
- The publisher reads the peer's advertised algorithms (PeerSetup) before serving a
  compress-hinted track. It forwards an already-DEFLATE cached frame verbatim when
  the peer can inflate it (relay passthrough, no inflate/deflate), and otherwise
  picks per frame whether DEFLATE actually shrinks the payload, sending `none` when
  it wouldn't (or the peer can't inflate). Empty payloads always use `none`.
- The subscriber decodes the per-frame field and caches each frame in that codec,
  so the cache holds compressed bytes (billing counts the wire size) and inflates
  only at delivery.

No datagram path exists in moq-net yet, so that part of the draft is N/A here. The
IETF moqt compression extension (moq-compression) is left for a follow-up; the IETF
path keeps sending verbatim, which is spec-compliant "extension not negotiated".

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

Author: kixelated

rs/moq-native/tests/broadcast.rs

@@ -355,11 +355,11 @@ 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.
+/// A modern (lite-05) subscriber caches a `compress` track's frames in the codec
+/// each frame names on the wire: a compressible frame is DEFLATE (so the cached
+/// byte count matches the wire/compressed size — the relay-billing invariant), a
+/// tiny frame DEFLATE would only enlarge opts out and is cached verbatim, and both
+/// inflate back to the original only when read (the decode-at-delivery contract).
 async fn lite05_compress_caches_compressed(scheme: &str) {
 	use moq_net::{Compression, Timescale, Timestamp};
 
@@ -378,6 +378,9 @@ async fn lite05_compress_caches_compressed(scheme: &str) {
 	// 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]);
+	// A second, tiny payload DEFLATE would only enlarge: the publisher must send it
+	// verbatim (per-frame opt-out) even though the track is compress-hinted.
+	let tiny = bytes::Bytes::from_static(b"hi");
 	let mut group = track.append_group().expect("failed to append group");
 	let mut writer = group
 		.create_frame(moq_net::Frame {
@@ -388,6 +391,15 @@ async fn lite05_compress_caches_compressed(scheme: &str) {
 		.expect("failed to create frame");
 	writer.write(payload.clone()).expect("failed to write frame");
 	writer.finish().expect("failed to finish frame");
+	let mut writer = group
+		.create_frame(moq_net::Frame {
+			size: tiny.len() as u64,
+			timestamp: Some(Timestamp::new(2_000, Timescale::MICRO).unwrap()),
+			compression: Compression::None,
+		})
+		.expect("failed to create tiny frame");
+	writer.write(tiny.clone()).expect("failed to write tiny frame");
+	writer.finish().expect("failed to finish tiny frame");
 	group.finish().expect("failed to finish group");
 
 	let mut server_config = moq_native::ServerConfig::default();
@@ -462,6 +474,15 @@ async fn lite05_compress_caches_compressed(scheme: &str) {
 	let decoded = frame_sub.read_all().await.expect("failed to read frame");
 	assert_eq!(decoded, payload);
 
+	// The tiny frame opted out: cached verbatim (no per-frame DEFLATE), reads back unchanged.
+	let mut tiny_sub = tokio::time::timeout(TIMEOUT, group_sub.next_frame())
+		.await
+		.expect("next_frame timed out")
+		.expect("next_frame failed")
+		.expect("expected a second frame");
+	assert_eq!(tiny_sub.compression, Compression::None);
+	assert_eq!(tiny_sub.read_all().await.expect("failed to read tiny frame"), tiny);
+
 	drop(session);
 	server_handle
 		.await

rs/moq-net/src/client.rs

@@ -161,6 +161,8 @@ impl Client {
 				let our_setup = lite::Setup {
 					probe: lite::ProbeLevel::Report,
 					path: self.setup_path.clone(),
+					// We can inflate DEFLATE, so a peer may compress what it sends us.
+					compression: vec![crate::Compression::Deflate],
 				};
 
 				let (recv_bw, connecting) = lite::start(