From 10dd1d7556fcbdd963b24ff198fdf6615fafdf0a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 20:37:52 -0700 Subject: [PATCH 01/11] fix(net): give an anonymous dial its own hop (#5025) (cherry picked from commit 2c0cf672829ffee492248a7b14264f6d6593ab65) Co-authored-by: Grok 4.7 Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/client.rs | 76 +++++---- rs/moq-net/src/ietf/publisher.rs | 8 +- rs/moq-net/src/ietf/subscriber.rs | 8 +- rs/moq-net/src/lite/publisher.rs | 7 +- rs/moq-net/src/lite/subscriber.rs | 12 +- rs/moq-net/tests/dial_split_horizon.rs | 205 +++++++++++++++++++++++++ rs/moq-tokio/src/client.rs | 4 +- 7 files changed, 265 insertions(+), 55 deletions(-) create mode 100644 rs/moq-net/tests/dial_split_horizon.rs diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 7574b768ca..e4a6497ca5 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -108,41 +108,47 @@ impl Client { self } - /// Assign an origin (hop) id to the peer, used whenever the peer doesn't declare - /// one itself. + /// Assign the identity this peer's routes are attributed to, overriding the fresh + /// per-dial default. /// - /// Some relays never declare their identity: moq-lite peers without the hops - /// extension, and moq-transport peers that don't negotiate the MoQ Cluster - /// extension (or predate it, on `moqt-16` and earlier). - /// Broadcasts received from such a peer are normally attributed to the reserved - /// Hop ID 0 ("unknown"), which identifies nothing: it never proves continuity, - /// so their advertisements neither splice nor survive a restart in place. This - /// knob pins a real identity instead, exactly as if the peer had declared it: + /// A dialed session whose peer declares no hop gets a random one for that + /// connection, the same way an accepted session does. That keeps a route + /// learned from the peer off the session that learned it, and keeps a + /// SUBSCRIBE that arrives on it from being routed back to it. An identity the + /// peer declares on the wire still wins. /// - /// - broadcasts received from the peer carry `origin` in their hop chains, so - /// every session dialing the same relay (with the same id) resolves to one - /// route and loop checks can recognize it; - /// - broadcasts whose hop chain already contains `origin` are neither announced - /// nor served back to the peer, preventing an echo through a relay that does - /// no loop detection of its own. - /// - /// An identity the peer does declare wins over this one. + /// Pass an id only for a peer whose identity the caller has actually + /// established. Two dials given the same hop are treated as one endpoint: + /// routes learned from either are kept off both, and content arriving on + /// either is interchangeable with the other's. That is the point when they + /// really are one peer reconnecting or running redundant links, and a bug + /// otherwise. Derive it from the authenticated identity, never from something + /// coarser like the remote address. pub fn with_peer_hop(mut self, hop: crate::Hop) -> Self { self.peer_hop = Some(hop); self } + /// The hop this connection attributes the peer to when the peer declares none. + /// + /// A caller-supplied id is stable across dials of this client. Otherwise each + /// connection gets a fresh one, matching the per-session default an accepted + /// session already has. + fn assigned_hop(&self) -> crate::Hop { + self.peer_hop.unwrap_or_else(crate::Hop::random) + } + /// The origin pair a session attaches, tagged and filtered. /// /// Reads through the publish (egress) consumer and writes through the /// subscribe (ingress) producer are attributed by the model through the /// stats context; one shared context, so presence and viewer counts are - /// never double-attributed across the two halves. An assigned peer identity - /// means subscriptions from the peer resolve to a source whose hop chain - /// excludes it, the same split-horizon rule applied when a peer declares - /// its own id; announce filtering is per-protocol and handled inside each - /// publisher. - fn origins(&self) -> (Option, Option) { + /// never double-attributed across the two halves. `peer_hop` is the identity + /// this connection attributes the peer to when the peer declares none, so + /// subscriptions from the peer resolve to a source whose route excludes it. + /// A declared identity still wins inside each publisher; announce filtering + /// is per-protocol and handled there too. + fn origins(&self, peer_hop: crate::Hop) -> (Option, Option) { if self.publish.is_none() && self.subscribe.is_none() { tracing::warn!("not publishing or consuming anything"); } @@ -151,7 +157,7 @@ impl Client { .subscribe .clone() .map(|origin| origin.with_stats(self.stats.clone())); - let publish = publish.map(|origin| origin.excluding(self.peer_hop.unwrap_or(crate::Hop::UNKNOWN))); + let publish = publish.map(|origin| origin.excluding(peer_hop)); (publish, subscribe) } @@ -162,11 +168,12 @@ impl Client { runtime: Clock, session: S, version: lite::Version, + peer_hop: crate::Hop, ) -> Result<(Session, crate::Driver), Error> where S: crate::transport::poll::Session, { - let (publish, subscribe) = self.origins(); + let (publish, subscribe) = self.origins(peer_hop); // Advertise our capabilities (we report what the transport measures; we // don't pad) plus the request path on URI-less transports, and the @@ -192,7 +199,7 @@ impl Client { setup_stream: None, publish, subscribe, - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), version, our_setup, peer_setup: None, @@ -231,7 +238,7 @@ impl Client { _ => return Err(Error::Version), }; self.versions.select(Version::Lite(version)).ok_or(Error::Version)?; - self.start_lite(runtime, session, version) + self.start_lite(runtime, session, version, self.assigned_hop()) } /// Perform the MoQ handshake, returning the [`Session`] and its [`Driver`](crate::Driver). @@ -242,7 +249,8 @@ impl Client { S: crate::transport::poll::Boxable, { let runtime = Clock::new(now); - let (publish, subscribe) = self.origins(); + let peer_hop = self.assigned_hop(); + let (publish, subscribe) = self.origins(peer_hop); // If ALPN was used to negotiate the version, use the appropriate encoding. // Default to IETF 14 if no ALPN was used and we'll negotiate the version later. @@ -269,7 +277,7 @@ impl Client { client: true, publish: publish.clone(), subscribe: subscribe.clone(), - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), cost: self.cost, version: draft, path: self.setup_path.clone(), @@ -317,19 +325,19 @@ impl Client { _ => lite::Version::Lite05, }; self.versions.select(Version::Lite(version)).ok_or(Error::Version)?; - return self.start_lite(runtime, session, version); + return self.start_lite(runtime, session, version, peer_hop); } Some(ALPN_LITE_04) => { self.versions .select(Version::Lite(lite::Version::Lite04)) .ok_or(Error::Version)?; - return self.start_lite(runtime, session, lite::Version::Lite04); + return self.start_lite(runtime, session, lite::Version::Lite04, peer_hop); } Some(ALPN_LITE_03) => { self.versions .select(Version::Lite(lite::Version::Lite03)) .ok_or(Error::Version)?; - return self.start_lite(runtime, session, lite::Version::Lite03); + return self.start_lite(runtime, session, lite::Version::Lite03, peer_hop); } Some(ALPN_LITE) | None => { let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?; @@ -381,7 +389,7 @@ impl Client { setup_stream: Some(stream), publish: publish.clone(), subscribe: subscribe.clone(), - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), version: v, // This path only handles versions negotiated via the bidi SETUP exchange // (pre-lite-05), which have no Setup Stream. @@ -418,7 +426,7 @@ impl Client { client: true, publish: publish.clone(), subscribe: subscribe.clone(), - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), cost: self.cost, version: v, path: None, diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index d42749c8ea..3659fbe8e3 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -313,8 +313,8 @@ pub(super) struct Publisher { // origin we consume so it matches the local relay identity across every session, // which is what makes cross-session loop detection work. self_origin: crate::Hop, - // The identity assigned to the peer by the caller (`Client::with_peer_hop`, or - // the per-session default a server hands every request), used when the peer declares + // The identity assigned to the peer (a fresh per-session id, or one the caller + // pinned with `with_peer_hop`), used when the peer declares // none itself. A peer that negotiates the MoQ Cluster extension declares its own, // which wins unless it withheld it as the reserved 0. peer_hop: Option, @@ -420,8 +420,8 @@ where /// The Hop ID whose paths must not be advertised (or served) back to this peer. /// /// A peer that declared an identity supplies its own; otherwise fall back to the one - /// we assigned it (`Client::with_peer_hop` when dialing, `Request::with_peer_hop` - /// or a fresh per-session id when accepting), since moq-transport carries no identity + /// we assigned it (a fresh per-session id, or one pinned with `Client::with_peer_hop` + /// or `Request::with_peer_hop`), since moq-transport carries no identity /// of its own. A peer that declared the reserved 0 declared no identity, so it takes /// the fallback like any other anonymous peer. fn exclude(&self, peer: &cluster::Peer) -> crate::Hop { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1a3e128b7b..78d04114ad 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -467,12 +467,12 @@ pub(super) struct Subscriber { // The origin naming this link for split-horizon (`Route.via`) when the peer // declares none of its own (see `session_route`). Base moq-transport carries no // hop ids, so a peer only has an identity if it negotiated the MoQ Cluster - // extension or the caller assigned it one (`Client::with_peer_hop`). + // extension; otherwise this is the one the session assigned it (a fresh id per + // dialed or accepted session, unless the caller pinned one with `with_peer_hop`), + // or `Hop::UNKNOWN` (0) when none was assigned. // - // Otherwise this is `Hop::UNKNOWN` (0), the reserved "no identity" value. // The assigned id stays local: it is never written into a hop chain, so a peer - // that withheld an identity is not named on the wire. A server answers it per - // accepted session; a client only when it knows the peer. + // that withheld an identity is not named on the wire. session_origin: crate::Hop, // A random Hop ID of this connection's own, written as the first hop of any path // that arrives naming no publisher, so a publisher that reconnects reads downstream diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index abc7f01875..4ee625a571 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -51,8 +51,8 @@ struct Shared { // peer from a source whose chain excludes them, keeping the data plane on // the same split-horizon rule as the announces we send them. peer_setup: super::PeerSetup, - // The identity assigned to the peer by the caller (`Client::with_peer_hop`, or - // the per-session default a server hands every request), standing in wherever the + // The identity assigned to the peer (a fresh per-session id, or one the caller + // pinned with `with_peer_hop`), standing in wherever the // peer declines to declare one. Backs both the announce filter and the serving // origin, so a peer that names itself nowhere on the wire is still split-horizoned. peer_hop: Option, @@ -528,8 +528,7 @@ impl AnnounceServe { // The identity whose routes we filter out. Lite-04/05 carry it per // announce stream; lite-06+ reads the session-wide SETUP Hop // parameter, the same identity the subscribe path excludes. A peer that - // declares nothing falls back to the identity the caller assigned it - // (`with_peer_hop`), if any. + // declares nothing falls back to the identity the session assigned it. let assigned = self.shared.peer_hop.map(|origin| origin.id()).unwrap_or(0); if self.shared.version.has_exclude_hop() { let exclude_hop = match interest.exclude_hop { diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 92b01d1423..2d49255bae 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -61,13 +61,11 @@ pub(super) struct Subscriber { // carry real hop ids on the wire (Lite01/02/03), and for a peer that reports // 0 in AnnounceOk. Lite03 placeholders stay 0 and count as anonymous. // - // This is the peer's assigned identity (`peer_hop`) when the caller gave - // it one. Otherwise it is `Hop::UNKNOWN` (0), the reserved "no identity" value. - // - // Assigning one is the caller's call, not this layer's: a server gives every - // accepted session a fresh id so its routes are at least distinguishable from - // another session's, while a client only assigns one it knows out of band. The - // assigned id stays local and is never written into a hop chain. + // This is the peer's assigned identity (`peer_hop`): a fresh id per dialed or + // accepted session, so its routes are distinguishable from another session's, + // unless the caller pinned a stable one with `with_peer_hop`. Without one it is + // `Hop::UNKNOWN` (0), the reserved "no identity" value. The assigned id stays + // local and is never written into a hop chain. session_origin: crate::Hop, // A random Hop ID of this connection's own, written as the first hop of any chain // that names no publisher (Lite01/02, a Lite03 placeholder, or a peer reporting 0), diff --git a/rs/moq-net/tests/dial_split_horizon.rs b/rs/moq-net/tests/dial_split_horizon.rs new file mode 100644 index 0000000000..402d01ba98 --- /dev/null +++ b/rs/moq-net/tests/dial_split_horizon.rs @@ -0,0 +1,205 @@ +//! A dialed moq-transport session whose peer declares no hop must not be offered +//! a route learned from that peer, and a SUBSCRIBE that arrives for that route +//! must not be sent back to the peer that announced it. +//! +//! Draft 16 has no cluster extension, so neither side declares a hop. That is the +//! anonymous dial: the accepting side already stamps one, and the dialing side +//! has to as well. + +mod support; + +use std::time::Duration; + +use moq_net::{Client, Hop, Server, Session, Version, origin, stats}; +use support::{ + harness::{now, spawn}, + mock::create_mock_session_pair, +}; + +/// Long enough, in virtual time, for an announce or subscribe to cross the mock link. +const SETTLE: Duration = Duration::from_secs(1); + +fn version() -> Version { + "moq-transport-16".parse().unwrap() +} + +fn produce(hop: u64) -> origin::Producer { + let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + spawn(driver); + producer +} + +/// One dial from `dialer` to `edge`, held open for the test. +struct Link { + _client: Session, + _server: Session, + /// Stats on the accepted session, so an announce or subscribe that arrives + /// there is visible without reading the wire. + registry: stats::Registry, +} + +async fn dial(dialer: &origin::Producer, edge: &origin::Producer) -> Link { + let version = version(); + let registry = stats::Registry::new(stats::Config::new()); + let session_stats = registry.tier(stats::Tier::default()).session(""); + let (client_transport, server_transport) = create_mock_session_pair(Some(version.alpn())); + + // Cluster dials publish and subscribe through a peer handle, which is what + // records the announcing session on the route. + let dialer_origin = dialer.clone().peer(); + let edge_origin = edge.clone().peer(); + let client = Client::new() + .with_versions(version.into()) + .with_publisher(dialer_origin.consume().with_hidden(true)) + .with_subscriber(dialer_origin); + let server = Server::new() + .with_versions(version.into()) + .with_stats(session_stats) + .with_publisher(edge_origin.consume().with_hidden(true)) + .with_subscriber(edge_origin); + + let client = async { + let (session, driver) = client + .connect(now(), client_transport) + .await + .expect("client handshake failed"); + spawn(driver); + session + }; + let server = async { + let (session, driver) = server + .accept(now(), server_transport) + .await + .expect("server handshake failed"); + spawn(driver); + session + }; + let (client, server) = futures::join!(client, server); + Link { + _client: client, + _server: server, + registry, + } +} + +fn traffic(registry: &stats::Registry, role: stats::Role) -> stats::Traffic { + registry + .snapshot() + .traffic() + .into_iter() + .find(|(_, row, _)| *row == role) + .map(|(_, _, traffic)| traffic) + .unwrap_or_default() +} + +/// The hop the dialer attributed `path` to, once that announce has arrived. +async fn learned_hop(dialer: &origin::Producer, path: &str) -> Hop { + let mut announced = dialer.consume().announced(); + let event = moq_net_sim::timeout(SETTLE, async { + loop { + let event = announced.next().await.expect("announce cursor closed"); + let announce = match event { + moq_net::announce::Event::Start(announce) | moq_net::announce::Event::Update(announce) => announce, + moq_net::announce::Event::End(_) => continue, + }; + if announce.prefix.as_str() == path { + return announce; + } + } + }) + .await + .unwrap_or_else(|_| panic!("{path}: the dialer never learned the namespace")); + + match event.route.source() { + origin::Source::Peer(hop) => hop, + origin::Source::Local => panic!("{path}: a route learned from the peer looks local"), + } +} + +/// Draft-16 edges dialed by one origin each get their own hop. The edge that +/// announced a namespace is not told about it, and the other edge is. A +/// subscribe from that other edge is fetched from the announcer and is not +/// sent back to the subscriber. +#[moq_net_sim::test] +async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { + let dialer = produce(1); + let publisher = produce(2); + let subscriber = produce(3); + let to_publisher = dial(&dialer, &publisher).await; + let to_subscriber = dial(&dialer, &subscriber).await; + + let broadcast = publisher.create_broadcast("room").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(origin::Route::default()).unwrap(); + + let publisher_hop = learned_hop(&dialer, "room").await; + assert_ne!( + publisher_hop, + Hop::UNKNOWN, + "an anonymous dial still needs a hop, or its routes echo" + ); + + // The other edge learns the namespace, attributed to a different hop, so the + // two dials are not one endpoint. + let mut subscriber_announced = subscriber.consume().announced(); + let subscriber_saw = moq_net_sim::timeout(SETTLE, subscriber_announced.next()) + .await + .expect("the other edge never learned the namespace") + .expect("announce cursor closed"); + let moq_net::announce::Event::Start(subscriber_saw) = subscriber_saw else { + panic!("the other edge's first event was not the namespace: {subscriber_saw:?}"); + }; + assert_eq!(subscriber_saw.prefix.as_str(), "room"); + let subscriber_hop = match subscriber_saw.route.source() { + origin::Source::Peer(hop) => hop, + origin::Source::Local => panic!("the forwarded route looks local"), + }; + assert_ne!(subscriber_hop, publisher_hop, "each dial gets its own hop"); + assert_ne!(subscriber_hop, Hop::UNKNOWN); + + moq_net_sim::sleep(SETTLE).await; + + let echoed = traffic(&to_publisher.registry, stats::Role::Subscriber); + assert_eq!( + echoed.announces_started, 0, + "the announcing edge was offered its own namespace back" + ); + let forwarded = traffic(&to_subscriber.registry, stats::Role::Subscriber); + assert!( + forwarded.announces_started >= 1, + "the other edge was not offered the namespace" + ); + + // The subscribe arrives on the dialer's session with the other edge. It has + // to be fetched from the announcer, and not turned into a subscribe on the + // session it arrived on. + let viewed = subscriber + .consume() + .request_broadcast("room") + .await + .expect("subscriber resolves the namespace"); + let subscription = moq_net::track::Subscription::default(); + let _reader = viewed + .track("video") + .unwrap() + .subscribe(subscription) + .await + .expect("subscriber subscribes"); + moq_net_sim::timeout(SETTLE, track.demand().used()) + .await + .expect("the subscribe never reached the announcer") + .unwrap(); + + moq_net_sim::sleep(SETTLE).await; + + let from_announcer = traffic(&to_publisher.registry, stats::Role::Publisher); + assert_eq!( + from_announcer.subscriptions_started, 1, + "the subscribe was not routed to the edge that announced it" + ); + let back_to_subscriber = traffic(&to_subscriber.registry, stats::Role::Publisher); + assert_eq!( + back_to_subscriber.subscriptions_started, 0, + "the subscribe was routed back to the edge that sent it" + ); +} diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index d7c517d258..8b97eaf778 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -220,8 +220,8 @@ impl Client { self } - /// Assign an origin (hop) id to the peers this client dials, used whenever a - /// peer doesn't declare one itself; see [`moq_net::Client::with_peer_hop`]. + /// Pin the identity a dialed peer's routes are attributed to, for a peer whose + /// identity the caller has established; see [`moq_net::Client::with_peer_hop`]. pub fn with_peer_hop(mut self, hop: moq_net::Hop) -> Self { self.moq = self.moq.with_peer_hop(hop); self From a97a13a3a0296402457343d473be4acf999a7c37 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 10:34:58 -0700 Subject: [PATCH 02/11] fix(rtmp): reassemble interleaved chunk streams independently (#4735) (cherry picked from commit b0eca5022fa451881b7a7bb02e3f62d1bb957ac8) Backport drops the chunk_interleaving benchmark, its bench-rtmp recipe, and the nightly smoke step: nightly only runs on main, which keeps them. Co-Authored-By: Claude Opus 5.5 --- doc/bin/rtmp.md | 20 + .../rml/chunk_io/deserialization_errors.rs | 20 + rs/moq-rtmp/src/rml/chunk_io/deserializer.rs | 491 ++++++++++++++++-- rs/moq-rtmp/src/rml/mod.rs | 1 + rs/moq-rtmp/src/rml/sessions/client/mod.rs | 5 + rs/moq-rtmp/src/rml/sessions/server/mod.rs | 3 +- 6 files changed, 500 insertions(+), 40 deletions(-) diff --git a/doc/bin/rtmp.md b/doc/bin/rtmp.md index 2c20774728..affd226a03 100644 --- a/doc/bin/rtmp.md +++ b/doc/bin/rtmp.md @@ -41,3 +41,23 @@ can arrive unencrypted. The embedder can set `plaintext` to `false` to serve is logged with the peer address. Refusing plaintext without a TLS config fails at startup. FLAC and MP3 enhanced-audio payloads are dropped because hang has no catalog codec for them. + +Incoming chunk streams are reassembled independently, including interleaved +control, audio, and video messages. Each connection retains at most 256 chunk +stream IDs, including completed streams' header history needed by compressed +headers. Reusing an ID does not consume another slot; the connection releases +all slots on teardown. A 257th distinct ID is refused, even if its numeric ID is +small. The limit counts streams rather than constraining their numeric IDs. + +Incomplete messages reserve their declared payload lengths against a 64 MiB +connection budget before allocation. Completion or an RTMP Abort releases that reservation; +malformed input or exhaustion closes the connection and releases its parser +state. The 24-bit RTMP message length still allows a payload up to 16 MiB minus +one byte. Four maximal payloads fit, with exactly four bytes remaining. + +These internal limits leave room for the gateway's five chunk-stream roles +and [FFmpeg's fixed control/audio/video channels](https://github.com/FFmpeg/FFmpeg/blob/master/libavformat/rtmppkt.h) +as well as [OBS's control and media channels](https://github.com/obsproject/obs-studio/blob/master/plugins/obs-outputs/librtmp/rtmp.c). +They bound retained state independently of chunk size and retain support for +large keyframes. Undecoded input has a separate 64 MiB buffer limit. +Connections exceeding these limits are unsupported. diff --git a/rs/moq-rtmp/src/rml/chunk_io/deserialization_errors.rs b/rs/moq-rtmp/src/rml/chunk_io/deserialization_errors.rs index 15fd98fc44..2b7e052e3a 100644 --- a/rs/moq-rtmp/src/rml/chunk_io/deserialization_errors.rs +++ b/rs/moq-rtmp/src/rml/chunk_io/deserialization_errors.rs @@ -16,6 +16,26 @@ pub enum ChunkDeserializationError { #[error("Requested an invalid max chunk size of {chunk_size}. Expected 1 through 2147483647")] InvalidMaxChunkSize { chunk_size: usize }, + /// The connection has exhausted its retained chunk-stream header slots. + #[error("RTMP connection exceeded 256 chunk stream IDs")] + ChunkStreamLimit, + + /// The connection has exhausted its aggregate message reservation budget. + #[error("RTMP connection exceeded 64 MiB of incomplete message payloads")] + AssemblyLimit, + + /// Undecoded input exceeds the connection's buffer budget. + #[error("RTMP connection exceeded 64 MiB of undecoded input")] + InputLimit, + + /// A continuation changed an incomplete message's metadata. + #[error("RTMP chunk changed an incomplete message on csid {csid}")] + InconsistentMessage { csid: u32 }, + + /// An extended timestamp delta was smaller than its 24-bit marker. + #[error("RTMP extended timestamp {timestamp} is below 16777215")] + InvalidExtendedTimestamp { timestamp: u32 }, + /// An I/O error occurred while reading the input buffer #[error("{0}")] Io(#[from] io::Error), diff --git a/rs/moq-rtmp/src/rml/chunk_io/deserializer.rs b/rs/moq-rtmp/src/rml/chunk_io/deserializer.rs index 7894635757..25aa6fd45a 100644 --- a/rs/moq-rtmp/src/rml/chunk_io/deserializer.rs +++ b/rs/moq-rtmp/src/rml/chunk_io/deserializer.rs @@ -2,7 +2,7 @@ use super::chunk_header::{ChunkHeader, ChunkHeaderFormat}; use crate::rml::chunk_io::ChunkDeserializationError; use crate::rml::messages::MessagePayload; use byteorder::{BigEndian, LittleEndian, ReadBytesExt}; -use bytes::{BufMut, BytesMut}; +use bytes::BytesMut; use std::cmp::min; use std::collections::HashMap; use std::io::Cursor; @@ -10,6 +10,13 @@ use std::mem; const INITIAL_MAX_CHUNK_SIZE: usize = 128; const MAX_INITIAL_TIMESTAMP: u32 = 16777215; +// The gateway uses five CSIDs; retain ample room for control and media roles +// without allowing every wire ID to pin header history for the connection. +const MAX_CHUNK_STREAMS: usize = 256; +// Four maximal 24-bit messages fit. Charge declared lengths before allocating, +// so tiny initial chunks cannot pin unaccounted full-message reservations. +const MAX_ASSEMBLY_BYTES: usize = 64 * 1024 * 1024; +const MAX_INPUT_BYTES: usize = 64 * 1024 * 1024; /// Allows deserializing bytes representing RTMP chunks into RTMP message payloads. /// @@ -21,12 +28,18 @@ pub struct ChunkDeserializer { current_header_format: ChunkHeaderFormat, current_header: ChunkHeader, current_stage: ParseStage, - current_payload: MessagePayload, - current_payload_data: BytesMut, + assemblies: HashMap, + assembly_bytes: usize, buffer: BytesMut, previous_headers: HashMap, } +struct Assembly { + payload: MessagePayload, + data: BytesMut, + length: usize, +} + enum ParsedValue { NotEnoughBytes, Value { val: T, next_index: u32 }, @@ -61,8 +74,8 @@ impl ChunkDeserializer { current_stage: ParseStage::Csid, buffer: BytesMut::with_capacity(4096), previous_headers: HashMap::new(), - current_payload: MessagePayload::new(), - current_payload_data: BytesMut::new(), + assemblies: HashMap::new(), + assembly_bytes: 0, } } @@ -145,6 +158,19 @@ impl ChunkDeserializer { /// # } /// ``` pub fn get_next_message(&mut self, bytes: &[u8]) -> Result, ChunkDeserializationError> { + let result = self.read_message(bytes); + if result.is_err() { + // Errors terminate the connection. Release every other CSID's retained + // state as well as this chunk, before returning through the session. + *self = Self::new(); + } + result + } + + fn read_message(&mut self, bytes: &[u8]) -> Result, ChunkDeserializationError> { + if bytes.len() > MAX_INPUT_BYTES - self.buffer.len() { + return Err(ChunkDeserializationError::InputLimit); + } self.buffer.extend_from_slice(bytes); loop { @@ -191,6 +217,13 @@ impl ChunkDeserializer { self.max_chunk_size } + /// Discard an aborted CSID's incomplete payload while retaining its header history. + pub(crate) fn abort(&mut self, csid: u32) { + if let Some(assembly) = self.assemblies.remove(&csid) { + self.assembly_bytes -= assembly.length; + } + } + fn form_header(&mut self) -> Result { if self.buffer.len() < 1 { return Ok(ParseStageResult::NotEnoughBytes); @@ -202,6 +235,10 @@ impl ChunkDeserializer { ParsedValue::Value { val, next_index } => (val, next_index), }; + if !self.previous_headers.contains_key(&csid) && self.previous_headers.len() >= MAX_CHUNK_STREAMS { + return Err(ChunkDeserializationError::ChunkStreamLimit); + } + self.current_header = match self.current_header_format { ChunkHeaderFormat::Full => { let mut new_header = ChunkHeader::new(); @@ -226,10 +263,8 @@ impl ChunkDeserializer { // across multiple chunks. We need to be careful *NOT* to apply the delta to each // type 3 chunk that's trying to serve a single message, otherwise timestamps will // get out of control. - if self.current_payload_data.len() == 0 { - // Since we don't have any payload data yet, that means this is the first - // chunk of the message. As it's the first chunk this is the only time we should - // apply the previous header's delta to the timestamp + if !self.assemblies.contains_key(&self.current_header.chunk_stream_id) { + // Apply the inherited delta once, when this CSID starts a message. self.current_header.timestamp = self.current_header.timestamp + self.current_header.timestamp_field; } @@ -343,12 +378,15 @@ impl ChunkDeserializer { timestamp = cursor.read_u32::()?; } - // If the type 3 chunk is not the first chunk of a message, we just ignore it's extended timestamp because the timestamp of this message was already deserialized. + // A continuation's type 3 timestamp does not advance the message timestamp. if self.current_header_format == ChunkHeaderFormat::Full { self.current_header.timestamp.set(timestamp); - } else if self.current_payload_data.len() == 0 { + } else if !self.assemblies.contains_key(&self.current_header.chunk_stream_id) { // Since we already added the MAX_INITIAL_TIMESTAMP to the timestamp, only add the delta difference - self.current_header.timestamp = self.current_header.timestamp + (timestamp - MAX_INITIAL_TIMESTAMP); + let delta = timestamp + .checked_sub(MAX_INITIAL_TIMESTAMP) + .ok_or(ChunkDeserializationError::InvalidExtendedTimestamp { timestamp })?; + self.current_header.timestamp = self.current_header.timestamp + delta; } self.current_stage = ParseStage::MessagePayload; @@ -359,38 +397,52 @@ impl ChunkDeserializer { &mut self, message_to_return: &mut Option, ) -> Result { - let mut length = self.current_header.message_length as usize; - let current_payload_length = self.current_payload_data.len(); - let remaining_bytes = length - current_payload_length; - if length > self.max_chunk_size as usize { - length = min(remaining_bytes, self.max_chunk_size as usize); + let csid = self.current_header.chunk_stream_id; + let length = self.current_header.message_length as usize; + if let Some(assembly) = self.assemblies.get(&csid) { + if assembly.length != length + || assembly.payload.type_id != self.current_header.message_type_id + || assembly.payload.message_stream_id != self.current_header.message_stream_id + || assembly.payload.timestamp != self.current_header.timestamp + { + return Err(ChunkDeserializationError::InconsistentMessage { csid }); + } + } else { + if length > MAX_ASSEMBLY_BYTES - self.assembly_bytes { + return Err(ChunkDeserializationError::AssemblyLimit); + } + self.assembly_bytes += length; + self.assemblies.insert( + csid, + Assembly { + payload: MessagePayload { + timestamp: self.current_header.timestamp, + type_id: self.current_header.message_type_id, + message_stream_id: self.current_header.message_stream_id, + ..MessagePayload::new() + }, + data: BytesMut::with_capacity(length), + length, + }, + ); } - if self.buffer.len() < length { + let assembly = self.assemblies.get_mut(&csid).unwrap(); + let remaining = length + .checked_sub(assembly.data.len()) + .ok_or(ChunkDeserializationError::InconsistentMessage { csid })?; + let chunk_length = min(remaining, self.max_chunk_size); + if self.buffer.len() < chunk_length { return Ok(ParseStageResult::NotEnoughBytes); } - self.current_payload.timestamp = self.current_header.timestamp; - self.current_payload.type_id = self.current_header.message_type_id; - self.current_payload.message_stream_id = self.current_header.message_stream_id; - - // Make sure the we have enough capacity for the whole message data. This - // helps with performance when there are smaller chunk sizes. - if remaining_bytes > self.current_payload_data.remaining_mut() { - let capacity_needed = remaining_bytes - self.current_payload_data.remaining_mut(); - self.current_payload_data.reserve(capacity_needed); - } - - let bytes = self.buffer.split_to(length as usize); - self.current_payload_data.extend_from_slice(&bytes[..]); - - // Check if this completes the message - if self.current_payload_data.len() == self.current_header.message_length as usize { - let data = mem::replace(&mut self.current_payload_data, BytesMut::new()); - self.current_payload.data = data.freeze(); - - let payload = mem::replace(&mut self.current_payload, MessagePayload::new()); - *message_to_return = Some(payload) + let bytes = self.buffer.split_to(chunk_length); + assembly.data.extend_from_slice(&bytes); + if assembly.data.len() == length { + let mut assembly = self.assemblies.remove(&csid).unwrap(); + self.assembly_bytes -= assembly.length; + assembly.payload.data = assembly.data.freeze(); + *message_to_return = Some(assembly.payload); } // This completes the current chunk, so cycle the header into the map and start a new one @@ -462,6 +514,367 @@ mod tests { use byteorder::{BigEndian, LittleEndian, WriteBytesExt}; use std::io::{Cursor, Write}; + fn interleaved_messages(b_len: usize, chunk_size: usize) { + let a = form_type_0_chunk(6, 100, 11, 9, &[0xa1; 300], chunk_size); + let b = form_type_0_chunk(4, 200, 22, 8, &vec![0xb2; b_len], chunk_size); + let boundary = 12 + chunk_size; + let bytes = [&a[..boundary], &b, &a[boundary..]].concat(); + let mut decoder = ChunkDeserializer::new(); + decoder.set_max_chunk_size(chunk_size).unwrap(); + let expected_b = MessagePayload { + timestamp: RtmpTimestamp::new(200), + type_id: 8, + message_stream_id: 22, + data: bytes::Bytes::from(vec![0xb2; b_len]), + }; + let expected_a = MessagePayload { + timestamp: RtmpTimestamp::new(100), + type_id: 9, + message_stream_id: 11, + data: bytes::Bytes::from(vec![0xa1; 300]), + }; + assert_eq!(decoder.get_next_message(&bytes).unwrap(), Some(expected_b)); + assert_eq!(decoder.get_next_message(&[]).unwrap(), Some(expected_a)); + assert_eq!(decoder.get_next_message(&[]).unwrap(), None); + } + + #[test] + fn interleaved_shorter_message_does_not_underflow() { + interleaved_messages(3, 128); + } + + #[test] + fn interleaved_longer_message_does_not_mix_payloads() { + // A's first chunk is shorter than B, which fits in a chunk after a size change. + let a = form_type_0_chunk(6, 100, 11, 9, &[0xa1; 200], 128); + let b = form_type_0_chunk(4, 200, 22, 8, &[0xb2; 160], 256); + let mut decoder = ChunkDeserializer::new(); + assert_eq!(decoder.get_next_message(&a[..140]).unwrap(), None); + decoder.set_max_chunk_size(256).unwrap(); + let result = decoder.get_next_message(&b).unwrap().unwrap(); + assert_eq!( + result, + MessagePayload { + timestamp: RtmpTimestamp::new(200), + type_id: 8, + message_stream_id: 22, + data: bytes::Bytes::from(vec![0xb2; 160]), + } + ); + let result = decoder.get_next_message(&a[140..]).unwrap().unwrap(); + assert_eq!( + result, + MessagePayload { + timestamp: RtmpTimestamp::new(100), + type_id: 9, + message_stream_id: 11, + data: bytes::Bytes::from(vec![0xa1; 200]), + } + ); + assert_eq!(decoder.get_next_message(&[]).unwrap(), None); + } + + fn drain(decoder: &mut ChunkDeserializer, bytes: &[u8], messages: &mut Vec) { + let mut next = decoder.get_next_message(bytes).unwrap(); + while let Some(message) = next { + messages.push(message); + next = decoder.get_next_message(&[]).unwrap(); + } + } + + #[test] + fn interleaved_compressed_headers_and_fragmented_input_match_independent_streams() { + let mut wire = Vec::new(); + let mut independent = [ChunkDeserializer::new(), ChunkDeserializer::new()]; + let mut expected = Vec::new(); + for decoder in &mut independent { + decoder.set_max_chunk_size(2).unwrap(); + } + for format in 0..4 { + let mut packets = Vec::new(); + for (index, csid) in [6, 4].into_iter().enumerate() { + let payload = vec![0xa0 + index as u8 + format; if format == 0 { 4 } else { 6 }]; + let (mut packet, header, extended) = match format { + 0 => ( + form_type_0_chunk(csid, 10 + index as u32, 11 + index as u32, 9, &payload, 2), + 12, + None, + ), + 1 => (form_type_1_chunk(csid, 16777220, 8, &payload), 12, Some(16777220)), + 2 => (form_type_2_chunk(csid, 10, &payload), 4, None), + _ => (form_type_3_chunk(csid, &payload, 2, None), 1, None), + }; + if format == 1 || format == 2 { + packet.truncate(header + 2); + packet.extend(form_type_3_chunk(csid, &payload[2..], 2, extended)); + } + let mut result = Vec::new(); + drain(&mut independent[index], &packet, &mut result); + assert_eq!(result.len(), 1); + assert_eq!(result[0].data.as_ref(), payload); + assert_eq!(result[0].message_stream_id, 11 + index as u32); + assert_eq!(result[0].type_id, if format == 0 { 9 } else { 8 }); + let timestamp = [10, 16777230, 16777240, 16777250][format as usize] + index as u32; + assert_eq!(result[0].timestamp, RtmpTimestamp::new(timestamp)); + packets.push((packet, header + 2, result.remove(0))); + } + wire.extend_from_slice(&packets[0].0[..packets[0].1]); + wire.extend_from_slice(&packets[1].0); + wire.extend_from_slice(&packets[0].0[packets[0].1..]); + expected.push(packets.remove(1).2); + expected.push(packets.remove(0).2); + } + for fragment in [1, 2, 7, wire.len()] { + let mut decoder = ChunkDeserializer::new(); + decoder.set_max_chunk_size(2).unwrap(); + let mut result = Vec::new(); + for bytes in wire.chunks(fragment) { + drain(&mut decoder, bytes, &mut result); + } + assert_eq!(result, expected); + assert_eq!(decoder.get_next_message(&[]).unwrap(), None); + assert!(decoder.assemblies.is_empty()); + assert_eq!(decoder.assembly_bytes, 0); + } + } + + fn assert_released(decoder: &ChunkDeserializer) { + assert!(decoder.assemblies.is_empty()); + assert!(decoder.previous_headers.is_empty()); + assert!(decoder.buffer.is_empty()); + assert_eq!(decoder.assembly_bytes, 0); + } + + #[test] + fn chunk_stream_limit_includes_incomplete_ids_and_releases_connection() { + let mut decoder = ChunkDeserializer::new(); + decoder.set_max_chunk_size(1).unwrap(); + for csid in 2..2 + MAX_CHUNK_STREAMS as u32 { + let packet = form_type_0_chunk(csid, 0, 1, 9, &[42; 2], 1); + let header = if csid < 64 { 12 } else { 13 }; + assert_eq!(decoder.get_next_message(&packet[..header + 1]).unwrap(), None); + } + assert_eq!(decoder.previous_headers.len(), MAX_CHUNK_STREAMS); + assert_eq!(decoder.assemblies.len(), MAX_CHUNK_STREAMS); + assert_eq!(decoder.assembly_bytes, MAX_CHUNK_STREAMS * 2); + let packet = form_type_0_chunk(258, 0, 1, 9, &[42], 1); + assert!(matches!( + decoder.get_next_message(&packet), + Err(ChunkDeserializationError::ChunkStreamLimit) + )); + assert_released(&decoder); + } + + #[test] + fn completed_ids_retain_headers_and_reuse_does_not_consume_slots() { + let mut decoder = ChunkDeserializer::new(); + for csid in 2..2 + MAX_CHUNK_STREAMS as u32 { + let packet = form_type_0_chunk(csid, 0, 1, 9, &[42], 128); + assert!(decoder.get_next_message(&packet).unwrap().is_some()); + } + for _ in 0..100 { + let packet = form_type_3_chunk(2, &[42], 128, None); + assert!(decoder.get_next_message(&packet).unwrap().is_some()); + assert_eq!(decoder.assembly_bytes, 0); + assert!(decoder.assemblies.is_empty()); + assert_eq!(decoder.previous_headers.len(), MAX_CHUNK_STREAMS); + } + let packet = form_type_0_chunk(258, 0, 1, 9, &[42], 128); + assert!(matches!( + decoder.get_next_message(&packet), + Err(ChunkDeserializationError::ChunkStreamLimit) + )); + assert_released(&decoder); + } + + #[test] + fn aggregate_declared_lengths_are_charged_before_allocation() { + let mut decoder = ChunkDeserializer::new(); + decoder.set_max_chunk_size(1).unwrap(); + for csid in 2..6 { + let mut first = form_type_0_chunk(csid, 0, 1, 9, &[42], 1); + first[4..7].copy_from_slice(&[0xff; 3]); + assert_eq!(decoder.get_next_message(&first).unwrap(), None); + } + assert_eq!(decoder.assembly_bytes, MAX_ASSEMBLY_BYTES - 4); + // Completion gives back the whole reservation, while the other IDs remain incomplete. + let small = form_type_0_chunk(6, 0, 1, 9, &[42; 4], 1); + assert!(decoder.get_next_message(&small).unwrap().is_some()); + assert_eq!(decoder.assembly_bytes, MAX_ASSEMBLY_BYTES - 4); + assert_eq!(decoder.get_next_message(&small[..13]).unwrap(), None); + assert_eq!(decoder.assembly_bytes, MAX_ASSEMBLY_BYTES); + assert_eq!( + decoder.assemblies.values().map(|a| a.data.capacity()).sum::(), + MAX_ASSEMBLY_BYTES + ); + let next = form_type_0_chunk(7, 0, 1, 9, &[42], 1); + assert!(matches!( + decoder.get_next_message(&next), + Err(ChunkDeserializationError::AssemblyLimit) + )); + assert_released(&decoder); + } + + #[test] + fn incomplete_message_reservations_are_reclaimed_on_repeated_reuse() { + let mut decoder = ChunkDeserializer::new(); + let packet = form_type_0_chunk(6, 0, 1, 9, &[42; 200], 128); + for _ in 0..100 { + assert_eq!(decoder.get_next_message(&packet[..140]).unwrap(), None); + assert_eq!(decoder.assembly_bytes, 200); + assert!(decoder.get_next_message(&packet[140..]).unwrap().is_some()); + assert_eq!(decoder.assembly_bytes, 0); + assert!(decoder.assemblies.is_empty()); + assert_eq!(decoder.previous_headers.len(), 1); + } + // A shared witness becomes unique only after teardown releases the assembly allocation. + assert_eq!(decoder.get_next_message(&packet[..140]).unwrap(), None); + let witness = decoder.assemblies.get_mut(&6).unwrap().data.split_to(1).freeze(); + assert!(!witness.is_unique()); + drop(decoder); + assert!(witness.is_unique()); + } + + #[test] + fn changed_incomplete_message_header_is_refused_and_releases_other_ids() { + for (offset, replacement) in [(6, 1), (7, 8), (8, 2), (3, 1)] { + let mut decoder = ChunkDeserializer::new(); + let packet = form_type_0_chunk(6, 0, 1, 9, &[42; 200], 128); + assert_eq!(decoder.get_next_message(&packet[..140]).unwrap(), None); + let other = form_type_0_chunk(4, 0, 1, 8, &[43; 200], 128); + assert_eq!(decoder.get_next_message(&other[..140]).unwrap(), None); + let mut changed = packet[..140].to_vec(); + changed[offset] = replacement; + assert!(matches!( + decoder.get_next_message(&changed), + Err(ChunkDeserializationError::InconsistentMessage { csid: 6 }) + )); + assert_released(&decoder); + } + } + + #[test] + fn input_budget_is_checked_before_buffering_and_releases_assemblies() { + let mut decoder = ChunkDeserializer::new(); + let packet = form_type_0_chunk(6, 0, 1, 9, &[42; 200], 128); + assert_eq!(decoder.get_next_message(&packet[..140]).unwrap(), None); + let complete = form_type_0_chunk(4, 0, 1, 99, &[43], 128); + let mut input = complete; + input.resize(MAX_INPUT_BYTES, 0); + assert!(decoder.get_next_message(&input).unwrap().is_some()); + assert_eq!(decoder.assembly_bytes, 200); + // Appending exactly the consumed bytes fits. The zero-filled suffix starts + // with a valid empty message on CSID 64, which consumes another 13 bytes. + assert!(decoder.get_next_message(&[0; 13]).unwrap().is_some()); + assert!(matches!( + decoder.get_next_message(&[0; 14]), + Err(ChunkDeserializationError::InputLimit) + )); + assert_released(&decoder); + } + + #[test] + fn extended_timestamp_subtraction_is_validated() { + let mut decoder = ChunkDeserializer::new(); + let first = form_type_0_chunk(6, 0, 1, 9, &[42], 128); + assert!(decoder.get_next_message(&first).unwrap().is_some()); + let mut next = form_type_1_chunk(6, MAX_INITIAL_TIMESTAMP + 1, 9, &[43]); + next[8..12].copy_from_slice(&1u32.to_be_bytes()); + assert!(matches!( + decoder.get_next_message(&next), + Err(ChunkDeserializationError::InvalidExtendedTimestamp { timestamp: 1 }) + )); + assert_released(&decoder); + } + + #[test] + fn largest_numeric_csid_and_zero_length_messages_are_supported() { + let mut decoder = ChunkDeserializer::new(); + let mut packet = vec![1, 255, 255]; + packet.extend_from_slice(&[0, 0, 10, 0, 0, 0, 9, 1, 0, 0, 0]); + let first = decoder.get_next_message(&packet).unwrap().unwrap(); + assert_eq!(first.message_stream_id, 1); + assert!(first.data.is_empty()); + assert_eq!(decoder.previous_headers.len(), 1); + assert!(decoder.previous_headers.contains_key(&65599)); + let next = decoder.get_next_message(&[0xc1, 255, 255]).unwrap().unwrap(); + assert_eq!(next.timestamp, RtmpTimestamp::new(20)); + assert!(next.data.is_empty()); + assert!(decoder.assemblies.is_empty()); + assert_eq!(decoder.assembly_bytes, 0); + assert_eq!(decoder.get_next_message(&[]).unwrap(), None); + } + + #[test] + fn abort_releases_only_the_target_assembly_in_both_session_roles() { + use crate::rml::sessions::{ + ClientSession, ClientSessionConfig, ClientSessionResult, ServerSession, ServerSessionConfig, + ServerSessionResult, + }; + let (mut server, _) = ServerSession::new(ServerSessionConfig::new()).unwrap(); + let (mut client, _) = ClientSession::new(ClientSessionConfig::new()).unwrap(); + let a = form_type_0_chunk(6, 100, 11, 99, &[0xa1; 200], 128); + let b = form_type_0_chunk(4, 200, 22, 99, &[0xb2; 200], 128); + let abort = form_type_0_chunk(2, 0, 0, 2, &6u32.to_be_bytes(), 128); + let replacement = form_type_2_chunk(6, 10, &[0xc3; 200]); + let replacement = [&replacement[..132], &form_type_3_chunk(6, &[0xc3; 72], 128, None)].concat(); + let bytes = [&a[..140], &b[..140], &abort, &replacement, &b[140..]].concat(); + let expected = [ + MessagePayload { + timestamp: RtmpTimestamp::new(110), + message_stream_id: 11, + type_id: 99, + data: bytes::Bytes::from(vec![0xc3; 200]), + }, + MessagePayload { + timestamp: RtmpTimestamp::new(200), + message_stream_id: 22, + type_id: 99, + data: bytes::Bytes::from(vec![0xb2; 200]), + }, + ]; + let server_results = server.handle_input(&bytes).unwrap(); + let client_results = client.handle_input(&bytes).unwrap(); + let server_payloads: Vec<_> = server_results + .into_iter() + .map(|result| match result { + ServerSessionResult::UnhandleableMessageReceived(payload) => payload, + other => panic!("unexpected result: {other:?}"), + }) + .collect(); + let client_payloads: Vec<_> = client_results + .into_iter() + .map(|result| match result { + ClientSessionResult::UnhandleableMessageReceived(payload) => payload, + other => panic!("unexpected result: {other:?}"), + }) + .collect(); + assert_eq!(server_payloads, expected); + assert_eq!(client_payloads, expected); + assert!(server.handle_input(&[]).unwrap().is_empty()); + assert!(client.handle_input(&[]).unwrap().is_empty()); + } + + #[test] + fn abort_reclaims_reservation_and_keeps_compressed_header_history() { + let mut decoder = ChunkDeserializer::new(); + for csid in [6, 4] { + let packet = form_type_0_chunk(csid, 0, 1, 9, &[42; 200], 128); + assert_eq!(decoder.get_next_message(&packet[..140]).unwrap(), None); + } + let witness = decoder.assemblies.get_mut(&6).unwrap().data.split_to(1).freeze(); + assert!(!witness.is_unique()); + decoder.abort(6); + assert!(witness.is_unique()); + assert_eq!(decoder.assembly_bytes, 200); + assert_eq!(decoder.assemblies.len(), 1); + assert!(decoder.assemblies.contains_key(&4)); + assert_eq!(decoder.previous_headers.len(), 2); + decoder.abort(6); + decoder.abort(65599); + assert_eq!(decoder.assembly_bytes, 200); + } + #[test] fn chunk_size_floor() { let mut decoder = ChunkDeserializer::new(); diff --git a/rs/moq-rtmp/src/rml/mod.rs b/rs/moq-rtmp/src/rml/mod.rs index d7a6427aea..5aa003d187 100644 --- a/rs/moq-rtmp/src/rml/mod.rs +++ b/rs/moq-rtmp/src/rml/mod.rs @@ -7,6 +7,7 @@ //! see LICENSE in this directory). The `amf0` module vendors rml_amf0 0.3.0 //! from the same project and license. Local patches include: //! - a 64-level AMF0 container nesting limit and a nonzero incoming chunk size. +//! - independent per-CSID message assembly with bounded connection state. //! - `sessions::ServerSession::set_connect_response_properties`, so the gateway //! can advertise enhanced-RTMP capabilities in the connect `_result`. //! - guards against two reachable panics on malformed untrusted input (short diff --git a/rs/moq-rtmp/src/rml/sessions/client/mod.rs b/rs/moq-rtmp/src/rml/sessions/client/mod.rs index df37f13d06..fbaf6c2ae3 100644 --- a/rs/moq-rtmp/src/rml/sessions/client/mod.rs +++ b/rs/moq-rtmp/src/rml/sessions/client/mod.rs @@ -156,6 +156,11 @@ impl ClientSession { RtmpMessage::SetChunkSize { size } => self.handle_set_chunk_size(size)?, + RtmpMessage::Abort { stream_id } => { + self.deserializer.abort(stream_id); + Vec::new() + } + _ => vec![ClientSessionResult::UnhandleableMessageReceived(payload)], }; diff --git a/rs/moq-rtmp/src/rml/sessions/server/mod.rs b/rs/moq-rtmp/src/rml/sessions/server/mod.rs index 4b25a89e3a..e7d62009bf 100644 --- a/rs/moq-rtmp/src/rml/sessions/server/mod.rs +++ b/rs/moq-rtmp/src/rml/sessions/server/mod.rs @@ -430,7 +430,8 @@ impl ServerSession { Ok(self.serializer.serialize(&payload, false, false)?) } - fn handle_abort_message(&self, _stream_id: u32) -> Result, ServerSessionError> { + fn handle_abort_message(&mut self, stream_id: u32) -> Result, ServerSessionError> { + self.deserializer.abort(stream_id); Ok(Vec::new()) } From 61825510a0a869cb5239c2658e53ffd814f3d8cf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 14:55:07 -0700 Subject: [PATCH 03/11] test(net): adapt dial_split_horizon to release's test harness Release predates moq_net_sim and announce::Event, so the backported test runs on paused tokio time and reads announce::Update/Kind instead. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/tests/dial_split_horizon.rs | 42 ++++++++++++-------------- 1 file changed, 20 insertions(+), 22 deletions(-) diff --git a/rs/moq-net/tests/dial_split_horizon.rs b/rs/moq-net/tests/dial_split_horizon.rs index 402d01ba98..914b52a9af 100644 --- a/rs/moq-net/tests/dial_split_horizon.rs +++ b/rs/moq-net/tests/dial_split_horizon.rs @@ -10,9 +10,9 @@ mod support; use std::time::Duration; -use moq_net::{Client, Hop, Server, Session, Version, origin, stats}; +use moq_net::{Client, Hop, Server, Session, Version, announce, origin, stats}; use support::{ - harness::{now, spawn}, + harness::{now, run}, mock::create_mock_session_pair, }; @@ -25,7 +25,7 @@ fn version() -> Version { fn produce(hop: u64) -> origin::Producer { let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); - spawn(driver); + tokio::spawn(run(driver)); producer } @@ -63,7 +63,7 @@ async fn dial(dialer: &origin::Producer, edge: &origin::Producer) -> Link { .connect(now(), client_transport) .await .expect("client handshake failed"); - spawn(driver); + tokio::spawn(run(driver)); session }; let server = async { @@ -71,10 +71,10 @@ async fn dial(dialer: &origin::Producer, edge: &origin::Producer) -> Link { .accept(now(), server_transport) .await .expect("server handshake failed"); - spawn(driver); + tokio::spawn(run(driver)); session }; - let (client, server) = futures::join!(client, server); + let (client, server) = tokio::join!(client, server); Link { _client: client, _server: server, @@ -95,15 +95,11 @@ fn traffic(registry: &stats::Registry, role: stats::Role) -> stats::Traffic { /// The hop the dialer attributed `path` to, once that announce has arrived. async fn learned_hop(dialer: &origin::Producer, path: &str) -> Hop { let mut announced = dialer.consume().announced(); - let event = moq_net_sim::timeout(SETTLE, async { + let event = tokio::time::timeout(SETTLE, async { loop { - let event = announced.next().await.expect("announce cursor closed"); - let announce = match event { - moq_net::announce::Event::Start(announce) | moq_net::announce::Event::Update(announce) => announce, - moq_net::announce::Event::End(_) => continue, - }; - if announce.prefix.as_str() == path { - return announce; + let update = announced.next().await.expect("announce cursor closed"); + if update.kind.is_active() && update.prefix.as_str() == path { + return update; } } }) @@ -120,7 +116,7 @@ async fn learned_hop(dialer: &origin::Producer, path: &str) -> Hop { /// announced a namespace is not told about it, and the other edge is. A /// subscribe from that other edge is fetched from the announcer and is not /// sent back to the subscriber. -#[moq_net_sim::test] +#[tokio::test(start_paused = true)] async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { let dialer = produce(1); let publisher = produce(2); @@ -142,13 +138,15 @@ async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { // The other edge learns the namespace, attributed to a different hop, so the // two dials are not one endpoint. let mut subscriber_announced = subscriber.consume().announced(); - let subscriber_saw = moq_net_sim::timeout(SETTLE, subscriber_announced.next()) + let subscriber_saw = tokio::time::timeout(SETTLE, subscriber_announced.next()) .await .expect("the other edge never learned the namespace") .expect("announce cursor closed"); - let moq_net::announce::Event::Start(subscriber_saw) = subscriber_saw else { - panic!("the other edge's first event was not the namespace: {subscriber_saw:?}"); - }; + assert_eq!( + subscriber_saw.kind, + announce::Kind::Announced, + "the other edge's first event was not the namespace: {subscriber_saw:?}" + ); assert_eq!(subscriber_saw.prefix.as_str(), "room"); let subscriber_hop = match subscriber_saw.route.source() { origin::Source::Peer(hop) => hop, @@ -157,7 +155,7 @@ async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { assert_ne!(subscriber_hop, publisher_hop, "each dial gets its own hop"); assert_ne!(subscriber_hop, Hop::UNKNOWN); - moq_net_sim::sleep(SETTLE).await; + tokio::time::sleep(SETTLE).await; let echoed = traffic(&to_publisher.registry, stats::Role::Subscriber); assert_eq!( @@ -185,12 +183,12 @@ async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { .subscribe(subscription) .await .expect("subscriber subscribes"); - moq_net_sim::timeout(SETTLE, track.demand().used()) + tokio::time::timeout(SETTLE, track.demand().used()) .await .expect("the subscribe never reached the announcer") .unwrap(); - moq_net_sim::sleep(SETTLE).await; + tokio::time::sleep(SETTLE).await; let from_announcer = traffic(&to_publisher.registry, stats::Role::Publisher); assert_eq!( From 1afd492a0e07a825b34429e2ca2a5d0b48ecbf17 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 14:58:25 -0700 Subject: [PATCH 04/11] fix(js): drop a headless subgroup before FIRST_OBJECT (backport #5019) On drafts 14-17 the subgroup header has no FIRST_OBJECT bit, so a stream that starts at the live edge reached Frame.decode with a non-zero first delta and failed the group. Run release's existing first-object peek on those drafts too, and drop the stream so the subscription resumes at the next group. (cherry picked from commit 321fc027405ce8e4e1cfc3fd87339e38f511fc53, adapted) Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/object.ts | 3 +- js/net/src/ietf/subscriber.test.ts | 59 +++++++++++++++++++++++++++--- js/net/src/ietf/subscriber.ts | 15 ++++---- 3 files changed, 63 insertions(+), 14 deletions(-) diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index 3f6f86f6a3..af7aaaacf9 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -25,7 +25,8 @@ const PROP_TIMESTAMP_DRAFT03 = 0x06n; // object 0 is still the head, and any other first ID is dropped. const FIRST_OBJECT_BIT = 0x40; -function hasFirstObjectBit(version: IetfVersion): boolean { +/** Whether the subgroup header carries FIRST_OBJECT: draft-18 and later, not drafts 14-17. */ +export function hasFirstObjectBit(version: IetfVersion): boolean { switch (version) { case Version.DRAFT_14: case Version.DRAFT_15: diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index ab3048e54b..ebd3cbc4ce 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -15,7 +15,7 @@ import { RequestError, RequestOk } from "./request.ts"; import { Subscribe, SubscribeOk, Unsubscribe } from "./subscribe.ts"; import { SubscribeNamespace, SubscribeNamespaceEntry, SubscribeNamespaceEntryDone } from "./subscribe_namespace.ts"; import { Subscriber } from "./subscriber.ts"; -import { ALPN, Version } from "./version.ts"; +import { ALPN, type IetfVersion, Version } from "./version.ts"; const VERSION = Version.DRAFT_19; @@ -860,20 +860,24 @@ function encodeObjects(deltas: number[]): Uint8Array { * A subscriber with one track subscribed and answered, which is what registers {@link ALIAS} * and lets a group stream naming it be handled. */ -async function subscribeTrack(): Promise<{ subscriber: Subscriber; track: track.Subscriber }> { - const pair = createMockTransportPair(ALPN.DRAFT_19); - const session = new NativeSession(pair.server, VERSION, true); +async function subscribeTrack( + version: IetfVersion = VERSION, +): Promise<{ subscriber: Subscriber; track: track.Subscriber }> { + const pair = createMockTransportPair(version === Version.DRAFT_16 ? ALPN.DRAFT_16 : ALPN.DRAFT_19); + const session = new NativeSession(pair.server, version, true); const subscriber = new Subscriber({ session }); const track = subscriber.consume(Path.from("room")).track("video").subscribe(); const peer = await nextStream(pair.client); if (!peer) throw new Error("the subscriber never opened a subscribe stream"); + peer.reader.version = version; + peer.writer.version = version; expect(await peer.reader.u53()).toBe(Subscribe.id); - const request = await Subscribe.decode(peer.reader, VERSION); + const request = await Subscribe.decode(peer.reader, version); await peer.writer.u53(SubscribeOk.id); - await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, VERSION); + await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, version); return { subscriber, track }; } @@ -1030,6 +1034,49 @@ test("a clear FIRST_OBJECT past object 0 is dropped", async () => { track.close(); }); +/** + * Drafts 14-17 have no FIRST_OBJECT bit, so a subgroup that starts at the live edge + * arrives with `firstObject` forced on and a non-zero first delta. That stream is the + * in-progress group: drop it, keep the subscription, and deliver the next group, which + * starts at object 0. A gap after an object was delivered still fails that group. + */ +test("a draft without FIRST_OBJECT drops a subgroup that starts mid-group", async () => { + const version = Version.DRAFT_16; + const { subscriber, track } = await subscribeTrack(version); + + // The header cannot say otherwise on this draft: decode reports firstObject. + const whole = groupFlags(true); + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 3, subGroupId: 0, publisherPriority: 0, flags: whole }), + new Reader(undefined, encodeObjects([2, 0]), version), + ); + expect(track.latest()).toBeUndefined(); + expect(track.closed.peek()).toBeUndefined(); + + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 4, subGroupId: 0, publisherPriority: 0, flags: whole }), + new Reader(undefined, encodeObjects([0, 0]), version), + ); + + const ordered = track.ordered(); + const group = await ordered.nextGroup(); + expect(group?.sequence).toBe(4); + expect(await group?.readString()).toBe("object 0"); + + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 5, subGroupId: 0, publisherPriority: 0, flags: whole }), + new Reader(undefined, encodeObjects([0, 5]), version), + ); + const gapped = await ordered.nextGroup(); + expect(gapped?.sequence).toBe(5); + expect(await gapped?.readString()).toBe("object 0"); + await expect(gapped?.readFrameSequence()).rejects.toThrow(/object IDs must start at 0/); + expect(track.closed.peek()).toBeUndefined(); + + ordered.close(); + track.close(); +}); + /** * FIRST_OBJECT is the publisher's claim, and the object ids are what actually happened. A * peer that sets the bit and then starts at object 5 is contradicting itself, so the group diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 473af22e40..f8784cf637 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -17,7 +17,7 @@ import type { Session } from "./adapter.ts"; import { DuplicateTrackAlias, RetiredTrackAlias, TrackAliases } from "./aliases.ts"; import * as Cluster from "./cluster.ts"; import { requestReason, toRequestCode } from "./error.ts"; -import { Frame, type Group as GroupMessage } from "./object.ts"; +import { Frame, type Group as GroupMessage, hasFirstObjectBit } from "./object.ts"; import { fromWire, toWire } from "./priority.ts"; import { type Publish, PublishDone, PublishError, publishDoneClean } from "./publish.ts"; import { @@ -1046,18 +1046,19 @@ export class Subscriber { // or a stream with no object, has a hole at the front: drop it and pick up at // the next group. // - // This only saves reading a stream we would throw away. The bit is the publisher's - // claim, so what is enforced is the object ids themselves: `Frame.decode` holds every - // object to starting at 0 and incrementing by 1, whatever the header said and on the - // drafts that have no such bit to read. - if (!group.flags.firstObject) { + // Drafts before the bit cannot say this in the header, so the first ID is the only + // signal: a non-zero one is the same hole and is dropped the same way, but an empty + // stream there is still a group. A later gap, or a header that claimed the group + // starts at object 0, still fails it: `Frame.decode` refuses every non-zero delta. + const legacy = !hasFirstObjectBit(this.#session.version); + if (!group.flags.firstObject || legacy) { let id: bigint | undefined; try { id = await stream.peekU62(); } catch (err: unknown) { if (!(err instanceof UnexpectedEnd)) throw err; } - if (id !== 0n) { + if (id !== 0n && !(legacy && id === undefined)) { console.debug(`dropping a group with no head: alias=${group.trackAlias} group=${group.groupId}`); stream.stop(new Error("a group must start at object 0")); return; From 64eee3d29674d0f4a52ff25cbfd7062beb6fb716 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 10:04:26 -0700 Subject: [PATCH 05/11] fix(net): accept an End of Group status on a marked subgroup stream (#5081) Backport to release. The Rust regression test writes the draft-18 status object inline with release's encode form and runs under tokio::test, since release lacks the end_marker helper and moq_net_sim. Quest edits dropped. (cherry picked from commit 52a2a8953910feb1e627883524fdd5ff2bbf19f7) Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/object.ts | 5 +++- js/net/src/ietf/subscriber.test.ts | 28 +++++++++++++++++++++ rs/moq-net/src/ietf/subscriber.rs | 39 +++++++++++++++++++++++++++--- 3 files changed, 68 insertions(+), 4 deletions(-) diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index 3f6f86f6a3..511c39e853 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -343,11 +343,14 @@ export class Frame { // Defined on every implemented draft, whether or not the header marks the group's end. if (status === END_OF_TRACK) return new Frame({ endOfTrack: true }); + // Allowed even when the header marks the group's end: that bit only lets a FIN imply it, + // and imquic sends both. + if (status === GROUP_END) return new Frame(); if (flags.hasEnd) { // Empty frame if (status === 0) return new Frame({ payload: new Uint8Array(0), timestamp }); - } else if (status === 0 || status === GROUP_END) { + } else if (status === 0) { // TODO status === 0 should be an empty frame, but moq-rs seems to be sending it incorrectly on group end. return new Frame(); } diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index ab3048e54b..c8e1b73894 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -1005,6 +1005,34 @@ test("a clear FIRST_OBJECT at object 0 is the whole group", async () => { track.close(); }); +/** + * The END_OF_GROUP header bit only lets a FIN imply the group's end, so an explicit + * END_OF_GROUP status on the same stream ends the group there rather than failing it. + */ +test("an END_OF_GROUP status on a marked stream finishes the group", async () => { + const { subscriber, track } = await subscribeTrack(); + + const flags = groupFlags(true); + expect(flags.hasEnd).toBe(true); + // Objects 0..4, then object 5 as a zero-length END_OF_GROUP (0x3) status. + const objects = new Uint8Array([...encodeObjects([0, 0, 0, 0, 0]), 0, 0, 0x3]); + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 3, subGroupId: 0, publisherPriority: 0, flags }), + new Reader(undefined, objects, VERSION), + ); + + const group = await track.ordered().nextGroup(); + expect(group?.sequence).toBe(3); + if (!group) return; + for (let i = 0; i < 5; i++) { + expect(await group.readString()).toBe("object 0"); + } + expect(await group.readFrame()).toBeUndefined(); + expect(await group.closed).toBeNull(); + + track.close(); +}); + /** * A clear FIRST_OBJECT whose first ID is not 0 still has a hole at the front, so the * stream is dropped and the track resumes at the next group. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1a3e128b7b..1f4bc2854c 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2647,7 +2647,6 @@ where struct GroupIngest { runtime: crate::time::Clock, has_extensions: bool, - has_end: bool, timescale: Option, version: Version, prior_object: Option, @@ -2683,7 +2682,6 @@ impl GroupIngest { Self { runtime: subscriber.runtime.clone(), has_extensions: group.flags.has_extensions, - has_end: group.flags.has_end, timescale, version: subscriber.version, prior_object: None, @@ -3505,7 +3503,9 @@ impl GroupIngest { let frame = group.create_frame_owned(frame::Info { size: 0, timestamp }, &self.budget)?; frame.finish()?; self.phase = IngestPhase::Delta; - } else if status == END_OF_GROUP && !self.has_end { + } else if status == END_OF_GROUP { + // Allowed even when the header marks the group's end: that bit only + // lets a FIN imply it, and imquic sends both. self.phase = IngestPhase::Finished(Ended::Group); } else if status == END_OF_TRACK { // Defined on every implemented draft, whether or not the header marks @@ -7050,6 +7050,39 @@ mod stitch_tests { assert_eq!(frames[1].1, b"next"); } + /// The END_OF_GROUP header bit only lets a FIN imply the group's end, so an explicit + /// END_OF_GROUP status on the same stream ends the group there rather than failing it. + #[tokio::test] + async fn an_end_of_group_status_on_a_marked_stream_finishes_the_group() { + use futures::FutureExt; + + let payloads: [&[u8]; 5] = [b"o0", b"o1", b"o2", b"o3", b"o4"]; + let mut bytes = draft18_subgroup(SEQUENCE, true, 0, &payloads); + // Object 5: the id delta, an empty extension block, a zero size, and the status. + for value in [0u64, 0, 0, END_OF_GROUP] { + value.encode(&mut bytes, Version::Draft18).unwrap(); + } + let flags = ietf::GroupFlags::decode(u64::from(bytes[0]), Version::Draft18).unwrap(); + assert!(flags.has_end, "the header marks the group's end"); + + let mut h = Harness::new(Fill::Done, vec![bytes]); + h.subscriber.version = Version::Draft18; + let mut consumer = h.track.subscribe(None); + let mut stream = read_draft18(&h.session).await; + + h.subscriber.clone().recv_group(&mut stream).await.expect("the group"); + + // The stream was read to its end, so everything is already delivered. + let mut group = consumer.recv_group().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(group.sequence, SEQUENCE); + for payload in payloads { + let frame = group.read_frame().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(frame.payload.as_ref(), payload); + } + let end = group.read_frame().now_or_never().expect("the group is closed"); + assert!(matches!(end, Ok(None)), "the group is finished, not aborted: {end:?}"); + } + /// A clear FIRST_OBJECT whose first ID is not 0 still has a hole at the front, so /// the stream is dropped and the track resumes at the next group. #[tokio::test] From 952ecb7283dfe29a25b5ab06c77aa51f6b7793a8 Mon Sep 17 00:00:00 2001 From: Jose Alejandro Chavarria Madriz <48740224+Jachm11@users.noreply.github.com> Date: Fri, 9 Oct 2026 00:22:22 -0600 Subject: [PATCH 06/11] fix(hang): accept AV1 profile 2 codec strings (#5064) Co-authored-by: Jchavarria Co-authored-by: Codex (cherry picked from commit 7556594007f9ef1bb772f5b34f93f5a2961b0752) Co-Authored-By: Claude Opus 5.5 --- rs/hang/src/catalog/video/av1.rs | 33 ++++++++++++++++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/rs/hang/src/catalog/video/av1.rs b/rs/hang/src/catalog/video/av1.rs index 1b2ba03cf6..06db4d37f7 100644 --- a/rs/hang/src/catalog/video/av1.rs +++ b/rs/hang/src/catalog/video/av1.rs @@ -81,7 +81,7 @@ impl std::fmt::Display for AV1 { lazy_static::lazy_static! { static ref AV1_REGEX: regex::Regex = regex::Regex::new(&r" ^av01 - \.(?[01]) + \.(?[0-2]) \.(?\d{2})(?\w) \.(?\d{2}) (? @@ -151,7 +151,7 @@ impl Default for AV1 { mod test { use std::str::FromStr; - use crate::catalog::VideoCodec; + use crate::catalog::{VideoCodec, VideoConfig}; use super::*; @@ -199,4 +199,33 @@ mod test { let output = decoded.to_string(); assert_eq!(output, encoded); } + + #[test] + fn test_av1_profiles_roundtrip() { + for (encoded, profile) in [ + ("av01.0.00M.08", 0), + ("av01.1.00M.10.0.000.01.01.01.0", 1), + ("av01.2.00M.12", 2), + ("av01.2.00M.08.0.100.02.02.02.0", 2), + ("av01.2.00M.10.0.100.09.16.09.0", 2), + ("av01.2.00M.12.0.100.09.16.09.0", 2), + ("av01.2.00M.12.0.000.09.16.09.0", 2), + ] { + let decoded = VideoCodec::from_str(encoded).expect("failed to parse"); + assert!(matches!(&decoded, VideoCodec::AV1(av1) if av1.profile == profile)); + assert_eq!(decoded.to_string(), encoded); + + let config = VideoConfig::new(decoded); + let json = serde_json::to_string(&config).unwrap(); + let output: VideoConfig = serde_json::from_str(&json).unwrap(); + assert_eq!(output, config); + } + } + + #[test] + fn test_av1_invalid_profile() { + for profile in 3..=9 { + assert!(VideoCodec::from_str(&format!("av01.{profile}.00M.08")).is_err()); + } + } } From 7711e52ff7df2bdb775988790b01cadf0b685534 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:18:03 -0700 Subject: [PATCH 07/11] fix(net): read each IETF uni stream's type in its own task (backport #5086) run_unis waited for each accepted uni stream's type before accepting the next. QUIC opens lower-numbered streams implicitly, so the stream it waited on could be one whose bytes sat behind connection flow control held by later streams' unread data, deadlocking the connection with any IETF peer. Each stream now reads its type and is served in its own child. A stream that breaks the session records the error in a shared slot, which ends the loop as before. Backports only the session.rs half of #5086; the resume.rs half fixes main-only code. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/session.rs | 211 ++++++++++++++++++++++----------- 1 file changed, 144 insertions(+), 67 deletions(-) diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 293345ace1..976363242f 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -721,10 +721,18 @@ struct UniSetup { } /// Accept incoming uni streams, including SETUP and GOAWAY on draft-17+. +/// +/// Each stream reads its own type in its own child. QUIC opens every lower-numbered +/// stream once a higher one arrives, so the next stream accepted can be one whose bytes +/// the peer has not sent yet, held back by connection flow control that the later +/// streams' unread data is using up. Waiting here for its type would deadlock the +/// connection. async fn run_unis(mut session: S, subscriber: Subscriber, setup: UniSetup) -> Result<(), Error> where S: crate::transport::poll::Boxable, { + use std::task::Poll; + let UniSetup { peer: peer_setup, read: setup_read, @@ -735,24 +743,42 @@ where } = setup; let outer_version = crate::Version::Ietf(version); let mut tasks = TaskSet::owned(); - // A gated server accept already read the peer's one SETUP off its own uni stream, - // so anything arriving here is a second one. - let mut seen_setup = setup_read; + let uni = Uni { + session: session.clone(), + subscriber: subscriber.clone(), + peer_setup, + // A gated server accept already read the peer's one SETUP off its own uni stream, + // so anything arriving here is a second one. + seen_setup: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(setup_read)), + goaway, + version, + }; + // The first stream to break the session, recorded by the child that read it. + let fatal = kio::Shared::>::default(); let mut early = early.into_iter(); loop { - let mut reader = match early.next() { + let reader = match early.next() { Some(reader) => reader, None => { let recv = tasks .drive(|waiter| { + let broken = fatal.poll(waiter, |fatal| match fatal.is_some() { + true => Poll::Ready(()), + false => Poll::Pending, + }); + if let Poll::Ready(mut fatal) = broken + && let Some(err) = fatal.take() + { + return Poll::Ready(Err(err)); + } let mut cx = std::task::Context::from_waker(waiter.waker()); - session.poll_accept_uni(&mut cx) + session.poll_accept_uni(&mut cx).map(Ok) }) .await; let recv = match recv { - Ok(recv) => recv, - Err(err) => { + Ok(Ok(recv)) => recv, + Ok(Err(err)) => { let err = Error::from_transport(err); // Settle tracks and open groups before the owned receive tasks drop // their cancellation guards, which would otherwise record Cancel. @@ -763,33 +789,75 @@ where } return Err(err); } + Err(fatal) => return Err(fatal), }; Reader::new(recv, outer_version) } }; + + let uni = uni.clone(); + let fatal = fatal.clone(); + tasks.push(async move { + if let Err(err) = uni.serve(reader).await { + fatal.lock().get_or_insert(err); + } + }); + } +} + +/// What serving one incoming uni stream needs from its session. +struct Uni { + session: S, + subscriber: Subscriber, + peer_setup: Option, + /// Exactly one SETUP per endpoint. + seen_setup: std::sync::Arc, + goaway: crate::goaway::Protocol, + version: Version, +} + +impl Clone for Uni { + fn clone(&self) -> Self { + Self { + session: self.session.clone(), + subscriber: self.subscriber.clone(), + peer_setup: self.peer_setup.clone(), + seen_setup: self.seen_setup.clone(), + goaway: self.goaway.clone(), + version: self.version, + } + } +} + +impl Uni { + /// Read the stream's type and serve it, failing only when the stream breaks the session. + async fn serve(self, mut reader: Reader) -> Result<(), Error> { + let Self { + mut session, + mut subscriber, + peer_setup, + seen_setup, + goaway, + version, + } = self; + // A stream that dies before its type varint is that stream's failure, not the // session's. RESET_STREAM is how a peer drops a group, and QUIC does not order // the reset behind the data, so one can beat the first byte even of a stream - // the peer wrote to. Failing the loop here would tear down the whole session - // over a single stream the peer had already given up on. Only death is - // tolerated: bytes that arrive and do not parse stay session-fatal. + // the peer wrote to. Failing here would tear down the whole session over a + // single stream the peer had already given up on. Only death is tolerated: + // bytes that arrive and do not parse stay session-fatal. // // A transport error counts as death too. A reset whose code the transport cannot // place in the stream registry surfaces as one: over raw QUIC a moq-transport // peer resets with its own codes (moxygen's CANCELLED is 0x1), which the // WebTransport code mapping rejects. If the connection itself died, the next // accept reports it. - let kind: u64 = match tasks - .drive(|waiter| { - let mut cx = std::task::Context::from_waker(waiter.waker()); - reader.poll_decode_peek(&mut cx) - }) - .await - { + let kind: u64 = match std::future::poll_fn(|cx| reader.poll_decode_peek(cx)).await { Ok(kind) => kind, Err(err) if died_before_header(&err) => { tracing::debug!(%err, "dropping uni stream that died before its type"); - continue; + return Ok(()); } Err(err) => return Err(err), }; @@ -800,78 +868,66 @@ where }; match ty { - // SETUP then becomes the GOAWAY channel. We accept it in the background - // without blocking; the one thing that does need it (the MoQ Cluster - // negotiation) waits on `peer_setup` instead, so a slow SETUP delays - // announcements rather than the whole session. + // SETUP then becomes the GOAWAY channel. It is read in the background like any + // other stream; the one thing that does need it (the MoQ Cluster negotiation) + // waits on `peer_setup` instead, so a slow SETUP delays announcements rather + // than the whole session. UniType::Setup => { // Exactly one SETUP per endpoint. A second would let a peer restate its // declared identity mid-session, silently re-attributing every route // already built from the first. - if std::mem::replace(&mut seen_setup, true) { + if seen_setup.swap(true, std::sync::atomic::Ordering::Relaxed) { return Err(Error::ProtocolViolation); } - let peer_setup = peer_setup.clone(); - let mut session = session.clone(); - let goaway = goaway.clone(); - tasks.push(async move { - // The negotiation gates the announce and dispatch loops, so a SETUP we - // cannot read must end the session rather than leave them parked on a - // slot nothing will ever fill. - let msg = match reader.decode::().await { - Ok(msg) => msg, + // The negotiation gates the announce and dispatch loops, so a SETUP we + // cannot read must end the session rather than leave them parked on a + // slot nothing will ever fill. + let msg = match reader.decode::().await { + Ok(msg) => msg, + Err(err) => { + tracing::warn!(%err, "setup decode error"); + session.close(SessionError::ProtocolViolation.to_code(), "invalid setup"); + return Ok(()); + } + }; + + if let Some(peer_setup) = peer_setup { + let peer = match decode_peer_setup(msg.parameters, version) { + Ok(peer) => peer, Err(err) => { - tracing::warn!(%err, "setup decode error"); - session.close(SessionError::ProtocolViolation.to_code(), "invalid setup"); - return; + tracing::warn!(%err, "setup parameter decode error"); + session.close(SessionError::ProtocolViolation.to_code(), "invalid setup parameters"); + return Ok(()); } }; + peer_setup.set(peer); + } - if let Some(peer_setup) = peer_setup { - let peer = match decode_peer_setup(msg.parameters, version) { - Ok(peer) => peer, - Err(err) => { - tracing::warn!(%err, "setup parameter decode error"); - session.close(SessionError::ProtocolViolation.to_code(), "invalid setup parameters"); - return; - } - }; - peer_setup.set(peer); - } - - // Monitor for GOAWAY after setup completes. - if let Err(err) = run_goaway(reader.with_version(version), version, goaway).await { - tracing::warn!(%err, "goaway error"); - } - }); + // Monitor for GOAWAY after setup completes. + if let Err(err) = run_goaway(reader.with_version(version), version, goaway).await { + tracing::warn!(%err, "goaway error"); + } } UniType::Subgroup => { - let mut sub = subscriber.clone(); - tasks.push(async move { - let mut reader = reader.with_version(version); - let res = sub.recv_group(&mut reader).await; - stop_on_error(&mut reader, res); - }); + let mut reader = reader.with_version(version); + let res = subscriber.recv_group(&mut reader).await; + stop_on_error(&mut reader, res); } // A fill fetch stream carries the head of the group a draft-20 subscription // joined part way through. One answering no fill of ours is refused inside. UniType::Fetch => { - let mut sub = subscriber.clone(); - tasks.push(async move { - let mut reader = reader.with_version(version); - let res = sub.recv_fill(&mut reader).await; - stop_on_error(&mut reader, res); - }); + let mut reader = reader.with_version(version); + let res = subscriber.recv_fill(&mut reader).await; + stop_on_error(&mut reader, res); } // The receiver MUST discard padding. We read it to the end rather than cancel, // so a peer probing for bandwidth gets the throughput it is measuring. UniType::Padding => { - tasks.push(async move { - while let Ok(Some(_)) = std::future::poll_fn(|cx| reader.poll_read_chunk(cx, usize::MAX)).await {} - }); + while let Ok(Some(_)) = std::future::poll_fn(|cx| reader.poll_read_chunk(cx, usize::MAX)).await {} } } + Ok(()) } } @@ -1543,6 +1599,27 @@ mod tests { assert_eq!(log.closes(), vec![], "one dropped group may not close the session"); } + /// A stream whose type has not arrived must not hold up the streams accepted after it. + /// + /// QUIC opens every lower-numbered stream when a higher one arrives, so the next stream + /// accepted can be one whose bytes the peer has not sent yet, held back by connection + /// flow control that the later streams' unread data is using up. Waiting on its type + /// before reading them deadlocks the connection. + #[tokio::test(start_paused = true)] + async fn a_silent_stream_does_not_hold_up_the_next() { + let header = subgroup_header(Version::Draft19, 7, 0).await; + let session = + crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_unis(vec![Vec::new(), header]); + let (log, result) = drive_unis(Version::Draft19, session, None, Some(7)).await; + + assert_eq!( + log.stops(), + vec![crate::ietf::error::CANCELLED], + "the stream behind the silent one was never read", + ); + assert!(result.is_none(), "the silent stream ended the session: {result:?}"); + } + /// A non-zero subgroup is refused on its own stream, never by closing the session. #[tokio::test(start_paused = true)] async fn a_non_zero_subgroup_is_stopped_without_closing_the_session() { From d073cc8cc47e6fdf1c89da9afa441be9e03fdfe4 Mon Sep 17 00:00:00 2001 From: bgreenway Date: Thu, 8 Oct 2026 00:05:23 -0500 Subject: [PATCH 08/11] fix(moq-mux): configure H.265 from the SPS head when scuffle_h265 refuses the rest (#4924) Co-authored-by: Luke Curley (cherry picked from commit d653d678f508177d26b766b431648ef7af84c873) Co-Authored-By: Claude Opus 5.5 --- rs/moq-gst/src/sink/pad.rs | 39 +++++ rs/moq-mux/src/codec/h265/import.rs | 32 ++-- rs/moq-mux/src/codec/h265/mod.rs | 98 ++++++++--- rs/moq-mux/src/codec/h265/sps.rs | 259 ++++++++++++++++++++++++++++ 4 files changed, 389 insertions(+), 39 deletions(-) create mode 100644 rs/moq-mux/src/codec/h265/sps.rs diff --git a/rs/moq-gst/src/sink/pad.rs b/rs/moq-gst/src/sink/pad.rs index 210bfce422..21931ee17a 100644 --- a/rs/moq-gst/src/sink/pad.rs +++ b/rs/moq-gst/src/sink/pad.rs @@ -1333,6 +1333,45 @@ mod tests { .unwrap(); } + // An H.265 camera whose SPS VUI zeroes the colour fields on 4:2:0 (`matrix_coeffs` 0, which + // ITU-T H.265 E.3.1 forbids there; a Viewtron IP-PTZ-440 sends it) still resolves its rendition + // instead of failing the pad with "h265: failed to parse SPS NAL unit". + #[test] + fn h265_with_zeroed_vui_colour_resolves_the_rendition() { + gst::init().unwrap(); + let (broadcast, catalog) = producers(); + let mut pad = Pad::new(); + let caps = gst::Caps::builder("video/x-h265") + .field("stream-format", "byte-stream") + .field("alignment", "au") + .build(); + pad.observe_caps(&broadcast, &catalog, producer_options(&caps, Some("camera"))); + pad.observe_segment(time_segment()); + let vps: &[u8] = &[ + 0x40, 0x01, 0x0c, 0x01, 0xff, 0xff, 0x01, 0x60, 0x00, 0x00, 0x03, 0x00, 0x90, 0x00, 0x00, 0x03, 0x00, 0x00, + 0x03, 0x00, 0x5d, 0x95, 0x98, 0x09, + ]; + let sps: &[u8] = &[ + 0x42, 0x01, 0x01, 0x01, 0x60, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, + 0x96, 0xa0, 0x03, 0xc0, 0x80, 0x11, 0x07, 0xcb, 0x8a, 0xad, 0x3b, 0xa2, 0x4b, 0xb9, 0x08, 0x00, 0x00, 0x03, + 0x00, 0x20, 0x05, 0x26, 0x5c, 0x00, 0x33, 0x7f, 0x98, 0x01, + ]; + let pps: &[u8] = &[0x44, 0x01, 0xc1, 0x72, 0xb4, 0x62, 0x40]; + let idr: &[u8] = &[0x26, 0x01, 0x80, 0xaa]; + let mut au = Vec::new(); + for nal in [vps, sps, pps, idr] { + au.extend_from_slice(&[0, 0, 0, 1]); + au.extend_from_slice(nal); + } + let outcome = pad + .push_buffer(Bytes::from(au), Some(gst::ClockTime::ZERO), None, None, Instant::now()) + .unwrap(); + assert_eq!(outcome, PushOutcome::Published); + assert!(!pad.is_failed()); + let config = catalog.snapshot().video.renditions.get("camera").cloned().unwrap(); + assert_eq!((config.coded_width, config.coded_height), (Some(1920), Some(1080))); + } + // A real IDR AU emits a frame to the published track (not just a rendition off the SPS). #[tokio::test] async fn frame_through_h264_emits_a_frame() { diff --git a/rs/moq-mux/src/codec/h265/import.rs b/rs/moq-mux/src/codec/h265/import.rs index 13ff53a150..0cc85badeb 100644 --- a/rs/moq-mux/src/codec/h265/import.rs +++ b/rs/moq-mux/src/codec/h265/import.rs @@ -15,7 +15,6 @@ //! that whoever owns the split drives via [`decode`](Import::decode). use bytes::Bytes; -use scuffle_h265::SpsNALUnit; use super::{Error, split::nal_unit_type}; use crate::Result; @@ -263,22 +262,29 @@ pub fn config(buf: &[u8]) -> Result { /// The config an inline SPS describes (`in_band: true`, the hev1 shape). No `description`: the /// parameter sets ride with every keyframe. fn config_from_sps(sps_nal: &[u8]) -> Result { - let sps = SpsNALUnit::parse(&mut &sps_nal[..]).map_err(|_| Error::SpsParse)?; - let profile = &sps.rbsp.profile_tier_level.general_profile; - let vui_data = sps.rbsp.vui_parameters.as_ref().map(VuiData::new).unwrap_or_default(); + let super::Sps { head, full } = super::Sps::parse(sps_nal)?; + // The VUI (frame rate, aspect ratio) only comes with scuffle_h265's full parse. + let vui_data = full + .as_ref() + .and_then(|sps| sps.vui_parameters.as_ref()) + .map(VuiData::new) + .unwrap_or_default(); let mut config = hang::catalog::VideoConfig::new(hang::catalog::H265 { in_band: true, // An inline SPS is the hev1 shape; hvc1 configs come from `config_from_hvcc`. - profile_space: profile.profile_space, - profile_idc: profile.profile_idc, - profile_compatibility_flags: profile.profile_compatibility_flag.bits().to_be_bytes(), - tier_flag: profile.tier_flag, - level_idc: profile.level_idc.ok_or(Error::MissingLevelIdc)?, - constraint_flags: super::pack_constraint_flags(profile), + profile_space: head.profile_space, + profile_idc: head.profile_idc, + profile_compatibility_flags: head.profile_compatibility_flags, + tier_flag: head.tier_flag, + level_idc: head.level_idc, + constraint_flags: head.constraint_flags, }); - config.coded_width = Some(sps.rbsp.cropped_width() as u32); - config.coded_height = Some(sps.rbsp.cropped_height() as u32); - config.framerate = super::sps_period(&sps.rbsp).map(|(units, scale)| scale as f64 / units as f64); + config.coded_width = Some(head.width); + config.coded_height = Some(head.height); + config.framerate = full + .as_ref() + .and_then(super::sps_period) + .map(|(units, scale)| scale as f64 / units as f64); config.display_aspect_width = vui_data.display_ratio_width; config.display_aspect_height = vui_data.display_ratio_height; Ok(config) diff --git a/rs/moq-mux/src/codec/h265/mod.rs b/rs/moq-mux/src/codec/h265/mod.rs index d0f420e274..9b69657729 100644 --- a/rs/moq-mux/src/codec/h265/mod.rs +++ b/rs/moq-mux/src/codec/h265/mod.rs @@ -10,6 +10,7 @@ mod export; mod import; mod split; +mod sps; pub use export::*; pub use import::*; @@ -17,6 +18,7 @@ pub use split::*; use bytes::{Buf, BufMut, Bytes, BytesMut}; use scuffle_h265::{NALUnitType, SpsNALUnit}; +use sps::Sps; /// Wrap one hvc1 (length-prefixed NALU) access unit as a single /// [`Frame`](crate::container::Frame), with the keyframe flag set when it @@ -197,20 +199,19 @@ impl Hvcc { pub(crate) fn config_from_hvcc(hvcc: &[u8]) -> Result { let params = Hvcc::parse(hvcc)?; let sps_nal = params.sps.first().ok_or(Error::MissingSps)?; - let sps = SpsNALUnit::parse(&mut &sps_nal[..]).map_err(|_| Error::SpsParse)?; - let profile = &sps.rbsp.profile_tier_level.general_profile; + let sps = Sps::parse(sps_nal)?.head; let mut config = hang::catalog::VideoConfig::new(hang::catalog::H265 { in_band: false, - profile_space: profile.profile_space, - profile_idc: profile.profile_idc, - profile_compatibility_flags: profile.profile_compatibility_flag.bits().to_be_bytes(), - tier_flag: profile.tier_flag, - level_idc: profile.level_idc.ok_or(Error::MissingLevelIdc)?, - constraint_flags: pack_constraint_flags(profile), + profile_space: sps.profile_space, + profile_idc: sps.profile_idc, + profile_compatibility_flags: sps.profile_compatibility_flags, + tier_flag: sps.tier_flag, + level_idc: sps.level_idc, + constraint_flags: sps.constraint_flags, }); - config.coded_width = Some(sps.rbsp.cropped_width() as u32); - config.coded_height = Some(sps.rbsp.cropped_height() as u32); + config.coded_width = Some(sps.width); + config.coded_height = Some(sps.height); config.description = Some(Bytes::copy_from_slice(hvcc)); Ok(config) } @@ -410,12 +411,8 @@ pub(crate) fn build_hvcc(vps_nals: &[Bytes], sps_nals: &[Bytes], pps_nals: &[Byt } } - let sps = SpsNALUnit::parse(&mut &first_sps[..]).map_err(|_| Error::SpsParse)?; - let profile = &sps.rbsp.profile_tier_level.general_profile; - let level_idc = profile.level_idc.ok_or(Error::MissingLevelIdc)?; - let constraint_flags = pack_constraint_flags(profile); - let compat = profile.profile_compatibility_flag.bits().to_be_bytes(); - let num_temporal_layers = sps.rbsp.sps_max_sub_layers_minus1 + 1; + let sps = Sps::parse(first_sps)?.head; + let num_temporal_layers = sps.max_sub_layers_minus1 + 1; let params_len: usize = vps_nals .iter() @@ -425,17 +422,17 @@ pub(crate) fn build_hvcc(vps_nals: &[Bytes], sps_nals: &[Bytes], pps_nals: &[Byt .sum(); let mut out = BytesMut::with_capacity(23 + 3 * 3 + params_len); out.put_u8(1); // configurationVersion - out.put_u8(((profile.profile_space & 0x3) << 6) | ((profile.tier_flag as u8) << 5) | (profile.profile_idc & 0x1f)); - out.put_slice(&compat); - out.put_slice(&constraint_flags); - out.put_u8(level_idc); + out.put_u8(((sps.profile_space & 0x3) << 6) | ((sps.tier_flag as u8) << 5) | (sps.profile_idc & 0x1f)); + out.put_slice(&sps.profile_compatibility_flags); + out.put_slice(&sps.constraint_flags); + out.put_u8(sps.level_idc); out.put_u16(0xf000); // min_spatial_segmentation_idc unknown out.put_u8(0xfc); // parallelismType mixed - out.put_u8(0xfc | (sps.rbsp.chroma_format_idc & 0x3)); - out.put_u8(0xf8 | (sps.rbsp.bit_depth_luma_minus8 & 0x7)); - out.put_u8(0xf8 | (sps.rbsp.bit_depth_chroma_minus8 & 0x7)); + out.put_u8(0xfc | (sps.chroma_format_idc & 0x3)); + out.put_u8(0xf8 | (sps.bit_depth_luma_minus8 & 0x7)); + out.put_u8(0xf8 | (sps.bit_depth_chroma_minus8 & 0x7)); out.put_u16(0); // avgFrameRate unspecified - out.put_u8(((num_temporal_layers & 0x7) << 3) | ((sps.rbsp.sps_temporal_id_nesting_flag as u8) << 2) | 0x3); + out.put_u8(((num_temporal_layers & 0x7) << 3) | ((sps.temporal_id_nesting_flag as u8) << 2) | 0x3); out.put_u8(3); // numOfArrays (VPS, SPS, PPS) for (nal_type, nals) in [ @@ -494,8 +491,8 @@ pub(crate) fn pack_constraint_flags(profile: &scuffle_h265::Profile) -> [u8; 6] flags } -/// Real parameter sets from a single-frame x265 encode (1280x720, Main profile), -/// for tests that need an SPS that scuffle_h265 can actually parse. +/// Real parameter sets, mostly from a single-frame x265 encode (1280x720, Main profile), +/// for tests that need an SPS rather than hand-built bytes. #[cfg(test)] pub(crate) mod fixtures { use bytes::Bytes; @@ -518,6 +515,16 @@ pub(crate) mod fixtures { 0x00, 0x00, 0x0c, 0xb0, 0x0a, 0x48, 0x2f, 0x00, 0x07, 0xa1, 0x20, 0x00, 0xf4, 0x24, 0x40, ]; + /// SPS from a Viewtron IP-PTZ-440 (1920x1080 Main@L5.0, 10 fps). Its VUI sets + /// `colour_description_present_flag` with `colour_primaries`, `transfer_characteristics` + /// and `matrix_coeffs` all 0 on a 4:2:0 stream. ITU-T H.265 E.3.1 forbids `matrix_coeffs` + /// 0 there, so scuffle_h265 refuses the SPS; libavcodec and GStreamer accept it. + pub(crate) const SPS_ZERO_COLOUR: &[u8] = &[ + 0x42, 0x01, 0x01, 0x01, 0x60, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, + 0x96, 0xa0, 0x03, 0xc0, 0x80, 0x11, 0x07, 0xcb, 0x8a, 0xad, 0x3b, 0xa2, 0x4b, 0xb9, 0x08, 0x00, 0x00, 0x03, + 0x00, 0x20, 0x05, 0x26, 0x5c, 0x00, 0x33, 0x7f, 0x98, 0x01, + ]; + /// An hvcC record built from the real parameter sets (4-byte NALU lengths). pub(crate) fn hvcc() -> Bytes { super::build_hvcc( @@ -649,6 +656,45 @@ mod tests { assert_eq!(params[2].as_ref(), pps); } + /// hev1: an inline SPS whose VUI carries zeroed colour fields on 4:2:0 still configures the + /// rendition from the fields before the VUI. + #[test] + fn hev1_config_tolerates_zeroed_colour_description() { + let mut annexb = Vec::new(); + for nal in [fixtures::VPS, fixtures::SPS_ZERO_COLOUR, fixtures::PPS] { + annexb.extend_from_slice(&[0, 0, 0, 1]); + annexb.extend_from_slice(nal); + } + let config = config(&annexb).expect("the SPS head is well-formed"); + let hang::catalog::VideoCodec::H265(h265) = &config.codec else { + panic!("expected H.265 codec") + }; + assert!(h265.in_band); + assert_eq!(h265.to_string(), "hev1.1.6.L150"); + assert_eq!((config.coded_width, config.coded_height), (Some(1920), Some(1080))); + } + + /// hvc1: the same SPS builds an hvcC and resolves an out-of-band config. The record header + /// matches, byte for byte, the one GStreamer's h265parse builds from this SPS. + #[test] + fn hvcc_tolerates_zeroed_colour_description() { + let hvcc = build_hvcc( + &[Bytes::from_static(fixtures::VPS)], + &[Bytes::from_static(fixtures::SPS_ZERO_COLOUR)], + &[Bytes::from_static(fixtures::PPS)], + ) + .expect("the SPS head is well-formed"); + assert_eq!( + &hvcc[..23], + &[ + 0x01, 0x01, 0x60, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x96, 0xf0, 0x00, 0xfc, 0xfd, + 0xf8, 0xf8, 0x00, 0x00, 0x0f, 0x03, + ] + ); + let config = config_from_hvcc(&hvcc).unwrap(); + assert_eq!((config.coded_width, config.coded_height), (Some(1920), Some(1080))); + } + /// The SPS and catalog paths keep `profile_compatibility_flags` as raw /// big-endian bytes, including across codec-string formatting and parsing. #[test] diff --git a/rs/moq-mux/src/codec/h265/sps.rs b/rs/moq-mux/src/codec/h265/sps.rs new file mode 100644 index 0000000000..9e2c15e47b --- /dev/null +++ b/rs/moq-mux/src/codec/h265/sps.rs @@ -0,0 +1,259 @@ +//! The SPS fields moq-mux reads, with a fallback for an SPS that scuffle_h265 refuses. +//! +//! scuffle_h265 turns semantic constraints deep in the SPS (VUI colour fields, HRD, extensions) +//! into parse errors, and real encoders break some of them: cameras ship `matrix_coeffs` 0 on +//! 4:2:0, which ITU-T H.265 E.3.1 forbids and libavcodec and GStreamer decode anyway. The codec +//! string, coded size and hvcC header only need the SPS head, which ends at +//! `bit_depth_chroma_minus8`. So when scuffle_h265 refuses a value (`InvalidData`), [`Sps::parse`] +//! reads the head itself. A truncated SPS, or one whose head is invalid, is still an error. + +use h264_parser::{bitreader::BitReader, eg::read_ue}; +use scuffle_h265::{NALUnitType, SpsNALUnit}; + +use super::{Error, Result}; + +/// The SPS fields up to `bit_depth_chroma_minus8` (ITU-T H.265 7.3.2.2.1). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct SpsHead { + pub profile_space: u8, + pub tier_flag: bool, + pub profile_idc: u8, + pub profile_compatibility_flags: [u8; 4], + /// As [`pack_constraint_flags`](super::pack_constraint_flags) packs them. + pub constraint_flags: [u8; 6], + pub level_idc: u8, + pub max_sub_layers_minus1: u8, + pub temporal_id_nesting_flag: bool, + pub chroma_format_idc: u8, + pub bit_depth_luma_minus8: u8, + pub bit_depth_chroma_minus8: u8, + /// The picture size inside the conformance window. + pub width: u32, + pub height: u32, +} + +/// A parsed SPS: its head, plus scuffle_h265's full parse (with the VUI) when it accepted the SPS. +pub(crate) struct Sps { + pub head: SpsHead, + pub full: Option, +} + +impl Sps { + /// Parse an SPS NAL unit, keeping only its head when scuffle_h265 refuses a later value. + pub(crate) fn parse(nal: &[u8]) -> Result { + match SpsNALUnit::parse(&mut &nal[..]) { + Ok(sps) => Ok(Self { + head: SpsHead::from_rbsp(&sps.rbsp)?, + full: Some(sps.rbsp), + }), + // A value scuffle_h265 refused, not a bitstream that ran out. + Err(err) if err.kind() == std::io::ErrorKind::InvalidData => { + let head = SpsHead::parse(nal).ok_or(Error::SpsParse)?; + tracing::warn!(%err, "H.265 SPS refused by scuffle_h265; using its head, without the VUI"); + Ok(Self { head, full: None }) + } + Err(_) => Err(Error::SpsParse), + } + } +} + +impl SpsHead { + fn from_rbsp(rbsp: &scuffle_h265::SpsRbsp) -> Result { + let profile = &rbsp.profile_tier_level.general_profile; + Ok(Self { + profile_space: profile.profile_space, + tier_flag: profile.tier_flag, + profile_idc: profile.profile_idc, + profile_compatibility_flags: profile.profile_compatibility_flag.bits().to_be_bytes(), + constraint_flags: super::pack_constraint_flags(profile), + level_idc: profile.level_idc.ok_or(Error::MissingLevelIdc)?, + max_sub_layers_minus1: rbsp.sps_max_sub_layers_minus1, + temporal_id_nesting_flag: rbsp.sps_temporal_id_nesting_flag, + chroma_format_idc: rbsp.chroma_format_idc, + bit_depth_luma_minus8: rbsp.bit_depth_luma_minus8, + bit_depth_chroma_minus8: rbsp.bit_depth_chroma_minus8, + width: rbsp.cropped_width() as u32, + height: rbsp.cropped_height() as u32, + }) + } + + /// Read the head of an SPS NAL unit (header included). `None` when the NAL is not an SPS, ends + /// before `bit_depth_chroma_minus8`, or breaks a constraint on the head. + fn parse(nal: &[u8]) -> Option { + let [header, header2, payload @ ..] = nal else { + return None; + }; + // forbidden_zero_bit, nal_unit_type, and TemporalId 0 for an SPS (7.4.2.2). + if header & 0x80 != 0 || super::split::nal_unit_type(*header) != NALUnitType::SpsNut || header2 & 0x07 != 1 { + return None; + } + let rbsp = h264_parser::nal::ebsp_to_rbsp(payload); + Self::read(&mut BitReader::new(&rbsp)).ok()? + } + + /// Read the head from the SPS RBSP: `Ok(None)` for an out-of-range value. + fn read(r: &mut BitReader<'_>) -> h264_parser::Result> { + r.skip_bits(4)?; // sps_video_parameter_set_id + let max_sub_layers_minus1 = r.read_bits(3)? as u8; + let temporal_id_nesting_flag = r.read_flag()?; + // 7.4.3.2.1: at most 7 sub-layers, and a single sub-layer is nested. + if max_sub_layers_minus1 > 6 || (max_sub_layers_minus1 == 0 && !temporal_id_nesting_flag) { + return Ok(None); + } + + // profile_tier_level(1, sps_max_sub_layers_minus1), 7.3.3 + let profile_space = r.read_bits(2)? as u8; + let tier_flag = r.read_flag()?; + let profile_idc = r.read_bits(5)? as u8; + let profile_compatibility_flags = r.read_bits(32)?.to_be_bytes(); + // progressive_source, interlaced_source, non_packed_constraint and frame_only_constraint. + let constraint_flags = [(r.read_bits(4)? as u8) << 4, 0, 0, 0, 0, 0]; + r.skip_bits(44)?; // the other 43 constraint and reserved bits, and general_inbld_flag + let level_idc = r.read_u8()?; + let sub_layers = usize::from(max_sub_layers_minus1); + let mut present = [(false, false); 6]; + for flags in &mut present[..sub_layers] { + *flags = (r.read_flag()?, r.read_flag()?); + } + if sub_layers > 0 { + r.skip_bits(2 * (8 - u32::from(max_sub_layers_minus1)))?; // reserved_zero_2bits + } + for &(profile, level) in &present[..sub_layers] { + // A sub-layer profile is the same 88 bits as the general one; sub_layer_level_idc is 8. + r.skip_bits(88 * u32::from(profile) + 8 * u32::from(level))?; + } + + let sps_seq_parameter_set_id = read_ue(r)?; + let chroma_format_idc = read_ue(r)?; + if sps_seq_parameter_set_id > 15 || chroma_format_idc > 3 { + return Ok(None); + } + if chroma_format_idc == 3 { + r.skip_bits(1)?; // separate_colour_plane_flag + } + let pic_width = read_ue(r)?; + let pic_height = read_ue(r)?; + // conformance_window_flag, then the left, right, top and bottom offsets. + let [left, right, top, bottom] = if r.read_flag()? { + [read_ue(r)?, read_ue(r)?, read_ue(r)?, read_ue(r)?] + } else { + [0; 4] + }; + let bit_depth_luma_minus8 = read_ue(r)?; + let bit_depth_chroma_minus8 = read_ue(r)?; + if bit_depth_luma_minus8 > 8 || bit_depth_chroma_minus8 > 8 { + return Ok(None); + } + + // The window offsets count chroma samples: SubWidthC and SubHeightC from Table 6-1. + let sub_width_c = if matches!(chroma_format_idc, 1 | 2) { 2 } else { 1 }; + let sub_height_c = if chroma_format_idc == 1 { 2 } else { 1 }; + let width = pic_width.checked_sub(left.saturating_add(right).saturating_mul(sub_width_c)); + let height = pic_height.checked_sub(top.saturating_add(bottom).saturating_mul(sub_height_c)); + let (Some(width @ 1..), Some(height @ 1..)) = (width, height) else { + return Ok(None); + }; + + Ok(Some(Self { + profile_space, + tier_flag, + profile_idc, + profile_compatibility_flags, + constraint_flags, + level_idc, + max_sub_layers_minus1, + temporal_id_nesting_flag, + chroma_format_idc: chroma_format_idc as u8, + bit_depth_luma_minus8: bit_depth_luma_minus8 as u8, + bit_depth_chroma_minus8: bit_depth_chroma_minus8 as u8, + width, + height, + })) + } +} + +#[cfg(test)] +mod tests { + use super::super::fixtures; + use super::*; + + /// The x265 fixture with `sps_max_sub_layers_minus1 = 2`: sub-layer 0 signals its profile, + /// sub-layer 1 its level, so the head walks both sub-layer branches of 7.3.3. + const SPS_SUB_LAYERS: &[u8] = &[ + 0x42, 0x01, 0x05, 0x01, 0x60, 0x00, 0x00, 0x03, 0x00, 0x90, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x5d, + 0x90, 0x00, 0x01, 0x60, 0x00, 0x00, 0x03, 0x00, 0x90, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x5d, 0xa0, + 0x02, 0x80, 0x80, 0x2d, 0x16, 0x51, 0x59, 0xa4, 0x93, 0x2b, 0xc0, 0x5a, 0x02, 0x00, 0x00, 0x03, 0x00, 0x02, + 0x00, 0x00, 0x03, 0x00, 0x3c, 0x10, + ]; + + /// As [`SPS_SUB_LAYERS`], but sub-layer 0 signals both its profile and its level. + const SPS_SUB_LAYER_PROFILE_AND_LEVEL: &[u8] = &[ + 0x42, 0x01, 0x05, 0x01, 0x60, 0x00, 0x00, 0x03, 0x00, 0x90, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x5d, + 0xd0, 0x00, 0x01, 0x60, 0x00, 0x00, 0x03, 0x00, 0x90, 0x00, 0x00, 0x03, 0x00, 0x00, 0x03, 0x00, 0x5a, 0x5d, + 0xa0, 0x02, 0x80, 0x80, 0x2d, 0x16, 0x51, 0x59, 0xa4, 0x93, 0x2b, 0xc0, 0x5a, 0x02, 0x00, 0x00, 0x03, 0x00, + 0x02, 0x00, 0x00, 0x03, 0x00, 0x3c, 0x10, + ]; + + /// The head reader agrees with scuffle_h265 on every SPS scuffle_h265 accepts. + #[test] + fn head_matches_the_full_parse() { + for nal in [fixtures::SPS, fixtures::SPS_PYRAMID, SPS_SUB_LAYERS] { + let sps = Sps::parse(nal).unwrap(); + assert!(sps.full.is_some(), "scuffle_h265 accepts this SPS"); + assert_eq!(SpsHead::parse(nal), Some(sps.head)); + } + assert_eq!(SpsHead::parse(SPS_SUB_LAYERS).unwrap().max_sub_layers_minus1, 2); + } + + /// scuffle_h265 0.2.2 reads that `sub_layer_level_idc` twice and refuses the SPS; the head + /// reads it as GStreamer's parser does. + #[test] + fn sub_layer_profile_and_level() { + let expected = SpsHead { + max_sub_layers_minus1: 2, + ..SpsHead::parse(fixtures::SPS).unwrap() + }; + assert_eq!(Sps::parse(SPS_SUB_LAYER_PROFILE_AND_LEVEL).unwrap().head, expected); + } + + /// Zeroed VUI colour fields on 4:2:0: scuffle_h265 refuses the SPS, the head carries on. + #[test] + fn zeroed_colour_description_falls_back_to_the_head() { + let sps = Sps::parse(fixtures::SPS_ZERO_COLOUR).unwrap(); + assert!(sps.full.is_none(), "scuffle_h265 refuses this SPS"); + assert_eq!( + sps.head, + SpsHead { + profile_space: 0, + tier_flag: false, + profile_idc: 1, + profile_compatibility_flags: [0x60, 0, 0, 0], + constraint_flags: [0; 6], + level_idc: 150, + max_sub_layers_minus1: 0, + temporal_id_nesting_flag: true, + chroma_format_idc: 1, + bit_depth_luma_minus8: 0, + bit_depth_chroma_minus8: 0, + width: 1920, + height: 1080, + } + ); + } + + /// The fallback covers a refusal after the head, not a truncated or invalid head. + #[test] + fn truncated_or_invalid_head_is_still_an_error() { + assert!(Sps::parse(&fixtures::SPS[..8]).is_err()); + assert!(Sps::parse(&fixtures::SPS_ZERO_COLOUR[..20]).is_err()); + assert!(Sps::parse(&fixtures::SPS[..fixtures::SPS.len() - 6]).is_err()); + // TemporalId 1, then a single sub-layer without temporal nesting. + for (index, byte) in [(1, 0x02), (2, 0x00)] { + let mut nal = fixtures::SPS_ZERO_COLOUR.to_vec(); + nal[index] = byte; + assert!(Sps::parse(&nal).is_err()); + } + assert_eq!(SpsHead::parse(fixtures::PPS), None); + assert_eq!(SpsHead::parse(fixtures::VPS), None); + } +} From 8e3b23261c039fae9ba8f2d49fb9cd886e6df63b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:26:42 -0700 Subject: [PATCH 09/11] fix(net): keep IETF subscriptions alive after a request FIN (backport #4658) On drafts 19-22 a requester may FIN its request stream right after SUBSCRIBE (draft-21 section 6.4.2.2). The publisher treated that FIN as a cancellation, so no media flowed. On draft 17+ any REQUEST_UPDATE byte also made Reader::poll_closed fail and silently ended the subscription. Port the subscription half of #4658 to release: - From draft 19, a clean FIN on a SUBSCRIBE or SUBSCRIBE_NAMESPACE stream stops reading but keeps serving; RESET_STREAM or STOP_SENDING cancels. Drafts 14-18 keep FIN cancellation. - From draft 17, a subscription REQUEST_UPDATE is parsed: a priority change is applied and acknowledged with REQUEST_OK, anything else is refused with NOT_SUPPORTED and ends the subscription with PUBLISH_DONE UPDATE_FAILED. - moq-tokio's wrapped send stream observes STOP_SENDING while idle, which the Rust subscriber's FIN plus STOP_SENDING unsubscribe needs once a FIN no longer cancels. Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 7 + rs/moq-net/src/ietf/mod.rs | 1 + rs/moq-net/src/ietf/publish.rs | 3 + rs/moq-net/src/ietf/publisher.rs | 293 +++++++++++++++++++++++--- rs/moq-net/src/ietf/request_stream.rs | 105 +++++++++ rs/moq-tokio/src/transport.rs | 81 +++---- 6 files changed, 419 insertions(+), 71 deletions(-) create mode 100644 rs/moq-net/src/ietf/request_stream.rs diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 7da4a25b11..37a959a3ac 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -26,6 +26,13 @@ maps everything else to "not supported" or a harmless equivalent. The [moq-lite page](/concept/moq-lite#what-moq-lite-leaves-out) lists the differences. +On draft 19 and later, a requester can FIN its request stream while a Rust +publisher keeps its `SUBSCRIBE` or `SUBSCRIBE_NAMESPACE` active; cancellation +uses `RESET_STREAM` or `STOP_SENDING`. From draft 17, a subscription +`REQUEST_UPDATE` can change subscriber priority; other changes are refused with +`NOT_SUPPORTED` and end the subscription with `UPDATE_FAILED`. Drafts 14 to 18 +retain FIN cancellation. + Rust and JavaScript subscribers accept object extension blocks up to 64 KiB. This is an implementation limit, not a limit in the IETF draft. A larger declared block stops its subgroup stream with `MALFORMED_TRACK` before reading diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index 944c277893..5968bef179 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -26,6 +26,7 @@ mod publish; mod publish_namespace; mod publisher; mod request; +mod request_stream; mod session; pub mod solicit; mod subscribe; diff --git a/rs/moq-net/src/ietf/publish.rs b/rs/moq-net/src/ietf/publish.rs index 706fa5a16f..379dbb7bd1 100644 --- a/rs/moq-net/src/ietf/publish.rs +++ b/rs/moq-net/src/ietf/publish.rs @@ -128,6 +128,8 @@ pub(crate) enum PublishDoneStatus { Unauthorized, /// The track is no longer being published. TrackEnded, + /// A requested subscription update could not be applied. + UpdateFailed, } impl PublishDoneStatus { @@ -148,6 +150,7 @@ impl PublishDoneStatus { Self::InternalError => 0x0, Self::Unauthorized => 0x1, Self::TrackEnded => 0x2, + Self::UpdateFailed => 0x8, }, } } diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index d42749c8ea..90cb597812 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -256,13 +256,18 @@ impl Target { } } - /// Ready when the peer ends this loop by closing the stream it asked on. + /// Ready when the peer ends this loop by cancelling the stream it asked on. /// /// An unsolicited loop has no stream of its own to watch and parks here: the /// session driver polling it is what drops it when the session ends. - fn poll_closed(&mut self, cx: &mut std::task::Context<'_>) -> Poll> { + fn poll_closed( + &mut self, + finished: &mut bool, + version: Version, + cx: &mut std::task::Context<'_>, + ) -> Poll> { match self.stream() { - Some(stream) => stream.reader.poll_closed(cx), + Some(stream) => super::request_stream::poll_cancel(stream, finished, version, cx), None => Poll::Pending, } } @@ -696,37 +701,22 @@ where }) .await?; - // Run the track, cancelling on reader close (Unsubscribe or stream close). + let mut request_finished = false; + + // Serve the track while reading updates; on draft-19+ only abrupt closure cancels. // The fill (when one was requested) runs alongside on its own fetch stream; // its failures reset that stream and never touch the subscription. let mut track_serve = TrackServe::new(self.session.clone(), track, request_id, self.version, range, timescale); - let served = { - let serve = async { - match fill { - Some((fill, cache, timescale)) => { - let fill = self.run_fill(request_id, priority, fill, cache, timescale); - let track = kio::wait(|waiter| track_serve.poll(waiter)); - let (res, filled) = futures::join!(track, fill); - (res, filled) - } - None => (kio::wait(|waiter| track_serve.poll(waiter)).await, false), - } - }; - let mut serve = std::pin::pin!(serve); - let mut closed_session = self.session.clone(); - kio::wait(|waiter| { - if let Poll::Ready(served) = waiter.poll_future(serve.as_mut()) { - return Poll::Ready(Some(served)); - } - let mut cx = std::task::Context::from_waker(waiter.waker()); - if stream.reader.poll_closed(&mut cx).is_ready() || closed_session.poll_closed(&mut cx).is_ready() { - return Poll::Ready(None); - } - Poll::Pending - }) - .await + let fill = async { + match fill { + Some((fill, cache, timescale)) => self.run_fill(request_id, priority, fill, cache, timescale).await, + None => false, + } }; + let served = self + .run_subscription(&mut stream, &mut track_serve, &mut request_finished, fill) + .await; let completed = served.is_some(); let (res, filled) = served.unwrap_or((Ok(()), false)); @@ -746,7 +736,9 @@ where return Poll::Ready(res); } let mut cx = std::task::Context::from_waker(waiter.waker()); - if stream.reader.poll_closed(&mut cx).is_ready() || closed_session.poll_closed(&mut cx).is_ready() { + if super::request_stream::poll_cancel(&mut stream, &mut request_finished, self.version, &mut cx) + .is_ready() || closed_session.poll_closed(&mut cx).is_ready() + { return Poll::Ready(Err(Error::Cancel)); } Poll::Pending @@ -765,6 +757,7 @@ where // Send PublishDone let (status, reason) = match &res { Ok(()) => (ietf::PublishDoneStatus::TrackEnded, "track ended"), + Err(Error::Unsupported) => (ietf::PublishDoneStatus::UpdateFailed, "update failed"), Err(_) => (ietf::PublishDoneStatus::InternalError, "internal error"), }; let _ = stream.writer.encode(&ietf::PublishDone::ID).await; @@ -788,6 +781,107 @@ where } } + /// Serve a subscription until the track and any fill finish, applying the requester's + /// updates. `None` when the requester or session cancelled it. + /// + /// Drafts 14-16 deliver follow-ups on a virtual stream, where anything still ends the + /// subscription. From draft 17 a REQUEST_UPDATE is read and answered: a priority + /// change is applied, and anything else is refused and ends the subscription with + /// UPDATE_FAILED rather than being dropped silently. + async fn run_subscription( + &self, + stream: &mut Stream, + serve: &mut TrackServe, + finished: &mut bool, + fill: impl std::future::Future, + ) -> Option<(Result<(), Error>, bool)> { + let legacy = matches!(self.version, Version::Draft14 | Version::Draft15 | Version::Draft16); + let mut session = self.session.clone(); + let mut fill = std::pin::pin!(fill); + let mut filled = None; + let mut served = None; + loop { + let event = kio::wait(|waiter| { + let mut cx = std::task::Context::from_waker(waiter.waker()); + if session.poll_closed(&mut cx).is_ready() { + return Poll::Ready(Err(Error::Cancel)); + } + // Virtual writers on drafts 14-16 are already closed while the request lives. + if !legacy && stream.writer.poll_closed(&mut cx).is_ready() { + return Poll::Ready(Err(Error::Cancel)); + } + if !*finished { + if legacy { + if stream.reader.poll_closed(&mut cx).is_ready() { + return Poll::Ready(Err(Error::Cancel)); + } + } else { + match stream + .reader + .poll_decode_maybe::(&mut cx) + { + Poll::Ready(Ok(Some(update))) => return Poll::Ready(Ok(Some(update))), + Poll::Ready(Ok(None)) if !super::request_stream::fin_cancels(self.version) => { + *finished = true + } + Poll::Ready(Ok(None)) => return Poll::Ready(Err(Error::Cancel)), + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Pending => {} + } + } + } + if filled.is_none() + && let Poll::Ready(done) = waiter.poll_future(fill.as_mut()) + { + filled = Some(done); + } + if served.is_none() + && let Poll::Ready(done) = serve.poll(waiter) + { + served = Some(done); + } + match (&served, filled) { + (Some(Ok(())), Some(_)) => Poll::Ready(Ok(None)), + (Some(Err(err)), Some(_)) => Poll::Ready(Err(err.clone())), + _ => Poll::Pending, + } + }) + .await; + let filled = filled.unwrap_or(false); + let update = match event { + Ok(None) => return Some((Ok(()), filled)), + Err(Error::Cancel) => return None, + Err(err) => return Some((Err(err), filled)), + Ok(Some(update)) => update, + }; + if update.unsupported { + let result = self + .write_subscribe_error( + &mut stream.writer, + serve.request_id, + &Error::Unsupported, + "REQUEST_UPDATE parameters not supported", + ) + .await; + return Some((result.and(Err(Error::Unsupported)), filled)); + } + if let Some(priority) = update.priority { + let mut subscription = serve.track.subscription(); + subscription.priority = super::priority::from_wire(priority); + if let Err(err) = serve.track.update(subscription) { + return Some((Err(err), filled)); + } + } + let acked = async { + stream.writer.encode(&ietf::RequestOk::ID).await?; + stream.writer.encode(&ietf::RequestOk { request_id: None }).await + }; + if let Err(err) = acked.await { + return Some((Err(err), filled)); + } + } + } + /// Reject a SUBSCRIBE, ending the request stream. /// /// Takes the whole stream because delivering the error is the other half of the job: @@ -2080,9 +2174,10 @@ where let mut retry = crate::runtime::Deadline::new(&self.runtime); let mut retry_at: Option = None; let mut retry_delay = RETRY_BASE; + let mut finished = false; - // Stream updates (origin route (un)announces), bailing if the peer closes - // its side first. + // Stream updates (origin route (un)announces), bailing if the peer cancels + // its request first. let res = loop { match ns.watched.values().any(|watch| watch.deferred) { // Arm on the edge, so a turn that changes nothing else doesn't push the @@ -2099,7 +2194,7 @@ where let Namespaces { target, .. } = &mut ns; kio::wait(|waiter| { let mut cx = std::task::Context::from_waker(waiter.waker()); - if let Poll::Ready(res) = target.poll_closed(&mut cx) { + if let Poll::Ready(res) = target.poll_closed(&mut finished, self.version, &mut cx) { return Poll::Ready(NamespaceEvent::Closed(res)); } if let Poll::Ready(update) = announced.poll_next(waiter) { @@ -4394,6 +4489,138 @@ mod serve_tests { "the cap excludes cc" ); } + + const ALL_VERSIONS: [Version; 9] = [ + Version::Draft14, + Version::Draft15, + Version::Draft16, + Version::Draft17, + Version::Draft18, + Version::Draft19, + Version::Draft20, + Version::Draft21, + Version::Draft22, + ]; + + /// Drafts 14-18 cancel a request on its requester's FIN; later drafts only end updates. + fn fin_cancels(version: Version) -> bool { + matches!( + version, + Version::Draft14 | Version::Draft15 | Version::Draft16 | Version::Draft17 | Version::Draft18 + ) + } + + /// Poll `run` while letting spawned tasks progress; whether it finished. + async fn drive(mut run: std::pin::Pin<&mut F>) -> bool { + for _ in 0..50 { + if futures::poll!(run.as_mut()).is_ready() { + return true; + } + tokio::task::yield_now().await; + } + false + } + + /// Start a live subscription on a request stream opened from `requester`, whose script + /// is what the requester sends after SUBSCRIBE. + async fn start_subscription( + h: &Serve, + mut requester: ScriptedSession, + ) -> impl std::future::Future> + use<> { + settle().await; + let stream = Stream::open(&mut requester, h.publisher.version).await.unwrap(); + h.publisher + .clone() + .run_subscribe_stream(stream, subscribe(Filter::NextObject, None)) + } + + /// Draft-21 section 6.4.2.2 lets a requester FIN right after SUBSCRIBE. From draft 19 + /// that only ends its updates, so media keeps flowing; earlier drafts cancel. + #[tokio::test(start_paused = true)] + async fn requester_fin_is_version_gated_for_subscriptions() { + for version in ALL_VERSIONS { + let h = serve(version); + let run = start_subscription(&h, ScriptedSession::per_stream_eof(vec![vec![]])).await; + let mut run = std::pin::pin!(run); + let ended = drive(run.as_mut()).await; + assert_eq!(ended, fin_cancels(version), "{version}"); + if !ended { + let mut group = h.track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(timestamp(), b"after FIN".as_slice()).unwrap(); + group.finish().unwrap(); + assert!(!drive(run.as_mut()).await, "{version}"); + assert_eq!(occurrences(&h.log, b"after FIN"), 1, "{version}"); + } + } + } + + /// A reset request stream still cancels the subscription on every draft. + #[tokio::test(start_paused = true)] + async fn requester_reset_cancels_subscriptions() { + for version in ALL_VERSIONS { + let h = serve(version); + let run = start_subscription(&h, ScriptedSession::per_stream_reset(vec![vec![]])).await; + assert!(drive(std::pin::pin!(run)).await, "{version}"); + } + } + + /// A REQUEST_UPDATE carrying a priority is applied and acknowledged, and the + /// subscription keeps going. The second, empty update changes nothing. + #[tokio::test(start_paused = true)] + async fn subscription_update_applies_priority_without_ending_the_request() { + for version in [Version::Draft19, Version::Draft22] { + let h = serve(version); + let requester = ScriptedSession::per_stream(vec![vec![0x02, 0, 4, 2, 1, 0x20, 10, 0x02, 0, 2, 4, 0]]); + let run = start_subscription(&h, requester.clone()).await; + let mut run = std::pin::pin!(run); + assert!(!drive(run.as_mut()).await, "{version}"); + assert_eq!(h.track.subscription().unwrap().priority, 245, "{version}"); + // REQUEST_OK with no parameters, once per update. + assert_eq!(occurrences(&requester.log, &[0x07, 0, 1, 0]), 2, "{version}"); + } + } + + /// An update we cannot apply is refused NOT_SUPPORTED and ends the subscription with + /// UPDATE_FAILED, rather than silently dropping it. + #[tokio::test(start_paused = true)] + async fn unsupported_subscription_update_fails_the_subscription() { + for version in [Version::Draft17, Version::Draft19, Version::Draft22] { + let h = serve(version); + // FORWARD=0 asks to pause delivery, which this publisher does not support. + let script = match version { + // Draft 17 carries a Required Request ID Delta after the Request ID. + Version::Draft17 => vec![0x02, 0, 5, 2, 0, 1, 0x10, 0], + _ => vec![0x02, 0, 4, 2, 1, 0x10, 0], + }; + let requester = ScriptedSession::per_stream(vec![script]); + let run = start_subscription(&h, requester.clone()).await; + run.await.unwrap_err(); + assert_eq!( + occurrences(&requester.log, b"REQUEST_UPDATE parameters not supported"), + 1, + "{version}" + ); + assert_eq!(occurrences(&requester.log, b"update failed"), 1, "{version}"); + } + } + + /// SUBSCRIBE_NAMESPACE follows the same gate as SUBSCRIBE. + #[tokio::test(start_paused = true)] + async fn namespace_requester_fin_is_version_gated() { + for version in ALL_VERSIONS { + let h = serve(version); + let mut requester = ScriptedSession::per_stream_eof(vec![vec![]]); + let stream = Stream::open(&mut requester, version).await.unwrap(); + let msg = ietf::SubscribeNamespace { + request_id: RequestId(0), + namespace: crate::Path::new(""), + hidden: false, + }; + let mut run = std::pin::pin!(h.publisher.clone().run_subscribe_namespace_stream(stream, msg)); + let ended = drive(run.as_mut()).await; + assert_eq!(ended, fin_cancels(version), "{version}"); + } + } } #[cfg(test)] diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs new file mode 100644 index 0000000000..bf98967a39 --- /dev/null +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -0,0 +1,105 @@ +use std::task::{Context, Poll}; + +use super::{Message, Version}; +use crate::{ + Error, + coding::{Decode, DecodeError, Stream}, +}; + +/// Whether the requester's FIN cancels its request; later drafts only end its updates. +pub(super) fn fin_cancels(version: Version) -> bool { + matches!( + version, + Version::Draft14 | Version::Draft15 | Version::Draft16 | Version::Draft17 | Version::Draft18 + ) +} + +/// Watch abrupt cancellation after the requester has finished sending messages. +pub(super) fn poll_cancel( + stream: &mut Stream, + finished: &mut bool, + version: Version, + cx: &mut Context<'_>, +) -> Poll> { + if !*finished { + match stream.reader.poll_closed(cx) { + Poll::Ready(Ok(())) if !fin_cancels(version) => *finished = true, + Poll::Ready(result) => return Poll::Ready(result), + Poll::Pending => {} + } + } + // Virtual writers on drafts 14-16 are already closed while their control request lives. + if !matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) { + return stream.writer.poll_closed(cx); + } + Poll::Pending +} + +/// A framed draft-17+ REQUEST_UPDATE on a subscription, preserving omitted preferences. +#[derive(Debug)] +pub(super) struct Update { + pub(super) priority: Option, + /// It asks for something other than a priority change, which we cannot apply. + pub(super) unsupported: bool, +} + +impl Decode for Update { + fn decode(r: &mut R, version: Version) -> Result { + if u64::decode(r, version)? != super::SubscribeUpdate::ID { + return Err(DecodeError::InvalidValue); + } + let size = u16::decode(r, version)? as usize; + if r.remaining() < size { + return Err(DecodeError::Short); + } + let mut data = r.copy_to_bytes(size); + // The complete frame is present; a short field inside it is malformed. + Self::decode_body(&mut data, version).map_err(DecodeError::complete) + } +} + +impl Update { + fn decode_body(data: &mut bytes::Bytes, version: Version) -> Result { + use super::{Fill, Filter, Opaque, RequestId}; + use bytes::Buf; + + RequestId::decode(data, version)?; + if version == Version::Draft17 { + // Required Request ID Delta, removed in draft 18. + u64::decode(data, version)?; + } + decode_params!(data, version, + 0x02 => object_timeout: Option, + 0x03 => token: Vec, + 0x06 => subgroup_timeout: Option, + 0x10 => forward: Option, + 0x20 => priority: Option, + 0x21 => filter: Option, + 0x23 => fill: Option, + 0x25 => subgroup_filter: Vec, + 0x26 => object_filter: Vec, + 0x27 => priority_filter: Vec, + 0x28 => property_filter: Vec, + 0x29 => track_filter: Vec, + 0x32 => new_group: Option, + ); + if data.has_remaining() { + return Err(DecodeError::Long); + } + Ok(Self { + priority, + unsupported: forward == Some(false) + || filter.is_some() + || fill.is_some() + || object_timeout.is_some() + || subgroup_timeout.is_some() + || !token.is_empty() + || !subgroup_filter.is_empty() + || !object_filter.is_empty() + || !priority_filter.is_empty() + || !property_filter.is_empty() + || !track_filter.is_empty() + || new_group.is_some(), + }) + } +} diff --git a/rs/moq-tokio/src/transport.rs b/rs/moq-tokio/src/transport.rs index 3912a2402b..31d7b34d6b 100644 --- a/rs/moq-tokio/src/transport.rs +++ b/rs/moq-tokio/src/transport.rs @@ -55,16 +55,8 @@ const CLOSE_LINGER: std::time::Duration = std::time::Duration::from_secs(10); /// watch parks until the caller's reads drain the backlog, deliberately /// trading watch liveness on an undrained flooding stream for a memory /// bound (every driver in this crate drains its reads alongside the watch). -/// A send stream only -/// starts the real watch once `finish` or `reset` makes it terminal; before -/// that a closure surfaces as an error on the next write instead of waking an -/// idle watch. The send-side gap this leaves: a peer that resets a stream -/// sitting idle (no pending write, not finished) does not wake a parked -/// `poll_closed`, so a driver waiting on "next frame or peer close" learns of -/// the closure only when the next write fails. The drivers' other arms keep -/// the session making progress; the stream itself lingers until then. A -/// native poll implementation observes closure without owning the stream, -/// which is why a backend that has one is not wrapped. +/// Send-side watches are interruptible, so an idle watch can observe peer +/// cancellation while later writes reclaim the stream. pub struct Session { session: S, accept_uni: OpSlot>, @@ -231,9 +223,6 @@ pub struct SendStream { priority: Option, finish: bool, reset: Option, - /// Whether `finish` or `reset` has been observed, so no further writes can - /// come and the closed() watch may safely take ownership of the stream. - terminal: bool, /// Bytes already transmitted but not yet reported to the caller: a write the /// stored future finished (possibly inside /// [`poll_closed`](wt_poll::SendStream::poll_closed)) while the caller held a @@ -273,7 +262,6 @@ impl SendStream { priority: None, finish: false, reset: None, - terminal: false, completed: Bytes::new(), interrupt: None, } @@ -374,17 +362,22 @@ impl wt_poll::SendStream for SendS } } }, - SendState::Closing(mut fut) => match fut.as_mut().poll(cx) { - Poll::Pending => { - *self.state.get_mut().unwrap() = Some(SendState::Closing(fut)); - return Poll::Pending; + SendState::Closing(mut fut) => { + if let Some(tx) = self.interrupt.take() { + let _ = tx.send(()); } - Poll::Ready((mut stream, _res)) => { - self.interrupt = None; - self.settle(&mut stream); - *self.state.get_mut().unwrap() = Some(SendState::Idle(stream)); + match fut.as_mut().poll(cx) { + Poll::Pending => { + *self.state.get_mut().unwrap() = Some(SendState::Closing(fut)); + return Poll::Pending; + } + Poll::Ready((mut stream, _res)) => { + self.interrupt = None; + self.settle(&mut stream); + *self.state.get_mut().unwrap() = Some(SendState::Idle(stream)); + } } - }, + } } } } @@ -405,18 +398,21 @@ impl wt_poll::SendStream for SendS } fn finish(&mut self) -> Result<(), Self::Error> { - self.terminal = true; match self.state.get_mut().unwrap().as_mut() { Some(SendState::Idle(stream)) => stream.finish(), _ => { self.finish = true; + if matches!(self.state.get_mut().unwrap(), Some(SendState::Closing(_))) + && let Some(tx) = self.interrupt.take() + { + let _ = tx.send(()); + } Ok(()) } } } fn reset(&mut self, code: u32) { - self.terminal = true; match self.state.get_mut().unwrap().as_mut() { Some(SendState::Idle(stream)) => stream.reset(code), _ => { @@ -435,14 +431,6 @@ impl wt_poll::SendStream for SendS match self.state.get_mut().unwrap().take().expect("in-flight") { SendState::Idle(mut stream) => { self.settle(&mut stream); - // Only a finished (or reset) stream starts the real closed() - // watch: that future owns the stream, and a stream the caller - // may still write to must stay reclaimable. Before that, - // closure surfaces as an error on the next operation instead. - if !self.terminal { - *self.state.get_mut().unwrap() = Some(SendState::Idle(stream)); - return Poll::Pending; - } // Interruptible like a write: a late reset (the group machines // cancel a finished stream this way) must not wait for a FIN // acknowledgement that may never come. @@ -839,6 +827,7 @@ mod tests { resets: Arc>>, /// The peer never acknowledges the FIN: closed() stays pending forever. never_ack: Arc, + stopped: Arc, } impl web_transport_trait::SendStream for FakeSend { @@ -871,10 +860,16 @@ mod tests { } async fn closed(&mut self) -> Result<(), Self::Error> { - match self.finished.load(Ordering::SeqCst) && !self.never_ack.load(Ordering::SeqCst) { - true => Ok(()), - false => std::future::pending().await, - } + std::future::poll_fn(|_| { + if self.stopped.load(Ordering::SeqCst) { + return Poll::Ready(Err(FakeError)); + } + match self.finished.load(Ordering::SeqCst) && !self.never_ack.load(Ordering::SeqCst) { + true => Poll::Ready(Ok(())), + false => Poll::Pending, + } + }) + .await } } @@ -931,7 +926,7 @@ mod tests { let mut send = SendStream::new(fake.clone()); let mut cx = cx(); - // A pre-terminal closed watch stays pending without consuming the stream. + // An idle closed watch stays pending until the peer cancels. assert!(send.poll_closed(&mut cx).is_pending()); // The write proceeds; the old ownership-transfer watch deadlocked here. @@ -939,6 +934,16 @@ mod tests { assert_eq!(fake.writes.lock().unwrap().as_slice(), b"hello"); } + #[test] + fn send_closed_observes_stop_while_idle() { + let fake = FakeSend::default(); + let mut send = SendStream::new(fake.clone()); + let mut cx = cx(); + assert!(send.poll_closed(&mut cx).is_pending()); + fake.stopped.store(true, Ordering::SeqCst); + assert!(matches!(send.poll_closed(&mut cx), Poll::Ready(Err(_)))); + } + // After finish() the real closed() watch runs and resolves. #[test] fn send_closed_resolves_after_finish() { From dd88fa3b144bac3010497d2aeda3671ffa21a613 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 16:17:41 -0700 Subject: [PATCH 10/11] fix(transcode): refuse a fetch that starts mid-group (#4812) Drops the quest/ file change, since release does not carry that quest. (cherry picked from commit 4ae871c947a58732cac9fd9d4d82b619695f639e) Co-Authored-By: Claude Opus 5.5 --- rs/moq-transcode/src/lib.rs | 123 +++++++++++++++++++++++++++++++++++ rs/moq-transcode/src/rung.rs | 18 ++++- 2 files changed, 140 insertions(+), 1 deletion(-) diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index 2fbbd371cb..9587f07b59 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/rs/moq-transcode/src/lib.rs @@ -1271,6 +1271,129 @@ mod tests { transcoder.abort(); } + /// A fetch that starts partway through a group is refused rather than served. + /// A fresh encode of that group need not match the bytes of the head a reader + /// holds, so its tail cannot continue it. The reader moves on to the next group. + #[tokio::test] + async fn a_mid_group_fetch_is_refused() { + let source = source_catalog(320, 240); + // Serve any source group on demand, so a fetch that is not refused has + // something to transcode and resolves. + let source_fetches = source._track.dynamic(); + let server = tokio::spawn(async move { + while let Ok(request) = source_fetches.requested_group().await { + let mut group = request.accept(None).unwrap(); + write_keyframe(&mut group); + group.finish().unwrap(); + } + }); + + let config = Config { + ladder: Ladder::new([Rung::new(120, moq_net::bandwidth::Rate::from_bps(100_000))]).unwrap(), + encoder: moq_video::encode::Kind::Software, + decoder: moq_video::decode::Kind::Software, + source: None, + ..Default::default() + }; + let output = moq_net::broadcast::Info::default().produce(); + let consumer = output.consume(); + let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config)); + + let catalog = loop { + match consumer.track(hang::Catalog::DEFAULT_NAME) { + Ok(track) => break track, + Err(moq_net::Error::NotFound) => tokio::task::yield_now().await, + Err(err) => panic!("catalog track: {err}"), + } + }; + let mut catalogs = moq_mux::catalog::hang::Consumer::<()>::new(catalog.subscribe(None).await.unwrap()); + await_catalog(&mut catalogs, |snapshot| { + snapshot.video.renditions.contains_key("video/120p") + }) + .await; + + // The source publishes no live groups, so group 7 is a cache miss that + // reaches the rung's fetch handler. + let rung = consumer.track("video/120p").unwrap(); + rung.query().await.unwrap(); + + let partial = rung + .fetch_group(7, moq_net::group::Fetch::default().with_frame_start(2)) + .await; + match partial { + Err(moq_net::Error::NotFound) => {} + Err(err) => panic!("expected a NotFound refusal, got {err}"), + Ok(_) => panic!("served a fetch that starts mid-group"), + } + + // The whole group still serves. + let mut whole = rung.fetch_group(7, None).await.unwrap(); + while whole.read_frame().await.unwrap().is_some() {} + assert!(whole.finished().await.unwrap() > 0, "the whole group had no frames"); + + server.abort(); + transcoder.abort(); + } + + /// Two transcoders fed the same source publish groups that mirror the + /// source's sequences and timestamps rather than anything numbered per + /// instance. + #[tokio::test] + async fn two_instances_mirror_source_groups() { + let source = source_broadcast(2, 5); + let config = Config { + ladder: Ladder::new([Rung::new(120, moq_net::bandwidth::Rate::from_bps(100_000))]).unwrap(), + encoder: moq_video::encode::Kind::Software, + decoder: moq_video::decode::Kind::Software, + source: None, + ..Default::default() + }; + + // Mirrored from the source, not numbered by either instance. + let expected: Vec<_> = (0..2u64) + .map(|sequence| { + ( + sequence, + (0..5).map(|i| (sequence * 5 + i) as u128 * 33_333).collect::>(), + ) + }) + .collect(); + + for _ in 0..2 { + let output = moq_net::broadcast::Info::default().produce(); + let consumer = output.consume(); + let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config.clone())); + + let track = loop { + match consumer.track(hang::Catalog::DEFAULT_NAME) { + Ok(track) => break track, + Err(moq_net::Error::NotFound) => tokio::task::yield_now().await, + Err(err) => panic!("catalog track: {err}"), + } + }; + let mut catalogs = moq_mux::catalog::hang::Consumer::<()>::new(track.subscribe(None).await.unwrap()); + await_catalog(&mut catalogs, |snapshot| { + snapshot.video.renditions.contains_key("video/120p") + }) + .await; + + let rung = consumer.track("video/120p").unwrap(); + let mut groups = Vec::new(); + for sequence in 0..2 { + let mut fetched = rung.fetch_group(sequence, None).await.unwrap(); + let mut timestamps = Vec::new(); + while let Some(payload) = fetched.read_frame().await.unwrap() { + let frame = hang::container::Frame::decode(payload.payload).unwrap(); + timestamps.push(frame.timestamp.as_micros()); + } + groups.push((fetched.sequence, timestamps)); + } + + assert_eq!(groups, expected); + transcoder.abort(); + } + } + /// A source whose codec description changes rebuilds the shared decode, so /// every rung retires with it. The picture may not have moved at all, so shape /// alone would hand the replacements the names that just ended. They have to be diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 1fc217c823..4176e0eba6 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -7,7 +7,9 @@ //! source track (mirroring the aggregate subscription) and transcodes group //! for group until the track goes `unused` again. //! - A fetch of a specific group (`requested_group`) fetches that same group -//! from the source and transcodes just that group with a fresh encoder. +//! from the source and transcodes just that group with a fresh encoder. A +//! fetch that starts mid-group is refused: a fresh encode's frames are only +//! valid after the head that same encode produced. //! //! Output groups mirror the source group sequence numbers 1:1, so a fetch for //! output group N maps to source group N and a player switching renditions @@ -481,6 +483,20 @@ fn spawn_fetch( /// "the handler vanished" and hides the actual decode/encode/source failure from /// the waiting consumer. async fn fetch(rung: Rung, request: moq_net::group::Request) -> Result<(), Error> { + // A fresh encode of this group need not match the bytes the reader's head + // came from (the live encoder, or an earlier fetch), so its tail cannot + // continue that head. Refuse it so the reader moves on to the next group + // boundary instead of decoding a splice. + if request.frame_start() != 0 { + tracing::debug!( + sequence = request.sequence(), + frame = request.frame_start(), + "refusing a fetch that starts mid-group" + ); + request.reject(moq_net::Error::NotFound); + return Ok(()); + } + let options = moq_net::group::Fetch::default().with_priority(request.priority()); let mut source = match rung.source.fetch_group(request.sequence(), options).await { Ok(source) => source, From 65b661758da5024adb0345db15062fd4bc0092c5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 15:33:29 -0700 Subject: [PATCH 11/11] ci: the back-merge no longer enables auto-merge (backport #4619) Co-Authored-By: Claude Opus 5.5 --- sh/gh/back-merge.sh | 24 ++++++------------------ 1 file changed, 6 insertions(+), 18 deletions(-) diff --git a/sh/gh/back-merge.sh b/sh/gh/back-merge.sh index 3021f3e318..2a7142b42f 100755 --- a/sh/gh/back-merge.sh +++ b/sh/gh/back-merge.sh @@ -1,7 +1,8 @@ #!/usr/bin/env bash # -# Open (or refresh) the pull request that merges `release` back into `main`, -# and queue it to land as a merge commit once Check and Test pass. +# Open (or refresh) the pull request that merges `release` back into `main`. +# Check's `land` job merges it as a merge commit once Check and Test pass on its +# head, bypassing the squash-only merge queue. # # The pull request's head is its own branch rather than `release` itself: the # repository deletes head branches on merge, and a conflict is resolved by @@ -45,25 +46,12 @@ esac # Only a pull request from this repository's own branch: `gh pr list --head` # matches the branch name alone, so a fork's branch of the same name would pass. -find_pr() { - gh api "repos/$repo/pulls?state=open&base=main&head=${repo%%/*}:$branch" \ - --jq "[.[] | select(.head.repo.full_name == \"$repo\")][0].number // empty" -} - -pr=$(find_pr) +pr=$(gh api "repos/$repo/pulls?state=open&base=main&head=${repo%%/*}:$branch" \ + --jq "[.[] | select(.head.repo.full_name == \"$repo\")][0].number // empty") if [[ -z "$pr" ]]; then gh pr create --repo "$repo" --head "$branch" --base main \ --title "chore: merge release into main" \ --body "Carries what \`release\` published (versions, CHANGELOGs, backports) back to trunk. Opened by the Back-merge workflow. -**Merge with a merge commit. Never squash.** A squash leaves the merge base at the last cut, so the next back-merge conflicts. On a conflict, merge \`main\` into \`$branch\` and resolve it there." - pr=$(find_pr) -fi -if [[ -z "$pr" ]]; then - echo "error: no pull request from $repo:$branch into main" >&2 - exit 1 +Check's \`land\` job merges it as a merge commit once Check and Test pass. **Never squash.** A squash leaves the merge base at the last cut, so the next back-merge conflicts. On a conflict, merge \`main\` into \`$branch\` and resolve it there." fi - -# Queue exactly the commit this run produced, so a later push cannot ride along. -head=$(gh api "repos/$repo/git/ref/heads/$branch" --jq .object.sha) -gh pr merge --repo "$repo" "$pr" --auto --merge --match-head-commit "$head"