From 40f31456a90634839039ea83f0495cf02fcca075 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 20:50:22 -0700 Subject: [PATCH 01/15] quest: claim m0/audit-interop Co-Authored-By: Claude Opus 5.5 From ca221d7992fe7b6a2e319d34f40ab5aa029943fd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 20:59:43 -0700 Subject: [PATCH 02/15] fix(net): never offer moq-00 over WebTransport, and scope the CLI subscriber Draft-14 predates WebTransport subprotocols, so a peer that accepts moq-00 picks draft-14 over a newer draft both support. Offer moql instead and negotiate in SETUP, as js/net does. Draft-14 and draft-15 cannot ask for the empty namespace, so the CLI subscriber discovered nothing from moxygen. Scope it to the consumed broadcasts, as the HLS and RTC exports scope theirs. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/main.rs | 37 ++++++++++++++++++++++++++++---- rs/moq-net/src/version.rs | 27 +++++++++++++++++++++++ rs/moq-tokio/src/quic/backend.rs | 4 ++-- rs/moq-wasm/src/transport.rs | 2 +- 4 files changed, 63 insertions(+), 7 deletions(-) diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 196863afe11..97c9299e905 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -417,6 +417,17 @@ impl Directions { } } +/// The broadcasts `names` and everything beneath them, as patterns to scope an origin. +fn consumed(names: impl IntoIterator>) -> anyhow::Result { + names + .into_iter() + .map(|name| { + let name = name.as_ref(); + moq_net::Pattern::subtree(name).with_context(|| format!("invalid broadcast name `{name}`")) + }) + .collect() +} + /// Attach the shared Origin to the MoQ network: dial a relay, accept inbound /// sessions, mesh with the LAN, or any combination. /// @@ -439,11 +450,16 @@ impl Directions { /// the estimate is only meaningful across everything sharing it. A stage that built /// its own would split its own tracks correctly and still oversubscribe every other /// stage on the same connection, which is the whole problem. +/// +/// The dialed relay is asked only for `consumed`, the broadcasts the stages read. +/// moq-transport drafts 14 and 15 cannot ask for the empty namespace, so an +/// unscoped subscriber discovers nothing from a peer that announces nothing unasked. async fn spawn_moq( moq: &MoqSide, net: &Net, cluster: moq_relay::cluster::Cluster, directions: Directions, + consumed: &moq_net::Patterns, tasks: &mut JoinSet>, ) -> anyhow::Result { let cluster = cluster @@ -479,7 +495,8 @@ async fn spawn_moq( dialer = dialer.with_publisher(origin.consume()); } if directions.consume { - dialer = dialer.with_subscriber(origin.clone()); + let scoped = origin.scope("", consumed).context("failed to scope the origin to the consumed broadcasts")?; + dialer = dialer.with_subscriber(scoped); } // Both directions ride one connection, so this dials directly rather than @@ -548,7 +565,7 @@ async fn run_play(moq: MoqSide, args: play::Args, net: Net) -> anyhow::Result<() client, connection, .. - } = spawn_moq(&moq, &net, cluster.clone(), directions, &mut tasks).await?; + } = spawn_moq(&moq, &net, cluster.clone(), directions, &consumed([&name])?, &mut tasks).await?; let result = play::run(origin.consume(), name, args, tasks).await; detach(connection, client, &cluster).await; @@ -576,13 +593,23 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res let mut client = None; let mut connection = None; let result = async { + let exports = stages.iter().filter(|stage| matches!(stage, Command::Export(_))); + let consumed = consumed(exports.map(|stage| stage.broadcast(&moq)))?; let Attached { bandwidth, origin, client: dialer, connection: attached, shutdown, - } = spawn_moq(&moq, &net, cluster.clone(), Directions::of(&stages), &mut tasks).await?; + } = spawn_moq( + &moq, + &net, + cluster.clone(), + Directions::of(&stages), + &consumed, + &mut tasks, + ) + .await?; client = dialer; connection = attached; @@ -1275,7 +1302,8 @@ mod tests { publish: true, consume: false, }; - let Err(err) = spawn_moq(&invocation.moq, &net, cluster, directions, &mut tasks).await else { + let consumed = moq_net::Patterns::new(); + let Err(err) = spawn_moq(&invocation.moq, &net, cluster, directions, &consumed, &mut tasks).await else { panic!("the occupied port must fail startup"); }; assert!(err.to_string().contains("failed to bind listeners"), "{err:#}"); @@ -1431,6 +1459,7 @@ mod tests { publish: true, consume: false, }, + &moq_net::Patterns::new(), &mut tasks, ) .await diff --git a/rs/moq-net/src/version.rs b/rs/moq-net/src/version.rs index 4dd475fd29f..76df116f58b 100644 --- a/rs/moq-net/src/version.rs +++ b/rs/moq-net/src/version.rs @@ -331,6 +331,25 @@ impl Versions { alpns } + /// Compute the unique WebTransport subprotocols to offer for these versions. + /// + /// Draft-14 predates WebTransport subprotocols: `moq-00` is only its QUIC ALPN, and a + /// peer that accepts it over WebTransport picks draft-14 over a newer draft both support. + /// It rides `moql` and negotiates in SETUP instead, as `js/net` does. + pub fn web_protocols(&self) -> Vec<&'static str> { + let mut protocols = Vec::new(); + for v in &self.0 { + let protocol = match v { + Version::Ietf(ietf::Version::Draft14) => ALPN_LITE, + v => v.alpn(), + }; + if !protocols.contains(&protocol) { + protocols.push(protocol); + } + } + protocols + } + /// Return only versions present in both self and other, or `None` if the intersection is empty. pub fn filter(&self, other: &Versions) -> Option { let filtered: Vec = self.0.iter().filter(|v| other.0.contains(v)).copied().collect(); @@ -403,6 +422,14 @@ mod tests { assert_eq!(ALPNS.first(), Some(&newest.alpn())); } + #[test] + fn web_protocols_never_offer_moq_00() { + let draft14 = Version::Ietf(ietf::Version::Draft14); + assert!(!Versions::all().web_protocols().contains(&ALPN_14)); + assert_eq!(Versions::from(draft14).web_protocols(), [ALPN_LITE]); + assert_eq!(Versions::all().web_protocols().len(), Versions::all().alpns().len() - 1); + } + #[test] fn lite_07_wip_is_opt_in_only() { let wip = Version::Lite(lite::Version::Lite07); diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index e7f5b64cb9b..558c56e5165 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -480,8 +480,8 @@ impl Client { let session = match url.scheme() { "https" => { let mut request = web::proto::ConnectRequest::new(url.clone()); - for alpn in versions.alpns() { - request = request.with_protocol(alpn.to_string()); + for protocol in versions.web_protocols() { + request = request.with_protocol(protocol.to_string()); } web::Session::connect(connection, request) .await diff --git a/rs/moq-wasm/src/transport.rs b/rs/moq-wasm/src/transport.rs index 6549215f5e6..987eea77db6 100644 --- a/rs/moq-wasm/src/transport.rs +++ b/rs/moq-wasm/src/transport.rs @@ -26,7 +26,7 @@ pub struct Options { /// Without this the browser negotiates no subprotocol and the session has no /// version to start from. fn builder() -> ClientBuilder { - ClientBuilder::new().with_protocols(moq_net::ALPNS.iter().copied()) + ClientBuilder::new().with_protocols(moq_net::Versions::all().web_protocols()) } /// Open a browser WebTransport connection to `url`. From cec5e344cdffc89a8fd1949f7a1ea9318904b351 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 20:59:43 -0700 Subject: [PATCH 03/15] fix(tokio): read WT_SESSION_GONE as the session's close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit draft-ietf-webtrans-http3 §4.4 resets and stops a closed session's streams with WT_SESSION_GONE, which surfaced as an invalid RESET_STREAM. Report the session's close reason instead, and log unknown capsules at DEBUG. Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/src/quic/web/error.rs | 42 ++++++++++++++++++++++++++++ rs/moq-tokio/src/quic/web/recv.rs | 13 +++++++-- rs/moq-tokio/src/quic/web/send.rs | 13 +++++++-- rs/moq-tokio/src/quic/web/session.rs | 2 +- 4 files changed, 63 insertions(+), 7 deletions(-) diff --git a/rs/moq-tokio/src/quic/web/error.rs b/rs/moq-tokio/src/quic/web/error.rs index 1679d8dbe94..c2fc53a8ec3 100644 --- a/rs/moq-tokio/src/quic/web/error.rs +++ b/rs/moq-tokio/src/quic/web/error.rs @@ -68,6 +68,9 @@ impl From for SessionError { } } +/// The HTTP/3 code a closed session's streams are reset and stopped with. +const WT_SESSION_GONE: u64 = 0x170d7b68; + /// Why a session closed, shared by the session and its streams. The first close wins. /// /// It also knows whether the session is raw QUIC, which decides how its close and stream @@ -107,6 +110,24 @@ impl CloseReason { } } + /// The session's close reason when the peer reset or stopped a stream with + /// WT_SESSION_GONE, which tears down the streams of a closed session + /// (draft-ietf-webtrans-http3 §4.4), or None for any other code. + /// + /// The reset can beat the close capsule; until the capsule lands, it reads as a + /// clean close, as a CONNECT stream that ends without one does. + pub(crate) fn session_gone(&self, code: crate::quic::driver::VarInt) -> Option { + if self.raw || code.into_inner() != WT_SESSION_GONE { + return None; + } + Some( + self.reason + .get() + .cloned() + .unwrap_or_else(|| WebTransportError::Closed(0, String::new()).into()), + ) + } + /// Record the close reason, returning false if one was already recorded. pub(crate) fn set(&self, err: SessionError) -> bool { self.reason.set(err).is_ok() @@ -381,4 +402,25 @@ mod tests { let invalid = crate::quic::driver::VarInt::from_u64(1 << 40).unwrap(); assert_eq!(close.decode_stream_code(invalid), None); } + + /// WT_SESSION_GONE reports the session's close, before and after its capsule arrives. + #[test] + fn session_gone_reads_as_the_close() { + let gone = crate::quic::driver::VarInt::from_u64(WT_SESSION_GONE).unwrap(); + let close = CloseReason::new(false); + assert!(matches!( + close.session_gone(gone), + Some(SessionError::WebTransportError(WebTransportError::Closed(0, _))) + )); + + close.set(WebTransportError::Closed(7, "bye".into()).into()); + assert!(matches!( + close.session_gone(gone), + Some(SessionError::WebTransportError(WebTransportError::Closed(7, reason))) if reason == "bye" + )); + assert!(close.session_gone(close.encode_stream_code(7)).is_none()); + + // Raw QUIC has no HTTP/3 code space, so the same number is an application code. + assert!(CloseReason::new(true).session_gone(gone).is_none()); + } } diff --git a/rs/moq-tokio/src/quic/web/recv.rs b/rs/moq-tokio/src/quic/web/recv.rs index 1e46bac784b..a444ae51f56 100644 --- a/rs/moq-tokio/src/quic/web/recv.rs +++ b/rs/moq-tokio/src/quic/web/recv.rs @@ -21,12 +21,16 @@ impl RecvStream { Self { inner: stream, close } } - /// Decode the peer's reset code, and report a session error as the session's close reason. + /// Decode the peer's reset code, and report a session error, or a reset for a session + /// that is gone, as the session's close reason. fn map_error(&self, e: crate::quic::driver::ReadError) -> ReadError { match e { crate::quic::driver::ReadError::Reset(code) => match self.close.decode_stream_code(code) { Some(code) => ReadError::Reset(code), - None => ReadError::InvalidReset(code), + None => match self.close.session_gone(code) { + Some(err) => ReadError::SessionError(err), + None => ReadError::InvalidReset(code), + }, }, e => match ReadError::from(e) { ReadError::SessionError(e) => ReadError::SessionError(self.close.map(e)), @@ -82,7 +86,10 @@ impl RecvStream { pub async fn received_reset(&mut self) -> Result, SessionError> { match self.inner.received_reset().await { Ok(None) => Ok(None), - Ok(Some(code)) => Ok(self.close.decode_stream_code(code)), + Ok(Some(code)) => match self.close.session_gone(code) { + Some(err) => Err(err), + None => Ok(self.close.decode_stream_code(code)), + }, Err(crate::quic::driver::ResetError::ConnectionLost(conn_err)) => Err(self.close.map(conn_err.into())), Err(crate::quic::driver::ResetError::ZeroRttRejected) => unreachable!("0-RTT not supported"), } diff --git a/rs/moq-tokio/src/quic/web/send.rs b/rs/moq-tokio/src/quic/web/send.rs index 5fecef3597c..07f4835c757 100644 --- a/rs/moq-tokio/src/quic/web/send.rs +++ b/rs/moq-tokio/src/quic/web/send.rs @@ -34,12 +34,16 @@ impl SendStream { Self { stream, close, header } } - /// Decode the peer's stop code, and report a session error as the session's close reason. + /// Decode the peer's stop code, and report a session error, or a stop for a session + /// that is gone, as the session's close reason. fn map_error(&self, e: crate::quic::driver::WriteError) -> WriteError { match e { crate::quic::driver::WriteError::Stopped(code) => match self.close.decode_stream_code(code) { Some(code) => WriteError::Stopped(code), - None => WriteError::InvalidStopped(code), + None => match self.close.session_gone(code) { + Some(err) => WriteError::SessionError(err), + None => WriteError::InvalidStopped(code), + }, }, e => match WriteError::from(e) { WriteError::SessionError(e) => WriteError::SessionError(self.close.map(e)), @@ -70,7 +74,10 @@ impl SendStream { /// supported. pub async fn stopped(&self) -> Result, SessionError> { match self.stream.stopped().await { - Ok(Some(code)) => Ok(self.close.decode_stream_code(code)), + Ok(Some(code)) => match self.close.session_gone(code) { + Some(err) => Err(err), + None => Ok(self.close.decode_stream_code(code)), + }, Ok(None) => Ok(None), Err(crate::quic::driver::StoppedError::ConnectionLost(conn_err)) => Err(self.close.map(conn_err.into())), Err(crate::quic::driver::StoppedError::ZeroRttRejected) => unreachable!("0-RTT not supported"), diff --git a/rs/moq-tokio/src/quic/web/session.rs b/rs/moq-tokio/src/quic/web/session.rs index 5215933272a..c3020f3ba45 100644 --- a/rs/moq-tokio/src/quic/web/session.rs +++ b/rs/moq-tokio/src/quic/web/session.rs @@ -195,7 +195,7 @@ impl Session { } Ok(Some(web_transport_proto::Capsule::Grease { .. })) => {} Ok(Some(web_transport_proto::Capsule::Unknown { typ, payload })) => { - tracing::warn!(%typ, size = payload.len(), "unknown capsule"); + tracing::debug!(%typ, size = payload.len(), "unknown capsule"); } Ok(None) => return None, // The capsule reader reports a lost stream as `UnexpectedEnd`, so ask the From d3317ceec8292e0d536c8a9226e62f1d556adf4b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 21:17:39 -0700 Subject: [PATCH 04/15] perf(js/net): write each control message in one write Each Writer primitive issued its own stream write, so an IETF control message went out as a type, a u16 length, and a body. Message.write now buffers the type and the encoded message and writes them once, as Rust does, and Message.encode (both IETF and lite) writes its size and body in one write too. A message that fails to encode still leaves nothing on the stream, and two writers can no longer interleave inside a message. Writer.collect replaces the scratch Writers that ietf/message.ts, lite/message.ts, ietf/object.ts, and lite/subscriber.ts each built. Co-Authored-By: Claude Opus 5.5 --- js/net/src/connection/accept.ts | 9 ++-- js/net/src/connection/connect.ts | 4 +- js/net/src/connection/handshake.ts | 4 +- js/net/src/ietf/adapter.ts | 5 ++- js/net/src/ietf/filter.ts | 13 +----- js/net/src/ietf/message.test.ts | 38 +++++++++++++++++ js/net/src/ietf/message.ts | 66 +++++++++--------------------- js/net/src/ietf/object.ts | 22 +--------- js/net/src/ietf/publisher.ts | 43 ++++++++----------- js/net/src/ietf/request_stream.ts | 6 +-- js/net/src/ietf/subscriber.test.ts | 9 ++-- js/net/src/ietf/subscriber.ts | 46 ++++++++------------- js/net/src/lite/message.ts | 51 ++++------------------- js/net/src/lite/subscriber.ts | 6 +-- js/net/src/stream.ts | 28 +++++++++++++ 15 files changed, 148 insertions(+), 202 deletions(-) create mode 100644 js/net/src/ietf/message.test.ts diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index ed89187f938..f0333fe94b3 100644 --- a/js/net/src/connection/accept.ts +++ b/js/net/src/connection/accept.ts @@ -1,4 +1,5 @@ import * as Ietf from "../ietf/index.ts"; +import * as Message from "../ietf/message.ts"; import * as Lite from "../lite/index.ts"; import type { Consumer as OriginConsumer, Producer as OriginProducer } from "../origin.ts"; import { Stream } from "../stream.ts"; @@ -159,8 +160,6 @@ async function acceptSetup( const client = await Ietf.ClientSetup.decode(stream.reader, version); - await stream.writer.u53(Lite.StreamId.ServerCompat); - const encoder = new TextEncoder(); const params = new Ietf.SetupOptions(); params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true)); @@ -171,7 +170,7 @@ async function acceptSetup( Ietf.activeCountIntoSetup(params, version); const server = new Ietf.ServerSetup({ version, parameters: params }); - await server.encode(stream.writer, version); + await Message.write(stream.writer, Lite.StreamId.ServerCompat, server, version); const maxRequestId = client.parameters.getVarint(Ietf.SetupOption.MaxRequestId) ?? 0n; @@ -225,8 +224,6 @@ async function acceptNegotiated( selectedVersion = match; } - await stream.writer.u53(Lite.StreamId.ServerCompat); - const encoder = new TextEncoder(); const params = new Ietf.SetupOptions(); params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true)); @@ -237,7 +234,7 @@ async function acceptNegotiated( Ietf.activeCountIntoSetup(params, setupVersion); const server = new Ietf.ServerSetup({ version: selectedVersion, parameters: params }); - await server.encode(stream.writer, setupVersion); + await Message.write(stream.writer, Lite.StreamId.ServerCompat, server, setupVersion); if (Object.values(Lite.Version).includes(selectedVersion as Lite.Version)) { return new Lite.Connection({ diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index 509d8451b2d..0fc12ff7bfe 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -2,6 +2,7 @@ import { Milli, Nano, sleep } from "@moq/clock"; import Session, { type Version as QmuxVersion } from "@moq/qmux"; import { error, fromClose, ProtocolViolation, sessionCause } from "../error.ts"; import * as Ietf from "../ietf/index.ts"; +import * as Message from "../ietf/message.ts"; import * as Lite from "../lite/index.ts"; import type { Consumer as OriginConsumer, Producer as OriginProducer } from "../origin.ts"; import { Stream } from "../stream.ts"; @@ -422,7 +423,6 @@ async function negotiate( } const stream = await Stream.open(session, { version: setupVersion }); - await stream.writer.u53(Lite.StreamId.ClientCompat); const encoder = new TextEncoder(); @@ -443,7 +443,7 @@ async function negotiate( : setup, parameters: params, }); - await client.encode(stream.writer, setupVersion); + await Message.write(stream.writer, Lite.StreamId.ClientCompat, client, setupVersion); const serverCompat = await stream.reader.u53(); if (serverCompat !== Lite.StreamId.ServerCompat) { diff --git a/js/net/src/connection/handshake.ts b/js/net/src/connection/handshake.ts index 1107f2fb915..1923cc77970 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -1,6 +1,7 @@ import { closeReason, ProtocolViolation, SessionCode, StreamError } from "../error.ts"; import { type Hop, randomHop } from "../hop.ts"; import * as Ietf from "../ietf/index.ts"; +import * as Message from "../ietf/message.ts"; import { Reader, Stream, UnexpectedEnd, Writer } from "../stream.ts"; import * as Extensions from "./extensions.ts"; @@ -82,8 +83,7 @@ async function sendSetup(transport: WebTransport, version: Ietf.IetfVersion, set // Via Writer.open for its deadline: a peer can advertise a stream limit of zero and // never raise it, which would otherwise hang the handshake rather than failing it. const writer = await Writer.open(transport, { version }); - await writer.u53(Ietf.Setup.id); // 0x2F00 stream type - await setupMsg.encode(writer, version); + await Message.write(writer, Ietf.Setup.id, setupMsg, version); // 0x2F00 stream type return writer; } diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index 048eb4f0b6d..281dc96f587 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -6,6 +6,7 @@ import { type OpenOptions, Reader, Stream, type Writer } from "../stream.ts"; import { untilAborted } from "../util/abort.ts"; import * as Varint from "../varint.ts"; import { GoAway } from "./goaway.ts"; +import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; import { MaxRequestId } from "./request.ts"; import { refuseUpdate } from "./request_stream.ts"; @@ -759,8 +760,8 @@ export class ControlStreamAdapter implements Session { const max = this.#peerMax; await this.#writeMutex.runExclusive(async () => { if (this.#closed) return; - await this.#writer.u53(MaxRequestId.id); - await new MaxRequestId({ requestId: max }).encode(this.#writer, this.version); + const grant = new MaxRequestId({ requestId: max }); + await Message.write(this.#writer, MaxRequestId.id, grant, this.version); }); } } catch { diff --git a/js/net/src/ietf/filter.ts b/js/net/src/ietf/filter.ts index 0eebd2b47c7..fe7af4dd478 100644 --- a/js/net/src/ietf/filter.ts +++ b/js/net/src/ietf/filter.ts @@ -5,7 +5,7 @@ */ import { ProtocolViolation } from "../error.ts"; -import type { Reader } from "../stream.ts"; +import { concat, type Reader } from "../stream.ts"; import * as Varint from "../varint.ts"; import { type IetfVersion, Version } from "./version.ts"; @@ -89,17 +89,6 @@ function unvarint(buf: Uint8Array, version: IetfVersion): [bigint, Uint8Array] { return usesLeadingOnes(version) ? Varint.decodeLeadingOnes(buf) : Varint.decodeBigInt(buf); } -function concat(parts: Uint8Array[]): Uint8Array { - const total = parts.reduce((n, p) => n + p.length, 0); - const out = new Uint8Array(total); - let offset = 0; - for (const part of parts) { - out.set(part, offset); - offset += part.length; - } - return out; -} - function endDelta(startGroup: bigint, endGroup: bigint): bigint { if (endGroup < startGroup) { throw new Error(`filter range runs backwards: ${startGroup} > ${endGroup}`); diff --git a/js/net/src/ietf/message.test.ts b/js/net/src/ietf/message.test.ts new file mode 100644 index 00000000000..cda580c04db --- /dev/null +++ b/js/net/src/ietf/message.test.ts @@ -0,0 +1,38 @@ +import { expect, test } from "bun:test"; +import { Reader, Writer } from "../stream.ts"; +import * as Message from "./message.ts"; +import { RequestError } from "./request.ts"; +import { type IetfVersion, Version } from "./version.ts"; + +/** A writer that keeps every chunk its stream is handed. */ +function recording(version: IetfVersion): { writer: Writer; chunks: Uint8Array[] } { + const chunks: Uint8Array[] = []; + const writer = new Writer( + new WritableStream({ write: (chunk) => void chunks.push(new Uint8Array(chunk)) }), + version, + ); + return { writer, chunks }; +} + +test("a control message goes out in one write, and one that fails to encode writes nothing", async () => { + for (const version of [Version.DRAFT_14, Version.DRAFT_17]) { + const { writer, chunks } = recording(version); + const err = new RequestError({ requestId: 3n, errorCode: 2, reasonPhrase: "nope" }); + await Message.write(writer, RequestError.id, err, version); + expect(chunks.length).toBe(1); + + const reader = new Reader(undefined, chunks[0], version); + expect(await reader.u53()).toBe(RequestError.id); + const decoded = await RequestError.decode(reader, version); + expect(decoded.reasonPhrase).toBe("nope"); + expect(await reader.done()).toBe(true); + + // A message written without a type, as on a stream whose header named it, is one write too. + await err.encode(writer, version); + expect(chunks.length).toBe(2); + + const huge = new RequestError({ requestId: 3n, errorCode: 2, reasonPhrase: "x".repeat(65_536) }); + await expect(Message.write(writer, RequestError.id, huge, version)).rejects.toThrow("Message too large"); + expect(chunks.length).toBe(2); + } +}); diff --git a/js/net/src/ietf/message.ts b/js/net/src/ietf/message.ts index 4a080e9a979..989f955eaae 100644 --- a/js/net/src/ietf/message.ts +++ b/js/net/src/ietf/message.ts @@ -1,56 +1,30 @@ -import { Reader, Writer } from "../stream.ts"; +import { concat, encodeVarint, Reader, Writer } from "../stream.ts"; import { MalformedAuthToken } from "./token.ts"; +import type { IetfVersion } from "./version.ts"; -// Encodes a message with a u16 (16-bit) size prefix as per draft-14. A type `id` is written -// only once the body fits, so an oversized message leaves nothing on the stream. +// Encodes a message with a u16 (16-bit) size prefix as per draft-14, after its type `id` if +// given. The whole message goes out in one write, so an oversized one leaves nothing on the stream. export async function encode(writer: Writer, f: (w: Writer) => Promise, id?: number) { - let scratch = new Uint8Array(); - - const temp = new Writer( - new WritableStream({ - write(chunk: Uint8Array) { - const needed = scratch.byteLength + chunk.byteLength; - if (needed > scratch.buffer.byteLength) { - // Resize the buffer to the needed size. - const capacity = Math.max(needed, scratch.buffer.byteLength * 2); - const newBuffer = new ArrayBuffer(capacity); - const newScratch = new Uint8Array(newBuffer, 0, needed); - - // Copy the old data into the new buffer. - newScratch.set(scratch); - - // Copy the new chunk into the new buffer. - newScratch.set(chunk, scratch.byteLength); - - scratch = newScratch; - } else { - // Copy chunk data into buffer - scratch = new Uint8Array(scratch.buffer, 0, needed); - scratch.set(chunk, needed - chunk.byteLength); - } - }, - }), - writer.version, - ); - - try { - await f(temp); - } finally { - temp.close(); + const body = await Writer.collect(writer.version, f); + if (body.byteLength > 65535) { + throw new Error(`Message too large: ${body.byteLength} bytes (max 65535)`); } - await temp.closed; - - // Check that message fits in u16 - if (scratch.byteLength > 65535) { - throw new Error(`Message too large: ${scratch.byteLength} bytes (max 65535)`); - } + const size = new Uint8Array([body.byteLength >> 8, body.byteLength & 0xff]); + const parts = id === undefined ? [size, body] : [encodeVarint(id, writer.version), size, body]; + await writer.write(concat(parts)); +} - if (id !== undefined) await writer.u53(id); +/** A message that encodes everything but its type. */ +export interface Encode { + encode(w: Writer, version: IetfVersion): Promise; +} - // Write u16 size (2 bytes, big-endian) - await writer.u16(scratch.byteLength); - await writer.write(scratch); +// Writes `msg` after the type `id` the caller picked for it, in one write, so a message that +// fails to encode leaves nothing on the stream. +export async function write(writer: Writer, id: number, msg: Encode, version: IetfVersion) { + const body = await Writer.collect(writer.version, (w) => msg.encode(w, version)); + await writer.write(concat([encodeVarint(id, writer.version), body])); } // Reads a message with a u16 size prefix. diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index ae61a0d96df..5088ff8a990 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -77,27 +77,7 @@ export async function encodeObjectExtensions( return new Uint8Array(); } - const chunks: Uint8Array[] = []; - const writer = new Writer( - new WritableStream({ - write(chunk) { - chunks.push(new Uint8Array(chunk)); - }, - }), - version, - ); - await encodeObjectTime(writer, timestamp, timescale, version); - writer.close(); - await writer.closed; - - const size = chunks.reduce((total, chunk) => total + chunk.byteLength, 0); - const result = new Uint8Array(size); - let offset = 0; - for (const chunk of chunks) { - result.set(chunk, offset); - offset += chunk.byteLength; - } - return result; + return Writer.collect(version, (w) => encodeObjectTime(w, timestamp, timescale, version)); } /** Decode the Timestamp an object's Properties block carries, consuming the whole block. */ diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 3047bb5d1db..2f84c893019 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -588,7 +588,6 @@ export class Publisher { : undefined; // Send SUBSCRIBE_OK - await request.writer.u53(SubscribeOk.id); const ok = new SubscribeOk({ requestId: version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 @@ -625,7 +624,7 @@ export class Publisher { // the track opts out of timestamps for this subscriber. {}, }); - await ok.encode(request.writer, version); + await Message.write(request.writer, SubscribeOk.id, ok, version); console.debug(`publish ok: broadcast=${name} track=${track.name}`); // Cancels groups still queued for a stream slot. Only the subscriber leaving counts: @@ -705,8 +704,8 @@ export class Publisher { if (params.maxAge !== undefined) track.update({ ...track.subscription.peek(), maxDelay: Milli(params.maxAge) }); if (version === Version.DRAFT_14) continue; - await request.writer.u53(RequestOk.id); - await new RequestOk({ requestId: ownRequestId }).encode(request.writer, version); + const ok = new RequestOk({ requestId: ownRequestId }); + await Message.write(request.writer, RequestOk.id, ok, version); } })(); const requestEnded = race([updates, request.ended]); @@ -847,7 +846,6 @@ export class Publisher { // Draft 14 registers no UPDATE_FAILED, so a failed update ends there as any error does. const updateFailed = publishError instanceof UpdateFailed && version !== Version.DRAFT_14; try { - await request.writer.u53(PublishDone.id); const done = new PublishDone({ requestId: version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16 @@ -869,7 +867,7 @@ export class Publisher { ? "internal error" : "track ended", }); - await done.encode(request.writer, version); + await Message.write(request.writer, PublishDone.id, done, version); } catch { // Stream might already be closed by peer. } @@ -1278,19 +1276,17 @@ export class Publisher { try { const answer = async (activeCount?: number) => { if (version === Version.DRAFT_14) { - await request.writer.u53(SubscribeNamespaceOk.id); const ok = new SubscribeNamespaceOk({ requestId: msg.requestId }); - await ok.encode(request.writer, version); + await Message.write(request.writer, SubscribeNamespaceOk.id, ok, version); return; } const parameters = new Parameters(); parameters.activeCount = activeCount; - await request.writer.u53(RequestOk.id); const ok = new RequestOk({ requestId: version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined, parameters, }); - await ok.encode(request.writer, version); + await Message.write(request.writer, RequestOk.id, ok, version); }; // MoQ Active Count: the OK counts what the first pass below sends, so it waits for @@ -1315,10 +1311,10 @@ export class Publisher { // publisher included, as Rust does. Only a legacy PUBLISH_NAMESPACE request can // be declined. const namespace = async (suffix: Path.Valid, snap: Advertised): Promise => { - await request.writer.u53(SubscribeNamespaceEntry.id); // Either extension puts the Parameters field on NAMESPACE, even an empty one. const params = this.#advert !== undefined || this.#epoch ? this.#params(snap.route) : undefined; - await new SubscribeNamespaceEntry({ suffix, params }).encode(request.writer, version); + const entry = new SubscribeNamespaceEntry({ suffix, params }); + await Message.write(request.writer, SubscribeNamespaceEntry.id, entry, version); return "held"; }; const target: Target = legacy @@ -1338,8 +1334,8 @@ export class Publisher { held: () => undefined, carry: () => undefined, withdraw: async (suffix) => { - await request.writer.u53(SubscribeNamespaceEntryDone.id); - await new SubscribeNamespaceEntryDone({ suffix }).encode(request.writer, version); + const done = new SubscribeNamespaceEntryDone({ suffix }); + await Message.write(request.writer, SubscribeNamespaceEntryDone.id, done, version); }, }; @@ -1424,8 +1420,7 @@ export class Publisher { // A new cursor starts its backoff fresh, as in Rust. retry = 0; } - await request.writer.u53(RequestOk.id); - await new RequestOk({ requestId: undefined }).encode(request.writer, version); + await Message.write(request.writer, RequestOk.id, new RequestOk({ requestId: undefined }), version); updating = readUpdate(); } } @@ -1781,8 +1776,7 @@ export class Publisher { await deadline.send( stream, async () => { - await opened.writer.u53(PublishNamespace.id); - await msg.encode(opened.writer, version); + await Message.write(opened.writer, PublishNamespace.id, msg, version); }, legacy ? () => this.#done(opened, path, requestId) : undefined, ); @@ -1906,12 +1900,9 @@ export class Publisher { (async (): Promise => { const writer = request.stream.writer; request.updating = true; - await writer.u53(PublishNamespaceUpdate.id); const namespaceRequestId = version === Version.DRAFT_16 ? request.requestId : undefined; - await new PublishNamespaceUpdate({ requestId, namespaceRequestId, update, token }).encode( - writer, - version, - ); + const msg = new PublishNamespaceUpdate({ requestId, namespaceRequestId, update, token }); + await Message.write(writer, PublishNamespaceUpdate.id, msg, version); const response = await (request.reply ?? request.stream.next()); if (response === undefined) { @@ -2008,9 +1999,8 @@ export class Publisher { /** Write the PUBLISH_NAMESPACE_DONE that withdraws a draft-14-16 request. */ async #done(request: RequestStream, path: Path.Valid, requestId: bigint) { try { - await request.writer.u53(PublishNamespaceDone.id); const done = new PublishNamespaceDone({ trackNamespace: path, requestId }); - await done.encode(request.writer, this.#session.version); + await Message.write(request.writer, PublishNamespaceDone.id, done, this.#session.version); } catch { // Stream might already be closed } @@ -2100,8 +2090,7 @@ export class Publisher { } answering = true; - await request.writer.u53(TrackStatusOk.id(version)); - await answer.encode(request.writer, version); + await Message.write(request.writer, TrackStatusOk.id(version), answer, version); // The answer is all this stream carries. request.close(); await request.writer.closed.catch(() => undefined); diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts index 0bec16b6c47..3149ded7acb 100644 --- a/js/net/src/ietf/request_stream.ts +++ b/js/net/src/ietf/request_stream.ts @@ -48,12 +48,12 @@ const ERROR_14: Record = { }; async function writeRefusal(w: Writer, version: IetfVersion, kind: RequestKind, refusal: Refusal): Promise { - await w.u53(version === Version.DRAFT_14 ? ERROR_14[kind] : RequestError.id); - await new RequestError({ + const err = new RequestError({ requestId: version <= Version.DRAFT_16 ? refusal.requestId : undefined, errorCode: toRequestCode(refusal.condition, kind, version), reasonPhrase: refusal.reasonPhrase, - }).encode(w, version); + }); + await Message.write(w, version === Version.DRAFT_14 ? ERROR_14[kind] : RequestError.id, err, version); } /** diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index c9f960a3adc..15c6e52e5bb 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -1836,13 +1836,14 @@ test("returning demand survives a blocked unsubscribe", async () => { const open = session.openBi.bind(session); const opening = spyOn(session, "openBi").mockImplementationOnce(() => { const stream = open(); - const write = stream.writer.u53.bind(stream.writer); - spyOn(stream.writer, "u53").mockImplementation(async (value) => { - if (value === Unsubscribe.id) { + const write = stream.writer.write.bind(stream.writer); + // Each message is one write that leads with its one-byte type. + spyOn(stream.writer, "write").mockImplementation(async (chunk) => { + if (chunk[0] === Unsubscribe.id) { cancelStarted.resolve(); await releaseCancel.promise; } - await write(value); + await write(chunk); }); const close = stream.close.bind(stream); spyOn(stream, "close").mockImplementation(() => { diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index ce472b2678c..60d26c0f89e 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -44,6 +44,7 @@ import { DuplicateTrackAlias, RetiredTrackAlias, TrackAliases } from "./aliases. import * as Cluster from "./cluster.ts"; import { ObjectDatagram } from "./datagram.ts"; import { requestReason, toRequestCode } from "./error.ts"; +import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; import { decodeObjectTime, @@ -545,17 +546,11 @@ export class Subscriber { version === Version.DRAFT_16 || version === Version.DRAFT_17 ) { - await request.writer.u53(SubscribeNamespaceLegacy.id); - await new SubscribeNamespaceLegacy({ namespace: prefix, requestId, hidden }).encode( - request.writer, - version, - ); + const msg = new SubscribeNamespaceLegacy({ namespace: prefix, requestId, hidden }); + await Message.write(request.writer, SubscribeNamespaceLegacy.id, msg, version); } else { - await request.writer.u53(SubscribeNamespace.id); - await new SubscribeNamespace({ namespace: prefix, requestId, hidden }).encode( - request.writer, - version, - ); + const msg = new SubscribeNamespace({ namespace: prefix, requestId, hidden }); + await Message.write(request.writer, SubscribeNamespace.id, msg, version); } console.debug(`subscribe_namespace written: requestId=${requestId}`); @@ -563,9 +558,8 @@ export class Subscriber { // For v14/v15: send UnsubscribeNamespace before closing if (version === Version.DRAFT_14 || version === Version.DRAFT_15) { try { - await request.writer.u53(UnsubscribeNamespace.id); const unsub = new UnsubscribeNamespace({ requestId }); - await unsub.encode(request.writer, version); + await Message.write(request.writer, UnsubscribeNamespace.id, unsub, version); } catch { // Stream might already be closed } @@ -1076,8 +1070,7 @@ export class Subscriber { // Counted before writing: the answer can be read before the write's promise resolves. if (answered) unanswered.update((count) => count + 1); try { - await stream.writer.u53(SubscribeUpdate.id); - await update.encode(stream.writer, version); + await Message.write(stream.writer, SubscribeUpdate.id, update, version); } catch { // The publisher stopped reading; its responses decide how the track ends. return; @@ -1102,8 +1095,7 @@ export class Subscriber { if (version !== Version.DRAFT_14 && version !== Version.DRAFT_15 && version !== Version.DRAFT_16) return; try { - await stream.writer.u53(Unsubscribe.id); - await new Unsubscribe({ requestId }).encode(stream.writer, version); + await Message.write(stream.writer, Unsubscribe.id, new Unsubscribe({ requestId }), version); } catch { // The stream may already be gone; there is nothing further to tell the peer. } @@ -1141,7 +1133,6 @@ export class Subscriber { // won the await above. No response data is released from the canceled setup. if (state.cancelled && !legacy) throw new Error("subscribe cancelled before it was sent"); - await state.stream.writer.u53(Subscribe.id); const msg = new Subscribe({ requestId, trackNamespace: broadcast, @@ -1160,7 +1151,7 @@ export class Subscriber { filter: joinFilter(version), epoch, }); - await msg.encode(state.stream.writer, version); + await Message.write(state.stream.writer, Subscribe.id, msg, version); state.sent = true; // The caller may have given up while that write was in flight. It deliberately left @@ -1300,15 +1291,13 @@ export class Subscriber { // because consumers may trigger Subscribe writes that would // interleave with our OK on the control stream. if (version === Version.DRAFT_14) { - await request.writer.u53(PublishNamespaceOk.id); const ok = new PublishNamespaceOk({ requestId: msg.requestId }); - await ok.encode(request.writer, version); + await Message.write(request.writer, PublishNamespaceOk.id, ok, version); } else { - await request.writer.u53(RequestOk.id); const ok = new RequestOk({ requestId: version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined, }); - await ok.encode(request.writer, version); + await Message.write(request.writer, RequestOk.id, ok, version); } // Only now is the advertisement ours to announce, for the reason above: a @@ -1374,12 +1363,12 @@ export class Subscriber { }); // Draft-16's advertisement outlives its stream, so withdraw it explicitly. if (keyed) { - await request.writer.u53(PublishNamespaceCancel.id); - await new PublishNamespaceCancel({ + const cancel = new PublishNamespaceCancel({ requestId: msg.requestId, errorCode: toRequestCode("malformed_auth_token", "publish_namespace", version), reasonPhrase: err.message, - }).encode(request.writer, version); + }); + await Message.write(request.writer, PublishNamespaceCancel.id, cancel, version); } request.close(); break; @@ -1417,11 +1406,8 @@ export class Subscriber { // Nothing here can fail to apply, so every update is acknowledged. A leaf // routes nothing, so a repricing changes nothing it holds; consumers still // hear the new route so a forwarder can reprice. - await request.writer.u53(RequestOk.id); - await new RequestOk({ requestId: keyed ? update.requestId : undefined }).encode( - request.writer, - version, - ); + const ok = new RequestOk({ requestId: keyed ? update.requestId : undefined }); + await Message.write(request.writer, RequestOk.id, ok, version); } } finally { if (legacy) this.#legacyRequests.delete(path); diff --git a/js/net/src/lite/message.ts b/js/net/src/lite/message.ts index 45656636ca6..90cf5ede494 100644 --- a/js/net/src/lite/message.ts +++ b/js/net/src/lite/message.ts @@ -1,4 +1,4 @@ -import { Reader, Writer } from "../stream.ts"; +import { concat, encodeVarint, Reader, Writer } from "../stream.ts"; import { decodeError } from "../util/decode_error.ts"; /** @@ -7,55 +7,20 @@ import { decodeError } from "../util/decode_error.ts"; */ export const MAX_MESSAGE_SIZE = 0xffff; -// Encodes a message with a varint size prefix, refusing a body the peer would reject. A type -// `id` is written only once the body fits, so a refused message leaves nothing on the stream. +// Encodes a message with a varint size prefix, refusing a body the peer would reject. The whole +// message, type `id` included, goes out in one write, so a refused one leaves nothing on the stream. export async function encode( writer: Writer, f: (w: Writer) => Promise, { id, max = MAX_MESSAGE_SIZE }: { id?: number; max?: number } = {}, ) { - let scratch = new Uint8Array(); - - const temp = new Writer( - new WritableStream({ - write(chunk: Uint8Array) { - const needed = scratch.byteLength + chunk.byteLength; - if (needed > scratch.buffer.byteLength) { - // Resize the buffer to the needed size. - const capacity = Math.max(needed, scratch.buffer.byteLength * 2); - const newBuffer = new ArrayBuffer(capacity); - const newScratch = new Uint8Array(newBuffer, 0, needed); - - // Copy the old data into the new buffer. - newScratch.set(scratch); - - // Copy the new chunk into the new buffer. - newScratch.set(chunk, scratch.byteLength); - - scratch = newScratch; - } else { - // Copy chunk data into buffer - scratch = new Uint8Array(scratch.buffer, 0, needed); - scratch.set(chunk, needed - chunk.byteLength); - } - }, - }), - writer.version, - ); - - await f(temp); - temp.close(); - await temp.closed; - - if (scratch.byteLength > max) { - throw new Error(`message too large: ${scratch.byteLength} bytes (max ${max})`); + const body = await Writer.collect(writer.version, f); + if (body.byteLength > max) { + throw new Error(`message too large: ${body.byteLength} bytes (max ${max})`); } - if (id !== undefined) await writer.u53(id); - await writer.u53(scratch.byteLength); - if (scratch.byteLength > 0) { - await writer.write(scratch); - } + const size = encodeVarint(body.byteLength, writer.version); + await writer.write(concat(id === undefined ? [size, body] : [encodeVarint(id, writer.version), size, body])); } // Reads a message with a varint size prefix, refusing an oversized body before reading it. diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 48e8d18d05e..8629d096cd7 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -1370,13 +1370,11 @@ export class Subscriber { endGroup: inclusiveGroupEnd(bounds.end), }); // Encoded apart from the write, so a bound the wire can't carry still fails the track. - const encoded: Uint8Array[] = []; - const buffer = new WritableStream({ write: (chunk) => void encoded.push(chunk.slice()) }); - await update.encode(new Writer(buffer, this.version), this.version); + const encoded = await Writer.collect(this.version, (w) => update.encode(w, this.version)); // A failed write means the publisher stopped reading, as it does once it finishes the // track. The responses decide how the track ends, cleanly at the FIN or with a reset's error. try { - for (const chunk of encoded) await stream.writer.write(chunk); + await stream.writer.write(encoded); } catch { return; } diff --git a/js/net/src/stream.ts b/js/net/src/stream.ts index 221ebb61b61..f33f1945cef 100644 --- a/js/net/src/stream.ts +++ b/js/net/src/stream.ts @@ -138,6 +138,17 @@ export function encodeVarint(v: number | bigint, version: StreamVersion): Uint8A return encodeTo(new ArrayBuffer(9), parts.hi, lo, version); } +/** The parts joined into one buffer, so they can go out in one write. */ +export function concat(parts: Uint8Array[]): Uint8Array { + const out = new Uint8Array(parts.reduce((n, p) => n + p.byteLength, 0)); + let offset = 0; + for (const part of parts) { + out.set(part, offset); + offset += part.byteLength; + } + return out; +} + /** * The `WebTransportSendStream` that every outgoing stream is, narrowed to the one attribute * this package uses. The DOM types available here still describe them as plain @@ -808,6 +819,23 @@ export class Writer { this.#writer.abort(withCode(reason, this.version)).catch(() => void 0); } + /** Run `f` against a Writer that buffers in memory, returning everything it wrote as one chunk. */ + static async collect(version: StreamVersion, f: (w: Writer) => Promise): Promise { + const chunks: Uint8Array[] = []; + // Copied, since each primitive reuses the Writer's scratch buffer. + const w = new Writer( + new WritableStream({ write: (chunk) => void chunks.push(chunk.slice()) }), + version, + ); + try { + await f(w); + } finally { + w.close(); + } + await w.closed; + return concat(chunks); + } + /** * Open an outgoing unidirectional stream. * @param quic - The session to open it on From a4d491e3af07236d2b04a5c9cff1c6573c70f7f2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 21:18:21 -0700 Subject: [PATCH 05/15] style(cli): format the consumed scope Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/main.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 97c9299e905..a30ab3f5b22 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -495,7 +495,9 @@ async fn spawn_moq( dialer = dialer.with_publisher(origin.consume()); } if directions.consume { - let scoped = origin.scope("", consumed).context("failed to scope the origin to the consumed broadcasts")?; + let scoped = origin + .scope("", consumed) + .context("failed to scope the origin to the consumed broadcasts")?; dialer = dialer.with_subscriber(scoped); } From 893dd410211fecd5de7dac9ffa6b4a3ffd26c295 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 21:21:27 -0700 Subject: [PATCH 06/15] test(smoke): report the last stage a failed subscriber reached The first error-looking line was often teardown noise after the timeout. Lead with the stage (connected, SUBSCRIBE sent, SUBSCRIBE_OK, headless groups) and keep the first error, minus capsule teardown, as the detail. Co-Authored-By: Claude Opus 5.5 --- test/smoke/smoke.sh | 29 ++++++++++++++++++++++++----- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/test/smoke/smoke.sh b/test/smoke/smoke.sh index 4d52831b4bc..9665da9d6f6 100755 --- a/test/smoke/smoke.sh +++ b/test/smoke/smoke.sh @@ -847,14 +847,33 @@ negotiated() { } failure_reason() { - # failure_reason : the first error-looking line of a cell log, minus its - # timestamp/level/target prefix and the browser's [page] tag. The first error is - # usually the cause; later ones tend to be the teardown it triggered. + # failure_reason : the last stage the subscriber reached, then the first + # error-looking line as the detail. The first error alone is often the teardown + # after the timeout, which hides how far the cell got. Stages come from the + # lines the Rust CLI, @moq/net, and the browser page already log: connected, + # SUBSCRIBE sent, SUBSCRIBE_OK, and a group dropped for starting mid-group. [[ -f "$1" ]] || return 0 - sed 's/\x1b\[[0-9;]*m//g' "$1" | + local log stage track detail + log=$(sed 's/\x1b\[[0-9;]*m//g' "$1") + track=$(grep -oE 'subscribe start(ed|:).*track=[^ ]+' <<<"$log" | tail -1 | sed -E 's/.*track=//' || true) + if ! grep -qE 'connected version=|negotiated protocol:' <<<"$log"; then + stage="not connected" + elif [[ -z "$track" ]]; then + stage="connected, nothing subscribed" + elif grep -qE "abandoned before it was accepted|waiting for SUBSCRIBE_OK" <<<"$log" || + { grep -q 'subscribe start:' <<<"$log" && ! grep -qF "track=$track" <<<"$(grep 'subscribe ok:' <<<"$log")"; }; then + stage="no SUBSCRIBE_OK for $track" + elif grep -q 'dropping a group with no head' <<<"$log"; then + stage="SUBSCRIBE_OK for $track, only headless groups" + else + stage="SUBSCRIBE_OK for $track, no objects" + fi + # Teardown noise once the timeout kills the client says nothing about the cause. + detail=$(grep -vE 'failed to read capsule|unknown capsule' <<<"$log" | grep -m1 -E '(^|[^a-z])(err|error|Error|ERROR|failed|Failed|timed out)([^a-z]|$)' | sed -E 's/^[0-9T:.-]+Z +[A-Z]+ +[a-z0-9_:]+: //; s/^\[page\] //; s/^[[:space:]]+//' | - tr '\t' ' ' | cut -c1-240 || true + tr '\t' ' ' || true) + echo "$stage${detail:+: $detail}" | cut -c1-240 } log_dir() { From b5dc8abc0395163d6d063f8ae389ee36cbed5106 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 21:28:48 -0700 Subject: [PATCH 07/15] test(smoke): drop the Cloudflare draft-16 extras Cloudflare's draft-16 relays require a provisioned token in the URL path and close anything else with "scope resolution failed", so these cells could never pass. The registry lists every other public relay. Co-Authored-By: Claude Opus 5.5 --- test/smoke/relays.sh | 24 ++++++++---------------- 1 file changed, 8 insertions(+), 16 deletions(-) diff --git a/test/smoke/relays.sh b/test/smoke/relays.sh index 0fb23ede492..4ef67fab6cf 100755 --- a/test/smoke/relays.sh +++ b/test/smoke/relays.sh @@ -21,12 +21,6 @@ SMOKE_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) REGISTRY="${RELAYS_REGISTRY:-https://raw.githubusercontent.com/englishm/moq-interop-runner/main/implementations.json}" REQUIRED="moq-dev-rs" REQUIRED_EXPLICIT=0 # set by --required; the default only gates when --only selects it -# Public endpoints the registry doesn't list (yet): key, display name, URL. They -# merge with the registry by URL, so an upstream registration supersedes these. -EXTRA_ENDPOINTS=( - $'moq-rs\tmoq-rs (draft-16)\thttps://draft-16.cloudflare.mediaoverquic.com/moq' - $'moq-rs\tmoq-rs (draft-16)\tmoqt://draft-16.cloudflare.mediaoverquic.com:443' -) # Relays that speak moq-dev's WebSocket binding. Each of their http(s) endpoints # gains a ws(s) twin. Add a key here when that implementation actually serves it. WEBSOCKET_KEYS="moq-dev-rs" @@ -121,16 +115,14 @@ seen_url() { # One row per endpoint: key, display name, url. Deduplicated by URL (an entry # can list the same URL twice under different transport labels); the transport # is taken from the URL scheme rather than the registry's label. -{ - # The registry is third-party data: fall back to the key for a missing name, - # and drop entries without a usable URL. - jq -r '.implementations | to_entries[] | .key as $k | - (.value.name | if type == "string" and . != "" then . else $k end) as $n | - (.value.roles.relay.remote // [])[] | - select((.url | type) == "string" and (.url | test("^[a-z][a-z0-9+.-]*://"))) | - [$k, $n, .url] | @tsv' "$TMP/registry.json" - printf '%s\n' "${EXTRA_ENDPOINTS[@]}" -} | awk -F'\t' '!seen[$3]++' >"$TMP/endpoints.tsv" +# The registry is third-party data: fall back to the key for a missing name, +# and drop entries without a usable URL. +jq -r '.implementations | to_entries[] | .key as $k | + (.value.name | if type == "string" and . != "" then . else $k end) as $n | + (.value.roles.relay.remote // [])[] | + select((.url | type) == "string" and (.url | test("^[a-z][a-z0-9+.-]*://"))) | + [$k, $n, .url] | @tsv' "$TMP/registry.json" | + awk -F'\t' '!seen[$3]++' >"$TMP/endpoints.tsv" relay_args=() : >"$TMP/selected.tsv" From 554bfc7aa6bd580e4a0b0ec3cfb919b26a9a30fa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 21:49:05 -0700 Subject: [PATCH 08/15] fix(js/net): dial a moqt:// URL over WebTransport at https MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A browser has no raw QUIC, and the WebTransport CONNECT carries the URL's scheme, so moq-relay refused moqt:// (draft-18 §3.1.3 says to swap in https). This also covers a GOAWAY redirect to a moqt:// URI. A failed race now names each transport's error instead of Promise.any's empty message. Co-Authored-By: Claude Opus 5.5 --- doc/concept/transport.md | 3 ++- js/net/src/connection/connect.test.ts | 37 +++++++++++++++++++++++++++ js/net/src/connection/connect.ts | 21 +++++++++++++-- 3 files changed, 58 insertions(+), 3 deletions(-) diff --git a/doc/concept/transport.md b/doc/concept/transport.md index cddcb32583f..9e55121898f 100644 --- a/doc/concept/transport.md +++ b/doc/concept/transport.md @@ -51,7 +51,8 @@ Native clients dial the relay directly with the MoQ ALPN, skipping the HTTP/3 handshake. The URL path and `?jwt=` token travel in the MoQ `SETUP` message instead of a `CONNECT` request, so the server sees the same thing either way. `moq-tokio` dials WebTransport for `https://` URLs; use `moql://` or `moqt://` -for raw QUIC. +for raw QUIC. A browser cannot dial raw QUIC, so `@moq/net` reaches a `moqt://` +or `moql://` URL over WebTransport at the same host, port, and path. ## TCP and Unix sockets (local workers) diff --git a/js/net/src/connection/connect.test.ts b/js/net/src/connection/connect.test.ts index d69e39883f2..382099eb3a0 100644 --- a/js/net/src/connection/connect.test.ts +++ b/js/net/src/connection/connect.test.ts @@ -91,6 +91,43 @@ test("WebTransport negotiates lite-07-wip only when explicitly offered", async ( expect(version).toBe("moq-lite-07-wip"); }); +test("a raw QUIC URL dials WebTransport over https", async () => { + const pair = createMockTransportPair(ALPN_06); + const original = globalThis.WebTransport; + let dialed: string | undefined; + + function StubWebTransport(this: unknown, target: string | URL) { + dialed = target.toString(); + return pair.client; + } + globalThis.WebTransport = StubWebTransport as unknown as typeof WebTransport; + + try { + const connection = await connect(new URL("moqt://example.com:4443/anon?jwt=x"), { + websocket: { enabled: false }, + }); + connection.abort(); + } finally { + globalThis.WebTransport = original; + } + + expect(dialed).toBe("https://example.com:4443/anon?jwt=x"); +}); + +test("a failed connect names why the transport failed", async () => { + const original = globalThis.WebTransport; + function FailingWebTransport(this: unknown): never { + throw new Error("handshake refused"); + } + globalThis.WebTransport = FailingWebTransport as unknown as typeof WebTransport; + + try { + await expect(connect(url, { websocket: { enabled: false } })).rejects.toThrow("handshake refused"); + } finally { + globalThis.WebTransport = original; + } +}); + test("connect logs the relay URL without its credentials", async () => { const pair = createMockTransportPair(ALPN_05); const restoreTransport = stubWebTransport(pair.client); diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index 0fc12ff7bfe..11a35aa0642 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -262,7 +262,19 @@ export async function connect({ url, ...props }: ConnectProps): Promise, abort: Promise): Promise { +/** + * The URL a browser transport dials. A raw QUIC `moqt:` or `moql:` URL is reached over + * WebTransport with its scheme swapped for https (draft-ietf-moq-transport-18 §3.1.3); the + * CONNECT carries the scheme, so a server refuses the raw one. Rebuilt rather than assigned, + * since the URL setter refuses a change from a non-special scheme to a special one. + */ +function dialable(url: URL): URL { + if (url.protocol !== "moqt:" && url.protocol !== "moql:") return url; + return new URL(`https:${url.href.slice(url.protocol.length)}`); +} + +async function connectInner(raw: URL, props: Omit, abort: Promise): Promise { + const url = dialable(raw); const wiring: SessionProps = { publish: props.publish, client: true, @@ -307,7 +319,12 @@ async function connectInner(url: URL, props: Omit, abort: P ? [websocket, webtransport] : [webtransport] : [websocket], - ); + ).catch((err: unknown) => { + // Promise.any rejects with an empty message; name each transport's failure instead. + if (!(err instanceof AggregateError)) throw err; + const reasons = err.errors.map((e: unknown) => (e instanceof Error ? e.message : String(e))); + throw new Error(`no transport connected: ${reasons.join("; ")}`, { cause: err }); + }); done(); if (!session) throw new Error("no transport available"); From 09c6fac8c0107a218f9bc01e138bb9840c85f002 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 21:53:54 -0700 Subject: [PATCH 09/15] quest: finish the m0 interop audit and schedule the m1 run Co-Authored-By: Claude Opus 5.5 --- quest/m0/README.md | 1 - quest/m0/audit-interop.md | 114 --------------------- quest/m1/README.md | 1 + quest/m1/audit-interop.md | 119 ++++++++++++++++++++++ quest/m1/interop-coverage/README.md | 2 +- quest/m1/interop-coverage/runner-cases.md | 2 +- quest/m1/interop-runner-nightly.md | 4 +- 7 files changed, 124 insertions(+), 119 deletions(-) delete mode 100644 quest/m0/audit-interop.md create mode 100644 quest/m1/audit-interop.md diff --git a/quest/m0/README.md b/quest/m0/README.md index 6a2c86043a4..6d34ae5f3fa 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -51,4 +51,3 @@ Each lands its fixes in its own PR and plans only what needs a decision. - [CI cost and latency](/quest/m0/ci/README.md) - cloud CI runs only in the merge queue as one `just ci` pass, expensive checks run nightly and gate `release`, caches move to R2, and runners are chosen last - [Simplification audit](/quest/m0/audit-simplify.md) - overbuilt code and APIs are collapsed, merged, or narrowed into a net simpler whole - [Lint override and deprecation audit](/quest/m0/audit-lints.md) - every lint suppression is still needed and says why, nothing calls a deprecated API, and stale Rust overrides fail the build -- [Interop audit](/quest/m0/audit-interop.md) - our interop runner entries match what we ship, and every red interop cell with a moq.dev side is fixed, reported, or noted as down diff --git a/quest/m0/audit-interop.md b/quest/m0/audit-interop.md deleted file mode 100644 index 10d79371249..00000000000 --- a/quest/m0/audit-interop.md +++ /dev/null @@ -1,114 +0,0 @@ -# [M] Interop audit - -## Goal - -Our moq.dev entries in [englishm/moq-interop-runner](https://github.com/englishm/moq-interop-runner) -match what we ship: the client builds on the latest published moq-tokio and -`@moq/net`, the declared draft versions, and the relay endpoints. Every red -cell with a moq.dev side on and on the -runner's report is classified from its logs and either fixed (ours), -reported upstream (theirs), noted as down, or noted as unsupported (a -feature outside the moq-net API). - -## Plan - -Recurs once per major release (decided 2026-10-21). The PR that finishes -this run also adds the next one with the same file name at the end of the next -milestone's `Required` (m0 to m1, m1 to m2). The successor keeps the Goal -and this paragraph; the rest of its Plan is rewritten with what this run -learned: baselines, results, and what to check first. A successor placed in m2 -also lists the [release cut](/quest/m2/release/cut.md) under `Required`, so it -waits for the release in flight to ship. Only `/quest-delete` ends the -recurrence. Fixes land in the run's PR, one commit each, rather than as follow-ups. -Only a change that needs a maintainer decision (public API, scope) or is -independently landable and too large to review alongside the rest becomes a -follow-up through `/quest-plan`. - -Decided with the maintainer on 2026-10-21: - -- Scope: the relays page (our published Rust, browser, and Bun clients - through every registered public relay), plus every cell of the runner's - report with a moq.dev side: our clients, our docker relay, or - `cdn.moq.dev`, including third-party clients failing against our relay. -- Sync new registry entries. Once the [runner nightly](/quest/m1/interop-runner-nightly.md) - lands, it runs the runner's HEAD, so check that newly registered clients - and docker-only relays show up in its report, and ask upstream for a - published image for any it lists as not run. On the relays page, drop an `EXTRA_ENDPOINTS` entry - in `test/smoke/relays.sh` once upstream registers it, and give new relays - the per-relay config they need (`publish_path`, `WEBSOCKET_KEYS`). -- Standing approval to open PRs on englishm/moq-interop-runner that touch only - the `moq-dev-*` builds and registry entries, and to file reproducer issues - on other implementations' repos. Every post ends with the AI marker. - -Both reports run published packages, so a fix turns green only after it -ships. Verify a fix from `main` instead: `test/smoke/smoke.sh` with -`RELAY_BIN` and `MOQ_BIN`, or `test/interop`, pointed at the failing relay. -Classify from the cell's logs, not its summary: `failure_reason` in -`test/smoke/smoke.sh` takes the first error-looking line, which is often -teardown noise after the 30 s timeout. - -### Check first - -From the relays run of 2026-10-21 ([run](https://github.com/moq-dev/moq/actions/runs/39608934024), -moq-cli 0.14.7, `@moq/net` 0.4.4), reproduced live against each relay. -Ours, to fix in this run: - -- The Rust client offers `moq-00` over WebTransport - (`rs/moq-net/src/version.rs`), so moxygen picks draft-14, where we send an - empty-prefix SUBSCRIBE_NAMESPACE that is illegal before draft-16 (moxygen - answers 0x10 "empty"; Cloudflare draft-14 never answers). `main` skips the - empty prefix (#5018) and then announces nothing. Scope the CLI - subscriber's origin to `--broadcast` as the RTC and HLS importers do, and - stop offering `moq-00` over WebTransport. -- WT_SESSION_GONE (0x170d7b68) resets surface as `invalid RESET_STREAM` - transport errors (`rs/moq-tokio/src/quic/web/recv.rs`). Per - draft-ietf-webtrans-http3 §4.4 an unmappable code is still a reset; map - SESSION_GONE to the session's close reason. Log unknown capsules at - DEBUG, not WARN. -- js/net writes each control message as id, length, and body in separate - writes (`js/net/src/ietf/message.ts`, `connection/handshake.ts`). Encode - once and write once, as Rust does, which saves frames and syscalls. It - does not fix red5: QUIC keeps no write boundaries, so red5's one-frame - SETUP parser (reported upstream below) can still stall on a split. -- The JS subscriber joins video unfiltered before draft-20 - (`js/net/src/ietf/subscribe.ts`) and then stops every headless group - stream. Asking for Next Group Start avoids requesting what it discards. -- The JS publisher refuses every FETCH, so a relay that fills with an - upstream FETCH still fails against it. Owned by - [JavaScript FETCH](/quest/m1/js-fetch.md). -- `failure_reason` should report the last stage the subscriber reached - (connected, announced, SUBSCRIBE_OK, no objects) instead of the first - error line. -- Our registry entries are stale: moq-dev-rs declares drafts 14-20 and - moq-dev-js 14-18, though both negotiate draft-22. -- Runner report of 2026-10-19: moq-relay fails `data-subgroup-basic` - (owned by [runner cases](/quest/m1/interop-coverage/runner-cases.md)) and - answers `rendezvous-timeout` with 0x10 instead of TIMEOUT. Cloudflare - moq-rs fails ten cases because moq-relay declines PUBLISH: unsupported, - since PUBLISH is not in the moq-net API. -- Unchased: a late-joining Rust subscriber on moqx got an empty catalog - group and its joining FETCH never yielded the object. - -Theirs, to report upstream: - -- imquic does not replay existing namespaces to a later - SUBSCRIBE_NAMESPACE, so a subscriber that arrives after the publisher sees - nothing on either transport; the WebTransport/QUIC split is that race. -- MOQtail draft-18 never forwards SUBSCRIBE and never answers it. -- Nokia and MOQtail draft-22 narrow a downstream join to Next Object with no - FILL or upstream FETCH (draft-22 §3.2.3, §7.4), so a catalog written once - never arrives. MOQtail resets with INTERNAL_ERROR (0). -- aiomoqt, moxygen, and moqx end the whole subscription when a subscriber - sends STOP_SENDING on one subgroup stream (draft-22 §11.3.2 ends only that - subgroup). -- red5 parses SETUP only when it arrives in one stream frame (RFC 9000 §2). - -Down or misconfigured: quiche-moq (connect idle timeout), Cloudflare -draft-16 (closes with "scope resolution failed"; check the URL in -`EXTRA_ENDPOINTS`). - -## Related - -- [Runner nightly](/quest/m1/interop-runner-nightly.md) - runs every registered client and docker relay with a published image against us -- [Runner teardown](/quest/m3/interop-runner-approval.md) - the runner's moq.dev clients move to the withdrawing `close()` once it ships -- [Moxygen minimal interop](/quest/m1/smoke-moxygen.md) - Moxygen's filter type, waiting upstream diff --git a/quest/m1/README.md b/quest/m1/README.md index 002fcd7fce8..7f92fedf263 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -143,6 +143,7 @@ lite-07 stream type here. - [Artifact size audit](/quest/m1/audit-size.md) - binary, FFI, WASM, and bundle growth since the m0 run is explained, fixed, or planned - [Compile time audit](/quest/m1/audit-compile-time.md) - build-time hotspots are measured against the m0 baselines and the worst are fixed or planned - [Performance audit](/quest/m1/audit-perf.md) - benchmarks that regressed since the m0 run are explained, fixed, or planned, and unbenched hot paths get coverage +- [Interop audit](/quest/m1/audit-interop.md) - our interop runner entries match what we ship, and every red interop cell with a moq.dev side is fixed, reported, or noted as down - [Quest tree audit](/quest/m1/audit-quests.md) - the tree agrees with itself and with main ## Related diff --git a/quest/m1/audit-interop.md b/quest/m1/audit-interop.md new file mode 100644 index 00000000000..8019208c378 --- /dev/null +++ b/quest/m1/audit-interop.md @@ -0,0 +1,119 @@ +# [M] Interop audit + +## Goal + +Our moq.dev entries in [englishm/moq-interop-runner](https://github.com/englishm/moq-interop-runner) +match what we ship: the client builds on the latest published moq-tokio and +`@moq/net`, the declared draft versions, and the relay endpoints. Every red +cell with a moq.dev side on and on the +runner's report is classified from its logs and either fixed (ours), +reported upstream (theirs), noted as down, or noted as unsupported (a +feature outside the moq-net API). + +## Plan + +Recurs once per major release (decided 2026-10-21). The PR that finishes +this run also adds the next one with the same file name at the end of the next +milestone's `Required` (m0 to m1, m1 to m2). The successor keeps the Goal +and this paragraph; the rest of its Plan is rewritten with what this run +learned: baselines, results, and what to check first. A successor placed in m2 +also lists the [release cut](/quest/m2/release/cut.md) under `Required`, so it +waits for the release in flight to ship. Only `/quest-delete` ends the +recurrence. Fixes land in the run's PR, one commit each, rather than as follow-ups. +Only a change that needs a maintainer decision (public API, scope) or is +independently landable and too large to review alongside the rest becomes a +follow-up through `/quest-plan`. + +Decided with the maintainer on 2026-10-21: + +- Scope: the relays page (our published Rust, browser, and Bun clients + through every registered public relay), plus every cell of the runner's + report with a moq.dev side: our clients, our docker relay, or + `cdn.moq.dev`, including third-party clients failing against our relay. +- Sync new registry entries. Once the [runner nightly](/quest/m1/interop-runner-nightly.md) + lands, it runs the runner's HEAD, so check that newly registered clients + and docker-only relays show up in its report, and ask upstream for a + published image for any it lists as not run. The relays page reads every + remote relay from the registry; give new relays the per-relay config they + need (`publish_path`, `WEBSOCKET_KEYS` in `test/smoke/relays.sh`). +- Standing approval to open PRs on englishm/moq-interop-runner that touch only + the `moq-dev-*` builds and registry entries, and to file reproducer issues + on other implementations' repos. Every post ends with the AI marker. + +Both reports run published packages, so a fix turns green only after it +ships. Verify a fix from `main` instead: `MOQ_BIN=target/debug/moq just +relays --only ` from `test/smoke`, or `test/interop`, pointed at the +failing relay. The relays page's reason column now leads with the last stage +the subscriber reached (connected, SUBSCRIBE sent, SUBSCRIBE_OK, headless +groups), but still read the cell's log before classifying. + +### Check first + +The m0 run (2026-10-22) fixed on `main`, so confirm each turned green once +released: + +- moxygen: the Rust client offered `moq-00` over WebTransport and sent an + empty-prefix SUBSCRIBE_NAMESPACE on draft-14. Both cells passed live from + `main`. +- js/net dialed `moqt://` as a WebTransport CONNECT with `:scheme: moqt`, + which moq-relay refuses; the runner's moq-dev-js `remote-quic` cells + failed with an empty message. +- `rendezvous-timeout` answered 0x10 instead of TIMEOUT (fixed in + [moq-dev/moq#5616](https://github.com/moq-dev/moq/pull/5616) before this run). +- `invalid RESET_STREAM: 386759528` (WT_SESSION_GONE) now reads as the + session's close. +- js/net writes each control message in one write; red5 parsed SETUP only + from one stream frame, so recheck red5 WebTransport with js-bun. + +Waiting on others: + +- The runner's moq-dev client images were last built 2026-09-03 (the Rust + one reports moq-native 0.19.15). Asked in + [englishm/moq-interop-runner#170](https://github.com/englishm/moq-interop-runner/pull/170), + which also declares drafts 14-22; until a rebuild, moq-dev client cells + test stale code. [#171](https://github.com/englishm/moq-interop-runner/pull/171) + stops the JS client's http fingerprint fetch against non-moq.dev relays. +- Reported: [moqtail/moqtail#450](https://github.com/moqtail/moqtail/issues/450) + (draft-18 never answers SUBSCRIBE), + [openmoq/moqx#852](https://github.com/openmoq/moqx/issues/852) and + [gmarzot/aiomoqt#123](https://github.com/gmarzot/aiomoqt/issues/123) + (STOP_SENDING on one subgroup ends the subscription, as + [facebookexperimental/moxygen#252](https://github.com/facebookexperimental/moxygen/issues/252)), + [yatagai-mm/Moqtopus#2](https://github.com/yatagai-mm/Moqtopus/issues/2) + (closes on a request before SETUP completes), + [meetecho/imquic#42](https://github.com/meetecho/imquic/issues/42) and + [Eyevinn/moqlivemock#157](https://github.com/Eyevinn/moqlivemock/issues/157) + (interop clients drop the URL path), plus moxygen's filter type 250 + ([facebookexperimental/moxygen#225](https://github.com/facebookexperimental/moxygen/issues/225)), + which also fails moqx's client against our relay. +- The JS publisher refuses every FETCH, so every relay that fills a late + joiner with an upstream FETCH (imquic, MOQtail draft-22, moqt-nr draft-22, + aiomoqt, red5) fails `js-web` publishing. Owned by + [JavaScript FETCH](/quest/m1/js-fetch.md); recheck those cells after it + lands before blaming the relay. MOQtail resets that subscription with + error 0. + +Unsupported or down, not worth rechecking unless something changes: +PUBLISH and SUBSCRIBE_TRACKS cases (`publish-track-*`, `namespace-discovery`, +Cloudflare's PUBLISH cases), Cloudflare draft-14 (no SUBSCRIBE_NAMESPACE, per +its feature matrix), Cloudflare draft-16 (needs a provisioned token; dropped +from the relays page), quiche-moq (connect idle timeout), xquic (its docker +relay binds loopback; its clients are raw QUIC only), and moqtopus and +xquic-draft-18 over WebTransport. + +Not yet reported, for lack of a repo or a reproduction: Nokia (moqt-nr) and +MOQtail draft-22 narrowing a downstream join to Next Object with no FILL, +red5 (its repo is not public), and imquic not replaying namespaces to a later +SUBSCRIBE_NAMESPACE (passed on 2026-10-22). + +Open from the m0 run: the JS subscriber still joins unfiltered before +draft-20, so a relay starting mid-group yields headless groups it stops. Next +Group Start would lose a write-once catalog without a joining FETCH, so this +waits on [JavaScript FETCH](/quest/m1/js-fetch.md). + +## Related + +- [Runner nightly](/quest/m1/interop-runner-nightly.md) - runs every registered client and docker relay with a published image against us +- [Runner teardown](/quest/m3/interop-runner-approval.md) - the runner's moq.dev clients move to the withdrawing `close()` once it ships +- [Moxygen minimal interop](/quest/m1/smoke-moxygen.md) - Moxygen's filter type, waiting upstream +- [JavaScript FETCH](/quest/m1/js-fetch.md) - the JS publisher serves FETCH, which most red `js-web` publisher cells need diff --git a/quest/m1/interop-coverage/README.md b/quest/m1/interop-coverage/README.md index 8045013a6df..5fe73319a65 100644 --- a/quest/m1/interop-coverage/README.md +++ b/quest/m1/interop-coverage/README.md @@ -48,5 +48,5 @@ coverage on `main` at `3ab6cbf81d`. ## Related -- [Interop audit](/quest/m0/audit-interop.md) - classifies the red cells these cases produce +- [Interop audit](/quest/m1/audit-interop.md) - classifies the red cells these cases produce - [moq-transport passthrough](/quest/m1/ietf-extension/README.md) - Moxygen's moq-test conformance through moq-relay diff --git a/quest/m1/interop-coverage/runner-cases.md b/quest/m1/interop-coverage/runner-cases.md index b4df1e8cbaa..5cc63b4b414 100644 --- a/quest/m1/interop-coverage/runner-cases.md +++ b/quest/m1/interop-coverage/runner-cases.md @@ -12,7 +12,7 @@ moq-relay. Decided with the maintainer on 2026-10-21. These cases exist in the runner's `docs/TEST-SPECIFICATIONS.md`, so implement them in our clients -there rather than in-tree; the [interop audit](/quest/m0/audit-interop.md)'s +there rather than in-tree; the [interop audit](/quest/m1/audit-interop.md)'s standing approval covers PRs touching the `moq-dev-*` builds. As of the runner report of 2026-10-19, moq-relay fails diff --git a/quest/m1/interop-runner-nightly.md b/quest/m1/interop-runner-nightly.md index ae70629664e..6cbcebcc8e9 100644 --- a/quest/m1/interop-runner-nightly.md +++ b/quest/m1/interop-runner-nightly.md @@ -22,7 +22,7 @@ Decided with the maintainer on 2026-10-21. registry's client contract. - Track the runner's HEAD, not a pinned commit. The registry lives in the runner's checkout, so a pin would freeze it and hide every implementation - registered after it until the next [interop audit](/quest/m0/audit-interop.md). + registered after it until the next [interop audit](/quest/m1/audit-interop.md). The Cloudflare and Moxygen smoke lanes already build and run unpinned third-party HEAD the same way. - Split the work the way `interop.yml` does, so upstream code never runs @@ -48,4 +48,4 @@ Public API: none. Wire: none. ## Related -- [Interop audit](/quest/m0/audit-interop.md) - audits this job's moq.dev cells each run +- [Interop audit](/quest/m1/audit-interop.md) - audits this job's moq.dev cells each run From 2291042e98d9be46d05d5d962cbaeae3cb7c2c57 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 23:15:08 -0700 Subject: [PATCH 10/15] refactor(net): keep the WebTransport offer inside moq-tokio and moq-wasm Drop the public `Versions::web_protocols()`: the two WebTransport clients map `moq-00` to `moql` themselves, so moq-net's API is unchanged. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/version.rs | 27 --------------------------- rs/moq-tokio/src/quic/backend.rs | 28 +++++++++++++++++++++++++++- rs/moq-wasm/src/transport.rs | 6 +++++- 3 files changed, 32 insertions(+), 29 deletions(-) diff --git a/rs/moq-net/src/version.rs b/rs/moq-net/src/version.rs index 76df116f58b..4dd475fd29f 100644 --- a/rs/moq-net/src/version.rs +++ b/rs/moq-net/src/version.rs @@ -331,25 +331,6 @@ impl Versions { alpns } - /// Compute the unique WebTransport subprotocols to offer for these versions. - /// - /// Draft-14 predates WebTransport subprotocols: `moq-00` is only its QUIC ALPN, and a - /// peer that accepts it over WebTransport picks draft-14 over a newer draft both support. - /// It rides `moql` and negotiates in SETUP instead, as `js/net` does. - pub fn web_protocols(&self) -> Vec<&'static str> { - let mut protocols = Vec::new(); - for v in &self.0 { - let protocol = match v { - Version::Ietf(ietf::Version::Draft14) => ALPN_LITE, - v => v.alpn(), - }; - if !protocols.contains(&protocol) { - protocols.push(protocol); - } - } - protocols - } - /// Return only versions present in both self and other, or `None` if the intersection is empty. pub fn filter(&self, other: &Versions) -> Option { let filtered: Vec = self.0.iter().filter(|v| other.0.contains(v)).copied().collect(); @@ -422,14 +403,6 @@ mod tests { assert_eq!(ALPNS.first(), Some(&newest.alpn())); } - #[test] - fn web_protocols_never_offer_moq_00() { - let draft14 = Version::Ietf(ietf::Version::Draft14); - assert!(!Versions::all().web_protocols().contains(&ALPN_14)); - assert_eq!(Versions::from(draft14).web_protocols(), [ALPN_LITE]); - assert_eq!(Versions::all().web_protocols().len(), Versions::all().alpns().len() - 1); - } - #[test] fn lite_07_wip_is_opt_in_only() { let wip = Version::Lite(lite::Version::Lite07); diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index 558c56e5165..76faa170db5 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -480,7 +480,7 @@ impl Client { let session = match url.scheme() { "https" => { let mut request = web::proto::ConnectRequest::new(url.clone()); - for protocol in versions.web_protocols() { + for protocol in web_protocols(versions) { request = request.with_protocol(protocol.to_string()); } web::Session::connect(connection, request) @@ -553,6 +553,22 @@ impl Error { } } +/// The WebTransport subprotocols to offer for `versions`. +/// +/// Draft-14 predates WebTransport subprotocols: `moq-00` is only its QUIC ALPN, and a peer that +/// accepts it over WebTransport picks draft-14 over a newer draft both support. It rides `moql` +/// and negotiates in SETUP instead, as `js/net` does. +fn web_protocols(versions: &moq_net::Versions) -> Vec<&'static str> { + let mut protocols = Vec::new(); + for alpn in versions.alpns() { + let protocol = if alpn == "moq-00" { "moql" } else { alpn }; + if !protocols.contains(&protocol) { + protocols.push(protocol); + } + } + protocols +} + fn map_client_error(err: web::ClientError) -> Error { match client_status(&err).and_then(crate::ConnectError::from_status_u16) { Some(rejected) => rejected.into(), @@ -1051,6 +1067,16 @@ mod tests { use super::*; use url::Url; + #[test] + fn web_protocols_never_offer_moq_00() { + let all = moq_net::Versions::all(); + assert!(!web_protocols(&all).contains(&"moq-00")); + assert_eq!(web_protocols(&all).len(), all.alpns().len() - 1); + + let draft14: moq_net::Version = "moq-transport-14".parse().unwrap(); + assert_eq!(web_protocols(&draft14.into()), ["moql"]); + } + #[tokio::test] async fn pinned_ipv6_connection() { pinned_connection("[::1]:0", None).await; diff --git a/rs/moq-wasm/src/transport.rs b/rs/moq-wasm/src/transport.rs index 987eea77db6..21c34845f6d 100644 --- a/rs/moq-wasm/src/transport.rs +++ b/rs/moq-wasm/src/transport.rs @@ -25,8 +25,12 @@ pub struct Options { /// Advertise every version `moq-net` supports, so the peer picks one via ALPN. /// Without this the browser negotiates no subprotocol and the session has no /// version to start from. +/// +/// Draft-14 predates WebTransport subprotocols: `moq-00` is only its QUIC ALPN, and a +/// peer that accepts it here picks draft-14 over a newer draft both support. It rides +/// `moql` (also offered) and negotiates in SETUP instead, as `js/net` does. fn builder() -> ClientBuilder { - ClientBuilder::new().with_protocols(moq_net::Versions::all().web_protocols()) + ClientBuilder::new().with_protocols(moq_net::ALPNS.iter().copied().filter(|alpn| *alpn != "moq-00")) } /// Open a browser WebTransport connection to `url`. From 17d3414d1afaa1da66a4e89632b0dcc87449dcc4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 23:45:32 -0700 Subject: [PATCH 11/15] test(smoke): don't claim SUBSCRIBE_OK for a Rust subscriber The Rust CLI logs SUBSCRIBE_OK only at DEBUG, so at INFO a Rust cell that sent SUBSCRIBE and timed out was reported as accepted. Say the OK was not logged unless a positive marker appears. Co-Authored-By: Claude Opus 5.5 --- test/smoke/smoke.sh | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/test/smoke/smoke.sh b/test/smoke/smoke.sh index 9665da9d6f6..9c018539b5e 100755 --- a/test/smoke/smoke.sh +++ b/test/smoke/smoke.sh @@ -851,7 +851,8 @@ failure_reason() { # error-looking line as the detail. The first error alone is often the teardown # after the timeout, which hides how far the cell got. Stages come from the # lines the Rust CLI, @moq/net, and the browser page already log: connected, - # SUBSCRIBE sent, SUBSCRIBE_OK, and a group dropped for starting mid-group. + # SUBSCRIBE sent, SUBSCRIBE_OK (not logged by the Rust CLI at INFO), and a group + # dropped for starting mid-group. [[ -f "$1" ]] || return 0 local log stage track detail log=$(sed 's/\x1b\[[0-9;]*m//g' "$1") @@ -863,10 +864,15 @@ failure_reason() { elif grep -qE "abandoned before it was accepted|waiting for SUBSCRIBE_OK" <<<"$log" || { grep -q 'subscribe start:' <<<"$log" && ! grep -qF "track=$track" <<<"$(grep 'subscribe ok:' <<<"$log")"; }; then stage="no SUBSCRIBE_OK for $track" - elif grep -q 'dropping a group with no head' <<<"$log"; then - stage="SUBSCRIBE_OK for $track, only headless groups" else - stage="SUBSCRIBE_OK for $track, no objects" + # The Rust CLI logs SUBSCRIBE_OK only at DEBUG, so at INFO its acceptance is unknown. + local ok="SUBSCRIBE_OK for $track" + grep -qE 'subscribe start:|received subscribe ok' <<<"$log" || ok="SUBSCRIBE for $track, no SUBSCRIBE_OK logged" + if grep -q 'dropping a group with no head' <<<"$log"; then + stage="$ok, only headless groups" + else + stage="$ok, no objects" + fi fi # Teardown noise once the timeout kills the client says nothing about the cause. detail=$(grep -vE 'failed to read capsule|unknown capsule' <<<"$log" | From 826473fa598bd995d40efeb8a4e6ce150d1e7dc2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 23:53:35 -0700 Subject: [PATCH 12/15] fix(tokio): report SessionGone until the close capsule arrives A WT_SESSION_GONE reset can beat the CloseWebTransportSession capsule. Reporting it as `Closed(0, "")` claimed a clean close and hid the real code; `WebTransportError::SessionGone` says only that the session is gone, and the recorded reason replaces it once known. Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/src/quic/web/error.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/rs/moq-tokio/src/quic/web/error.rs b/rs/moq-tokio/src/quic/web/error.rs index c2fc53a8ec3..8f6a8fcf704 100644 --- a/rs/moq-tokio/src/quic/web/error.rs +++ b/rs/moq-tokio/src/quic/web/error.rs @@ -114,8 +114,8 @@ impl CloseReason { /// WT_SESSION_GONE, which tears down the streams of a closed session /// (draft-ietf-webtrans-http3 §4.4), or None for any other code. /// - /// The reset can beat the close capsule; until the capsule lands, it reads as a - /// clean close, as a CONNECT stream that ends without one does. + /// The reset can beat the close capsule, so until it lands the code and reason are + /// unknown: [`WebTransportError::SessionGone`], never a guessed clean close. pub(crate) fn session_gone(&self, code: crate::quic::driver::VarInt) -> Option { if self.raw || code.into_inner() != WT_SESSION_GONE { return None; @@ -124,7 +124,7 @@ impl CloseReason { self.reason .get() .cloned() - .unwrap_or_else(|| WebTransportError::Closed(0, String::new()).into()), + .unwrap_or_else(|| WebTransportError::SessionGone.into()), ) } @@ -139,7 +139,7 @@ impl CloseReason { let conn = match &err { SessionError::ConnectionError(conn) | SessionError::SendDatagramError(crate::quic::driver::SendDatagramError::ConnectionLost(conn)) => Some(conn), - SessionError::WebTransportError(WebTransportError::Closed(..)) => None, + SessionError::WebTransportError(WebTransportError::Closed(..) | WebTransportError::SessionGone) => None, _ => return err, }; @@ -171,6 +171,10 @@ pub enum WebTransportError { #[error("unknown session")] UnknownSession, + /// The peer tore down the session's streams before its close reason arrived. + #[error("session gone")] + SessionGone, + #[error("read error: {0}")] ReadError(#[from] crate::quic::driver::ReadExactError), @@ -410,7 +414,7 @@ mod tests { let close = CloseReason::new(false); assert!(matches!( close.session_gone(gone), - Some(SessionError::WebTransportError(WebTransportError::Closed(0, _))) + Some(SessionError::WebTransportError(WebTransportError::SessionGone)) )); close.set(WebTransportError::Closed(7, "bye".into()).into()); From c0164fa78b56eb735790a39067c2cb3cccc568b0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 14:13:13 -0700 Subject: [PATCH 13/15] revert(cli): drop the subscriber scope moq-lite clients already ask only for watched prefixes (#6165), and moq-mux Source watches the sibling broadcasts a catalog references, so the subtree scope only lost those siblings. The moxygen draft-14 cell stays red until IETF lazy discovery asks for watched prefixes too. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/main.rs | 39 ++++----------------------------------- 1 file changed, 4 insertions(+), 35 deletions(-) diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index a30ab3f5b22..196863afe11 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -417,17 +417,6 @@ impl Directions { } } -/// The broadcasts `names` and everything beneath them, as patterns to scope an origin. -fn consumed(names: impl IntoIterator>) -> anyhow::Result { - names - .into_iter() - .map(|name| { - let name = name.as_ref(); - moq_net::Pattern::subtree(name).with_context(|| format!("invalid broadcast name `{name}`")) - }) - .collect() -} - /// Attach the shared Origin to the MoQ network: dial a relay, accept inbound /// sessions, mesh with the LAN, or any combination. /// @@ -450,16 +439,11 @@ fn consumed(names: impl IntoIterator>) -> anyhow::Result< /// the estimate is only meaningful across everything sharing it. A stage that built /// its own would split its own tracks correctly and still oversubscribe every other /// stage on the same connection, which is the whole problem. -/// -/// The dialed relay is asked only for `consumed`, the broadcasts the stages read. -/// moq-transport drafts 14 and 15 cannot ask for the empty namespace, so an -/// unscoped subscriber discovers nothing from a peer that announces nothing unasked. async fn spawn_moq( moq: &MoqSide, net: &Net, cluster: moq_relay::cluster::Cluster, directions: Directions, - consumed: &moq_net::Patterns, tasks: &mut JoinSet>, ) -> anyhow::Result { let cluster = cluster @@ -495,10 +479,7 @@ async fn spawn_moq( dialer = dialer.with_publisher(origin.consume()); } if directions.consume { - let scoped = origin - .scope("", consumed) - .context("failed to scope the origin to the consumed broadcasts")?; - dialer = dialer.with_subscriber(scoped); + dialer = dialer.with_subscriber(origin.clone()); } // Both directions ride one connection, so this dials directly rather than @@ -567,7 +548,7 @@ async fn run_play(moq: MoqSide, args: play::Args, net: Net) -> anyhow::Result<() client, connection, .. - } = spawn_moq(&moq, &net, cluster.clone(), directions, &consumed([&name])?, &mut tasks).await?; + } = spawn_moq(&moq, &net, cluster.clone(), directions, &mut tasks).await?; let result = play::run(origin.consume(), name, args, tasks).await; detach(connection, client, &cluster).await; @@ -595,23 +576,13 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res let mut client = None; let mut connection = None; let result = async { - let exports = stages.iter().filter(|stage| matches!(stage, Command::Export(_))); - let consumed = consumed(exports.map(|stage| stage.broadcast(&moq)))?; let Attached { bandwidth, origin, client: dialer, connection: attached, shutdown, - } = spawn_moq( - &moq, - &net, - cluster.clone(), - Directions::of(&stages), - &consumed, - &mut tasks, - ) - .await?; + } = spawn_moq(&moq, &net, cluster.clone(), Directions::of(&stages), &mut tasks).await?; client = dialer; connection = attached; @@ -1304,8 +1275,7 @@ mod tests { publish: true, consume: false, }; - let consumed = moq_net::Patterns::new(); - let Err(err) = spawn_moq(&invocation.moq, &net, cluster, directions, &consumed, &mut tasks).await else { + let Err(err) = spawn_moq(&invocation.moq, &net, cluster, directions, &mut tasks).await else { panic!("the occupied port must fail startup"); }; assert!(err.to_string().contains("failed to bind listeners"), "{err:#}"); @@ -1461,7 +1431,6 @@ mod tests { publish: true, consume: false, }, - &moq_net::Patterns::new(), &mut tasks, ) .await From 0109bf0063ada8a5d57c5116087cc753b0cb8be7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 14:15:29 -0700 Subject: [PATCH 14/15] quest: promote IETF lazy discovery to m1 It owns the moxygen draft-14 CLI cell now that the CLI does no subscriber scoping of its own. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 + quest/m1/audit-interop.md | 10 +++++++--- quest/{m2 => m1}/ietf-lazy-discovery.md | 10 ++++++++++ quest/m2/README.md | 1 - quest/m2/keyed-prefix-polling.md | 2 +- 5 files changed, 19 insertions(+), 5 deletions(-) rename quest/{m2 => m1}/ietf-lazy-discovery.md (63%) diff --git a/quest/m1/README.md b/quest/m1/README.md index b5ec6ab5134..6b6c36107b8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -155,6 +155,7 @@ tracks, so per-group stats stop costing cluster announces. - [Drop dead items](/quest/m1/drop-dead-items.md) - the audit's uncalled public items and test helpers are removed - [Warmup in the hang draft](/quest/m1/warmup-draft.md) - the draft lets a dual-decoder consumer present before `warmup`, matching the shipping `decode::Consumer` - [API-diff check](/quest/m1/api-diff.md) - a recipe diffs every published package's API against its last release, run on PRs into release and against main before the cut +- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - moq-transport sessions ask for announcements only under watched prefixes, starting with UNSUBSCRIBE_NAMESPACE on d14/15 - [Interop coverage](/quest/m1/interop-coverage/README.md) - the riskiest moq-transport features moq-net exposes get cross-implementation tests over the even drafts 14 to 22 - [Runner nightly](/quest/m1/interop-runner-nightly.md) - a nightly job runs the upstream interop runner against our relay and clients, so newly registered implementations with a published image are tested without a change here - [One Annex-B exporter](/quest/m1/annexb-export.md) - H.264 and H.265 Annex-B export share one private lifecycle, which also follows a rebound rendition diff --git a/quest/m1/audit-interop.md b/quest/m1/audit-interop.md index 8019208c378..4d93aee879e 100644 --- a/quest/m1/audit-interop.md +++ b/quest/m1/audit-interop.md @@ -52,9 +52,12 @@ groups), but still read the cell's log before classifying. The m0 run (2026-10-22) fixed on `main`, so confirm each turned green once released: -- moxygen: the Rust client offered `moq-00` over WebTransport and sent an - empty-prefix SUBSCRIBE_NAMESPACE on draft-14. Both cells passed live from - `main`. +- moxygen: the Rust client offered `moq-00` over WebTransport, so moxygen + picked draft-14 over draft-16. That cell passed live from `main`. The + draft-14 QUIC cell stays red until [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) + lands: the CLI asks draft-14 for no namespace, so it discovers nothing + from moxygen. A CLI subtree scope was tried and dropped, since it lost + the sibling broadcasts a catalog references. - js/net dialed `moqt://` as a WebTransport CONNECT with `:scheme: moqt`, which moq-relay refuses; the runner's moq-dev-js `remote-quic` cells failed with an empty message. @@ -117,3 +120,4 @@ waits on [JavaScript FETCH](/quest/m1/js-fetch.md). - [Runner teardown](/quest/m3/interop-runner-approval.md) - the runner's moq.dev clients move to the withdrawing `close()` once it ships - [Moxygen minimal interop](/quest/m1/smoke-moxygen.md) - Moxygen's filter type, waiting upstream - [JavaScript FETCH](/quest/m1/js-fetch.md) - the JS publisher serves FETCH, which most red `js-web` publisher cells need +- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - asks for watched prefixes, which the red moxygen draft-14 cell needs diff --git a/quest/m2/ietf-lazy-discovery.md b/quest/m1/ietf-lazy-discovery.md similarity index 63% rename from quest/m2/ietf-lazy-discovery.md rename to quest/m1/ietf-lazy-discovery.md index e9a2cc2f401..aa9f21c0653 100644 --- a/quest/m2/ietf-lazy-discovery.md +++ b/quest/m1/ietf-lazy-discovery.md @@ -13,6 +13,16 @@ discovery lazy and kept IETF eager: lazy SUBSCRIBE_NAMESPACE needs an UNSUBSCRIBE_NAMESPACE path on drafts 14 and 15, and peers still push unsolicited PUBLISH_NAMESPACE. +In m1 because it owns a red interop cell (decided 2026-10-22): `moq` +against moxygen on draft-14 discovers nothing. moq-net never asks draft-14 +or 15 for the empty namespace, and moxygen announces nothing unasked. The +CLI does no scoping of its own: the subtree scope tried for this in +[moq-dev/moq#6255](https://github.com/moq-dev/moq/pull/6255) lost the +sibling broadcasts a catalog references, such as a transcode's `.` source. +Asking for the watched prefixes fixes the cell without that loss; verify +it with `MOQ_BIN=target/debug/moq just relays --only moxygen` from +`test/smoke`. + - First step: Rust sends UNSUBSCRIBE_NAMESPACE on d14/15. It never does today ([moq-dev/moq#6133](https://github.com/moq-dev/moq/pull/6133)). - Then drive SUBSCRIBE_NAMESPACE from the origin's watched prefixes, as the diff --git a/quest/m2/README.md b/quest/m2/README.md index a584c58217b..1b0dff7391e 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -80,7 +80,6 @@ ranked just below deterministic TS legs, which it requires. - [Rust-backed @moq/net](/quest/m2/rs2ts/README.md) - measure browser baselines, complete WASM bindings, and adopt Rust behind @moq/net before the later JS-only stage - [Front parking](/quest/m2/origin-front-parks.md) - an unroutable request waits on a front instead of re-asking on every route-table move - [Route wakes](/quest/m2/route-wakes.md) - a route change wakes only the fronts it can move, so pool churn stops scaling with served paths -- [IETF lazy discovery](/quest/m2/ietf-lazy-discovery.md) - moq-transport sessions ask for announcements only under watched prefixes, starting with UNSUBSCRIBE_NAMESPACE on d14/15 - [Draft 14-16 session errors](/quest/m2/ietf-adapter-session-errors.md) - the moq-transport adapter keeps session-level error causes, so a terminal rejection is not redialed - [Keyed prefix polling](/quest/m2/keyed-prefix-polling.md) - `SubscriberDriver` polls only the prefixes a wake concerns, if a bench shows the slope - [Route health](/quest/m2/route-health/README.md) - epoch-only active/passive track recovery over metadata-only subscriptions diff --git a/quest/m2/keyed-prefix-polling.md b/quest/m2/keyed-prefix-polling.md index 5d04843514a..45e1fca87d4 100644 --- a/quest/m2/keyed-prefix-polling.md +++ b/quest/m2/keyed-prefix-polling.md @@ -22,4 +22,4 @@ Public API: none. Wire: none. ## Related -- [IETF lazy discovery](/quest/m2/ietf-lazy-discovery.md) - more prefixes per session once IETF is lazy too +- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - more prefixes per session once IETF is lazy too From b84f6114ddf7b2b823ee4cf805b79d38069d441d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 14:16:09 -0700 Subject: [PATCH 15/15] style(js/net): format the merged control writes Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.ts | 123 +++++++++++++++++++++++------------ 1 file changed, 80 insertions(+), 43 deletions(-) diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index d2d838fa669..4c872f82851 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -462,7 +462,12 @@ export class Publisher { const key = JSON.stringify([msg.trackNamespace, msg.trackName]); const claimed = this.#session.version === Version.DRAFT_18; if (claimed && this.#trackClaims.get(key) === "publish") { - await Message.write(request.writer, RequestError.id, new RequestError({ errorCode: 0x19, reasonPhrase: "track already PUBLISHed" }), this.#session.version); + await Message.write( + request.writer, + RequestError.id, + new RequestError({ errorCode: 0x19, reasonPhrase: "track already PUBLISHed" }), + this.#session.version, + ); return; } // Only a subscription we would serve holds a slot, as in Rust: a refused one ends at @@ -2198,7 +2203,12 @@ export class Publisher { (prefix) => Path.hasPrefix(prefix, msg.prefix) || Path.hasPrefix(msg.prefix, prefix), ) ) { - await Message.write(request.writer, RequestError.id, new RequestError({ errorCode: 0x30, reasonPhrase: "overlapping SUBSCRIBE_TRACKS prefix" }), version); + await Message.write( + request.writer, + RequestError.id, + new RequestError({ errorCode: 0x30, reasonPhrase: "overlapping SUBSCRIBE_TRACKS prefix" }), + version, + ); return; } this.#listings.add(msg.prefix); @@ -2461,25 +2471,30 @@ export class Publisher { parameters.subscriptionFilter = subscription.filter; } const largest = wireOf(source).largest(name); - await Message.write(request.writer, Publish.id, new Publish({ - requestId: id, - trackNamespace: namespace, - trackName: name, - trackAlias: id, - groupOrder: GroupOrder.Descending, - contentExists: !!largest, - largest: largest && { groupId: BigInt(largest.group), objectId: BigInt(largest.frame) }, - forward: false, - parameters, - properties: subscription.propertiesWanted - ? { - priority: toWire(info.priority), - groupOrder: Properties.DESCENDING, - timescale: info.timescale ?? undefined, - maxCacheDuration: info.maxAge === undefined ? undefined : BigInt(info.maxAge), - } - : {}, - }), version); + await Message.write( + request.writer, + Publish.id, + new Publish({ + requestId: id, + trackNamespace: namespace, + trackName: name, + trackAlias: id, + groupOrder: GroupOrder.Descending, + contentExists: !!largest, + largest: largest && { groupId: BigInt(largest.group), objectId: BigInt(largest.frame) }, + forward: false, + parameters, + properties: subscription.propertiesWanted + ? { + priority: toWire(info.priority), + groupOrder: Properties.DESCENDING, + timescale: info.timescale ?? undefined, + maxCacheDuration: info.maxAge === undefined ? undefined : BigInt(info.maxAge), + } + : {}, + }), + version, + ); hold.signal.removeEventListener("abort", cancelled); stop(); // Once sent, the PUBLISH survives cancellation of its listing. @@ -2507,13 +2522,18 @@ export class Publisher { let forward = !Filter.isDraft20(version) && params.forward !== false; for (;;) { if (!this.#gate.allows(namespace) || unlisted.peek() !== undefined) { - await Message.write(request.writer, PublishDone.id, new PublishDone({ - statusCode: this.#gate.allows(namespace) - ? PublishDoneStatus.INTERNAL_ERROR - : PublishDoneStatus.UNAUTHORIZED, - streamCount: BigInt(state.streams.opened), - reasonPhrase: "track no longer listed or granted", - }), version); + await Message.write( + request.writer, + PublishDone.id, + new PublishDone({ + statusCode: this.#gate.allows(namespace) + ? PublishDoneStatus.INTERNAL_ERROR + : PublishDoneStatus.UNAUTHORIZED, + streamCount: BigInt(state.streams.opened), + reasonPhrase: "track no longer listed or granted", + }), + version, + ); return; } if (forward) { @@ -2534,11 +2554,16 @@ export class Publisher { await refuseUpdate(request.writer, version, "subscribe", this.#failed(err)); await this.#publishFailed(request, state.streams); } else { - await Message.write(request.writer, PublishDone.id, new PublishDone({ - statusCode: PublishDoneStatus.INTERNAL_ERROR, - streamCount: BigInt(state.streams.opened), - reasonPhrase: "resume failed", - }), version); + await Message.write( + request.writer, + PublishDone.id, + new PublishDone({ + statusCode: PublishDoneStatus.INTERNAL_ERROR, + streamCount: BigInt(state.streams.opened), + reasonPhrase: "resume failed", + }), + version, + ); } return; } @@ -2562,11 +2587,18 @@ export class Publisher { gate.dispose(); } if (next === true) { - await Message.write(request.writer, PublishDone.id, new PublishDone({ - statusCode: revoked.peek() ? PublishDoneStatus.UNAUTHORIZED : PublishDoneStatus.INTERNAL_ERROR, - streamCount: BigInt(state.streams.opened), - reasonPhrase: "track no longer listed or granted", - }), version); + await Message.write( + request.writer, + PublishDone.id, + new PublishDone({ + statusCode: revoked.peek() + ? PublishDoneStatus.UNAUTHORIZED + : PublishDoneStatus.INTERNAL_ERROR, + streamCount: BigInt(state.streams.opened), + reasonPhrase: "track no longer listed or granted", + }), + version, + ); return; } if (!next) return request.ended; @@ -2627,11 +2659,16 @@ export class Publisher { } async #publishFailed(request: RequestStream, streams: StreamCount) { - await Message.write(request.writer, PublishDone.id, new PublishDone({ - statusCode: PublishDoneStatus.UPDATE_FAILED, - streamCount: BigInt(streams.opened), - reasonPhrase: "update failed", - }), this.#session.version); + await Message.write( + request.writer, + PublishDone.id, + new PublishDone({ + statusCode: PublishDoneStatus.UPDATE_FAILED, + streamCount: BigInt(streams.opened), + reasonPhrase: "update failed", + }), + this.#session.version, + ); } #applyPublished(msg: Subscribe, params: Parameters) {