From a56f60a6b07977559bfe97034483ff92c9036d3b Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Thu, 22 Oct 2026 18:08:09 -0700 Subject: [PATCH 1/3] chore(quest): claim quest/m1/test-flakes-2/auth-union-renewal.md Quest-Claim: 9cf04398847f47fb From 9894b111792ef6e44dcea380ea572e861bfebb95 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Thu, 22 Oct 2026 18:28:42 -0700 Subject: [PATCH 2/3] fix(net): never leave a torn request message on the stream The union renewal flake was the publisher writing REQUEST_OK's type and body in two writes: the shared write budget can yield to the host after any write, and under load the subscriber's FIN landed in that gap, so the publisher FINed after a bare type and the subscriber closed the session as malformed. #6255 already made every JS message one write; this pins that with a test that yields after every write, and closes the rest: - JS: the IETF subscriber no longer sends an update of defaults when its last subscriber leaves (the logged priority=255 was that), matching lite and Rust, and an update that fails to encode ends the track loudly instead of being swallowed. - Rust: every IETF control message buffers its type with its body, all or nothing, via Writer::buffer_message / encode_message, so a failed encode or a cancelled write can no longer strand a bare type for the next write or a FIN to complete. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EeyiEy5FWz3z44DYZVo7vU --- js/net/src/ietf/message.ts | 11 +- js/net/src/ietf/subscriber.ts | 17 ++- js/net/src/subscribe_priority.test.ts | 113 ++++++++++++++++++- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/auth-union-renewal.md | 44 -------- rs/moq-net/src/coding/writer.rs | 34 +++++- rs/moq-net/src/ietf/publisher.rs | 94 +++++++-------- rs/moq-net/src/ietf/publisher/tracks.rs | 34 +++--- rs/moq-net/src/ietf/request_stream.rs | 38 ++++--- rs/moq-net/src/ietf/subscriber.rs | 39 +++---- rs/moq-net/src/ietf/subscriber/tracks.rs | 50 ++++---- 11 files changed, 280 insertions(+), 195 deletions(-) delete mode 100644 quest/m1/test-flakes-2/auth-union-renewal.md diff --git a/js/net/src/ietf/message.ts b/js/net/src/ietf/message.ts index 989f955eaae..2a2de8b6423 100644 --- a/js/net/src/ietf/message.ts +++ b/js/net/src/ietf/message.ts @@ -1,4 +1,4 @@ -import { concat, encodeVarint, Reader, Writer } from "../stream.ts"; +import { concat, encodeVarint, Reader, type StreamVersion, Writer } from "../stream.ts"; import { MalformedAuthToken } from "./token.ts"; import type { IetfVersion } from "./version.ts"; @@ -23,8 +23,13 @@ export interface Encode { // 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])); + await writer.write(await frame(writer.version, id, msg, version)); +} + +// The one chunk `write` sends, for a caller that must tell a failed encode from a failed write. +export async function frame(stream: StreamVersion, id: number, msg: Encode, version: IetfVersion) { + const body = await Writer.collect(stream, (w) => msg.encode(w, version)); + return concat([encodeVarint(id, stream), body]); } // Reads a message with a u16 size prefix. diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index b8fc38f621d..6f42d69a69f 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -1061,10 +1061,12 @@ export class Subscriber { void stopped.then(() => stop.abort()); for (;;) { + // With no subscriber left there is nothing to follow: the subscription is ending, or + // keeps what it last asked for until one returns, as on lite and in Rust. const subscription = producer.subscription.peek(); const wanted = toWire(subscription?.priority ?? 0); - const wantedMaxAge = this.#maxAge ? (subscription?.maxDelay ?? Number.MAX_SAFE_INTEGER) : undefined; - if ((wanted === sent && wantedMaxAge === sentMaxAge) || unanswered.peek() > 0) { + const wantedMaxAge = this.#maxAge ? subscription?.maxDelay : undefined; + if (!subscription || (wanted === sent && wantedMaxAge === sentMaxAge) || unanswered.peek() > 0) { const disposes: Dispose[] = []; const wake = new Promise((resolve) => { disposes.push( @@ -1092,10 +1094,19 @@ export class Subscriber { const update = legacy ? new SubscribeUpdate({ requestId, ownRequestId: own, parameters }) : new SubscribeUpdate({ requestId: own, parameters }); + let message: Uint8Array; + try { + message = await Message.frame(stream.writer.version, SubscribeUpdate.id, update, version); + } catch (err) { + // Nothing reached the wire, but the track can no longer follow its subscribers. + console.warn(`subscribe update failed: id=${requestId} error=${reason(error(err))}`); + producer.close(error(err)); + return; + } // Counted before writing: the answer can be read before the write's promise resolves. if (answered) unanswered.update((count) => count + 1); try { - await Message.write(stream.writer, SubscribeUpdate.id, update, version); + await stream.writer.write(message); } catch { // The publisher stopped reading; its responses decide how the track ends. return; diff --git a/js/net/src/subscribe_priority.test.ts b/js/net/src/subscribe_priority.test.ts index 3b160162b01..b082c04aef0 100644 --- a/js/net/src/subscribe_priority.test.ts +++ b/js/net/src/subscribe_priority.test.ts @@ -2,15 +2,19 @@ // follows it as they come and go. The JS origin never serves a received route onward, so the // relay case lives with the Rust relay (rs/moq-net/tests/integration/subscribe_priority.rs). -import { expect, test } from "bun:test"; +import { expect, spyOn, test } from "bun:test"; +import { Manual } from "@moq/clock"; import { accept, connect, type Established } from "./connection/index.ts"; import * as Ietf from "./ietf/index.ts"; +import { SubscribeUpdate } from "./ietf/subscribe.ts"; import * as Lite from "./lite/index.ts"; import { createMockTransportPair, type MockTransport } from "./mock.ts"; import { Producer as OriginProducer } from "./origin.ts"; import * as Path from "./path.ts"; +import { Writer } from "./stream.ts"; import { Timescale } from "./time.ts"; import type { Producer as TrackProducer } from "./track.ts"; +import { nextTask } from "./util/yield.ts"; import { wireOf } from "./wire.ts"; const url = new URL("https://localhost:4443/test"); @@ -89,3 +93,110 @@ test.each([Lite.ALPN_07_WIP, ...Object.values(Ietf.ALPN)])( server.abort(); }, ); + +// A write may yield to the host after it lands (the shared write budget does under load), so the +// peer's end of a request can arrive between any two writes. With every write yielding, the last +// subscribers leaving while an update is answered must leave only whole messages on the stream: +// a torn one closes the session as malformed. +test.each(Object.values(Ietf.ALPN))("%s ends a subscription mid-update without a torn message", async (protocol) => { + const clock = new Manual(); + const restore = clock.install(); + const write = Writer.prototype.write; + const yielding = spyOn(Writer.prototype, "write").mockImplementation(async function (this: Writer, chunk) { + await write.call(this, chunk); + await nextTask(); + }); + const origin = new OriginProducer(); + const broadcast = origin.createBroadcast(path); + broadcast.announce(); + const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const [client, server] = await session(protocol, origin); + let closed: Error | null | undefined; + void client.closed.then((err) => { + closed = err; + }); + try { + const remote = wireOf(client).consume(path); + const low = remote.track("video").subscribe({ priority: LOW }); + await priority(video, LOW); + const high = remote.track("video").subscribe({ priority: HIGH }); + await priority(video, HIGH); + high.close(); + await priority(video, LOW); + // The publisher has read the update and is answering it as the subscription ends. + low.close(); + while (video.subscription.peek() !== undefined) await video.subscription.changed(); + // Each hop is a task or two away, so this lets the subscriber read all the publisher wrote. + for (let i = 0; i < 20; i++) await nextTask(); + expect(closed).toBeUndefined(); + remote.close(); + } finally { + yielding.mockRestore(); + broadcast.close(); + client.abort(); + server.abort(); + restore(); + } +}); + +// With no subscriber left there is no aggregate to follow, so the subscription ends without an +// update of defaults first, as on lite and in Rust. +test.each(Object.values(Ietf.ALPN))("%s sends no update when the last subscriber leaves", async (protocol) => { + const origin = new OriginProducer(); + const broadcast = origin.createBroadcast(path); + broadcast.announce(); + const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const [client, server] = await session(protocol, origin); + try { + const remote = wireOf(client).consume(path); + const track = remote.track("video").subscribe({ priority: LOW }); + await priority(video, LOW); + await track.info(); + track.close(); + while (video.subscription.peek() !== undefined) { + expect(video.subscription.peek()?.priority).toBe(LOW); + await video.subscription.changed(); + } + remote.close(); + } finally { + broadcast.close(); + client.abort(); + server.abort(); + } +}); + +// An update that fails to encode is a bug, not a publisher that stopped reading, so it ends the +// track with the error rather than leaving it at a stale priority. Nothing reaches the wire. +test.each(Object.values(Ietf.ALPN))("%s ends the track when an update fails to encode", async (protocol) => { + const origin = new OriginProducer(); + const broadcast = origin.createBroadcast(path); + broadcast.announce(); + const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const [client, server] = await session(protocol, origin); + let closed: Error | null | undefined; + void client.closed.then((err) => { + closed = err; + }); + const broken = new Error("broken update"); + const encode = spyOn(SubscribeUpdate.prototype, "encode").mockRejectedValue(broken); + try { + const remote = wireOf(client).consume(path); + const track = remote.track("video").subscribe({ priority: LOW }); + await priority(video, LOW); + await track.info(); + const high = remote.track("video").subscribe({ priority: HIGH }); + expect(await track.closed).toBe(broken); + expect(await high.closed).toBe(broken); + while (video.subscription.peek() !== undefined) { + expect(video.subscription.peek()?.priority).toBe(LOW); + await video.subscription.changed(); + } + expect(closed).toBeUndefined(); + remote.close(); + } finally { + encode.mockRestore(); + broadcast.close(); + client.abort(); + server.abort(); + } +}); diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index ca2565fd675..b282d750589 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -39,7 +39,6 @@ Public API: none. Wire: none. ## Required -- [Union renewal](/quest/m1/test-flakes-2/auth-union-renewal.md) - the JS subscriber never leaves a torn REQUEST_UPDATE on the stream - [One process](/quest/m1/test-flakes-2/one-process.md) - no test writes process-wide env or unwraps the crypto provider install - [Media resource instrumentation](/quest/m1/test-flakes-2/resource-instrumentation.md) - the media test stops seeing a playing player report nothing to release - [Interop cpp to js budget](/quest/m1/test-flakes-2/interop-cpp-js.md) - the cpp -> js cell finishes inside its 30 s budget, fixed at its cause diff --git a/quest/m1/test-flakes-2/auth-union-renewal.md b/quest/m1/test-flakes-2/auth-union-renewal.md deleted file mode 100644 index 2a0f0155158..00000000000 --- a/quest/m1/test-flakes-2/auth-union-renewal.md +++ /dev/null @@ -1,44 +0,0 @@ -# [S] A union renewal never sends a torn request update - -## Goal - -`js/net/src/auth.test.ts`'s "a request waits for its token and survives union -renewal" passes on every moq-transport draft every time. The subscriber never -leaves a request stream holding a partial message, so the publisher never -closes the session with "malformed request stream message" (ProtocolViolation, -code 3). - -## Plan - -Seen on 2026-10-22 on moqt-17 and moqt-22: about 4 failures in 120 runs on -main. The failure follows a logged `subscribe update priority=255`. - -Leads from reading the code (unverified): - -- `js/net/src/ietf/subscriber.ts:1098-1099` writes the message type - (`stream.writer.u53(SubscribeUpdate.id)`) and then the body in two awaits. - If the body's encode throws, the catch at `:1100-1103` returns silently with - the type already sent. Teardown then FINs the stream (`:970`), and the - publisher fails reading the size, which is exactly this error - (`js/net/src/ietf/request_stream.ts:124`, `:312-326`). -- The test sets no priority, and wire 255 means model priority -128 - (`js/net/src/ietf/priority.ts`). The idle aggregate is 0 (wire 127), so - either the aggregate is wrong mid-revocation or the log came from another - session. -- `first.close()` shrinks the grant union while `#runUpdates` may be writing - (`subscriber.ts:920-924`, `:961-973`). - -Reproduce it first: loop the test until it fails, capture which session logged -255 and where the write stopped. Fix it at the source. A message goes out -whole or not at all, and an encode failure fails loudly instead of being -swallowed. Then, if the 255 is real, correct the aggregate. Add a test with a -mocked clock that tears the stream mid-update. Check the Rust subscriber for -the same split write. - -The test's draft matrix runs only drafts 17 and 22, but -`SubscribeUpdate`'s encode has a separate draft 14-16 path (an extra request -ID and different fields). Add a representative draft 14-16 case, or every -supported IETF ALPN, to both the existing test and the new one, so the -legacy encoding is covered too. - -Public API: none. Wire: none. diff --git a/rs/moq-net/src/coding/writer.rs b/rs/moq-net/src/coding/writer.rs index 31a7f5d9304..983b54d5568 100644 --- a/rs/moq-net/src/coding/writer.rs +++ b/rs/moq-net/src/coding/writer.rs @@ -232,10 +232,23 @@ impl Writer { } impl Writer { - /// Encode an IETF `Message` to the stream, writing `[type_id][size][body]`. + /// Encode `msg` into the write buffer as `[id][size][body]`, to be sent by + /// [`Self::poll_flush`]. The type and body go in together or not at all, so a failed encode + /// never leaves a bare type for the next write to complete. `id` is explicit for the drafts + /// that send one body under several types. + pub fn buffer_message>(&mut self, id: u64, msg: &T) -> Result<(), Error> { + let version = self.version; + self.buffer_with(|w| { + w.varint(id)?; + msg.encode(w, version) + }) + } + + /// Encode an IETF `Message` to the stream as `[type_id][size][body]`, with the same + /// all-or-nothing buffering as [`Self::buffer_message`] and cancellation rules as [`Self::encode`]. pub async fn encode_message(&mut self, msg: &T) -> Result<(), Error> { - self.buffer_varint(T::ID)?; - self.encode(msg).await + self.buffer_message(T::ID, msg)?; + std::future::poll_fn(|cx| self.poll_flush(cx)).await } } @@ -346,8 +359,8 @@ mod tests { #[derive(Debug)] struct Poison; - impl Encode for Poison { - fn encode(&self, w: &mut Encoder<'_>, _: crate::lite::Version) -> Result<(), EncodeError> { + impl Encode for Poison { + fn encode(&self, w: &mut Encoder<'_>, _: V) -> Result<(), EncodeError> { w.slice(b"junk"); Err(EncodeError::BoundsExceeded) } @@ -367,6 +380,17 @@ mod tests { let mut cx = std::task::Context::from_waker(Waker::noop()); assert!(writer.poll_flush(&mut cx).is_ready()); assert_eq!(*log.writes.lock().unwrap(), vec![5, 7], "the partial encode leaked"); + + // A control message's type goes with its body: a bare type would make the next + // message, or a FIN, a torn one. + let mut writer = Writer::new(SinkSend::new(Log::default()), crate::ietf::Version::Draft17); + writer.buffer_varint(5).unwrap(); + writer.buffer_message(0x2, &Poison).unwrap_err(); + writer.buffer_varint(7).unwrap(); + + let log = writer.stream.as_ref().unwrap().log.clone(); + assert!(writer.poll_flush(&mut cx).is_ready()); + assert_eq!(*log.writes.lock().unwrap(), vec![5, 7], "the message type leaked"); } /// A flush interrupted mid-message resumes where it left off, so an diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 4e37dc9ecf7..32f4e754131 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1398,10 +1398,9 @@ where // Buffered whole before its first await, so however the subscription is // dropped from here, the request can still end on the wire. acked = Some(track_serve.opened.clone()); - stream.writer.buffer_varint(ietf::SubscribeOk::ID)?; stream .writer - .encode(&ietf::SubscribeOk { + .encode_message(&ietf::SubscribeOk { request_id: match self.version { Version::Draft14 | Version::Draft15 | Version::Draft16 => Some(request_id), _ => None, @@ -1538,10 +1537,9 @@ where Served::Paused => Ok(()), }; - let _ = stream.writer.varint(ietf::PublishDone::ID).await; let _ = stream .writer - .encode(&ietf::PublishDone { + .encode_message(&ietf::PublishDone { request_id: match self.version { Version::Draft14 | Version::Draft15 | Version::Draft16 => Some(request_id), _ => None, @@ -1877,9 +1875,8 @@ where request_id: RequestId, largest: Option, ) -> Result<(), Error> { - writer.buffer_varint(ietf::RequestOk::ID)?; writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest, request_id: matches!(self.version, Version::Draft15 | Version::Draft16).then_some(request_id), active: None, @@ -2686,10 +2683,9 @@ where // FETCH_OK on every draft, never REQUEST_OK: section 5.2 allows exactly one FETCH_OK or // REQUEST_ERROR in answer to a FETCH, and REQUEST_OK's own definition lists the other // requests it answers without ever naming this one. - stream.writer.varint(ietf::FetchOk::ID).await?; stream .writer - .encode(&ietf::FetchOk { + .encode_message(&ietf::FetchOk { request_id: match self.version { Version::Draft14 | Version::Draft15 | Version::Draft16 => Some(msg.request_id), _ => None, @@ -2963,8 +2959,10 @@ where largest: live_edge(&track).largest, properties, }; - stream.writer.varint(ietf::TrackStatusOk::id(self.version)).await?; - stream.writer.encode(&ok).await?; + stream + .writer + .buffer_message(ietf::TrackStatusOk::id(self.version), &ok)?; + std::future::poll_fn(|cx| stream.writer.poll_flush(cx)).await?; Ok::<(), Error>(()) }; let res = { @@ -3175,19 +3173,17 @@ where // A repeat under another epoch replaces the instance in place. if renamed && !replaced { tracing::debug!(broadcast = %absolute, "namespace_done for an epoch gained or lost"); - stream.writer.varint(ietf::NamespaceDone::ID).await?; stream .writer - .encode(&ietf::NamespaceDone { + .encode_message(&ietf::NamespaceDone { suffix: suffix.as_path(), }) .await?; } tracing::debug!(broadcast = %absolute, "namespace"); - stream.writer.varint(ietf::Namespace::ID).await?; stream .writer - .encode(&ietf::Namespace { + .encode_message(&ietf::Namespace { suffix: suffix.as_path(), params: negotiated.extended().then(|| advert.params()), }) @@ -3195,10 +3191,9 @@ where } (false, true) => { tracing::debug!(broadcast = %absolute, "namespace_done"); - stream.writer.varint(ietf::NamespaceDone::ID).await?; stream .writer - .encode(&ietf::NamespaceDone { + .encode_message(&ietf::NamespaceDone { suffix: suffix.as_path(), }) .await?; @@ -3250,19 +3245,15 @@ where }; let mut messages = super::request_stream::Messages::new(self.session.clone(), self.control.client()); - let sent = async { - request.writer.varint(ietf::PublishNamespace::ID).await?; - request - .writer - .encode(&ietf::PublishNamespace { - request_id, - track_namespace: path.as_path(), - params, - authorization_token: auth.as_ref().and_then(|auth| self.request_token(auth)), - }) - .await - } - .await; + let sent = request + .writer + .encode_message(&ietf::PublishNamespace { + request_id, + track_namespace: path.as_path(), + params, + authorization_token: auth.as_ref().and_then(|auth| self.request_token(auth)), + }) + .await; let answer = match sent { Ok(()) => self.read_response(&mut request, &mut messages).await, // A sequential-ID peer cannot accept any later request when this header failed @@ -3447,11 +3438,7 @@ where update.authorization_token = auth.as_ref().and_then(|auth| self.request_token(auth)); } tracing::debug!(broadcast = %self.origin.absolute(&request.path), "publish_namespace update"); - let written = async { - request.stream.writer.varint(ietf::PublishNamespaceUpdate::ID).await?; - request.stream.writer.encode(&update).await - } - .await; + let written = request.stream.writer.encode_message(&update).await; if let Err(err) = written { // The peer ended the request before the update reached it, which ends only this // advertisement: closing our side completes the withdrawal, and the retry @@ -3728,10 +3715,9 @@ where } } Target::Inline(stream) => { - stream.writer.varint(ietf::NamespaceDone::ID).await?; stream .writer - .encode(&ietf::NamespaceDone { + .encode_message(&ietf::NamespaceDone { suffix: suffix.as_path(), }) .await?; @@ -3913,19 +3899,17 @@ where // Send OK response match self.version { Version::Draft14 => { - stream.writer.varint(ietf::SubscribeNamespaceOk::ID).await?; stream .writer - .encode(&ietf::SubscribeNamespaceOk { + .encode_message(&ietf::SubscribeNamespaceOk { request_id: msg.request_id, }) .await?; } Version::Draft15 | Version::Draft16 => { - stream.writer.varint(ietf::RequestOk::ID).await?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest: None, request_id: Some(msg.request_id), active, @@ -3933,10 +3917,9 @@ where .await?; } _ => { - stream.writer.varint(ietf::RequestOk::ID).await?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest: None, request_id: None, active, @@ -4142,8 +4125,7 @@ where }); if let Some(error) = refusal { let stream = ns.target.stream().expect("update has a stream"); - stream.writer.varint(ietf::RequestError::ID).await?; - stream.writer.encode(&error).await?; + stream.writer.encode_message(&error).await?; // Dropping a finished writer resets it, which could discard the refusal. stream.writer.finish()?; break stream.writer.closed().await; @@ -4178,10 +4160,9 @@ where retry_delay = RETRY_BASE; } let stream = ns.target.stream().expect("update has a stream"); - stream.writer.varint(ietf::RequestOk::ID).await?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest: None, request_id: None, active: None, @@ -15341,14 +15322,17 @@ mod tests { for version in [Version::Draft14, Version::Draft15, Version::Draft19] { let origin = unheld_origin(); - let _cam = origin.announce("cam", crate::origin::Route::default()).unwrap(); + // A token too big for the header's u16 length, so the header fails after its stream + // opened and its request ID was spent. A message goes out whole, so this is the one + // failure a fresh in-memory stream can be made to show. + let auth = crate::auth::Handle::ietf(false); + let token = auth.present(bytes::Bytes::from(vec![0; 70_000]), false).unwrap(); + let _cam = origin + .announce("cam", crate::origin::Route::default().with_auth(token)) + .unwrap(); settle().await; - // One message at a time, so the header's body waits behind its type. - let (session, mut peer) = memory::pair(memory::Config { - stream_buffer: 0, - ..Default::default() - }); + let (session, mut peer) = memory::pair(memory::Config::default()); let publisher = Publisher::new( crate::time::sim(), session, @@ -15357,14 +15341,14 @@ mod tests { None, declared(Some(false)), version, - ); + ) + .with_auth(auth.clone()); let run = moq_clock::sim::spawn(publisher.run_publish_namespaces()); - let mut stream = moq_clock::timeout(Duration::from_secs(10), Stream::accept(&mut peer, version)) + moq_clock::timeout(Duration::from_secs(10), Stream::accept(&mut peer, version)) .await .expect("no advertisement") .unwrap(); - stream.reader.abort(&Error::Cancel); match version { Version::Draft14 | Version::Draft15 => { diff --git a/rs/moq-net/src/ietf/publisher/tracks.rs b/rs/moq-net/src/ietf/publisher/tracks.rs index e3a6a4144c3..165ee5792e1 100644 --- a/rs/moq-net/src/ietf/publisher/tracks.rs +++ b/rs/moq-net/src/ietf/publisher/tracks.rs @@ -264,10 +264,9 @@ where .discovery(!declared.hidden || self.origin.includes_hidden()); let mut announced = origin.announced(); - stream.writer.buffer_varint(ietf::RequestOk::ID)?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { request_id: None, active: None, largest: None, @@ -455,13 +454,15 @@ where ), }; let code = request::to_code(&err, request::Kind::SubscribeNamespace, self.version); - stream.writer.buffer_varint(ietf::RequestError::ID)?; - stream.writer.buffer(&ietf::RequestError { - request_id: None, - error_code: code, - reason_phrase: reason.into(), - retry_interval: 0, - })?; + stream.writer.buffer_message( + ietf::RequestError::ID, + &ietf::RequestError { + request_id: None, + error_code: code, + reason_phrase: reason.into(), + retry_interval: 0, + }, + )?; return Poll::Ready(Ok(())); } Ok(Some(Next::GoAway { redirect })) => goaways.receive(redirect)?, @@ -492,11 +493,13 @@ where break; }; drop(pending); - stream.writer.buffer_varint(ietf::PublishSkipped::ID)?; - stream.writer.buffer(&ietf::PublishSkipped { - suffix, - track_name: name.as_ref().into(), - })?; + stream.writer.buffer_message( + ietf::PublishSkipped::ID, + &ietf::PublishSkipped { + suffix, + track_name: name.as_ref().into(), + }, + )?; } Ok(()) } @@ -766,7 +769,6 @@ where self.pause_join(join.as_ref()); tracing::debug!(id = %request_id, broadcast = %absolute, track = %name, "publish started"); - stream.writer.buffer_varint(ietf::Publish::ID)?; let publish = ietf::Publish { request_id, track_namespace: namespace.clone(), @@ -782,7 +784,7 @@ where filter: Some(params.filter), }; let written = { - let mut writing = std::pin::pin!(stream.writer.encode(&publish)); + let mut writing = std::pin::pin!(stream.writer.encode_message(&publish)); kio::wait(|waiter| { if listed.poll_closed(waiter).is_ready() || live.poll_closed(waiter).is_ready() { return Poll::Ready(None); diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs index fbf2f3134b4..9d6627852df 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -103,17 +103,19 @@ pub(super) fn refuse_update( return Ok(()); } let request_id = super::RequestId::decode(&mut body.decoder(version), version).map_err(DecodeError::complete)?; - writer.buffer_varint(super::RequestError::ID)?; - writer.buffer(&super::RequestError { - request_id: Some(request_id), - error_code: super::error::request::to_code( - &Error::Unsupported, - super::error::request::Kind::Subscribe, - version, - ), - reason_phrase: "update not supported".into(), - retry_interval: 0, - }) + writer.buffer_message( + super::RequestError::ID, + &super::RequestError { + request_id: Some(request_id), + error_code: super::error::request::to_code( + &Error::Unsupported, + super::error::request::Kind::Subscribe, + version, + ), + reason_phrase: "update not supported".into(), + retry_interval: 0, + }, + ) } /// A message the adapter routed onto a draft 14-16 request after the request itself. @@ -292,8 +294,7 @@ pub(super) async fn reject( mut stream: Stream, error: super::RequestError<'_>, ) -> Result<(), Error> { - stream.writer.varint(super::RequestError::ID).await?; - stream.writer.encode(&error).await?; + stream.writer.encode_message(&error).await?; // A peer dropping the response after reading the refusal is a normal end. let _ = stream.writer.close().await; Ok(()) @@ -322,15 +323,16 @@ pub(super) async fn write_error( (Version::Draft14, Kind::TrackStatus) => super::TRACK_STATUS_ERROR_14, _ => super::RequestError::ID, }; - writer.buffer_varint(id)?; - writer - .encode(&super::RequestError { + writer.buffer_message( + id, + &super::RequestError { request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16).then_some(request_id), error_code: super::error::request::to_code(err, kind, version), reason_phrase: reason.into(), retry_interval: 0, - }) - .await + }, + )?; + std::future::poll_fn(|cx| writer.poll_flush(cx)).await } /// A framed subscription update, preserving omitted preferences. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 8eb9a27b13f..30981c7f387 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1192,8 +1192,7 @@ where subscribe_options: ietf::SubscribeOptions::Namespace, hidden, }; - stream.writer.varint(ietf::SubscribeNamespaceLegacy::ID).await?; - stream.writer.encode(&msg).await?; + stream.writer.encode_message(&msg).await?; } _ => { let msg = ietf::SubscribeNamespace { @@ -1201,8 +1200,7 @@ where namespace: prefix.clone(), hidden, }; - stream.writer.varint(ietf::SubscribeNamespace::ID).await?; - stream.writer.encode(&msg).await?; + stream.writer.encode_message(&msg).await?; } } @@ -2174,14 +2172,15 @@ where async fn write_ok(&self, stream: &mut Stream, request_id: RequestId) -> Result<(), Error> { match self.version { Version::Draft14 => { - stream.writer.buffer_varint(ietf::PublishNamespaceOk::ID)?; - stream.writer.encode(&ietf::PublishNamespaceOk { request_id }).await?; + stream + .writer + .encode_message(&ietf::PublishNamespaceOk { request_id }) + .await?; } Version::Draft15 | Version::Draft16 => { - stream.writer.buffer_varint(ietf::RequestOk::ID)?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest: None, request_id: Some(request_id), active: None, @@ -2189,10 +2188,9 @@ where .await?; } _ => { - stream.writer.buffer_varint(ietf::RequestOk::ID)?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest: None, request_id: None, active: None, @@ -2987,10 +2985,9 @@ where } None => { async { - stream.writer.varint(ietf::Subscribe::ID).await?; stream .writer - .encode(&ietf::Subscribe { + .encode_message(&ietf::Subscribe { max_age, new_group, request_id, @@ -3796,9 +3793,8 @@ where update: Update, ) -> Result<(), Error> { let request_id = self.control.next_request_id(&self.runtime).await?; - writer.varint(ietf::SubscribeUpdate::ID).await?; writer - .encode(&ietf::SubscribeUpdate { + .encode_message(&ietf::SubscribeUpdate { max_age: update.max_age, request_id, subscription_request_id: matches!(self.version, Version::Draft14 | Version::Draft15 | Version::Draft16) @@ -3864,8 +3860,7 @@ where writer: &mut crate::coding::Writer, request_id: RequestId, ) -> Result<(), Error> { - writer.varint(ietf::Unsubscribe::ID).await?; - writer.encode(&ietf::Unsubscribe { request_id }).await?; + writer.encode_message(&ietf::Unsubscribe { request_id }).await?; Ok(()) } @@ -3919,10 +3914,9 @@ where } } if let Err(err) = async { - stream.writer.varint(ietf::Fetch::ID).await?; stream .writer - .encode(&ietf::Fetch { + .encode_message(&ietf::Fetch { request_id: fetch_id, subscriber_priority: super::priority::to_wire( track.subscription().map(|s| s.priority).unwrap_or(0), @@ -4116,10 +4110,9 @@ where }; let written = async { - stream.writer.varint(ietf::TrackStatus::ID).await?; stream .writer - .encode(&ietf::TrackStatus { + .encode_message(&ietf::TrackStatus { request_id, track_namespace: broadcast_path.to_owned(), track_name: track_name.into(), @@ -5243,10 +5236,9 @@ where }, }; let res = async { - stream.writer.varint(ietf::Fetch::ID).await?; stream .writer - .encode(&ietf::Fetch { + .encode_message(&ietf::Fetch { request_id: fetch_id, subscriber_priority: super::priority::to_wire(request.priority()), group_order: GroupOrder::Ascending, @@ -5458,8 +5450,7 @@ where // The adapter has no transport reset, so deliver FETCH_CANCEL without a // subsequent writer drop discarding unacknowledged control bytes. let res = async { - stream.writer.varint(ietf::FetchCancel::ID).await?; - stream.writer.encode(&ietf::FetchCancel { request_id }).await?; + stream.writer.encode_message(&ietf::FetchCancel { request_id }).await?; stream.writer.close().await } .await; diff --git a/rs/moq-net/src/ietf/subscriber/tracks.rs b/rs/moq-net/src/ietf/subscriber/tracks.rs index 22a4417acac..91a0d6a809c 100644 --- a/rs/moq-net/src/ietf/subscriber/tracks.rs +++ b/rs/moq-net/src/ietf/subscriber/tracks.rs @@ -446,21 +446,23 @@ where let mut stream = stream?; // No draft with SUBSCRIBE_TRACKS has MAX_REQUEST_ID, so this never waits on credit. let request_id = kio::wait(|waiter| self.control.poll_next_request_id(waiter)).await; - stream.writer.buffer_varint(ietf::SubscribeTracks::ID)?; - stream.writer.buffer(&ietf::SubscribeTracks { - request_id, - prefix: prefix.as_path(), - forward: false, - subscriber_priority: None, - group_order: None, - filter: None, - fill: None, - properties_wanted: true, - new_group: None, - range_filters: false, - track_filter: false, - authorization_token: None, - })?; + stream.writer.buffer_message( + ietf::SubscribeTracks::ID, + &ietf::SubscribeTracks { + request_id, + prefix: prefix.as_path(), + forward: false, + subscriber_priority: None, + group_order: None, + filter: None, + fill: None, + properties_wanted: true, + new_group: None, + range_filters: false, + track_filter: false, + authorization_token: None, + }, + )?; tracing::debug!(prefix = %self.origin.absolute(prefix), id = %request_id, "subscribe_tracks started"); let mut messages = Messages::new(self.session.clone(), self.control.client()); @@ -652,10 +654,9 @@ where ); let mut messages = Messages::new(self.session.clone(), self.control.client()); let accepted = async { - stream.writer.varint(ietf::Subscribe::ID).await?; stream .writer - .encode(&ietf::Subscribe { + .encode_message(&ietf::Subscribe { max_age: None, new_group: None, request_id, @@ -874,18 +875,18 @@ where Version::Draft18 => ietf::PublishOk::ID, _ => ietf::RequestOk::ID, }; - stream.writer.varint(id).await?; - stream - .writer - .encode(&ietf::PublishOk { + stream.writer.buffer_message( + id, + &ietf::PublishOk { request_id: None, forward: false, subscriber_priority: None, group_order: None, filter: None, new_group: None, - }) - .await + }, + )?; + std::future::poll_fn(|cx| stream.writer.poll_flush(cx)).await } .await; if let Err(err) = ok { @@ -1157,9 +1158,8 @@ where forward: &Forward, ) -> Result<(), Error> { let request_id = self.control.next_request_id(&self.runtime).await?; - writer.varint(ietf::SubscribeUpdate::ID).await?; writer - .encode(&ietf::SubscribeUpdate { + .encode_message(&ietf::SubscribeUpdate { max_age: forward.max_age, request_id, subscription_request_id: None, From 8a71efa4a582546096c60201772583a53df5f760 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 17:23:09 -0700 Subject: [PATCH 3/3] fix(net): a legacy update that fails to encode ends the session Drafts 14 and 15 require the next request ID, so an update that spent one and never sent it would stall every later request. As in Rust's legacy header path, that now closes the session with an internal error. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/subscriber.ts | 18 ++++++++++++------ js/net/src/subscribe_priority.test.ts | 8 +++++++- 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 2a3304004c1..e0f1d73ef2a 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -13,6 +13,7 @@ import { error, ProtocolViolation, reason, + SessionCode, StreamCode, Stream as StreamError, sessionCause, @@ -466,11 +467,11 @@ export class Subscriber { } } - // The peer broke the protocol: close with the violation's code before the session's own - // close, which would send a clean one, as the connection does for its own violations. - #violated(err: ProtocolViolation) { + // Close with this code before the session's own close, which would send a clean one, as + // the connection does for its own violations. + #fail(code: SessionCode, reason: string) { try { - this.#quic?.close({ closeCode: err.code, reason: err.message }); + this.#quic?.close({ closeCode: code, reason }); } catch { // Already closed. } @@ -1102,6 +1103,11 @@ export class Subscriber { // Nothing reached the wire, but the track can no longer follow its subscribers. console.warn(`subscribe update failed: id=${requestId} error=${reason(error(err))}`); producer.close(error(err)); + // Drafts 14 and 15 require the next request ID, so one spent and never sent would + // stall every later request: as in Rust, that ends the session. + if (version === Version.DRAFT_14 || version === Version.DRAFT_15) { + this.#fail(SessionCode.Internal, "subscribe update failed to encode"); + } return; } // Counted before writing: the answer can be read before the write's promise resolves. @@ -1515,7 +1521,7 @@ export class Subscriber { // The control message establishing this alias can arrive after the data stream. subscription = await this.#aliases.get(group.trackAlias); } catch (err: unknown) { - if (err instanceof ProtocolViolation) this.#violated(err); + if (err instanceof ProtocolViolation) this.#fail(err.code, err.message); const e = await sessionCause(this.#quic, err); // Ours: we cancelled the subscription and the publisher has not stopped yet. // Anything else on this alias is the publisher sending data for a track it never @@ -1613,7 +1619,7 @@ export class Subscriber { // A group with no objects still exists. open().close(); } catch (err: unknown) { - if (err instanceof ProtocolViolation) this.#violated(err); + if (err instanceof ProtocolViolation) this.#fail(err.code, err.message); const e = await sessionCause(this.#quic, err); // The compact form names the subgroup by its first Object ID, so a nonzero one is // subgroup N: stop the stream before any group exists for it. diff --git a/js/net/src/subscribe_priority.test.ts b/js/net/src/subscribe_priority.test.ts index b082c04aef0..768d76c0178 100644 --- a/js/net/src/subscribe_priority.test.ts +++ b/js/net/src/subscribe_priority.test.ts @@ -5,6 +5,7 @@ import { expect, spyOn, test } from "bun:test"; import { Manual } from "@moq/clock"; import { accept, connect, type Established } from "./connection/index.ts"; +import { SessionCode, type Session as SessionError } from "./error.ts"; import * as Ietf from "./ietf/index.ts"; import { SubscribeUpdate } from "./ietf/subscribe.ts"; import * as Lite from "./lite/index.ts"; @@ -167,6 +168,7 @@ test.each(Object.values(Ietf.ALPN))("%s sends no update when the last subscriber // An update that fails to encode is a bug, not a publisher that stopped reading, so it ends the // track with the error rather than leaving it at a stale priority. Nothing reaches the wire. +// Drafts 14 and 15 require the next request ID, so the one the update spent ends the session. test.each(Object.values(Ietf.ALPN))("%s ends the track when an update fails to encode", async (protocol) => { const origin = new OriginProducer(); const broadcast = origin.createBroadcast(path); @@ -191,7 +193,11 @@ test.each(Object.values(Ietf.ALPN))("%s ends the track when an update fails to e expect(video.subscription.peek()?.priority).toBe(LOW); await video.subscription.changed(); } - expect(closed).toBeUndefined(); + if (protocol === Ietf.ALPN.DRAFT_14 || protocol === Ietf.ALPN.DRAFT_15) { + expect(((await server.closed) as SessionError | null)?.code).toBe(SessionCode.Internal); + } else { + expect(closed).toBeUndefined(); + } remote.close(); } finally { encode.mockRestore();