From 3602759ece040a31abc910f401c13c86f568b6d5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 05:34:29 -0700 Subject: [PATCH 1/5] chore(quest): claim quest/m1/cut-audit/announce-reencode.md Quest-Claim: d80c32e571813a07 From d6e0bcc5f684f192f915ddeb1f39e2f6e40e67ea Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 06:13:27 -0700 Subject: [PATCH 2/5] fix(net): skip unencodable namespace advertisements Co-Authored-By: GPT-6 --- js/net/src/ietf/namespace.test.ts | 39 ++++++- js/net/src/ietf/namespace.ts | 25 ++++- js/net/src/ietf/publisher.test.ts | 94 ++++++++++++++++ js/net/src/ietf/publisher.ts | 4 + js/net/src/lite/announce.test.ts | 13 ++- quest/m1/cut-audit/README.md | 1 - quest/m1/cut-audit/announce-path-bytes.md | 4 - quest/m1/cut-audit/announce-reencode.md | 37 ------- rs/moq-net/benches/path.rs | 23 ++++ rs/moq-net/src/fuzz.rs | 5 +- rs/moq-net/src/ietf/namespace.rs | 52 +++++++-- rs/moq-net/src/ietf/publisher.rs | 127 +++++++++++++++++++++- rs/moq-net/src/lite/announce.rs | 23 ++++ 13 files changed, 389 insertions(+), 58 deletions(-) delete mode 100644 quest/m1/cut-audit/announce-reencode.md diff --git a/js/net/src/ietf/namespace.test.ts b/js/net/src/ietf/namespace.test.ts index 9e89aefb4c1..8e3206a42af 100644 --- a/js/net/src/ietf/namespace.test.ts +++ b/js/net/src/ietf/namespace.test.ts @@ -1,5 +1,6 @@ import { expect, test } from "bun:test"; -import { fromTuple, reserved } from "./namespace.ts"; +import { Reader, Writer } from "../stream.ts"; +import { decode, encodable, encode, fromTuple, reserved } from "./namespace.ts"; import { Version } from "./version.ts"; test("reserved namespaces preserve tuple fields and draft boundaries", () => { @@ -18,3 +19,39 @@ test("reserved namespaces preserve tuple fields and draft boundaries", () => { expect(reserved(path, Version.DRAFT_18)).toBe(false); } }); + +test("namespace encoding uses decoded UTF-8 field bounds", async () => { + for (const [parts, fits] of [ + [["é".repeat(2048)], true], + [[`${"é".repeat(2048)}x`], false], + [["a".repeat(2048), "b".repeat(2048)], true], + [["a".repeat(2048), "b".repeat(2049)], false], + [["/".repeat(4096)], true], + [["\\".repeat(4097)], false], + [Array(32).fill("a"), true], + [Array(33).fill("a"), false], + ] as [string[], boolean][]) { + const namespace = fromTuple(parts); + expect(encodable(namespace)).toBe(fits); + const bytes: number[] = []; + const writer = new Writer( + new WritableStream({ + write: (chunk) => { + bytes.push(...chunk); + }, + }), + Version.DRAFT_22, + ); + try { + if (fits) { + await encode(writer, namespace); + expect(await decode(new Reader(undefined, Uint8Array.from(bytes), Version.DRAFT_22))).toBe(namespace); + } else { + await expect(encode(writer, namespace)).rejects.toThrow("namespace exceeds"); + expect(bytes).toEqual([]); + } + } finally { + await writer.close(); + } + } +}); diff --git a/js/net/src/ietf/namespace.ts b/js/net/src/ietf/namespace.ts index 44a4314346d..6660636a264 100644 --- a/js/net/src/ietf/namespace.ts +++ b/js/net/src/ietf/namespace.ts @@ -30,11 +30,8 @@ export function fromTuple(parts: string[]): Path.Valid { /** Encode a moq-net namespace as an IETF namespace tuple. */ export async function encode(w: Writer, namespace: Path.Valid): Promise { const parts = toTuple(namespace); - - // The IETF draft limits namespaces to 32 parts. - if (parts.length > Path.MAX_PARTS) { - throw new Error(`namespace exceeds ${Path.MAX_PARTS} parts`); - } + const error = check(parts); + if (error) throw new Error(error); await w.u53(parts.length); for (const part of parts) { @@ -44,6 +41,24 @@ export async function encode(w: Writer, namespace: Path.Valid): Promise { // Every draft caps a namespace, and a namespace plus track name, at 4096 bytes. const MAX_BYTES = 4096; +const utf8 = new TextEncoder(); + +function check(parts: string[]): string | undefined { + if (parts.length > Path.MAX_PARTS) return `namespace exceeds ${Path.MAX_PARTS} parts`; + let size = 0; + for (const part of parts) { + size += utf8.encode(part).length; + if (size > MAX_BYTES) return `namespace exceeds ${MAX_BYTES} bytes`; + } + return undefined; +} + +/** Whether one advertisement fits the namespace representation this wire carries. */ +export function encodable(namespace: Path.Valid): boolean { + // A UTF-16 unit costs at most three UTF-8 bytes; removing separators/escapes only shrinks it. + if (namespace.length <= MAX_BYTES / 3 && Path.check(namespace) === undefined) return true; + return check(toTuple(namespace)) === undefined; +} /** Decode an IETF namespace tuple into an escaped moq-net namespace. */ export async function decode(r: Reader): Promise { diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 72ca463f41f..369481a4427 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -194,6 +194,100 @@ function publisher( const IETF_VERSIONS = [Version.DRAFT_18, Version.DRAFT_22] as const; +test.each(IETF_VERSIONS)("%p: namespaces skip oversize routes and keep siblings flowing", async (version) => { + using _clock = manual(); + const pair = createMockTransportPair(version === Version.DRAFT_18 ? ALPN.DRAFT_18 : ALPN.DRAFT_22); + const session = new NativeSession(pair.server, version, true); + const { pub, origin } = publisher(pair.server, { session, requiresSolicitation: true, activeCount: true }); + const boundary = Path.from("b".repeat(4096)); + const held = [ + publish(origin, Path.from(`a${"é".repeat(2048)}`)), + publish(origin, boundary), + publish(origin, Path.from("sibling")), + ]; + try { + for (let attempt = 0; attempt < 2; attempt++) { + const subscription = await Stream.open(pair.client, { version }); + const accepted = await Stream.accept(pair.server, version); + if (!accepted) throw new Error("missing namespace request stream"); + const running = pub.runSubscribeNamespace( + new SubscribeNamespace({ requestId: BigInt(attempt), namespace: Path.empty() }), + accepted, + ); + const settled = running.catch(() => {}); + try { + expect(await subscription.reader.u53()).toBe(RequestOk.id); + const ok = await RequestOk.decode(subscription.reader, version); + const seen = new Set(); + for (let i = 0; i < 2; i++) { + expect(await subscription.reader.u53()).toBe(SubscribeNamespaceEntry.id); + seen.add((await SubscribeNamespaceEntry.decode(subscription.reader, version)).suffix); + } + expect(seen).toEqual(new Set([boundary, Path.from("sibling")])); + expect(ok.parameters.activeCount).toBe(2); + const late = publish(origin, Path.from("late")); + expect(await subscription.reader.u53()).toBe(SubscribeNamespaceEntry.id); + expect((await SubscribeNamespaceEntry.decode(subscription.reader, version)).suffix).toBe( + Path.from("late"), + ); + late.close(); + expect(await subscription.reader.u53()).toBe(SubscribeNamespaceEntryDone.id); + expect((await SubscribeNamespaceEntryDone.decode(subscription.reader, version)).suffix).toBe( + Path.from("late"), + ); + } finally { + subscription.close(); + await settled; + } + } + } finally { + for (const broadcast of held) broadcast.close(); + origin.close(); + pub.close(); + pair.client.close(); + pair.server.close(); + } +}); + +test.each(IETF_VERSIONS)("%p: unencodable namespaces spend no advertisement request ids", async (version) => { + using _clock = manual(); + const pair = createMockTransportPair(version === Version.DRAFT_18 ? ALPN.DRAFT_18 : ALPN.DRAFT_22); + const session = new NativeSession(pair.server, version, true); + const ids = spyOn(session, "nextRequestId"); + const { pub, origin } = publisher(pair.server, { session }); + const held = [ + publish(origin, Path.from(`a${"é".repeat(2048)}`)), + publish(origin, Path.from("first")), + publish(origin, Path.from("second")), + ]; + const peers: Stream[] = []; + const running = pub.runPublishNamespaces(); + const settled = running.catch(() => {}); + try { + const seen = new Set(); + for (let i = 0; i < 2; i++) { + const stream = await Stream.accept(pair.client, version); + if (!stream) throw new Error("missing advertisement stream"); + peers.push(stream); + expect(await stream.reader.u53()).toBe(PublishNamespace.id); + seen.add((await PublishNamespace.decode(stream.reader, version)).trackNamespace); + await stream.writer.u53(RequestOk.id); + await new RequestOk().encode(stream.writer, version); + } + expect(seen).toEqual(new Set([Path.from("first"), Path.from("second")])); + expect(ids).toHaveBeenCalledTimes(2); + } finally { + origin.close(); + for (const stream of peers) stream.close(); + await settled; + for (const broadcast of held) broadcast.close(); + pub.close(); + pair.client.close(); + pair.server.close(); + ids.mockRestore(); + } +}); + /** A request stream recording what the publisher writes, with the requester's side under test control. */ function requestStream(version: IetfVersion) { let input!: ReadableStreamDefaultController; diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 8719a3005b7..56e3c93d514 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -1631,6 +1631,9 @@ export class Publisher { const advertised = this.#advertised.peek(); const updated = advertised ? presented(prefix, advertised, carries) : new Map(); + for (const key of updated.keys()) { + if (!Namespace.encodable(key)) updated.delete(key); + } // A fresh pass sends a NAMESPACE for every entry, ahead of any change. if (!answered) { answered = true; @@ -1790,6 +1793,7 @@ export class Publisher { if ( (this.#hidden && hiddenBelow(Path.empty(), covered)) || !this.#gate.allows(covered) || + !Namespace.encodable(covered) || candidates.length === 0 ) continue; diff --git a/js/net/src/lite/announce.test.ts b/js/net/src/lite/announce.test.ts index 84d4ce421d4..5dad9f7e8f1 100644 --- a/js/net/src/lite/announce.test.ts +++ b/js/net/src/lite/announce.test.ts @@ -1,7 +1,7 @@ import { expect, test } from "bun:test"; import * as Epoch from "../epoch.ts"; import { ProtocolViolation } from "../error.ts"; -import { Cost, HopSchema, UNKNOWN_HOP } from "../hop.ts"; +import { Cost, HopSchema, MAX_HOPS, UNKNOWN_HOP } from "../hop.ts"; import * as Path from "../path.ts"; import { Reader, Writer } from "../stream.ts"; import { @@ -47,6 +47,17 @@ async function roundTrip(msg: AnnounceBroadcast, version: Version): Promise { + const message: AnnounceBroadcast = { + status: "active", + suffix: Path.from("é".repeat(Path.MAX_BYTES / 2)), + hops: Array.from({ length: MAX_HOPS }, (_, n) => HopSchema.parse((1n << 51n) + BigInt(n))), + epoch: Epoch.parse("01900000-0000-7000-8000-000000000001"), + cost: (1n << 62n) - 1n, + }; + expect(await roundTrip(message, Version.DRAFT_07)).toEqual(message); +}); + test("AnnounceBroadcast round-trips on draft-05", async () => { const hops = [HopSchema.parse(7n)]; const gotActive = await roundTrip({ status: "active", suffix: Path.from("room/cam"), hops }, Version.DRAFT_05); diff --git a/quest/m1/cut-audit/README.md b/quest/m1/cut-audit/README.md index 0542968a292..216371d071d 100644 --- a/quest/m1/cut-audit/README.md +++ b/quest/m1/cut-audit/README.md @@ -56,7 +56,6 @@ leaving JS-only redesigns of `js/net` to [rs2ts](/quest/m2/rs2ts/README.md). - [Counted field-less `trun` entries](/quest/m1/cut-audit/fmp4-sample-resolver.md) - a moof of field-less runs cannot materialize millions of sample entries - [A compressed announce path never outgrows its literal form](/quest/m1/cut-audit/announce-path-bytes.md) - lite-07 announce compression cannot build multi-MiB paths -- [A route too large to re-encode is skipped](/quest/m1/cut-audit/announce-reencode.md) - one oversized route never resets a downstream announce stream - [A reserved group survives a route returning](/quest/m1/cut-audit/reserved-route-return.md) - a route flip A to B to A no longer panics a peeked reader - [JS tracks end clean only on a deliberate finish](/quest/m1/cut-audit/js-close-cause.md) - local close, violations, and unanswered requests end with errors - [Withdrawing a moq-transport FETCH or TRACK_STATUS ends it](/quest/m1/cut-audit/ietf-withdraw-resolve.md) - withdrawal during resolution releases the request diff --git a/quest/m1/cut-audit/announce-path-bytes.md b/quest/m1/cut-audit/announce-path-bytes.md index ef8a86615ad..e917d253947 100644 --- a/quest/m1/cut-audit/announce-path-bytes.md +++ b/quest/m1/cut-audit/announce-path-bytes.md @@ -37,7 +37,3 @@ parts-only check. unpublished, so this changes no published version). Public API: none. Wire: one receiver rule in the lite-07 draft. - -## Related - -- [A route too large to re-encode is skipped](/quest/m1/cut-audit/announce-reencode.md) - the same size bound on the send side diff --git a/quest/m1/cut-audit/announce-reencode.md b/quest/m1/cut-audit/announce-reencode.md deleted file mode 100644 index 3fc30eb0720..00000000000 --- a/quest/m1/cut-audit/announce-reencode.md +++ /dev/null @@ -1,37 +0,0 @@ -# [S] A route too large to re-encode is skipped, not fatal to the stream - -## Goal - -When a relay cannot re-encode one route for a downstream announce stream -(the downstream's shorter prefix or an extra hop pushes the message past the -encode cap), that downstream simply does not see that route. Its other -routes keep flowing, and asking again does not fail the same way. The same -holds for moq-transport's namespace advertisement. - -## Plan - -Found by the 2026-10-24 pre-cut audit at 151a529670. Reproduced: announcing -`"x/" + "a"*65530` on the lite test harness ended the downstream announce -task. New on main: [moq-dev/moq#4820](https://github.com/moq-dev/moq/pull/4820) -lowered the message cap from 64 MiB to 64 KiB. - -`AnnounceRun::start` (`rs/moq-net/src/lite/publisher.rs:1194`) calls -`stream.writer.buffer(..)?`; `EncodeError::TooLarge` -(`lite/announce.rs:297`) propagates to `AnnounceServe::poll` (about line -906), which aborts the whole stream. Cluster peers and `moq ls` ask for the -root prefix, so one hostile publisher withdraws every route a relay sends -them. On lite-06+ the encoder has already spent an announce id. - -- `outgoing()` returns `None` for a route this wire cannot carry, like a - reflected or overlong hop chain. Encode before assigning the id. -- Never abort an announce stream for one route. -- Check the IETF publisher's namespace encoding against its 4096-byte limit - (`ietf/namespace.rs:66`) for the same failure, and fix it the same way. -- A test announces a near-cap suffix under a short downstream prefix and - asserts the sibling routes still arrive. - -Public API: none. Wire: none. - -## Related - -- [A compressed announce path never outgrows its literal form](/quest/m1/cut-audit/announce-path-bytes.md) - the receive-side bound diff --git a/rs/moq-net/benches/path.rs b/rs/moq-net/benches/path.rs index 566d08a3635..6dff0be4cda 100644 --- a/rs/moq-net/benches/path.rs +++ b/rs/moq-net/benches/path.rs @@ -192,9 +192,32 @@ fn fanout() { } } +fn namespaces() { + for (name, prefix) in [("short", SHORT), ("long", LONG)] { + for routes in [1, 64, 1024] { + let paths: Vec<_> = (0..routes) + .map(|i| Path::new(&format!("{prefix}/{i}")).unwrap()) + .collect(); + for peers in [1, 8, 32] { + allocs( + &format!("namespace/emit/{name}/{routes}r_{peers}p"), + 16, + routes * peers, + || { + for _ in 0..peers { + black_box(encode_namespaces(&paths)); + } + }, + ); + } + } + } +} + fn main() { paths(); decode(); requests(); fanout(); + namespaces(); } diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 9d61e7b21ca..320042ff79d 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -452,11 +452,14 @@ pub fn decode_announces(data: &[u8], compress: bool) -> Vec { resolved } -/// Encode a moq-transport NAMESPACE message for each of `paths`. +/// Encode the moq-transport NAMESPACE messages this wire can carry from `paths`. pub fn encode_namespaces(paths: &[Path]) -> Vec { let mut data = Vec::new(); let w = &mut Encoder::new(&mut data, BENCH_IETF.into()); for path in paths { + if !ietf::namespace::encodable(path) { + continue; + } let msg = ietf::Namespace { suffix: path.clone(), params: None, diff --git a/rs/moq-net/src/ietf/namespace.rs b/rs/moq-net/src/ietf/namespace.rs index f464793fa5e..594a50c2d87 100644 --- a/rs/moq-net/src/ietf/namespace.rs +++ b/rs/moq-net/src/ietf/namespace.rs @@ -49,13 +49,7 @@ fn from_tuple(parts: &[&str]) -> Result { /// Helper function to encode namespace as tuple of strings pub fn encode_namespace(w: &mut Encoder<'_>, namespace: &Path) -> Result<(), EncodeError> { - let parts = to_tuple(namespace); - - // The IETF draft limits namespaces to 32 parts. - if parts.len() > Path::MAX_PARTS { - return Err(BoundsExceeded.into()); - } - + let parts = fields(namespace)?; w.varint(parts.len() as u64)?; for part in parts { w.string(&part)?; @@ -63,6 +57,23 @@ pub fn encode_namespace(w: &mut Encoder<'_>, namespace: &Path) -> Result<(), Enc Ok(()) } +fn fields(namespace: &Path) -> Result, EncodeError> { + let parts = to_tuple(namespace); + if parts.len() > Path::MAX_PARTS || parts.iter().map(String::len).sum::() > MAX_BYTES { + return Err(BoundsExceeded.into()); + } + Ok(parts) +} + +/// Whether one advertisement fits the namespace representation this wire carries. +pub(crate) fn encodable(namespace: &Path) -> bool { + // Separators and escaping only shrink the wire fields, so ordinary paths need no tuple allocation. + if namespace.len() <= MAX_BYTES && namespace.parts().count() <= Path::MAX_PARTS { + return true; + } + fields(namespace).is_ok() +} + /// The most bytes a Track Namespace, or a whole Full Track Name, may hold on the wire. const MAX_BYTES: usize = 4096; @@ -271,4 +282,31 @@ mod tests { assert_eq!(decode_track_slice(&bytes).is_ok(), ok, "{namespace} + {name}"); } } + + /// Limits count decoded UTF-8 field bytes, not escaped path bytes or separators. + #[test] + fn encoding_uses_the_same_field_bounds_as_decoding() { + for (parts, fits) in [ + (vec!["é".repeat(2048)], true), + (vec![format!("{}x", "é".repeat(2048))], false), + (vec!["a".repeat(2048), "b".repeat(2048)], true), + (vec!["a".repeat(2048), "b".repeat(2049)], false), + (vec!["/".repeat(4096)], true), + (vec!["\\".repeat(4097)], false), + (vec!["a".to_owned(); Path::MAX_PARTS], true), + (vec!["a".to_owned(); Path::MAX_PARTS + 1], false), + ] { + let refs: Vec<_> = parts.iter().map(String::as_str).collect(); + let path = from_tuple(&refs).unwrap(); + assert_eq!(encodable(&path), fits); + let mut bytes = Vec::new(); + let result = encode_namespace(&mut Encoder::new(&mut bytes, FORM), &path); + assert_eq!(result.is_ok(), fits); + if fits { + assert_eq!(decode_ns(&bytes), path); + } else { + assert!(bytes.is_empty(), "refuse before emitting namespace fields"); + } + } + } } diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index ebf1e666afc..593c7ddce7c 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -3675,7 +3675,11 @@ impl Publisher { path: &crate::Path, ) -> Option<(Advert, Option)> { let watch = ns.watched.get(suffix)?; - let advert = match ns.permitted(path) { + let wire_path = match &ns.target { + Target::Requests => path, + Target::Inline(_) => suffix, + }; + let advert = match ns.permitted(path) && super::namespace::encodable(wire_path) { true => self.select(&watch.route, &ns.peer, ns.epochs), false => Advert::None, }; @@ -4135,6 +4139,7 @@ impl Publisher { let count = initial .iter() .filter(|update| permit.matches(prefix.rejoin(&update.prefix).as_str())) + .filter(|update| super::namespace::encodable(&update.prefix)) .filter(|update| self.select(&update.route, &peer, declared.epoch).wanted()) .count(); (initial, Some(count as u64)) @@ -14244,6 +14249,126 @@ mod tests { slot } + /// An unencodable namespace cannot poison discovery or its initial active count. + #[moq_clock::test] + async fn namespaces_skip_oversize_routes_and_keep_siblings_flowing() { + for version in [Version::Draft18, Version::Draft22] { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let too_long = format!("a{}", "é".repeat(2048)); + let boundary = "b".repeat(4096); + let _oversize = origin + .announce(crate::Path::new(&too_long).unwrap(), Default::default()) + .unwrap(); + let _boundary = origin + .announce(crate::Path::new(&boundary).unwrap(), Default::default()) + .unwrap(); + let _sibling = origin.announce("sibling", Default::default()).unwrap(); + let gate = kio::Producer::new(true); + let session = SinkSession::gated_bi(gate.consume()); + let log = session.log.clone(); + let setup = peer::PeerSetup::default(); + setup.set(peer::Peer { + solicit: Some(true), + active_count: true, + ..Default::default() + }); + let publisher = Publisher::new( + crate::time::sim(), + crate::driver::Transport::new(session.clone()), + origin.consume(), + Control::new(false), + None, + setup, + version, + ); + for attempt in 0..2 { + log.writes.lock().unwrap().clear(); + let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), version) + .await + .unwrap(); + let mut run = std::pin::pin!(publisher.clone().run_subscribe_namespace_stream( + stream, + ietf::SubscribeNamespace { + request_id: RequestId(1 + attempt), + namespace: crate::Path::empty(), + hidden: false, + } + )); + assert!(futures::poll!(run.as_mut()).is_pending()); + let bytes = log.writes.lock().unwrap().clone(); + let events = namespace_events(&bytes, version); + assert_eq!(events.len(), 2, "only encodable namespaces are advertised"); + assert!(events.contains(&(ietf::Namespace::ID, boundary.clone()))); + assert!(events.contains(&(ietf::Namespace::ID, "sibling".to_owned()))); + let mut wire = crate::coding::Decoder::new(&bytes, version.into()); + assert_eq!(wire.varint().unwrap(), ietf::RequestOk::ID); + assert_eq!(ietf::RequestOk::decode(&mut wire, version).unwrap().active, Some(2)); + let late = origin.announce("late", Default::default()).unwrap(); + assert!(futures::poll!(run.as_mut()).is_pending()); + let events = namespace_events(&log.writes.lock().unwrap(), version); + assert!( + events.contains(&(ietf::Namespace::ID, "late".to_owned())), + "later siblings still flow" + ); + drop(late); + assert!(futures::poll!(run.as_mut()).is_pending()); + assert!( + namespace_events(&log.writes.lock().unwrap(), version) + .contains(&(ietf::NamespaceDone::ID, "late".to_owned())) + ); + } + } + } + + #[moq_clock::test] + async fn unencodable_namespaces_spend_no_advertisement_request_ids() { + for version in [Version::Draft18, Version::Draft22] { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let _oversize = origin + .announce( + crate::Path::new(&format!("a{}", "é".repeat(2048))).unwrap(), + Default::default(), + ) + .unwrap(); + let _first = origin.announce("first", Default::default()).unwrap(); + let _second = origin.announce("second", Default::default()).unwrap(); + let refusal = publish_namespace_error(version, 0).await; + let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![ + refusal.clone(), + refusal.clone(), + refusal, + ]); + let log = session.log.clone(); + let publisher = Publisher::new( + crate::time::sim(), + crate::driver::Transport::new(session), + origin.consume(), + Control::new(false), + None, + declared(Some(false)), + version, + ); + let reference = Control::new(false); + let expected_ids = [reference.next_request_id(), reference.next_request_id()]; + let mut run = std::pin::pin!(publisher.clone().run_publish_namespaces()); + assert!(futures::poll!(run.as_mut()).is_pending()); + let bytes = log.writes.lock().unwrap().clone(); + let mut wire = crate::coding::Decoder::new(&bytes, version.into()); + let mut seen = Vec::new(); + while !wire.is_empty() { + assert_eq!(wire.varint().unwrap(), ietf::PublishNamespace::ID); + let message = ietf::PublishNamespace::decode(&mut wire, version).unwrap(); + seen.push((message.request_id, message.track_namespace)); + } + assert_eq!(seen.len(), 2); + assert_eq!([seen[0].0, seen[1].0], expected_ids); + assert_eq!(publisher.control.next_request_id(), reference.next_request_id()); + let mut names: Vec<_> = seen.into_iter().map(|(_, path)| path.as_str().to_owned()).collect(); + names.sort(); + assert_eq!(names, ["first", "second"]); + } + } + /// An absent max age preserves everything the producer still retains. #[test] fn serving_subscription_keeps_retained_backlog() { diff --git a/rs/moq-net/src/lite/announce.rs b/rs/moq-net/src/lite/announce.rs index d210d77f823..eb06e6b385b 100644 --- a/rs/moq-net/src/lite/announce.rs +++ b/rs/moq-net/src/lite/announce.rs @@ -597,6 +597,29 @@ mod tests { use super::*; use bytes::Buf; + /// The model's path bound leaves room for every other literal START field. + #[test] + fn a_maximal_literal_start_fits_the_lite_message() { + let path = Path::new(&"é".repeat(Path::MAX_BYTES / 2)).unwrap(); + let hops = Hops::try_from( + (0..crate::origin::HopLimit::default().get()) + .map(|n| Hop::new((1 << 51) + n as u64).unwrap()) + .collect::>(), + ) + .unwrap(); + let version = Version::Lite07; + let message = AnnounceBroadcast::Active { + suffix: PathRef::literal(path.clone()), + hops: HopsRef::literal(hops), + epoch: Some("01900000-0000-7000-8000-000000000001".parse().unwrap()), + cost: Cost::MAX, + }; + let encoded = message.encode_bytes(version).unwrap(); + let (decoded, consumed) = AnnounceBroadcast::decode_slice(&encoded, version).unwrap(); + assert_eq!(consumed, encoded.len()); + assert!(matches!(decoded, AnnounceBroadcast::Active { suffix, .. } if suffix.rest == path)); + } + /// A legacy initial set is held to the announce budget before any suffix decodes, so /// a body of one-byte suffixes cannot become a list many times its size. #[test] From 7fed8cb679430354afacffc34b9532f0ae79b8c3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 06:42:19 -0700 Subject: [PATCH 3/5] test(net): preserve namespace boundary on strict paths Co-Authored-By: GPT-6 --- js/net/src/ietf/namespace.test.ts | 2 +- rs/moq-net/src/ietf/namespace.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/js/net/src/ietf/namespace.test.ts b/js/net/src/ietf/namespace.test.ts index 1bee5320b6d..69a7baa2629 100644 --- a/js/net/src/ietf/namespace.test.ts +++ b/js/net/src/ietf/namespace.test.ts @@ -25,7 +25,7 @@ test("namespace encoding uses decoded UTF-8 field bounds", async () => { [[`${"é".repeat(2048)}x`], false], [["a".repeat(2048), "b".repeat(2048)], true], [["a".repeat(2048), "b".repeat(2049)], false], - [["/".repeat(4096)], true], + [[`${"/".repeat(4095)}a`], true], [["\\".repeat(4097)], false], [Array(32).fill("a"), true], [Array(33).fill("a"), false], diff --git a/rs/moq-net/src/ietf/namespace.rs b/rs/moq-net/src/ietf/namespace.rs index f2809af8fcb..5ae3187446d 100644 --- a/rs/moq-net/src/ietf/namespace.rs +++ b/rs/moq-net/src/ietf/namespace.rs @@ -300,7 +300,7 @@ mod tests { (vec![format!("{}x", "é".repeat(2048))], false), (vec!["a".repeat(2048), "b".repeat(2048)], true), (vec!["a".repeat(2048), "b".repeat(2049)], false), - (vec!["/".repeat(4096)], true), + (vec![format!("{}a", "/".repeat(4095))], true), (vec!["\\".repeat(4097)], false), (vec!["a".to_owned(); Path::MAX_PARTS], true), (vec!["a".to_owned(); Path::MAX_PARTS + 1], false), From 85739e93cac996662954a465051a5d85fcd9c297 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 11:22:56 -0700 Subject: [PATCH 4/5] fix(net)!: bound paths to the moq-transport namespace limit Lower Path::MAX_BYTES to 4096 in Rust and JS, so every valid path fits a moq-transport namespace, and drop the advertisement preflight and skip logic. A namespace and track name over 4096 bytes together now fails its own SUBSCRIBE, TRACK_STATUS, FETCH, or PUBLISH at encode time instead of being sent to a peer that must close the session. Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-lite.md | 4 +- js/net/MIGRATE.md | 5 ++ js/net/api/index.api.md | 2 +- js/net/src/ietf/namespace.test.ts | 76 +++++++++-------- js/net/src/ietf/namespace.ts | 48 ++++++----- js/net/src/ietf/publish.ts | 3 +- js/net/src/ietf/publisher.test.ts | 94 --------------------- js/net/src/ietf/publisher.ts | 4 - js/net/src/ietf/subscribe.ts | 3 +- js/net/src/ietf/track.ts | 3 +- js/net/src/lite/announce.test.ts | 13 +-- js/net/src/path.ts | 9 ++- rs/moq-net/MIGRATE.md | 7 +- rs/moq-net/benches/path.rs | 23 ------ rs/moq-net/src/fuzz.rs | 15 ++-- rs/moq-net/src/ietf/fetch.rs | 8 +- rs/moq-net/src/ietf/namespace.rs | 90 +++++++++++++-------- rs/moq-net/src/ietf/publish.rs | 14 ++-- rs/moq-net/src/ietf/publisher.rs | 130 +----------------------------- rs/moq-net/src/ietf/subscribe.rs | 11 +-- rs/moq-net/src/ietf/subscriber.rs | 66 +++++++++++++++ rs/moq-net/src/ietf/track.rs | 5 +- rs/moq-net/src/lite/announce.rs | 23 ------ rs/moq-net/src/path/mod.rs | 6 +- 24 files changed, 241 insertions(+), 421 deletions(-) diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 09ef864b9fc..49a2714db34 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -442,7 +442,7 @@ A second ANNOUNCE_START for an already-advertised route is a protocol violation; The subscriber MUST close the session with a PROTOCOL_VIOLATION if it receives an ANNOUNCE_END, ANNOUNCE_UPDATE, or ANNOUNCE_RESTART referencing an Announce ID that was never assigned or already retired, an ANNOUNCE_START for a route that is already advertised, or any announcement before ANNOUNCE_OK. When the stream is closed, the subscriber MUST assume that all routes are now unavailable. -A broadcast path has at most 32 segments and at most 16,384 bytes; a receiver MUST close the session with a PROTOCOL_VIOLATION on a longer one, including one resolved by [Compression](#announce-compression). +A broadcast path has at most 32 segments and at most 4,096 bytes; a receiver MUST close the session with a PROTOCOL_VIOLATION on a longer one, including one resolved by [Compression](#announce-compression). A route covers a path when its prefix is a leading run of the path's segments; matching is per path segment, so a prefix never matches half a segment, and equality is byte-by-byte within each segment. A publisher answering a request stream presents each of its routes clamped to the intersection with the requested prefix: a route above the request's prefix appears as the request prefix itself (an empty suffix), which is exactly the covered set the subscriber may see. There MAY be multiple Announce Streams, potentially containing overlapping prefixes, that get their own ANNOUNCE_OK + announcements. @@ -1826,7 +1826,7 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 -- Limited a broadcast path to 32 segments and 16,384 bytes. +- Limited a broadcast path to 32 segments and 4,096 bytes. - A session without a negotiated subprotocol negotiates the version in a draft-14 CLIENT_SETUP and SERVER_SETUP exchange. Implementations apply this to moq-lite-03 through moq-lite-06 too. diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index e43e6f21b6a..d76fb52a939 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Paths fit moq-transport's 4096-byte namespace ([#6927](https://github.com/moq-dev/moq/pull/6927)) + +`Path.MAX_BYTES` is 4096, moq-transport's namespace limit, so every path fits every wire and an origin refuses a longer one. Keep broadcast paths within 4096 bytes. +A moq-transport namespace whose escaped path would exceed it is refused when decoded, and a SUBSCRIBE or PUBLISH whose namespace and track name exceed 4096 bytes together fails that request instead of being sent. + ### `Path.from` refuses empty segments ([#6648](https://github.com/moq-dev/moq/pull/6648)) `Path.from` throws on a component with a leading, trailing, or doubled `/` instead of collapsing it, so a path has one spelling, and `Path.decode` refuses one off the wire. A moq-transport namespace field ending in `/`, which escapes to an empty segment, is refused too. diff --git a/js/net/api/index.api.md b/js/net/api/index.api.md index 341dd9c39dc..48a1ebaa242 100644 --- a/js/net/api/index.api.md +++ b/js/net/api/index.api.md @@ -427,7 +427,7 @@ function isWebTransportSupported(): boolean; function join(path: Valid, other: Valid): Valid; -const MAX_BYTES = 16384; +const MAX_BYTES = 4096; const MAX_GROUP_CACHE_BYTES: number; diff --git a/js/net/src/ietf/namespace.test.ts b/js/net/src/ietf/namespace.test.ts index 69a7baa2629..12eab9058d8 100644 --- a/js/net/src/ietf/namespace.test.ts +++ b/js/net/src/ietf/namespace.test.ts @@ -1,6 +1,8 @@ import { expect, test } from "bun:test"; +import { ProtocolViolation } from "../error.ts"; +import * as Path from "../path.ts"; import { Reader, Writer } from "../stream.ts"; -import { decode, encodable, encode, fromTuple, reserved } from "./namespace.ts"; +import { decode, decodeTrack, encode, encodeTrack, fromTuple, reserved } from "./namespace.ts"; import { Version } from "./version.ts"; test("reserved namespaces preserve tuple fields and draft boundaries", () => { @@ -19,38 +21,42 @@ test("reserved namespaces preserve tuple fields and draft boundaries", () => { } }); -test("namespace encoding uses decoded UTF-8 field bounds", async () => { - for (const [parts, fits] of [ - [["é".repeat(2048)], true], - [[`${"é".repeat(2048)}x`], false], - [["a".repeat(2048), "b".repeat(2048)], true], - [["a".repeat(2048), "b".repeat(2049)], false], - [[`${"/".repeat(4095)}a`], true], - [["\\".repeat(4097)], false], - [Array(32).fill("a"), true], - [Array(33).fill("a"), false], - ] as [string[], boolean][]) { - const namespace = fromTuple(parts); - expect(encodable(namespace)).toBe(fits); - const bytes: number[] = []; - const writer = new Writer( - new WritableStream({ - write: (chunk) => { - bytes.push(...chunk); - }, - }), - Version.DRAFT_22, - ); - try { - if (fits) { - await encode(writer, namespace); - expect(await decode(new Reader(undefined, Uint8Array.from(bytes), Version.DRAFT_22))).toBe(namespace); - } else { - await expect(encode(writer, namespace)).rejects.toThrow("namespace exceeds"); - expect(bytes).toEqual([]); - } - } finally { - await writer.close(); - } - } +test("separators and escapes count toward a path but not its tuple fields", async () => { + const segments = Array(Path.MAX_PARTS - 1) + .fill("a") + .join("/"); + const deepest = Path.from(`${segments}/${"é".repeat(Path.MAX_BYTES / 2 - 31)}`); + expect(await roundTrip((w) => encode(w, deepest), decode)).toBe(deepest); + + expect(fromTuple([`${"/".repeat(Path.MAX_BYTES / 2 - 1)}a`]).length).toBe(Path.MAX_BYTES - 1); + expect(() => fromTuple([`${"/".repeat(Path.MAX_BYTES / 2)}a`])).toThrow(ProtocolViolation); }); + +test("a full track name counts field bytes and is refused before writing", async () => { + // 4095 path bytes holding 2048 field bytes. + const namespace = Path.from(`${"\\/".repeat(Path.MAX_BYTES / 2 - 1)}a`); + const name = "x".repeat(2048); + expect(await roundTrip((w) => encodeTrack(w, namespace, name), decodeTrack)).toEqual({ namespace, name }); + await expect(roundTrip((w) => encodeTrack(w, namespace, `${name}x`), decodeTrack)).rejects.toThrow( + "full track name exceeds", + ); +}); + +/** Write with `write`, then read the bytes back with `read`. */ +async function roundTrip(write: (w: Writer) => Promise, read: (r: Reader) => Promise): Promise { + const bytes: number[] = []; + const writer = new Writer( + new WritableStream({ + write: (chunk) => { + bytes.push(...chunk); + }, + }), + Version.DRAFT_22, + ); + try { + await write(writer); + } finally { + await writer.close(); + } + return read(new Reader(undefined, Uint8Array.from(bytes), Version.DRAFT_22)); +} diff --git a/js/net/src/ietf/namespace.ts b/js/net/src/ietf/namespace.ts index 52da6357631..8ef9d6c07a1 100644 --- a/js/net/src/ietf/namespace.ts +++ b/js/net/src/ietf/namespace.ts @@ -25,10 +25,12 @@ function toTuple(namespace: Path.Valid): string[] { /** * Convert IETF namespace tuple fields into the escaped moq-net path representation. * - * Throws when a field ends in `/`, which escapes to a path with an empty segment. + * Throws when a field ends in `/`, which escapes to a path with an empty segment, or when the + * separators and escapes push the path past {@link Path.MAX_BYTES}. */ export function fromTuple(parts: string[]): Path.Valid { const escaped = parts.map((part) => part.replaceAll("\\", "\\\\").replaceAll("/", "\\/")).join("/"); + if (tooLong(escaped)) throw new ProtocolViolation(`namespace exceeds ${Path.MAX_BYTES} bytes as a path`); try { return Path.from(escaped); } catch (cause) { @@ -38,37 +40,43 @@ export function fromTuple(parts: string[]): Path.Valid { /** Encode a moq-net namespace as an IETF namespace tuple. */ export async function encode(w: Writer, namespace: Path.Valid): Promise { + await writeFields(w, fields(namespace)); +} + +/** Encode a full track name: a namespace tuple then a track name, bounded together. */ +export async function encodeTrack(w: Writer, namespace: Path.Valid, name: string): Promise { + const parts = fields(namespace); + let size = utf8.encode(name).length; + for (const part of parts) size += utf8.encode(part).length; + if (size > MAX_BYTES) throw new Error(`full track name exceeds ${MAX_BYTES} bytes`); + await writeFields(w, parts); + await w.string(name); +} + +function fields(namespace: Path.Valid): string[] { + // The fields drop the path's separators and escapes, so a path within Path.MAX_BYTES fits. + if (tooLong(namespace)) throw new Error(`namespace exceeds ${Path.MAX_BYTES} bytes`); const parts = toTuple(namespace); - const error = check(parts); - if (error) throw new Error(error); + if (parts.length > Path.MAX_PARTS) throw new Error(`namespace exceeds ${Path.MAX_PARTS} parts`); + return parts; +} +async function writeFields(w: Writer, parts: string[]): Promise { await w.u53(parts.length); for (const part of parts) { await w.string(part); } } +// A UTF-16 unit is at most 3 UTF-8 bytes, so only a long path needs encoding to measure. +function tooLong(path: string): boolean { + return path.length > Path.MAX_BYTES / 3 && utf8.encode(path).length > Path.MAX_BYTES; +} + // Every draft caps a namespace, and a namespace plus track name, at 4096 bytes. const MAX_BYTES = 4096; const utf8 = new TextEncoder(); -function check(parts: string[]): string | undefined { - if (parts.length > Path.MAX_PARTS) return `namespace exceeds ${Path.MAX_PARTS} parts`; - let size = 0; - for (const part of parts) { - size += utf8.encode(part).length; - if (size > MAX_BYTES) return `namespace exceeds ${MAX_BYTES} bytes`; - } - return undefined; -} - -/** Whether one advertisement fits the namespace representation this wire carries. */ -export function encodable(namespace: Path.Valid): boolean { - // A UTF-16 unit costs at most three UTF-8 bytes; removing separators/escapes only shrinks it. - if (namespace.length <= MAX_BYTES / 3 && Path.check(namespace) === undefined) return true; - return check(toTuple(namespace)) === undefined; -} - /** Decode an IETF namespace tuple into an escaped moq-net namespace. */ export async function decode(r: Reader): Promise { return (await decodeFields(r)).namespace; diff --git a/js/net/src/ietf/publish.ts b/js/net/src/ietf/publish.ts index 1d8ca5f77ba..a89a33968e1 100644 --- a/js/net/src/ietf/publish.ts +++ b/js/net/src/ietf/publish.ts @@ -61,8 +61,7 @@ export class Publish { async #encode(w: Writer, version: IetfVersion): Promise { await w.u62(this.requestId); - await Namespace.encode(w, this.trackNamespace); - await w.string(this.trackName); + await Namespace.encodeTrack(w, this.trackNamespace, this.trackName); await w.u62(this.trackAlias); if (this.contentExists !== !!this.largest) { diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 369481a4427..72ca463f41f 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -194,100 +194,6 @@ function publisher( const IETF_VERSIONS = [Version.DRAFT_18, Version.DRAFT_22] as const; -test.each(IETF_VERSIONS)("%p: namespaces skip oversize routes and keep siblings flowing", async (version) => { - using _clock = manual(); - const pair = createMockTransportPair(version === Version.DRAFT_18 ? ALPN.DRAFT_18 : ALPN.DRAFT_22); - const session = new NativeSession(pair.server, version, true); - const { pub, origin } = publisher(pair.server, { session, requiresSolicitation: true, activeCount: true }); - const boundary = Path.from("b".repeat(4096)); - const held = [ - publish(origin, Path.from(`a${"é".repeat(2048)}`)), - publish(origin, boundary), - publish(origin, Path.from("sibling")), - ]; - try { - for (let attempt = 0; attempt < 2; attempt++) { - const subscription = await Stream.open(pair.client, { version }); - const accepted = await Stream.accept(pair.server, version); - if (!accepted) throw new Error("missing namespace request stream"); - const running = pub.runSubscribeNamespace( - new SubscribeNamespace({ requestId: BigInt(attempt), namespace: Path.empty() }), - accepted, - ); - const settled = running.catch(() => {}); - try { - expect(await subscription.reader.u53()).toBe(RequestOk.id); - const ok = await RequestOk.decode(subscription.reader, version); - const seen = new Set(); - for (let i = 0; i < 2; i++) { - expect(await subscription.reader.u53()).toBe(SubscribeNamespaceEntry.id); - seen.add((await SubscribeNamespaceEntry.decode(subscription.reader, version)).suffix); - } - expect(seen).toEqual(new Set([boundary, Path.from("sibling")])); - expect(ok.parameters.activeCount).toBe(2); - const late = publish(origin, Path.from("late")); - expect(await subscription.reader.u53()).toBe(SubscribeNamespaceEntry.id); - expect((await SubscribeNamespaceEntry.decode(subscription.reader, version)).suffix).toBe( - Path.from("late"), - ); - late.close(); - expect(await subscription.reader.u53()).toBe(SubscribeNamespaceEntryDone.id); - expect((await SubscribeNamespaceEntryDone.decode(subscription.reader, version)).suffix).toBe( - Path.from("late"), - ); - } finally { - subscription.close(); - await settled; - } - } - } finally { - for (const broadcast of held) broadcast.close(); - origin.close(); - pub.close(); - pair.client.close(); - pair.server.close(); - } -}); - -test.each(IETF_VERSIONS)("%p: unencodable namespaces spend no advertisement request ids", async (version) => { - using _clock = manual(); - const pair = createMockTransportPair(version === Version.DRAFT_18 ? ALPN.DRAFT_18 : ALPN.DRAFT_22); - const session = new NativeSession(pair.server, version, true); - const ids = spyOn(session, "nextRequestId"); - const { pub, origin } = publisher(pair.server, { session }); - const held = [ - publish(origin, Path.from(`a${"é".repeat(2048)}`)), - publish(origin, Path.from("first")), - publish(origin, Path.from("second")), - ]; - const peers: Stream[] = []; - const running = pub.runPublishNamespaces(); - const settled = running.catch(() => {}); - try { - const seen = new Set(); - for (let i = 0; i < 2; i++) { - const stream = await Stream.accept(pair.client, version); - if (!stream) throw new Error("missing advertisement stream"); - peers.push(stream); - expect(await stream.reader.u53()).toBe(PublishNamespace.id); - seen.add((await PublishNamespace.decode(stream.reader, version)).trackNamespace); - await stream.writer.u53(RequestOk.id); - await new RequestOk().encode(stream.writer, version); - } - expect(seen).toEqual(new Set([Path.from("first"), Path.from("second")])); - expect(ids).toHaveBeenCalledTimes(2); - } finally { - origin.close(); - for (const stream of peers) stream.close(); - await settled; - for (const broadcast of held) broadcast.close(); - pub.close(); - pair.client.close(); - pair.server.close(); - ids.mockRestore(); - } -}); - /** A request stream recording what the publisher writes, with the requester's side under test control. */ function requestStream(version: IetfVersion) { let input!: ReadableStreamDefaultController; diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 56e3c93d514..8719a3005b7 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -1631,9 +1631,6 @@ export class Publisher { const advertised = this.#advertised.peek(); const updated = advertised ? presented(prefix, advertised, carries) : new Map(); - for (const key of updated.keys()) { - if (!Namespace.encodable(key)) updated.delete(key); - } // A fresh pass sends a NAMESPACE for every entry, ahead of any change. if (!answered) { answered = true; @@ -1793,7 +1790,6 @@ export class Publisher { if ( (this.#hidden && hiddenBelow(Path.empty(), covered)) || !this.#gate.allows(covered) || - !Namespace.encodable(covered) || candidates.length === 0 ) continue; diff --git a/js/net/src/ietf/subscribe.ts b/js/net/src/ietf/subscribe.ts index d6cd89489fa..1668402d816 100644 --- a/js/net/src/ietf/subscribe.ts +++ b/js/net/src/ietf/subscribe.ts @@ -139,8 +139,7 @@ export class Subscribe { async #encode(w: Writer, version: IetfVersion): Promise { await w.u62(this.requestId); - await Namespace.encode(w, this.trackNamespace); - await w.string(this.trackName); + await Namespace.encodeTrack(w, this.trackNamespace, this.trackName); const params = new Parameters(); params.subscriberPriority = this.subscriberPriority; diff --git a/js/net/src/ietf/track.ts b/js/net/src/ietf/track.ts index c06ab1883ab..938c6cf9cdb 100644 --- a/js/net/src/ietf/track.ts +++ b/js/net/src/ietf/track.ts @@ -43,8 +43,7 @@ export class TrackStatusRequest { async #encode(w: Writer, version: IetfVersion): Promise { await w.u62(this.requestId); - await Namespace.encode(w, this.trackNamespace); - await w.string(this.trackName); + await Namespace.encodeTrack(w, this.trackNamespace, this.trackName); // Only the opt-out is worth bytes, and a peer before draft-22 would read // INCLUDE_PROPERTIES as unknown, which is a protocol violation. diff --git a/js/net/src/lite/announce.test.ts b/js/net/src/lite/announce.test.ts index ae1c2ab4377..d3d6e2a4d5a 100644 --- a/js/net/src/lite/announce.test.ts +++ b/js/net/src/lite/announce.test.ts @@ -1,7 +1,7 @@ import { expect, test } from "bun:test"; import * as Epoch from "../epoch.ts"; import { ProtocolViolation } from "../error.ts"; -import { Cost, HopSchema, MAX_HOPS, UNKNOWN_HOP } from "../hop.ts"; +import { Cost, HopSchema, UNKNOWN_HOP } from "../hop.ts"; import * as Path from "../path.ts"; import { Reader, Writer } from "../stream.ts"; import { @@ -47,17 +47,6 @@ async function roundTrip(msg: AnnounceBroadcast, version: Version): Promise { - const message: AnnounceBroadcast = { - status: "active", - suffix: Path.from("é".repeat(Path.MAX_BYTES / 2)), - hops: Array.from({ length: MAX_HOPS }, (_, n) => HopSchema.parse((1n << 51n) + BigInt(n))), - epoch: Epoch.parse("01900000-0000-7000-8000-000000000001"), - cost: (1n << 62n) - 1n, - }; - expect(await roundTrip(message, Version.DRAFT_07)).toEqual(message); -}); - test("AnnounceBroadcast round-trips on draft-05", async () => { const hops = [HopSchema.parse(7n)]; const gotActive = await roundTrip({ status: "active", suffix: Path.from("room/cam"), hops }, Version.DRAFT_05); diff --git a/js/net/src/path.ts b/js/net/src/path.ts index f1002dfa127..5cd36815cb6 100644 --- a/js/net/src/path.ts +++ b/js/net/src/path.ts @@ -54,8 +54,13 @@ export type Relative = string & { __brand: "Relative" }; */ export const MAX_PARTS = 32; -/** Maximum UTF-8 length of a path in bytes, leaving room in a 64 KiB moq-lite message for its other fields. */ -export const MAX_BYTES = 16384; +/** + * Maximum UTF-8 length of a path in bytes. + * + * Matches the IETF moq-transport limit of 4096 bytes for a namespace. Its tuple fields + * drop the path's separators and escapes, so every path fits both wires. + */ +export const MAX_BYTES = 4096; const utf8 = new TextEncoder(); diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index f5319291da5..b110770ddcc 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Paths fit moq-transport's 4096-byte namespace ([#6927](https://github.com/moq-dev/moq/pull/6927)) + +`Path::MAX_BYTES` is 4096, moq-transport's namespace limit, so every path fits every wire and building a longer one fails with `path::Invalid::TooLong`. Keep broadcast paths within 4096 bytes. +A moq-transport namespace whose escaped path would exceed it is refused when decoded, and a SUBSCRIBE, TRACK\_STATUS, FETCH, or PUBLISH whose namespace and track name exceed 4096 bytes together fails that request with `Error::Encode` instead of being sent. + ### Building a `Path` refuses empty segments ([#6648](https://github.com/moq-dev/moq/pull/6648)) `Path::new` and `TryFrom` fail with `path::Invalid::EmptySegment` on a leading, trailing, or doubled `/` instead of collapsing it, so a path has one spelling; the empty path is still the root. Origin entry points report it as `Error::EmptySegment`, and `Path::from_static` panics on it. @@ -20,7 +25,7 @@ The moql SETUP (no ALPN, or `moql`) now negotiates only moq-lite versions. ### `Path` and `Relative` have no lifetime -There is one owned `Path` type: a short path is stored inline and a long one shares a reference-counted buffer, so cloning stays cheap. Every path is at most `Path::MAX_BYTES` (16 KiB). +There is one owned `Path` type: a short path is stored inline and a long one shares a reference-counted buffer, so cloning stays cheap. Every path is at most `Path::MAX_BYTES`. - `PathOwned`, `Path<'a>`, and `Path<'static>` are all `Path`; `RelativeOwned` and `Relative<'a>` are `Relative`. - `Path::{to_owned, into_owned, borrow}` and the `Relative` equivalents are gone: drop the call, or `clone()` a `&Path` you need to keep. diff --git a/rs/moq-net/benches/path.rs b/rs/moq-net/benches/path.rs index 6dff0be4cda..566d08a3635 100644 --- a/rs/moq-net/benches/path.rs +++ b/rs/moq-net/benches/path.rs @@ -192,32 +192,9 @@ fn fanout() { } } -fn namespaces() { - for (name, prefix) in [("short", SHORT), ("long", LONG)] { - for routes in [1, 64, 1024] { - let paths: Vec<_> = (0..routes) - .map(|i| Path::new(&format!("{prefix}/{i}")).unwrap()) - .collect(); - for peers in [1, 8, 32] { - allocs( - &format!("namespace/emit/{name}/{routes}r_{peers}p"), - 16, - routes * peers, - || { - for _ in 0..peers { - black_box(encode_namespaces(&paths)); - } - }, - ); - } - } - } -} - fn main() { paths(); decode(); requests(); fanout(); - namespaces(); } diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 255afe1d6a2..34560d57f46 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -180,11 +180,11 @@ fn ietf_paths(data: &[u8], version: ietf::Version) { if let Ok(path) = ietf::namespace::decode_namespace(&mut Decoder::new(data, form)) { assert_strict(&path); let mut encoded = Vec::new(); - if ietf::namespace::encode_namespace(&mut Encoder::new(&mut encoded, form), &path).is_ok() { - let echo = ietf::namespace::decode_namespace(&mut Decoder::new(&encoded, form)) - .expect("could not decode our own namespace"); - assert_eq!(echo, path, "namespace did not survive a round trip"); - } + ietf::namespace::encode_namespace(&mut Encoder::new(&mut encoded, form), &path) + .expect("could not encode a decoded namespace"); + let echo = ietf::namespace::decode_namespace(&mut Decoder::new(&encoded, form)) + .expect("could not decode our own namespace"); + assert_eq!(echo, path, "namespace did not survive a round trip"); } if let Ok((path, _)) = ietf::namespace::decode_track(&mut Decoder::new(data, form)) { assert_strict(&path); @@ -495,14 +495,11 @@ pub fn decode_announces(data: &[u8], compress: bool) -> Vec { resolved } -/// Encode the moq-transport NAMESPACE messages this wire can carry from `paths`. +/// Encode a moq-transport NAMESPACE message for each of `paths`. pub fn encode_namespaces(paths: &[Path]) -> Vec { let mut data = Vec::new(); let w = &mut Encoder::new(&mut data, BENCH_IETF.into()); for path in paths { - if !ietf::namespace::encodable(path) { - continue; - } let msg = ietf::Namespace { suffix: path.clone(), params: None, diff --git a/rs/moq-net/src/ietf/fetch.rs b/rs/moq-net/src/ietf/fetch.rs index ba0b12238e9..de3334ca157 100644 --- a/rs/moq-net/src/ietf/fetch.rs +++ b/rs/moq-net/src/ietf/fetch.rs @@ -5,7 +5,7 @@ use crate::{ coding::{Decode, DecodeError, Decoder, Encode, EncodeError, Encoder}, ietf::{ Filter, GroupOrder, Location, Opaque, RequestId, - namespace::{decode_track, encode_namespace}, + namespace::{decode_track, encode_track}, }, }; @@ -54,8 +54,7 @@ impl Encode for FetchType<'_> { end, } => { w.u8(1); - encode_namespace(w, namespace)?; - w.string(track)?; + encode_track(w, namespace, track)?; start.encode(w, version)?; end.encode(w, version)?; } @@ -171,8 +170,7 @@ impl Message for Fetch<'_> { else { return Err(EncodeError::Version); }; - encode_namespace(w, namespace)?; - w.string(track)?; + encode_track(w, namespace, track)?; // INCLUDE_PROPERTIES defaults to 1, so only the opt-out is worth bytes. let include_properties = (!self.properties_wanted).then_some(false); encode_params!(w, version, diff --git a/rs/moq-net/src/ietf/namespace.rs b/rs/moq-net/src/ietf/namespace.rs index 5ae3187446d..6f17b30565a 100644 --- a/rs/moq-net/src/ietf/namespace.rs +++ b/rs/moq-net/src/ietf/namespace.rs @@ -28,7 +28,8 @@ fn to_tuple(namespace: &Path) -> Vec { parts } -// A field ending in `/` escapes to a path with an empty segment, so it is refused. +// A field ending in `/` escapes to a path with an empty segment, so it is refused, as is a +// namespace whose separators and escapes push its path past `Path::MAX_BYTES`. fn from_tuple(parts: &[&str]) -> Result { // Sized for the common case of no escapes, so building it rarely grows. let mut path = String::with_capacity(parts.iter().map(|part| part.len() + 1).sum()); @@ -48,40 +49,50 @@ fn from_tuple(parts: &[&str]) -> Result { Path::try_from(path) } -/// Helper function to encode namespace as tuple of strings +/// Encode a Track Namespace. pub fn encode_namespace(w: &mut Encoder<'_>, namespace: &Path) -> Result<(), EncodeError> { + write_fields(w, &fields(namespace)?) +} + +/// Encode a Full Track Name: a Track Namespace, then a Track Name, at most 4096 bytes together. +pub fn encode_track(w: &mut Encoder<'_>, namespace: &Path, name: &str) -> Result<(), EncodeError> { let parts = fields(namespace)?; - w.varint(parts.len() as u64)?; - for part in parts { - w.string(&part)?; + if parts.iter().map(String::len).sum::() + name.len() > MAX_BYTES { + return Err(BoundsExceeded.into()); } - Ok(()) + write_fields(w, &parts)?; + w.string(name) } fn fields(namespace: &Path) -> Result, EncodeError> { let parts = to_tuple(namespace); - if parts.len() > Path::MAX_PARTS || parts.iter().map(String::len).sum::() > MAX_BYTES { + // The fields drop the path's separators and escapes, so `Path::MAX_BYTES` already bounds + // their bytes. A path's depth is unbounded until it reaches a wire. + if parts.len() > Path::MAX_PARTS { return Err(BoundsExceeded.into()); } Ok(parts) } -/// Whether one advertisement fits the namespace representation this wire carries. -pub(crate) fn encodable(namespace: &Path) -> bool { - // Separators and escaping only shrink the wire fields, so ordinary paths need no tuple allocation. - if namespace.len() <= MAX_BYTES && namespace.parts().count() <= Path::MAX_PARTS { - return true; +fn write_fields(w: &mut Encoder<'_>, parts: &[String]) -> Result<(), EncodeError> { + w.varint(parts.len() as u64)?; + for part in parts { + w.string(part)?; } - fields(namespace).is_ok() + Ok(()) } /// The most bytes a Track Namespace, or a whole Full Track Name, may hold on the wire. const MAX_BYTES: usize = 4096; +// Every path encodes as a namespace. +const _: () = assert!(Path::MAX_BYTES <= MAX_BYTES); + /// Decode a namespace tuple, along with the sum of its field lengths. /// /// Every limit here closes the session as a PROTOCOL_VIOLATION (draft-22 section 8.7): /// more than 32 fields, an empty field, a field ending in `/`, or more than 4096 bytes. +/// So does a namespace too long as a [`Path`], which the draft allows but no path can hold. fn decode_fields(r: &mut Decoder<'_>) -> Result<(Path, usize), DecodeError> { let count = r.varint()?; @@ -292,29 +303,40 @@ mod tests { } } - /// Limits count decoded UTF-8 field bytes, not escaped path bytes or separators. + /// Separators and escapes count toward a path but not its tuple fields, so every path + /// encodes, while a namespace dense with escaped slashes decodes only if its path fits. #[test] - fn encoding_uses_the_same_field_bounds_as_decoding() { - for (parts, fits) in [ - (vec!["é".repeat(2048)], true), - (vec![format!("{}x", "é".repeat(2048))], false), - (vec!["a".repeat(2048), "b".repeat(2048)], true), - (vec!["a".repeat(2048), "b".repeat(2049)], false), - (vec![format!("{}a", "/".repeat(4095))], true), - (vec!["\\".repeat(4097)], false), - (vec!["a".to_owned(); Path::MAX_PARTS], true), - (vec!["a".to_owned(); Path::MAX_PARTS + 1], false), - ] { - let refs: Vec<_> = parts.iter().map(String::as_str).collect(); - let path = from_tuple(&refs).unwrap(); - assert_eq!(encodable(&path), fits); + fn path_bound_covers_namespace_fields() { + let segments = ["a"; Path::MAX_PARTS - 1].join("/"); + let deepest = format!("{segments}/{}", "é".repeat(Path::MAX_BYTES / 2 - 31)); + let deepest = Path::new(&deepest).unwrap(); + assert_eq!(deepest.len(), Path::MAX_BYTES); + assert_eq!(decode_ns(&encode_path(&deepest)), deepest); + + let fits = format!("{}a", "/".repeat(Path::MAX_BYTES / 2 - 1)); + assert_eq!(decode_ns(&encode_tuple(&[&fits])).len(), Path::MAX_BYTES - 1); + let too_long = format!("{}a", "/".repeat(Path::MAX_BYTES / 2)); + let err = decode_namespace(&mut Decoder::new(&encode_tuple(&[&too_long]), FORM)).unwrap_err(); + assert!(matches!(err, DecodeError::BoundsExceeded), "{err:?}"); + } + + /// A Full Track Name counts its namespace's field bytes plus its name, and one too long + /// is refused before anything is written. + #[test] + fn full_track_name_is_bounded_on_encode() { + // 4095 path bytes holding 2048 field bytes. + let namespace = Path::new(&format!("{}a", r"\/".repeat(Path::MAX_BYTES / 2 - 1))).unwrap(); + for (len, ok) in [(2048, true), (2049, false)] { + let name = "x".repeat(len); let mut bytes = Vec::new(); - let result = encode_namespace(&mut Encoder::new(&mut bytes, FORM), &path); - assert_eq!(result.is_ok(), fits); - if fits { - assert_eq!(decode_ns(&bytes), path); - } else { - assert!(bytes.is_empty(), "refuse before emitting namespace fields"); + let result = encode_track(&mut Encoder::new(&mut bytes, FORM), &namespace, &name); + assert_eq!(result.is_ok(), ok, "{len}"); + match ok { + true => assert_eq!( + decode_track(&mut Decoder::new(&bytes, FORM)).unwrap(), + (namespace.clone(), name) + ), + false => assert!(bytes.is_empty(), "refuse before writing"), } } } diff --git a/rs/moq-net/src/ietf/publish.rs b/rs/moq-net/src/ietf/publish.rs index c9610438e0a..6a8cac95485 100644 --- a/rs/moq-net/src/ietf/publish.rs +++ b/rs/moq-net/src/ietf/publish.rs @@ -110,7 +110,7 @@ use crate::{ coding::{Decode, DecodeError, Decoder, Encode, EncodeError, Encoder}, ietf::{ Filter, GroupOrder, Location, Properties, RequestId, - namespace::{decode_track, encode_namespace}, + namespace::{decode_track, encode_track}, }, }; @@ -213,8 +213,7 @@ impl Message for Publish<'_> { fn encode_msg(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { self.request_id.encode(w, version)?; - encode_namespace(w, &self.track_namespace)?; - w.string(&self.track_name)?; + encode_track(w, &self.track_namespace, &self.track_name)?; w.varint(self.track_alias)?; // GROUP_ORDER is illegal on PUBLISH on draft-18 (a peer closes the session with @@ -359,8 +358,7 @@ mod tests { let mut body = Vec::new(); let w = &mut Encoder::new(&mut body, version.into()); RequestId(1).encode(w, version)?; - super::super::namespace::encode_namespace(w, &crate::Path::from_static("broadcast"))?; - w.string("video")?; + super::super::namespace::encode_track(w, &crate::Path::from_static("broadcast"), "video")?; w.varint(1u64)?; // track alias // SUBSCRIBER_PRIORITY then LOCATION_FILTER, delta encoded from 0. @@ -382,8 +380,7 @@ mod tests { let mut body = Vec::new(); let w = &mut Encoder::new(&mut body, version.into()); RequestId(1).encode(w, version)?; - super::super::namespace::encode_namespace(w, &crate::Path::from_static("broadcast"))?; - w.string("video")?; + super::super::namespace::encode_track(w, &crate::Path::from_static("broadcast"), "video")?; w.varint(1u64)?; encode_params!(w, version, 0x20 => 128u8); Properties::default().encode(w, version)?; @@ -408,8 +405,7 @@ mod tests { let mut buf = Vec::new(); let w = &mut Encoder::new(&mut buf, version.into()); RequestId(1).encode(w, version).unwrap(); - encode_namespace(w, &Path::from_static("t")).unwrap(); - w.string("v").unwrap(); + encode_track(w, &Path::from_static("t"), "v").unwrap(); w.varint(1).unwrap(); w.varint(1).unwrap(); w.varint(key).unwrap(); diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index a974d2adcd3..7c682cb4b56 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -3675,11 +3675,7 @@ impl Publisher { path: &crate::Path, ) -> Option<(Advert, Option)> { let watch = ns.watched.get(suffix)?; - let wire_path = match &ns.target { - Target::Requests => path, - Target::Inline(_) => suffix, - }; - let advert = match ns.permitted(path) && super::namespace::encodable(wire_path) { + let advert = match ns.permitted(path) { true => self.select(&watch.route, &ns.peer, ns.epochs), false => Advert::None, }; @@ -4139,7 +4135,6 @@ impl Publisher { let count = initial .iter() .filter(|update| permit.matches(prefix.rejoin(&update.prefix).as_str())) - .filter(|update| super::namespace::encodable(&update.prefix)) .filter(|update| self.select(&update.route, &peer, declared.epoch).wanted()) .count(); (initial, Some(count as u64)) @@ -8714,8 +8709,7 @@ mod serve_tests { let encode = |body: &mut Vec| -> Result<(), crate::coding::EncodeError> { let w = &mut Encoder::new(body, version.into()); RequestId(REQUEST_ID).encode(w, version)?; - super::super::namespace::encode_namespace(w, &crate::Path::from_static("room"))?; - w.string("video")?; + super::super::namespace::encode_track(w, &crate::Path::from_static("room"), "video")?; encode_params!(w, version, 0x04 => 7u64, super::super::epoch::EPOCH_PARAM => Some(stale.clone()), @@ -14253,126 +14247,6 @@ mod tests { slot } - /// An unencodable namespace cannot poison discovery or its initial active count. - #[moq_clock::test] - async fn namespaces_skip_oversize_routes_and_keep_siblings_flowing() { - for version in [Version::Draft18, Version::Draft22] { - let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); - let too_long = format!("a{}", "é".repeat(2048)); - let boundary = "b".repeat(4096); - let _oversize = origin - .announce(crate::Path::new(&too_long).unwrap(), Default::default()) - .unwrap(); - let _boundary = origin - .announce(crate::Path::new(&boundary).unwrap(), Default::default()) - .unwrap(); - let _sibling = origin.announce("sibling", Default::default()).unwrap(); - let gate = kio::Producer::new(true); - let session = SinkSession::gated_bi(gate.consume()); - let log = session.log.clone(); - let setup = peer::PeerSetup::default(); - setup.set(peer::Peer { - solicit: Some(true), - active_count: true, - ..Default::default() - }); - let publisher = Publisher::new( - crate::time::sim(), - crate::driver::Transport::new(session.clone()), - origin.consume(), - Control::new(false), - None, - setup, - version, - ); - for attempt in 0..2 { - log.writes.lock().unwrap().clear(); - let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), version) - .await - .unwrap(); - let mut run = std::pin::pin!(publisher.clone().run_subscribe_namespace_stream( - stream, - ietf::SubscribeNamespace { - request_id: RequestId(1 + attempt), - namespace: crate::Path::empty(), - hidden: false, - } - )); - assert!(futures::poll!(run.as_mut()).is_pending()); - let bytes = log.writes.lock().unwrap().clone(); - let events = namespace_events(&bytes, version); - assert_eq!(events.len(), 2, "only encodable namespaces are advertised"); - assert!(events.contains(&(ietf::Namespace::ID, boundary.clone()))); - assert!(events.contains(&(ietf::Namespace::ID, "sibling".to_owned()))); - let mut wire = crate::coding::Decoder::new(&bytes, version.into()); - assert_eq!(wire.varint().unwrap(), ietf::RequestOk::ID); - assert_eq!(ietf::RequestOk::decode(&mut wire, version).unwrap().active, Some(2)); - let late = origin.announce("late", Default::default()).unwrap(); - assert!(futures::poll!(run.as_mut()).is_pending()); - let events = namespace_events(&log.writes.lock().unwrap(), version); - assert!( - events.contains(&(ietf::Namespace::ID, "late".to_owned())), - "later siblings still flow" - ); - drop(late); - assert!(futures::poll!(run.as_mut()).is_pending()); - assert!( - namespace_events(&log.writes.lock().unwrap(), version) - .contains(&(ietf::NamespaceDone::ID, "late".to_owned())) - ); - } - } - } - - #[moq_clock::test] - async fn unencodable_namespaces_spend_no_advertisement_request_ids() { - for version in [Version::Draft18, Version::Draft22] { - let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); - let _oversize = origin - .announce( - crate::Path::new(&format!("a{}", "é".repeat(2048))).unwrap(), - Default::default(), - ) - .unwrap(); - let _first = origin.announce("first", Default::default()).unwrap(); - let _second = origin.announce("second", Default::default()).unwrap(); - let refusal = publish_namespace_error(version, 0).await; - let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![ - refusal.clone(), - refusal.clone(), - refusal, - ]); - let log = session.log.clone(); - let publisher = Publisher::new( - crate::time::sim(), - crate::driver::Transport::new(session), - origin.consume(), - Control::new(false), - None, - declared(Some(false)), - version, - ); - let reference = Control::new(false); - let expected_ids = [reference.next_request_id(), reference.next_request_id()]; - let mut run = std::pin::pin!(publisher.clone().run_publish_namespaces()); - assert!(futures::poll!(run.as_mut()).is_pending()); - let bytes = log.writes.lock().unwrap().clone(); - let mut wire = crate::coding::Decoder::new(&bytes, version.into()); - let mut seen = Vec::new(); - while !wire.is_empty() { - assert_eq!(wire.varint().unwrap(), ietf::PublishNamespace::ID); - let message = ietf::PublishNamespace::decode(&mut wire, version).unwrap(); - seen.push((message.request_id, message.track_namespace)); - } - assert_eq!(seen.len(), 2); - assert_eq!([seen[0].0, seen[1].0], expected_ids); - assert_eq!(publisher.control.next_request_id(), reference.next_request_id()); - let mut names: Vec<_> = seen.into_iter().map(|(_, path)| path.as_str().to_owned()).collect(); - names.sort(); - assert_eq!(names, ["first", "second"]); - } - } - /// An absent max age preserves everything the producer still retains. #[test] fn serving_subscription_keeps_retained_backlog() { diff --git a/rs/moq-net/src/ietf/subscribe.rs b/rs/moq-net/src/ietf/subscribe.rs index b924dcef53b..9d64d99bbe3 100644 --- a/rs/moq-net/src/ietf/subscribe.rs +++ b/rs/moq-net/src/ietf/subscribe.rs @@ -12,7 +12,7 @@ use super::token::RequestToken; use super::Message; use super::epoch; -use super::namespace::{decode_track, encode_namespace}; +use super::namespace::{decode_track, encode_track}; use super::Version; @@ -128,8 +128,7 @@ impl Message for Subscribe<'_> { fn encode_msg(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { self.request_id.encode(w, version)?; - encode_namespace(w, &self.track_namespace)?; - w.string(&self.track_name)?; + encode_track(w, &self.track_namespace, &self.track_name)?; // FILL_PARAMETERS arrived in draft-22. Sending it to an older peer would be an // unknown parameter, which is a protocol violation, so it is dropped instead. @@ -419,8 +418,7 @@ mod tests { let mut buf = Vec::new(); let w = &mut Encoder::new(&mut buf, version.into()); RequestId(1).encode(w, version)?; - encode_namespace(w, &Path::from_static("test"))?; - w.string("video")?; + encode_track(w, &Path::from_static("test"), "video")?; encode_params!(w, version, 0x04 => millis, ); @@ -605,8 +603,7 @@ mod tests { let mut body = Vec::new(); let w = &mut Encoder::new(&mut body, version.into()); RequestId(1).encode(w, version)?; - encode_namespace(w, &crate::Path::from_static("broadcast"))?; - w.string("video")?; + encode_track(w, &crate::Path::from_static("broadcast"), "video")?; encode_params!(w, version, 0x02 => 5000u64, 0x06 => 9000u64, diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index fb202d52f97..5d4d9c5d880 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -16344,6 +16344,72 @@ mod track_property_tests { } } +#[cfg(test)] +mod full_track_name_tests { + use super::*; + use crate::{model::ProduceTest, transport::memory}; + + /// A Full Track Name too long for the wire fails only its own subscription, and the + /// session carries the next one. + #[moq_clock::test] + async fn an_unencodable_track_name_fails_only_its_subscription() { + for version in [Version::Draft18, Version::Draft22] { + let (session, peer) = memory::pair(memory::Config::default()); + let (tasks, _task_set) = TaskSet::new(); + let subscriber = Subscriber::new( + crate::time::sim(), + crate::driver::Transport::new(session.clone()), + crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + Control::new(false), + None, + peer::PeerSetup::declared(peer::Peer::default()), + crate::Hop::new(1).unwrap(), + None, + version, + tasks, + Default::default(), + ); + // One byte past 4096 with the 9-byte namespace. + let long = "x".repeat(4088); + for (name, fits) in [(long.as_str(), false), ("video", true)] { + let broadcast = crate::broadcast::Info::default().produce(); + let mut dynamic = broadcast.requests(); + let track = broadcast.consume().track(name).unwrap(); + let mut subscribing = std::pin::pin!(track.subscribe(None)); + let request = dynamic.requested_track().await.expect("no track requested"); + let mut subscriber = subscriber.clone(); + let serving = moq_clock::sim::spawn(async move { + subscriber + .run_subscribe(Path::from_static("broadcast"), dynamic, None, request) + .await; + }); + if fits { + // The refused request's stream was reset before its header, which a peer ignores. + let id = loop { + let mut stream = Stream::accept(&mut crate::driver::Transport::new(peer.clone()), version) + .await + .unwrap(); + if let Ok(id) = stream.reader.varint().await { + break id; + } + }; + assert_eq!(id, ietf::Subscribe::ID, "{version}"); + } else { + let refused = moq_clock::timeout(Duration::from_secs(1), &mut subscribing) + .await + .expect("the subscription was never refused"); + assert!( + matches!(refused, Err(Error::Encode(crate::coding::EncodeError::BoundsExceeded))), + "{version}: {:?}", + refused.map(|_| ()) + ); + } + serving.abort(); + } + } + } +} + #[cfg(test)] mod compression_negotiation_tests { use super::*; diff --git a/rs/moq-net/src/ietf/track.rs b/rs/moq-net/src/ietf/track.rs index c9937969f0b..a6dc1675c55 100644 --- a/rs/moq-net/src/ietf/track.rs +++ b/rs/moq-net/src/ietf/track.rs @@ -11,7 +11,7 @@ use crate::{ }; use super::Message; -use super::namespace::encode_namespace; +use super::namespace::encode_track; use super::Version; @@ -34,8 +34,7 @@ impl Message for TrackStatus<'_> { fn encode_msg(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { self.request_id.encode(w, version)?; - encode_namespace(w, &self.track_namespace)?; - w.string(&self.track_name)?; + encode_track(w, &self.track_namespace, &self.track_name)?; // Only the opt-out is worth bytes, and an older peer would read the // parameter as unknown, which is a protocol violation. diff --git a/rs/moq-net/src/lite/announce.rs b/rs/moq-net/src/lite/announce.rs index 83aa3d380d9..8b13570a2f5 100644 --- a/rs/moq-net/src/lite/announce.rs +++ b/rs/moq-net/src/lite/announce.rs @@ -597,29 +597,6 @@ mod tests { use super::*; use bytes::Buf; - /// The model's path bound leaves room for every other literal START field. - #[test] - fn a_maximal_literal_start_fits_the_lite_message() { - let path = Path::new(&"é".repeat(Path::MAX_BYTES / 2)).unwrap(); - let hops = Hops::try_from( - (0..crate::origin::HopLimit::default().get()) - .map(|n| Hop::new((1 << 51) + n as u64).unwrap()) - .collect::>(), - ) - .unwrap(); - let version = Version::Lite07; - let message = AnnounceBroadcast::Active { - suffix: PathRef::literal(path.clone()), - hops: HopsRef::literal(hops), - epoch: Some("01900000-0000-7000-8000-000000000001".parse().unwrap()), - cost: Cost::MAX, - }; - let encoded = message.encode_bytes(version).unwrap(); - let (decoded, consumed) = AnnounceBroadcast::decode_slice(&encoded, version).unwrap(); - assert_eq!(consumed, encoded.len()); - assert!(matches!(decoded, AnnounceBroadcast::Active { suffix, .. } if suffix.rest == path)); - } - /// A legacy initial set is held to the announce budget before any suffix decodes, so /// a body of one-byte suffixes cannot become a list many times its size. #[test] diff --git a/rs/moq-net/src/path/mod.rs b/rs/moq-net/src/path/mod.rs index 63612ca5bb4..d8ffaf11943 100644 --- a/rs/moq-net/src/path/mod.rs +++ b/rs/moq-net/src/path/mod.rs @@ -129,9 +129,9 @@ impl Path { /// Maximum length of a path in bytes. /// - /// A quarter of the 64 KiB moq-lite message limit, leaving room for every other - /// field of a message that carries a path. - pub const MAX_BYTES: usize = 16 * 1024; + /// Matches the IETF moq-transport limit of 4096 bytes for a namespace. Its tuple fields + /// drop the path's separators and escapes, so every path fits both wires. + pub const MAX_BYTES: usize = 4096; /// Create a new Path by copying a string slice. /// From d7fb02125901c7df99118df1c88cc5e9eb62ddcb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 12:38:00 -0700 Subject: [PATCH 5/5] docs(net): list TRACK_STATUS and drop stale 16 KiB path limit Co-Authored-By: Claude Opus 5.5 --- js/net/MIGRATE.md | 2 +- rs/moq-net/src/model/origin.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index d76fb52a939..ebee10186ff 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -7,7 +7,7 @@ How to update your code for each breaking change, newest first. ### Paths fit moq-transport's 4096-byte namespace ([#6927](https://github.com/moq-dev/moq/pull/6927)) `Path.MAX_BYTES` is 4096, moq-transport's namespace limit, so every path fits every wire and an origin refuses a longer one. Keep broadcast paths within 4096 bytes. -A moq-transport namespace whose escaped path would exceed it is refused when decoded, and a SUBSCRIBE or PUBLISH whose namespace and track name exceed 4096 bytes together fails that request instead of being sent. +A moq-transport namespace whose escaped path would exceed it is refused when decoded, and a SUBSCRIBE, TRACK\_STATUS, or PUBLISH whose namespace and track name exceed 4096 bytes together fails that request instead of being sent. ### `Path.from` refuses empty segments ([#6648](https://github.com/moq-dev/moq/pull/6648)) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 9551dc406af..0e9e3018f25 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2149,7 +2149,7 @@ impl Producer { /// producer may publish under (after [`scope`](Self::scope)) or beneath a /// [`mount`](Self::mount), /// [`Error::BoundsExceeded`] if the full rooted path exceeds - /// [`Path::MAX_PARTS`] or 16 KiB, [`Error::InvalidPath`] if it holds a segment no + /// [`Path::MAX_PARTS`] or [`Path::MAX_BYTES`], [`Error::InvalidPath`] if it holds a segment no /// pattern can spell (`*` or `**`), [`Error::NavigationSegment`] for a literal `.` or `..`, /// or [`Error::Closed`] once the origin's /// [`Driver`] has been dropped. @@ -2446,7 +2446,7 @@ impl Producer { /// `at` or `target` overlaps a mount point this producer already has, or `at` /// overlaps an existing target or its own: mounts never chain, in whatever order they are added, /// so a target is always read as the origin holds it. Mounts may share a target. - /// [`Error::BoundsExceeded`] if either rooted path exceeds [`Path::MAX_PARTS`] or 16 KiB, + /// [`Error::BoundsExceeded`] if either rooted path exceeds [`Path::MAX_PARTS`] or [`Path::MAX_BYTES`], /// and [`Error::InvalidPath`] if `at` holds a segment no pattern can spell. pub fn mount( &self,