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 464ec672e20..49a24192c9f 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. } @@ -1062,10 +1063,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( @@ -1093,10 +1096,24 @@ 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)); + // 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. 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; @@ -1504,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 @@ -1602,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 3b160162b01..51c6d2617da 100644 --- a/js/net/src/subscribe_priority.test.ts +++ b/js/net/src/subscribe_priority.test.ts @@ -2,15 +2,20 @@ // 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 { 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"; 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 +94,115 @@ 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 installed = 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(); + installed.close(); + } +}); + +// 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. +// 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); + 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(); + } + 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(); + 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 07510eb6942..fd6ab16c739 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 - [Media resource instrumentation](/quest/m1/test-flakes-2/resource-instrumentation.md) - the media test stops seeing a playing player report nothing to release - [Tone check grace](/quest/m1/test-flakes-2/tone-grace.md) - the interop tone check charges its grace only for audible playback, so a refilling ring under CPU starvation is not a failure - [Relay tests in one process](/quest/m1/test-flakes-2/relay-one-process.md) - moq-relay's integration binary passes under plain `cargo test`, not only nextest 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 c852833560e..3ef57596fbf 100644 --- a/rs/moq-net/src/coding/writer.rs +++ b/rs/moq-net/src/coding/writer.rs @@ -243,10 +243,27 @@ 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 start = self.buffer.len(); + self.buffer_varint(id)?; + // The body is buffered on its own so control stats count it without its type selector. + let version = self.version; + if let Err(err) = self.buffer_with(|w| msg.encode(w, version)) { + self.buffer.truncate(start); + return Err(err); + } + Ok(()) + } + + /// 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 } } @@ -357,8 +374,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.control = Some(crate::stats::Control::Probe); w.slice(b"junk"); Err(EncodeError::BoundsExceeded) @@ -379,6 +396,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"); } #[test] diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 0c35fca8791..e75db13498c 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -2217,8 +2217,7 @@ mod tests { request_id: RequestId(0), }; let size = msg.encode_bytes(version).unwrap().len() as u64; - virtual_stream.writer.buffer_varint(ietf::FetchCancel::ID).unwrap(); - virtual_stream.writer.encode(&msg).await.unwrap(); + virtual_stream.writer.encode_message(&msg).await.unwrap(); let raw = adapter.shared.control.pop().await.unwrap(); let log = Log::default(); let mut physical = Writer::new( diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 4996a1defb2..775baddce8c 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1356,10 +1356,9 @@ impl Publisher { // 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, @@ -1585,10 +1584,9 @@ impl Publisher { 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, @@ -2091,9 +2089,8 @@ impl Publisher { 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, @@ -2890,10 +2887,9 @@ impl Publisher { // 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, @@ -3176,8 +3172,10 @@ impl Publisher { 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 = { @@ -3388,19 +3386,17 @@ impl Publisher { // 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()), }) @@ -3408,10 +3404,9 @@ impl Publisher { } (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?; @@ -3463,19 +3458,15 @@ impl Publisher { }; 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 @@ -3660,11 +3651,7 @@ impl Publisher { 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 @@ -3951,10 +3938,9 @@ impl Publisher { } } Target::Inline(stream) => { - stream.writer.varint(ietf::NamespaceDone::ID).await?; stream .writer - .encode(&ietf::NamespaceDone { + .encode_message(&ietf::NamespaceDone { suffix: suffix.as_path(), }) .await?; @@ -4137,19 +4123,17 @@ impl Publisher { // 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, @@ -4157,10 +4141,9 @@ impl Publisher { .await?; } _ => { - stream.writer.varint(ietf::RequestOk::ID).await?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest: None, request_id: None, active, @@ -4366,8 +4349,7 @@ impl Publisher { }); 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; @@ -4399,10 +4381,9 @@ impl Publisher { 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, @@ -13301,19 +13282,26 @@ mod serve_tests { log.writes.lock().unwrap().clone() }; let request = |sent: &crate::lite::test_transport::Sent| h.log.trail().contains(&(0, sent.clone())); + // PUBLISH_DONE goes out whole, so its write leads with the type. + let sent_done = || { + h.log.trail().iter().any(|(stream, sent)| { + *stream == 0 + && matches!(sent, crate::lite::test_transport::Sent::Write(bytes) if bytes.starts_with(&done)) + }) + }; for _ in 0..50 { let _ = futures::poll!(acceptor.as_mut()); assert!( futures::poll!(serving.as_mut()).is_pending(), "the request let go of its stream while the acceptor held the renewal" ); - if request(&crate::lite::test_transport::Sent::Write(done.clone())) { + if sent_done() { break; } settle().await; } assert!( - request(&crate::lite::test_transport::Sent::Write(done.clone())), + sent_done(), "serving must end at the old deadline, not stall on the slow renewal" ); assert!( @@ -16032,14 +16020,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(), crate::driver::Transport::new(session.clone()), @@ -16048,17 +16039,17 @@ 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( + moq_clock::timeout( Duration::from_secs(10), Stream::accept(&mut crate::driver::Transport::new(peer.clone()), 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 155ed3f5d7e..5803cec28d4 100644 --- a/rs/moq-net/src/ietf/publisher/tracks.rs +++ b/rs/moq-net/src/ietf/publisher/tracks.rs @@ -262,10 +262,9 @@ impl Publisher { .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, @@ -453,13 +452,15 @@ impl Publisher { ), }; 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)?, @@ -490,11 +491,13 @@ impl Publisher { 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(()) } @@ -775,7 +778,6 @@ impl Publisher { 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(), @@ -791,7 +793,7 @@ impl Publisher { 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 bc87bcbedbc..ae9ee8a0be1 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -119,17 +119,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. @@ -317,8 +319,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(()) @@ -347,15 +348,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 7b8d2e2bd2b..d6561adaf95 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1205,8 +1205,7 @@ impl Subscriber { 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 { @@ -1214,8 +1213,7 @@ impl Subscriber { namespace: prefix.clone(), hidden, }; - stream.writer.varint(ietf::SubscribeNamespace::ID).await?; - stream.writer.encode(&msg).await?; + stream.writer.encode_message(&msg).await?; } } @@ -2193,14 +2191,15 @@ impl Subscriber { 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, @@ -2208,10 +2207,9 @@ impl Subscriber { .await?; } _ => { - stream.writer.buffer_varint(ietf::RequestOk::ID)?; stream .writer - .encode(&ietf::RequestOk { + .encode_message(&ietf::RequestOk { largest: None, request_id: None, active: None, @@ -3006,10 +3004,9 @@ impl Subscriber { } None => { async { - stream.writer.varint(ietf::Subscribe::ID).await?; stream .writer - .encode(&ietf::Subscribe { + .encode_message(&ietf::Subscribe { max_age, new_group, request_id, @@ -3799,9 +3796,8 @@ impl Subscriber { 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) @@ -3867,8 +3863,7 @@ impl Subscriber { 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(()) } @@ -3925,10 +3920,9 @@ impl Subscriber { } } 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), @@ -4134,10 +4128,9 @@ impl Subscriber { }; 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(), @@ -5265,10 +5258,9 @@ impl Subscriber { }, }; 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, @@ -5480,8 +5472,7 @@ impl Subscriber { // 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 cf755fd8a5e..d7445bb242c 100644 --- a/rs/moq-net/src/ietf/subscriber/tracks.rs +++ b/rs/moq-net/src/ietf/subscriber/tracks.rs @@ -429,21 +429,23 @@ impl Subscriber { let mut stream = request_stream::with_stats(stream?, self.origin.stats(), self.version); // 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()); @@ -631,10 +633,9 @@ impl Subscriber { ); 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, @@ -853,18 +854,18 @@ impl Subscriber { 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 { @@ -1136,9 +1137,8 @@ impl Subscriber { 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,