From 09558a2f6857f2b7f554006ff16247d50d8e3923 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:37:56 -0700 Subject: [PATCH 01/16] quest: claim ietf-fetch-only Co-Authored-By: Claude Opus 5.5 From c02ed6f6c4fa7491d673a0af0e76db1dba09e967 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:52:50 -0700 Subject: [PATCH 02/16] feat(moq-net): serve IETF TRACK_STATUS and keep timestamps without properties Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/adapter.rs | 8 + rs/moq-net/src/ietf/publisher.rs | 317 ++++++++++++++++++++++--------- rs/moq-net/src/ietf/track.rs | 239 ++++++++++++++++++++++- 3 files changed, 464 insertions(+), 100 deletions(-) diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index d72cc088e62..cbbb5f04fda 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -922,6 +922,14 @@ fn classify(type_id: u64, body: &Bytes, version: Version, namespaces: &Namespace } _ => Err(Error::UnexpectedMessage), }, + // TRACK_STATUS_OK and TRACK_STATUS_ERROR (v14 only): the one answer the request gets. + ietf::TrackStatusOk::ID_14 | ietf::TRACK_STATUS_ERROR_14 => match version { + Version::Draft14 => { + let id = decode_request_id(body, version)?; + Ok(Route::CloseStream(id)) + } + _ => Err(Error::UnexpectedMessage), + }, // PublishOk (0x1E) ietf::PublishOk::ID => { let id = decode_response_request_id(body, version)?; diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 54bf615531a..07142d10df0 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -37,6 +37,25 @@ fn serving_subscription(subscriber_priority: u8) -> Subscription { } } +/// The Track Properties an answer describing the track carries: SUBSCRIBE_OK and +/// TRACK_STATUS_OK. Empty when the requester opted out with INCLUDE_PROPERTIES=0, which +/// keeps the block present but empty. +/// +/// Declaring the timescale is what opts the track into timestamps; every object Timestamp +/// is in these units. We serve the newest group first, matching moq-lite. A draft without +/// room for a property drops it on encode. +fn track_properties(info: &track::Info, wanted: bool) -> ietf::Properties { + if !wanted { + return ietf::Properties::default(); + } + ietf::Properties { + max_cache_duration: info.max_age, + timescale: Some(info.timescale), + priority: Some(super::priority::to_wire(info.priority)), + group_order: Some(GroupOrder::Descending), + } +} + enum FillStep { Batch, Partial(frame::Consumer), @@ -562,9 +581,10 @@ where if !data.is_empty() { return Err(Error::WrongSize); } + tracing::debug!(message = ?msg, "received track_status"); async move { - if let Err(err) = this.reject_track_status(stream, msg.request_id).await { - tracing::debug!(%err, "track status refusal failed"); + if let Err(err) = this.run_track_status_stream(stream, msg).await { + tracing::debug!(%err, "track_status stream error"); } } .maybe_boxed() @@ -671,12 +691,12 @@ where None => track::Position::after_group(end.group), }); let _ = track.update(subscription); - // A Timestamp goes out only when this SUBSCRIBE_OK actually carries TIMESCALE. - // Drafts 14-16 never write that property, so their objects stay unstamped. - let timescale = msg - .properties_wanted - .then(|| track.info().timescale) - .filter(|_| ietf::Properties::sends_timescale(self.version)); + // A Timestamp goes out wherever the draft can declare its units. A subscriber that + // opted out of this SUBSCRIBE_OK's properties learns them from TRACK_STATUS, so + // the opt-out strips nothing from the objects. Drafts 14-16 never write TIMESCALE, + // so their objects stay unstamped. + let timescale = + Some(track.info().timescale).filter(|_| ietf::Properties::sends_timescale(self.version)); // Draft-20 replaced joining FETCH with subscription fills. Older drafts save // the same boundary used by the subscription so the two streams never overlap. @@ -715,21 +735,7 @@ where track_alias: request_id.0, // The subscription floor and FETCH/FILL cap use this same snapshot. largest: edge.largest, - properties: match msg.properties_wanted { - // Declaring the timescale is what opts the track into timestamps; every - // object Timestamp below is in these units. `timescale` is None when this - // version cannot send TIMESCALE. We serve the newest group first, - // matching moq-lite. - true => ietf::Properties { - max_cache_duration: track.info().max_age, - timescale, - priority: Some(super::priority::to_wire(track.info().priority)), - group_order: Some(GroupOrder::Descending), - }, - // INCLUDE_PROPERTIES=0. The field stays present but empty, which also - // means the track opts out of timestamps for this subscriber. - false => ietf::Properties::default(), - }, + properties: track_properties(track.info(), msg.properties_wanted), }) .await?; @@ -1595,16 +1601,87 @@ where Ok(Err(refusal)) } - async fn reject_track_status(&self, mut stream: Stream, request_id: RequestId) -> Result<(), Error> { - let error_code = request::to_code(&Error::Unsupported, request::Kind::TrackStatus, self.version); + /// Answer a TRACK_STATUS with what a SUBSCRIBE_OK for the track would carry, without + /// subscribing. + /// + /// The track resolves as it would for a SUBSCRIBE, so a relay asks its upstream for a + /// track it does not hold yet and the answer waits on that. The Largest Location is + /// whatever the cache holds now: nothing waits for a fresher one. + async fn run_track_status_stream( + self, + mut stream: Stream, + msg: ietf::TrackStatus<'_>, + ) -> Result<(), Error> { + let request_id = msg.request_id; + let broadcast = match self + .serving_origin() + .await + .request_broadcast(&msg.track_namespace) + .await + { + Ok(broadcast) => broadcast, + Err(err) => return self.reject_track_status(stream, request_id, &err).await, + }; + let track = match broadcast.track(&msg.track_name) { + Ok(track) => track, + Err(err) => return self.reject_track_status(stream, request_id, &err).await, + }; + + // The requester abandoning the request is owed nothing. + let info = { + let query = track.query(); + let mut finished = false; + kio::wait(|waiter| { + let mut cx = waiter.context(); + if super::request_stream::poll_cancel(&mut stream, &mut finished, self.version, &mut cx).is_ready() { + return Poll::Ready(None); + } + query.poll_ok(waiter).map(Some) + }) + .await + }; + let info = match info { + Some(Ok(info)) => info, + Some(Err(err)) => return self.reject_track_status(stream, request_id, &err).await, + None => return Ok(()), + }; + + stream.writer.varint(ietf::TrackStatusOk::id(self.version)).await?; + stream + .writer + .encode(&ietf::TrackStatusOk { + request_id: match self.version { + Version::Draft14 | Version::Draft15 | Version::Draft16 => Some(request_id), + _ => None, + }, + largest: live_edge(&track).largest, + properties: track_properties(&info, msg.properties_wanted), + }) + .await?; + + // The answer is all this stream carries. See [`Self::reject_subscribe`] for why the + // close is not optional. + let _ = stream.writer.close().await; + Ok(()) + } + + async fn reject_track_status( + &self, + mut stream: Stream, + request_id: RequestId, + err: &Error, + ) -> Result<(), Error> { + let error_code = request::to_code(err, request::Kind::TrackStatus, self.version); + let reason_phrase = err.to_string().into(); if self.version == Version::Draft14 { - stream.writer.varint(0x0fu64).await?; // TRACK_STATUS_ERROR has the SUBSCRIBE_ERROR body. + // TRACK_STATUS_ERROR has the SUBSCRIBE_ERROR body. + stream.writer.varint(ietf::TRACK_STATUS_ERROR_14).await?; stream .writer .encode(&ietf::SubscribeError { request_id, error_code, - reason_phrase: "TRACK_STATUS is not supported".into(), + reason_phrase, }) .await?; } else { @@ -1614,7 +1691,7 @@ where .encode(&ietf::RequestError { request_id: matches!(self.version, Version::Draft15 | Version::Draft16).then_some(request_id), error_code, - reason_phrase: "TRACK_STATUS is not supported".into(), + reason_phrase, retry_interval: 0, }) .await?; @@ -3689,68 +3766,125 @@ mod serve_tests { } } - /// TRACK_STATUS is recognized but not implemented, and must get a complete refusal. + const ALL_VERSIONS: [Version; 9] = [ + Version::Draft14, + Version::Draft15, + Version::Draft16, + Version::Draft17, + Version::Draft18, + Version::Draft19, + Version::Draft20, + Version::Draft21, + Version::Draft22, + ]; + + fn track_status(namespace: &str, wanted: bool, version: Version) -> Vec { + let msg = ietf::TrackStatus { + request_id: RequestId(REQUEST_ID), + track_namespace: crate::Path::new(namespace), + track_name: "video".into(), + properties_wanted: wanted, + }; + let mut body = Vec::new(); + msg.encode_msg(&mut Encoder::new(&mut body, version.into()), version) + .unwrap(); + body + } + + /// TRACK_STATUS gets what a SUBSCRIBE_OK would carry, as far as each draft's answer has + /// room, without subscribing to the track. INCLUDE_PROPERTIES=0 empties the properties. #[moq_net_sim::test] - async fn track_status_is_refused_on_every_draft() { - for version in [ - Version::Draft14, - Version::Draft15, - Version::Draft16, - Version::Draft17, - Version::Draft18, - Version::Draft19, - Version::Draft20, - Version::Draft21, - Version::Draft22, - ] { - let h = serve(version); - let stream = Stream::open(&mut h.session.clone(), version).await.unwrap(); - let msg = ietf::TrackStatus { - request_id: RequestId(REQUEST_ID), - track_namespace: crate::Path::new("live"), - track_name: "video".into(), - }; - let mut body = Vec::new(); - msg.encode_msg(&mut Encoder::new(&mut body, version.into()), version) - .unwrap(); - let mark = h.log.writes.lock().unwrap().len(); - h.publisher - .clone() - .handle_stream(ietf::TrackStatus::ID, ietf::Body(bytes::Bytes::from(body)), stream) - .unwrap() - .await; - let actual = h.log.writes.lock().unwrap()[mark..].to_vec(); - let expected = { - let log = crate::lite::test_transport::Log::default(); - let mut writer = - crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); - if version == Version::Draft14 { - writer.varint(0x0fu64).await.unwrap(); - writer - .encode(&ietf::SubscribeError { - request_id: RequestId(REQUEST_ID), - error_code: 0x3, - reason_phrase: "TRACK_STATUS is not supported".into(), - }) - .await - .unwrap(); - } else { - writer.varint(ietf::RequestError::ID).await.unwrap(); - writer - .encode(&ietf::RequestError { - request_id: matches!(version, Version::Draft15 | Version::Draft16) - .then_some(RequestId(REQUEST_ID)), - error_code: 0x3, - reason_phrase: "TRACK_STATUS is not supported".into(), - retry_interval: 0, - }) - .await - .unwrap(); - } - log.writes.lock().unwrap().clone() + async fn track_status_is_answered_on_every_draft() { + for version in ALL_VERSIONS { + for wanted in [true, false] { + let h = serve(version); + let mut group = h.track.create_group(group::Info { sequence: 4 }).unwrap(); + group.write_frame(timestamp(), b"a".as_slice()).unwrap(); + group.write_frame(timestamp(), b"b".as_slice()).unwrap(); + moq_net_sim::sleep(std::time::Duration::from_millis(1)).await; + + let stream = Stream::open(&mut h.session.clone(), version).await.unwrap(); + let mark = h.log.writes.lock().unwrap().len(); + let body = track_status("room", wanted, version); + h.publisher + .clone() + .handle_stream(ietf::TrackStatus::ID, ietf::Body(bytes::Bytes::from(body)), stream) + .unwrap() + .await; + assert!(h.log.resets().is_empty(), "{version}: the answer was reset"); + assert!(h.track.subscription().is_none(), "{version}: TRACK_STATUS subscribed"); + + let wire = h.log.writes.lock().unwrap()[mark..].to_vec(); + let mut buf = Decoder::new(&wire, version.into()); + assert_eq!(buf.varint().unwrap(), ietf::TrackStatusOk::id(version), "{version}"); + let ok = ietf::TrackStatusOk::decode(&mut buf, version).unwrap(); + assert!(buf.is_empty(), "{version}: trailing bytes"); + + let expected = match version { + // GROUP_ORDER is a field on draft-14 and a parameter on draft-15. + Version::Draft14 | Version::Draft15 => ietf::Properties { + group_order: Some(GroupOrder::Descending), + ..Default::default() + }, + Version::Draft16 | Version::Draft17 => ietf::Properties::default(), + // Only draft-20 can carry the opt-out. + _ if !wanted && Filter::is_draft20(version) => ietf::Properties::default(), + _ => track_properties(&track::Info::default(), true), + }; + assert_eq!( + ok, + ietf::TrackStatusOk { + request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) + .then_some(RequestId(REQUEST_ID)), + largest: Some(Location { group: 4, object: 1 }), + properties: expected, + }, + "{version} wanted={wanted}" + ); + } + } + } + + /// A TRACK_STATUS for a broadcast we do not serve is refused as a SUBSCRIBE would be. + #[moq_net_sim::test] + async fn track_status_for_a_missing_broadcast_is_refused() { + for version in ALL_VERSIONS { + let code = refusal(version, ietf::TrackStatus::ID, track_status("absent", true, version)).await; + let expected = match version { + Version::Draft14 => 0x4, + _ => 0x10, }; - assert_eq!(actual, expected, "{version}: wrong TRACK_STATUS refusal bytes"); - assert!(h.log.resets().is_empty(), "{version}: refusal was reset"); + assert_eq!(code, expected, "{version}"); + } + } + + /// Opting out of SUBSCRIBE_OK's properties does not strip the objects' timestamps: + /// their units come from TRACK_STATUS instead. + #[moq_net_sim::test] + async fn the_properties_opt_out_keeps_object_timestamps() { + let version = Version::Draft20; + let stamp = crate::Timestamp::from_millis(123_456).unwrap(); + let mut properties = Vec::new(); + ietf::encode_object_time( + &mut Encoder::new(&mut properties, version.into()), + stamp, + track::Info::default().timescale, + version, + ) + .unwrap(); + + for wanted in [true, false] { + let mut h = serve(version); + let mut group = h.track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(stamp, b"stamped".as_slice()).unwrap(); + group.finish().unwrap(); + + let mut msg = subscribe(Filter::Unfiltered, None); + msg.properties_wanted = wanted; + run_live(&mut h, msg).await; + + assert_eq!(occurrences(&h.log, b"stamped"), 1, "wanted={wanted}"); + assert_eq!(occurrences(&h.log, &properties), 1, "wanted={wanted}: object unstamped"); } } @@ -3821,7 +3955,7 @@ mod serve_tests { ]; #[rustfmt::skip] - let cases: [(&str, u64, &[u8]); 3] = [ + let cases: [(&str, u64, &[u8]); 2] = [ ("SUBSCRIBE with a Range Filter", ietf::Subscribe::ID, &[ 0x02, // Number of Parameters 0x03, 0x03, 0x03, 0x00, 0xAA, // AUTHORIZATION TOKEN @@ -3833,11 +3967,6 @@ mod serve_tests { 0x17, 0x01, 0x01, // LOCATION_FILTER (0x21): from one group back 0x14, 0x01, // INCLUDE_PROPERTIES (0x35) = 1 ]), - ("TRACK_STATUS with parameters", ietf::TrackStatus::ID, &[ - 0x02, // Number of Parameters - 0x03, 0x03, 0x03, 0x00, 0xAA, // AUTHORIZATION TOKEN - 0x32, 0x00, // INCLUDE_PROPERTIES (0x35) = 0 - ]), ]; // A standalone FETCH we would otherwise serve, carrying FILL_TIMEOUT=0: cache only, diff --git a/rs/moq-net/src/ietf/track.rs b/rs/moq-net/src/ietf/track.rs index 616b5bf5466..3d088e44a60 100644 --- a/rs/moq-net/src/ietf/track.rs +++ b/rs/moq-net/src/ietf/track.rs @@ -1,4 +1,4 @@ -//! IETF moq-transport track status messages (v14 + v15) +//! IETF moq-transport TRACK_STATUS and its answer. use std::borrow::Cow; @@ -7,7 +7,7 @@ use num_enum::{IntoPrimitive, TryFromPrimitive}; use crate::{ Path, coding::*, - ietf::{Filter, GroupOrder, RequestId, Subscribe}, + ietf::{Filter, GroupOrder, Location, Properties, RequestId, Subscribe, SubscribeOk}, }; use super::Message; @@ -23,6 +23,9 @@ pub struct TrackStatus<'a> { pub request_id: RequestId, pub track_namespace: Path<'a>, pub track_name: Cow<'a, str>, + /// Whether the requester wants Track Properties on the answer (INCLUDE_PROPERTIES, + /// draft-20). + pub properties_wanted: bool, } impl Message for TrackStatus<'_> { @@ -45,20 +48,147 @@ impl Message for TrackStatus<'_> { w.u8(0); // no parameters } _ => { - encode_params!(w, version,); + // Only the opt-out is worth bytes, and an older peer would read the + // parameter as unknown, which is a protocol violation. + let include_properties = (!self.properties_wanted && Filter::is_draft20(version)).then_some(false); + encode_params!(w, version, 0x35 => include_properties); } } Ok(()) } /// Every draft defines TRACK_STATUS as identical to SUBSCRIBE, so it decodes as one and - /// keeps only what names the track. We refuse the request, so the rest goes unread. + /// keeps what names the track and shapes the answer. The rest goes unread: there is no + /// subscription for it to configure. fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { let subscribe = Subscribe::decode_msg(r, version)?; Ok(Self { request_id: subscribe.request_id, track_namespace: subscribe.track_namespace, track_name: subscribe.track_name, + properties_wanted: subscribe.properties_wanted, + }) + } +} + +/// TRACK_STATUS_OK: what a SUBSCRIBE_OK would say about the track, with no subscription. +/// +/// Draft-14 gives it its own type ([`Self::ID_14`]) and the SUBSCRIBE_OK body, with a +/// Track Alias of 0. Later drafts answer with a REQUEST_OK, which gained Track Properties +/// in draft-18; before that only the message parameters carry anything. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TrackStatusOk { + /// Present on drafts 14-16 only. + pub request_id: Option, + + /// The largest Location in the track (LARGEST_OBJECT), absent while it has no content. + pub largest: Option, + + /// The Track Properties, on the drafts whose answer has room for them. + pub properties: Properties, +} + +/// TRACK_STATUS_ERROR on draft-14, with the SUBSCRIBE_ERROR body. Later drafts refuse +/// with REQUEST_ERROR. +pub const TRACK_STATUS_ERROR_14: u64 = 0x0F; + +impl TrackStatusOk { + /// The draft-14 message type. + pub const ID_14: u64 = 0x0E; + + /// The message type on `version`. + pub fn id(version: Version) -> u64 { + match version { + Version::Draft14 => Self::ID_14, + _ => Self::ID, + } + } + + /// Whether the answer carries a Track Properties block on `version`. + fn has_properties(version: Version) -> bool { + !matches!( + version, + Version::Draft14 | Version::Draft15 | Version::Draft16 | Version::Draft17 + ) + } +} + +impl Message for TrackStatusOk { + /// REQUEST_OK; see [`Self::id`]. + const ID: u64 = 0x07; + + fn encode_msg(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { + if version == Version::Draft14 { + return SubscribeOk { + request_id: self.request_id, + track_alias: 0, + largest: self.largest, + properties: self.properties, + } + .encode_msg(w, version); + } + + if matches!(version, Version::Draft15 | Version::Draft16) { + self.request_id + .expect("request_id required for draft14-16") + .encode(w, version)?; + } else { + assert!(self.request_id.is_none(), "request_id must be None for draft17+"); + } + + // The parameters SUBSCRIBE_OK carries: GROUP_ORDER is one only on draft-15, and a + // Track Property after. + let group_order = match version { + Version::Draft15 => self.properties.group_order, + _ => None, + }; + encode_params!(w, version, + 0x09 => self.largest, + 0x22 => group_order, + ); + + // Track Properties are the final field, so nothing may follow. + if Self::has_properties(version) { + self.properties.encode(w, version)?; + } + Ok(()) + } + + fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { + if version == Version::Draft14 { + let ok = SubscribeOk::decode_msg(r, version)?; + return Ok(Self { + request_id: ok.request_id, + largest: ok.largest, + properties: ok.properties, + }); + } + + let request_id = match version { + Version::Draft15 | Version::Draft16 => Some(RequestId::decode(r, version)?), + _ => None, + }; + // Read as SUBSCRIBE_OK's are: EXPIRES is ignored, and MAX_CACHE_DURATION is a + // parameter only on draft-15. + decode_params!(r, version, + 0x04 => max_cache_duration: Option, + 0x08 => _expires: Option, + 0x09 => largest: Option, + 0x22 => group_order: Option, + ); + let mut properties = match Self::has_properties(version) { + true => Properties::decode(r, version)?, + false => Properties::default(), + }; + properties.group_order = properties.group_order.or(group_order); + if version == Version::Draft15 { + properties.max_cache_duration = max_cache_duration.map(std::time::Duration::from_millis); + } + + Ok(Self { + request_id, + largest, + properties, }) } } @@ -107,6 +237,7 @@ mod tests { request_id: RequestId(1), track_namespace: Path::new("test/ns"), track_name: "video".into(), + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft14); @@ -123,6 +254,7 @@ mod tests { request_id: RequestId(1), track_namespace: Path::new("test/ns"), track_name: "video".into(), + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft15); @@ -139,6 +271,7 @@ mod tests { request_id: RequestId(1), track_namespace: Path::new("test/ns"), track_name: "video".into(), + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft17); @@ -155,6 +288,7 @@ mod tests { request_id: RequestId(1), track_namespace: Path::new("test/ns"), track_name: "video".into(), + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft16); @@ -171,6 +305,7 @@ mod tests { request_id: RequestId(1), track_namespace: Path::new("test/ns"), track_name: "video".into(), + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft18); @@ -181,9 +316,100 @@ mod tests { assert_eq!(decoded.track_name, "video"); } + const ALL: [Version; 9] = [ + Version::Draft14, + Version::Draft15, + Version::Draft16, + Version::Draft17, + Version::Draft18, + Version::Draft19, + Version::Draft20, + Version::Draft21, + Version::Draft22, + ]; + + /// INCLUDE_PROPERTIES only exists from draft-20, so the opt-out is dropped before it, + /// and the default (1) is never written. + #[test] + fn test_track_status_opts_out_of_properties_from_draft_20() { + for version in ALL { + for wanted in [true, false] { + let msg = TrackStatus { + request_id: RequestId(1), + track_namespace: Path::new("ns"), + track_name: "video".into(), + properties_wanted: wanted, + }; + let decoded: TrackStatus = decode_message(&encode_message(&msg, version), version) + .unwrap_or_else(|e| panic!("{version}: {e}")); + let expected = wanted || !Filter::is_draft20(version); + assert_eq!(decoded.properties_wanted, expected, "{version} wanted={wanted}"); + } + } + } + + /// TRACK_STATUS_OK carries what SUBSCRIBE_OK would, as far as each draft's answer + /// has room: draft-14 is a SUBSCRIBE_OK, drafts 15-17 a REQUEST_OK with only + /// parameters, and from draft-18 the REQUEST_OK ends with the Track Properties. + #[test] + fn test_track_status_ok_round_trip() { + let properties = Properties { + max_cache_duration: Some(std::time::Duration::from_secs(30)), + timescale: Some(crate::Timescale::MILLI), + priority: Some(7), + group_order: Some(GroupOrder::Descending), + }; + let largest = Some(Location { group: 5, object: 2 }); + + for version in ALL { + let msg = TrackStatusOk { + request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) + .then_some(RequestId(3)), + largest, + properties, + }; + let decoded: TrackStatusOk = decode_message(&encode_message(&msg, version), version) + .unwrap_or_else(|e| panic!("{version}: {e}")); + + let expected = match version { + Version::Draft14 | Version::Draft15 => Properties { + group_order: Some(GroupOrder::Descending), + ..Default::default() + }, + Version::Draft16 | Version::Draft17 => Properties::default(), + _ => properties, + }; + assert_eq!(decoded.request_id, msg.request_id, "{version}"); + assert_eq!(decoded.largest, largest, "{version}"); + assert_eq!(decoded.properties, expected, "{version}"); + } + } + + /// Draft-14's TRACK_STATUS_OK is a SUBSCRIBE_OK with Track Alias 0, under its own type. + #[test] + fn test_track_status_ok_draft_14_is_a_subscribe_ok() { + let msg = TrackStatusOk { + request_id: Some(RequestId(3)), + largest: None, + properties: Properties::default(), + }; + assert_eq!(TrackStatusOk::id(Version::Draft14), 0x0E); + assert_eq!(TrackStatusOk::id(Version::Draft15), 0x07); + let subscribe_ok = SubscribeOk { + request_id: Some(RequestId(3)), + track_alias: 0, + largest: None, + properties: Properties::default(), + }; + assert_eq!( + encode_message(&msg, Version::Draft14), + encode_message(&subscribe_ok, Version::Draft14) + ); + } + /// TRACK_STATUS is identical to SUBSCRIBE on every draft, so it carries whatever - /// fields and parameters a SUBSCRIBE can. A peer that sends them must still reach - /// our refusal rather than having its session closed. + /// fields and parameters a SUBSCRIBE can. A peer that sends them must still be + /// answered rather than having its session closed. #[test] fn test_track_status_carries_subscribe_fields() { // Request ID 1, Track Namespace ("live"), Track Name ("video"). @@ -219,6 +445,7 @@ mod tests { let msg = TrackStatus::decode_msg(&mut buf, version).unwrap_or_else(|e| panic!("{version}: {e}")); assert!(buf.is_empty(), "{version}: trailing bytes"); assert_eq!(msg.track_name, "video", "{version}"); + assert_eq!(msg.properties_wanted, version != Version::Draft20, "{version}"); } } } From 4dcd0436b4fc22b9a3f60307be4e8e0bcb94b647 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:01:33 -0700 Subject: [PATCH 03/16] feat(moq-net): learn fetch-only IETF tracks from TRACK_STATUS Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/subscriber.rs | 236 +++++++++++++++++++++++++++++- rs/moq-net/tests/fetch_only.rs | 166 +++++++++++++++++++++ 2 files changed, 397 insertions(+), 5 deletions(-) create mode 100644 rs/moq-net/tests/fetch_only.rs diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index ceb5bd8f5cd..211b3cac6cd 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -444,6 +444,27 @@ impl Target { } } +/// The model's info for a track the publisher described (SUBSCRIBE_OK or TRACK_STATUS_OK). +/// +/// The copy keeps microsecond timestamps whatever units the publisher declared: each +/// object converts on receipt. +fn accepted_info(max_age: Option, priority: Option) -> track::Info { + track::Info::default() + .with_timescale(Timescale::MICRO) + .with_max_age(max_age) + .with_priority(super::priority::from_wire(priority.unwrap_or(128))) +} + +/// What a TRACK_STATUS for a track nobody subscribes to came to. +enum Status { + /// Accepted as a copy with no subscription. + Known(Idle), + /// The publisher does not implement TRACK_STATUS, so a SUBSCRIBE has to learn the track. + Unsupported(track::Request), + /// The request was answered: refused, or abandoned. + Ended, +} + /// A copy with no subscription upstream, kept for fetches and a returning subscriber. struct Idle { track: track::Producer, @@ -1735,7 +1756,21 @@ where let track_name = request.name().to_owned(); // Group FETCHes for cache misses: standalone, so they outlive each subscription. let mut group_fetches = TaskSet::owned(); - let mut target = Target::Request(request); + + // Demand with nobody subscribing (fetches, or a TRACK_STATUS downstream) learns the + // track from TRACK_STATUS rather than a SUBSCRIBE: a finished track refuses one, and + // one made only to learn the track races the fetches it was for. + let mut target = match request.subscription() { + Some(_) => Target::Request(request), + None => match self.track_status(&broadcast_path, &track_name, request).await { + Status::Known(idle) => match self.linger(&broadcast_path, &track_name, idle, &mut group_fetches).await { + Some(next) => Target::Resume(next), + None => return, + }, + Status::Unsupported(request) => Target::Request(request), + Status::Ended => return, + }, + }; loop { let Some(idle) = self .subscribe_once(&broadcast_path, &track_name, target, &mut group_fetches) @@ -1971,10 +2006,7 @@ where priority, largest, } = accepted; - let info = track::Info::default() - .with_timescale(Timescale::MICRO) - .with_max_age(max_age) - .with_priority(super::priority::from_wire(priority.unwrap_or(128))); + let info = accepted_info(max_age, priority); // The copy already holds the track: it is current again up to the answer's Largest // Location once the join's head lands or a newer group does, since leaving cut the // group it was in. @@ -2544,6 +2576,118 @@ where } } + /// Learn a track nobody subscribes to from TRACK_STATUS, and accept it as a copy with no + /// subscription, which [`Self::linger`] serves fetches from and subscribes once someone + /// does. + async fn track_status(&mut self, broadcast_path: &Path<'_>, track_name: &str, request: track::Request) -> Status { + let request_id = match self.control.next_request_id(&self.runtime).await { + Ok(id) => id, + Err(err) => { + request.reject(err); + return Status::Ended; + } + }; + let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { + Ok(stream) => stream, + Err(err) => { + request.reject(err); + return Status::Ended; + } + }; + let written = async { + stream.writer.varint(ietf::TrackStatus::ID).await?; + stream + .writer + .encode(&ietf::TrackStatus { + request_id, + track_namespace: broadcast_path.to_owned(), + track_name: track_name.into(), + properties_wanted: true, + }) + .await + } + .await; + if let Err(err) = written { + request.reject(err); + return Status::Ended; + } + + // Nobody wanting the track any more ends the wait on a peer that may never answer. + let demand = request.demand(); + let response = { + let mut response = std::pin::pin!(self.read_track_status_response(&mut stream)); + loop { + let res = kio::wait(|waiter| { + // An answer that already arrived wins over abandonment. + if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { + return Poll::Ready(Some(res)); + } + demand.poll_unused(waiter).map(|_| None) + }) + .await; + match res { + Some(res) => break Some(res), + None if request.reject_unused(Error::Cancel) => break None, + // Demand returned in the gap. + None => continue, + } + } + }; + + let Some(response) = response else { + stream.reader.abort(&Error::Cancel); + stream.writer.abort(&Error::Cancel); + return Status::Ended; + }; + // The publisher has read the request, since it answered, and FINs after its answer: + // FIN our side too and let the stream drop. + let _ = stream.writer.finish(); + let ok = match response { + Ok(ok) => ok, + // A publisher that does not implement TRACK_STATUS still answers a SUBSCRIBE. + Err(Error::Unsupported) => return Status::Unsupported(request), + Err(err) => { + tracing::debug!(%err, broadcast = %self.origin.absolute(broadcast_path), track = %track_name, "track status refused"); + request.reject(err); + return Status::Ended; + } + }; + + tracing::debug!(broadcast = %self.origin.absolute(broadcast_path), track = %track_name, "track status known"); + let dynamic = request.dynamic(); + let mut track = request.accept(accepted_info(ok.properties.max_cache_duration, ok.properties.priority)); + // Nothing feeds this copy live: what it caches is fetched, never the live edge. + track.set_idle(); + Status::Known(Idle { + demand: track.demand(), + track, + dynamic, + timescale: ok.properties.timescale, + }) + } + + /// Read the answer to a TRACK_STATUS: TRACK_STATUS_OK, or the publisher's refusal as an + /// error. + async fn read_track_status_response(&self, stream: &mut Stream) -> Result { + let type_id = stream.reader.varint().await?; + let body: ietf::Body = stream.reader.decode().await?; + let mut data = body.decoder(self.version); + + let code = match type_id { + id if id == ietf::TrackStatusOk::id(self.version) => { + return Ok(ietf::TrackStatusOk::decode_msg(&mut data, self.version)?); + } + ietf::TRACK_STATUS_ERROR_14 if self.version == Version::Draft14 => { + ietf::SubscribeError::decode_msg(&mut data, self.version)?.error_code + } + ietf::RequestError::ID if self.version != Version::Draft14 => { + ietf::RequestError::decode_msg(&mut data, self.version)?.error_code + } + _ => return Err(Error::UnexpectedMessage), + }; + Err(request::from_code(code, request::Kind::TrackStatus, self.version)) + } + pub async fn recv_group(&mut self, stream: &mut Reader) -> Result<(), Error> { let mut group: ietf::GroupHeader = stream.decode().await?; @@ -5214,6 +5358,88 @@ mod tests { outstanding } + /// A publisher that does not implement TRACK_STATUS (an older release, or another + /// implementation) refuses it NOT_SUPPORTED. Fetch-only demand then learns the track + /// from a SUBSCRIBE, as it did before TRACK_STATUS was asked. + #[moq_net_sim::test] + async fn an_unsupported_track_status_falls_back_to_subscribe() { + for version in [ + Version::Draft14, + Version::Draft15, + Version::Draft17, + Version::Draft18, + Version::Draft22, + ] { + let log = crate::lite::test_transport::Log::default(); + let mut writer = + crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); + let reason = "TRACK_STATUS is not supported"; + match version { + Version::Draft14 => { + writer.varint(ietf::TRACK_STATUS_ERROR_14).await.unwrap(); + writer + .encode(&ietf::SubscribeError { + request_id: RequestId(0), + error_code: 0x3, + reason_phrase: reason.into(), + }) + .await + .unwrap(); + } + _ => { + writer.varint(ietf::RequestError::ID).await.unwrap(); + writer + .encode(&ietf::RequestError { + request_id: matches!(version, Version::Draft15 | Version::Draft16).then_some(RequestId(0)), + error_code: 0x3, + reason_phrase: reason.into(), + retry_interval: 0, + }) + .await + .unwrap(); + } + } + let refusal = log.writes.lock().unwrap().clone(); + + let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![refusal]); + let log = session.log.clone(); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let mut subscriber = Subscriber::new( + crate::time::Clock::sim(), + session, + crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + Control::new(None, false), + None, + peer::PeerSetup::default(), + crate::Hop::new(1).unwrap(), + None, + version, + tasks, + Default::default(), + ); + + let producer = crate::broadcast::Info::default().produce(); + let mut dynamic = producer.dynamic(); + let consumer = producer.consume(); + // Fetch-only demand: nobody subscribes. + let _fetching = consumer.track("video").unwrap().fetch_group(0, None); + let request = dynamic.requested_track().await.expect("no track requested"); + assert!(request.subscription().is_none()); + + let serving = moq_net_sim::spawn(async move { + subscriber.run_subscribe(Path::new("broadcast"), dynamic, request).await; + }); + settle().await; + + assert_eq!( + control_message_types(&log, version), + [ietf::TrackStatus::ID, ietf::Subscribe::ID], + "{version}: a refused TRACK_STATUS must fall back to SUBSCRIBE" + ); + serving.abort(); + } + } + /// The publisher opens no fetch stream for an empty range, so a fill against a track /// with no content is owed nothing. Leaving it outstanding would withhold every later /// group behind a head that is never coming. diff --git a/rs/moq-net/tests/fetch_only.rs b/rs/moq-net/tests/fetch_only.rs new file mode 100644 index 00000000000..a22f6cc96fe --- /dev/null +++ b/rs/moq-net/tests/fetch_only.rs @@ -0,0 +1,166 @@ +//! Fetching a group from a moq-transport publisher never subscribes to its track. +//! +//! The subscriber learns the track from TRACK_STATUS instead, so a finished track is still +//! fetchable, and a relay's fetch-only demand puts no live subscription upstream to race +//! the fetches that report where the track ends. + +mod support; + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use moq_net::{Hop, Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(5); + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +async fn resolve(origin: &moq_net::origin::Producer) -> moq_net::broadcast::Consumer { + let consumer = origin.consume(); + moq_net_sim::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves") +} + +/// Publish a finished track of three groups, and flag any subscription it ever gets. +fn publish_finished(origin: &moq_net::origin::Producer) -> (moq_net::broadcast::Producer, Arc) { + let broadcast = origin.create_broadcast("bcast").unwrap(); + let mut track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + for sequence in 0..3u64 { + let mut group = track.append_group().unwrap(); + for frame in 0..2 { + group + .write_frame(Timestamp::ZERO, format!("{sequence}-{frame}").into_bytes()) + .unwrap(); + } + group.finish().unwrap(); + } + track.finish().unwrap(); + + let subscribed = Arc::new(AtomicBool::new(false)); + let flag = subscribed.clone(); + drop(moq_net_sim::spawn(async move { + while let Ok(subscription) = track.subscription_changed().await { + if subscription.is_some() { + flag.store(true, Ordering::SeqCst); + } + } + })); + (broadcast, subscribed) +} + +async fn fetch(broadcast: &moq_net::broadcast::Consumer, sequence: u64) -> Result>, moq_net::Error> { + let mut group = moq_net_sim::timeout(TIMEOUT, async { + broadcast.track("video").unwrap().fetch_group(sequence, None).await + }) + .await + .expect("fetch timed out")?; + let mut frames = Vec::new(); + while let Some(frame) = moq_net_sim::timeout(TIMEOUT, group.read_frame()) + .await + .expect("read timed out")? + { + frames.push(frame.payload.to_vec()); + } + Ok(frames) +} + +fn frames(sequence: u64) -> Vec> { + (0..2).map(|frame| format!("{sequence}-{frame}").into_bytes()).collect() +} + +/// Drafts whose FETCH we serve, and the newest, whose FETCH we refuse until draft-20 +/// FETCH lands: it must still not subscribe. +const VERSIONS: &[(&str, bool)] = &[ + ("moq-transport-14", true), + ("moq-transport-15", true), + ("moq-transport-16", true), + ("moq-transport-17", true), + ("moq-transport-18", true), + ("moq-transport-19", true), + ("moq-transport-22", false), +]; + +async fn direct(version: &str, served: bool) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let (broadcast, subscribed) = publish_finished(&publisher); + + let client = produce_origin(2); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(client.clone()); + let pair = connect_mock(options).await; + + let remote = resolve(&client).await; + let fetched = fetch(&remote, 1).await; + match served { + true => assert_eq!(fetched.expect("fetch"), frames(1), "{version}"), + false => assert!(fetched.is_err(), "{version}: draft-20 FETCH is refused"), + } + assert!(!subscribed.load(Ordering::SeqCst), "{version}: the fetch subscribed"); + + drop((remote, pair, broadcast, client, publisher)); +} + +/// A finished track is fetched without a SUBSCRIBE, which it would refuse. +#[moq_net_sim::test] +async fn a_fetch_learns_the_track_without_subscribing() { + let mut failures = Vec::new(); + for (version, served) in VERSIONS { + if moq_net_sim::spawn(direct(version, *served)).await.is_err() { + failures.push(*version); + } + } + assert!(failures.is_empty(), "failed: {failures:?}"); +} + +async fn relayed(version: &str) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let (broadcast, subscribed) = publish_finished(&publisher); + + let relay = produce_origin(2); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let upstream = connect_mock(options).await; + + let client = produce_origin(3); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(client.clone()); + let downstream = connect_mock(options).await; + + let remote = resolve(&client).await; + for sequence in [2, 0] { + assert_eq!(fetch(&remote, sequence).await.expect("fetch"), frames(sequence), "{version}"); + } + assert!(!subscribed.load(Ordering::SeqCst), "{version}: the relay subscribed upstream"); + + drop((remote, downstream, upstream, broadcast, client, relay, publisher)); +} + +/// A relay serves a downstream fetch with a fetch upstream, never a subscription. +#[moq_net_sim::test] +async fn a_relay_fetches_without_subscribing_upstream() { + let mut failures = Vec::new(); + for (version, served) in VERSIONS { + if *served && moq_net_sim::spawn(relayed(version)).await.is_err() { + failures.push(*version); + } + } + assert!(failures.is_empty(), "failed: {failures:?}"); +} From ee8656b37ba833eba301634f7b22214775c87540 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:03:04 -0700 Subject: [PATCH 04/16] docs: TRACK_STATUS for fetch-only demand; carry the opt-out into its own quest Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 13 +++++++- quest/m1/README.md | 2 +- quest/m1/fetch-ok-properties.md | 2 +- quest/m1/ietf-fetch-only.md | 49 ----------------------------- quest/m1/ietf-properties-opt-out.md | 39 +++++++++++++++++++++++ quest/m1/untimed-model.md | 7 ++--- 6 files changed, 56 insertions(+), 56 deletions(-) delete mode 100644 quest/m1/ietf-fetch-only.md create mode 100644 quest/m1/ietf-properties-opt-out.md diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 9b4730a7953..3919f383866 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -60,6 +60,17 @@ any `FETCH` on draft-20 and later, which moved the range into `LOCATION_FILTER`. A standalone `FETCH` carries no timestamps, since no `SUBSCRIBE_OK` declared a timescale for it. +A Rust subscriber learns a track that is only fetched, never subscribed, from +`TRACK_STATUS` instead of a `SUBSCRIBE`, so a finished track stays fetchable +and a relay puts no live subscription upstream to race its fetches. A +publisher that refuses `TRACK_STATUS` with `NOT_SUPPORTED` gets the +`SUBSCRIBE` instead. The Rust publisher answers `TRACK_STATUS` with what its +`SUBSCRIBE_OK` would carry, as far as the draft's answer has room: the Largest +Location on every draft, and the Track Properties from draft-18, when the +answer gained them. A `SUBSCRIBE` that sets `INCLUDE_PROPERTIES` to 0 gets an +empty properties block, but its objects keep their Timestamps, whose units +`TRACK_STATUS` reports. + On drafts 14–19, the Rust publisher also serves relative and absolute joining `FETCH` requests for `NextObject` subscriptions, for the subscription group's saved prefix only, while the subscription delivers later objects. One reaching @@ -91,7 +102,7 @@ A legal request that is not served is refused on its own with `NOT_SUPPORTED`, leaving the session open: a `SUBSCRIBE` with `FORWARD=0`, a `SUBSCRIBE` or `FETCH` carrying Range Filters (no `MAX_FILTER_RANGES` is advertised), a `FETCH` carrying `FILL_TIMEOUT` (Timed-Out gaps are not written), -`TRACK_STATUS`, `SUBSCRIBE_TRACKS` (draft-18 and later), and the `FETCH` +`TRACK_STATUS` to a JavaScript publisher, `SUBSCRIBE_TRACKS` (draft-18 and later), and the `FETCH` forms above. `NEW_GROUP_REQUEST` is ignored, as the draft allows a publisher without dynamic groups to do. A parameter the negotiated draft does not define still closes the session with diff --git a/quest/m1/README.md b/quest/m1/README.md index 21d2307fea9..6ed8db97bea 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -83,7 +83,7 @@ blocks. The quests that gated m0 lines moved under them. - [JS data consumer timestamps](/quest/m1/js-data-consumer-timestamps.md) - @moq/json and @moq/flate consumers return each value's timestamp, with snapshot `next()` and `latest()` - [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) - a draft-20+ FETCH within one group is served from its LOCATION_FILTER, as older drafts are - [FETCH_OK properties](/quest/m1/fetch-ok-properties.md) - our FETCH_OK carries the track properties SUBSCRIBE_OK does, as draft 16+ requires -- [Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md) - a relay fetches from an IETF upstream without subscribing, finished tracks included, with End of Track always reported +- [Properties opt-out](/quest/m1/ietf-properties-opt-out.md) - our IETF subscriber learns track properties once, from TRACK_STATUS, once the opt-out stops costing released publishers their properties and timestamps - [IETF object gaps](/quest/m1/ietf-object-gaps.md) - a gapped object ID is refused loudly like a subgroup in Rust and JS, and an overflowing one closes the session in Rust - [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - `@moq/net` sends and receives datagram groups over moq-transport, like Rust - [Datagrams are live-only](/quest/m1/datagram-unfetchable.md) - no FETCH, replay to a new subscriber, or cache fill ever returns a datagram group diff --git a/quest/m1/fetch-ok-properties.md b/quest/m1/fetch-ok-properties.md index c2b03a2f72f..9eed24f5e5d 100644 --- a/quest/m1/fetch-ok-properties.md +++ b/quest/m1/fetch-ok-properties.md @@ -51,4 +51,4 @@ the block. Interop: run `just test interop --all`. ## Related -- [Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md) - asks TRACK_STATUS for the same properties when a publisher omits them +- [Properties opt-out](/quest/m1/ietf-properties-opt-out.md) - our subscriber opting out of these properties, to learn them from TRACK_STATUS instead diff --git a/quest/m1/ietf-fetch-only.md b/quest/m1/ietf-fetch-only.md deleted file mode 100644 index c904d30e600..00000000000 --- a/quest/m1/ietf-fetch-only.md +++ /dev/null @@ -1,49 +0,0 @@ -# [M] Fetch without SUBSCRIBE - -## Goal - -A relay serves a FETCH-only demand for an IETF upstream track without -SUBSCRIBE upstream. A finished upstream track can still be fetched, and a -FETCH_OK that reaches the end of the track says so every time. - -## Plan - -The origin splices a route only once the track's info is known. moq-lite gets -it from TRACK_INFO, so its fetch-only demand never subscribes. IETF has no -TRACK_INFO, so today fetch-only demand still makes the relay SUBSCRIBE upstream. -Two symptoms follow: - -- A finished upstream track refuses the SUBSCRIBE, so its groups cannot be - fetched either. -- The live subscription races the group FETCHes. End of Track is known only - once an upstream FETCH_OK reports it, so a downstream FETCH that runs to the - end can answer before that and leave End of Track unset. moxygen's "FETCH - with large objects" case flakes on this. - -TRACK_STATUS is the likely source of the info. The publisher refuses it today, -so both sides are in scope. Keep the SUBSCRIBE path for real subscription -demand. - -Decided (2026-10-01): - -- Our subscriber learns a track's properties from TRACK_STATUS, sent in - parallel with every FETCH or SUBSCRIBE. From draft 20 it sets - INCLUDE_PROPERTIES to 0 on those requests, so the properties aren't sent - twice. Before draft 20 there's no opt-out, so it ignores the duplicate. -- First decouple object timestamps from INCLUDE_PROPERTIES: the publisher - derives its serving timescale from `properties_wanted` - (`publisher.rs:628`), so opting out today strips the Timestamp property - from every object and the subscriber falls back to arrival time. Test - TRACK_STATUS plus the opt-out with source timestamps that differ from - arrival. -- The publisher serves TRACK_STATUS. Its reply carries what it would set in a - SUBSCRIBE_OK (d22 §9.13), from the same code that fills SUBSCRIBE_OK and - FETCH_OK. -- The publisher still follows the spec for subscribers that don't opt out: - [FETCH_OK properties](/quest/m1/fetch-ok-properties.md). - -## Related - -- [FETCH_OK properties](/quest/m1/fetch-ok-properties.md) - the publisher side of the same properties - -- [JavaScript FETCH](/quest/m1/js-fetch.md) - the browser publisher answers these fetches diff --git a/quest/m1/ietf-properties-opt-out.md b/quest/m1/ietf-properties-opt-out.md new file mode 100644 index 00000000000..f8c8308a8c6 --- /dev/null +++ b/quest/m1/ietf-properties-opt-out.md @@ -0,0 +1,39 @@ +# [S] IETF subscriber opts out of duplicate properties + +## Goal + +Our IETF subscriber learns a track's properties once, from TRACK_STATUS, and +from draft 20 opts out of the copy in SUBSCRIBE_OK and FETCH_OK with +INCLUDE_PROPERTIES = 0, without losing properties or object timestamps +against a publisher that refuses TRACK_STATUS. + +## Plan + +Decided (2026-10-01), carried over from fetch-only demand (#PR_NUMBER): send +TRACK_STATUS in parallel with every SUBSCRIBE or FETCH, set +INCLUDE_PROPERTIES to 0 on those from draft 20, and ignore the duplicate +before draft 20. The publisher half already landed there: our publisher +serves TRACK_STATUS, and the opt-out no longer strips object timestamps. +Fetch-only demand already asks TRACK_STATUS instead of subscribing. + +Open, for the maintainer (found 2026-10-07): every released moq-rs +publisher, drafts 20-22 included, refuses TRACK_STATUS with NOT_SUPPORTED, +and on INCLUDE_PROPERTIES = 0 strips the Timestamp from every object. A +subscriber that opts out against one loses max cache duration, priority, +group order, and timestamps on a published version. Each SUBSCRIBE also +gains a request, which spends request-ID credit. Options: + +- Opt out only once this session's peer has answered a TRACK_STATUS, so the + first request on a session pays the duplicate. +- Opt out unconditionally, once releases that serve TRACK_STATUS are old + enough to drop. +- Drop the opt-out: SUBSCRIBE_OK and FETCH_OK keep carrying the properties, + and TRACK_STATUS stays for fetch-only demand. Then delete this quest. + +Test TRACK_STATUS plus the opt-out with source timestamps that differ from +arrival, and against a publisher that refuses TRACK_STATUS. + +## Related + +- [FETCH_OK properties](/quest/m1/fetch-ok-properties.md) - the FETCH_OK half of the same properties +- [Untimed model](/quest/m1/untimed-model.md) - decides what a track without declared units looks like to the subscriber diff --git a/quest/m1/untimed-model.md b/quest/m1/untimed-model.md index 93387fa29c7..8b9ce3fee24 100644 --- a/quest/m1/untimed-model.md +++ b/quest/m1/untimed-model.md @@ -113,10 +113,10 @@ Things to look out for: Media consumers that need a time refuse an untimed frame, except for the end-marker rule above. -- A FETCH learns track units from SUBSCRIBE_OK (a joining or fill FETCH). - A standalone FETCH that learns none when it is accepted is untimed (decided +- A FETCH learns track units from SUBSCRIBE_OK (a joining or fill FETCH), and + fetch-only demand from the TRACK_STATUS_OK its track is accepted with. A + standalone FETCH that learns none when it is accepted is untimed (decided 2026-10-05), until [FETCH_OK properties](/quest/m1/fetch-ok-properties.md) - or TRACK_STATUS ([Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md)) carry them. Test that a track accepted with units still yields timestamps, alongside the untimed cases. - Which objects the malformed rule covers. Status-only objects (End of @@ -157,7 +157,6 @@ published drafts change as the timestamp draft says. ## Related -- [Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md) - opting out of properties today falls back to arrival time - [Publisher timeliness](/quest/m1/qos/publisher-timeliness.md) - relay ingest measures arrival minus timestamp, and must skip untimed frames - [One max_age meaning](/quest/m1/cache-max-age.md) - gives untimed groups a wall-clock staleness rule, fixes the failover stall this introduces, and replaces the latest-group start - [Translator](/quest/m1/rs2ts/translator.md) - flags the nested `Option` in first-start resolution this touches From 594e25e99ef73b093fdc625169ffe4dcb5e55d53 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:05:22 -0700 Subject: [PATCH 05/16] style: rustfmt Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 3 +-- rs/moq-net/src/ietf/subscriber.rs | 5 ++++- rs/moq-net/src/ietf/track.rs | 4 ++-- rs/moq-net/tests/fetch_only.rs | 11 +++++++++-- 4 files changed, 16 insertions(+), 7 deletions(-) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 07142d10df0..83b3881890d 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -695,8 +695,7 @@ where // opted out of this SUBSCRIBE_OK's properties learns them from TRACK_STATUS, so // the opt-out strips nothing from the objects. Drafts 14-16 never write TIMESCALE, // so their objects stay unstamped. - let timescale = - Some(track.info().timescale).filter(|_| ietf::Properties::sends_timescale(self.version)); + let timescale = Some(track.info().timescale).filter(|_| ietf::Properties::sends_timescale(self.version)); // Draft-20 replaced joining FETCH with subscription fills. Older drafts save // the same boundary used by the subscription so the two streams never overlap. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 211b3cac6cd..498bd54ddb4 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1763,7 +1763,10 @@ where let mut target = match request.subscription() { Some(_) => Target::Request(request), None => match self.track_status(&broadcast_path, &track_name, request).await { - Status::Known(idle) => match self.linger(&broadcast_path, &track_name, idle, &mut group_fetches).await { + Status::Known(idle) => match self + .linger(&broadcast_path, &track_name, idle, &mut group_fetches) + .await + { Some(next) => Target::Resume(next), None => return, }, diff --git a/rs/moq-net/src/ietf/track.rs b/rs/moq-net/src/ietf/track.rs index 3d088e44a60..5d82d976d1f 100644 --- a/rs/moq-net/src/ietf/track.rs +++ b/rs/moq-net/src/ietf/track.rs @@ -368,8 +368,8 @@ mod tests { largest, properties, }; - let decoded: TrackStatusOk = decode_message(&encode_message(&msg, version), version) - .unwrap_or_else(|e| panic!("{version}: {e}")); + let decoded: TrackStatusOk = + decode_message(&encode_message(&msg, version), version).unwrap_or_else(|e| panic!("{version}: {e}")); let expected = match version { Version::Draft14 | Version::Draft15 => Properties { diff --git a/rs/moq-net/tests/fetch_only.rs b/rs/moq-net/tests/fetch_only.rs index a22f6cc96fe..dc531089d99 100644 --- a/rs/moq-net/tests/fetch_only.rs +++ b/rs/moq-net/tests/fetch_only.rs @@ -146,9 +146,16 @@ async fn relayed(version: &str) { let remote = resolve(&client).await; for sequence in [2, 0] { - assert_eq!(fetch(&remote, sequence).await.expect("fetch"), frames(sequence), "{version}"); + assert_eq!( + fetch(&remote, sequence).await.expect("fetch"), + frames(sequence), + "{version}" + ); } - assert!(!subscribed.load(Ordering::SeqCst), "{version}: the relay subscribed upstream"); + assert!( + !subscribed.load(Ordering::SeqCst), + "{version}: the relay subscribed upstream" + ); drop((remote, downstream, upstream, broadcast, client, relay, publisher)); } From 4ff8c9068207431c967c7cb2402c3f9730d4daf5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:07:37 -0700 Subject: [PATCH 06/16] fix(moq-net): fail a pending TRACK_STATUS with the session Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 2 +- rs/moq-net/src/ietf/subscriber.rs | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 83b3881890d..9eef587f68f 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -695,7 +695,7 @@ where // opted out of this SUBSCRIBE_OK's properties learns them from TRACK_STATUS, so // the opt-out strips nothing from the objects. Drafts 14-16 never write TIMESCALE, // so their objects stay unstamped. - let timescale = Some(track.info().timescale).filter(|_| ietf::Properties::sends_timescale(self.version)); + let timescale = ietf::Properties::sends_timescale(self.version).then_some(track.info().timescale); // Draft-20 replaced joining FETCH with subscription fills. Older drafts save // the same boundary used by the subscription so the two streams never overlap. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 498bd54ddb4..c7a21dc7348 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2615,8 +2615,10 @@ where return Status::Ended; } - // Nobody wanting the track any more ends the wait on a peer that may never answer. + // Nobody wanting the track any more ends the wait on a peer that may never answer, + // and the session ending fails it with the session's reason. let demand = request.demand(); + let mut closed = self.session.clone(); let response = { let mut response = std::pin::pin!(self.read_track_status_response(&mut stream)); loop { @@ -2625,6 +2627,9 @@ where if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { return Poll::Ready(Some(res)); } + if let Poll::Ready(err) = closed.poll_closed(&mut waiter.context()) { + return Poll::Ready(Some(Err(Error::from_transport(err)))); + } demand.poll_unused(waiter).map(|_| None) }) .await; From 047e1ae060809a045a2850fb09c5ee5da2986a80 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:12:27 -0700 Subject: [PATCH 07/16] quest: link the opt-out quest to its PR Co-Authored-By: Claude Opus 5.5 --- quest/m1/ietf-properties-opt-out.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/quest/m1/ietf-properties-opt-out.md b/quest/m1/ietf-properties-opt-out.md index f8c8308a8c6..b4a7dedd0e4 100644 --- a/quest/m1/ietf-properties-opt-out.md +++ b/quest/m1/ietf-properties-opt-out.md @@ -9,7 +9,7 @@ against a publisher that refuses TRACK_STATUS. ## Plan -Decided (2026-10-01), carried over from fetch-only demand (#PR_NUMBER): send +Decided (2026-10-01), carried over from fetch-only demand (#4974): send TRACK_STATUS in parallel with every SUBSCRIBE or FETCH, set INCLUDE_PROPERTIES to 0 on those from draft 20, and ignore the duplicate before draft 20. The publisher half already landed there: our publisher From 39dfccf0ccf4cfe4f4991207c7cd42fda0315541 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 10:39:14 -0700 Subject: [PATCH 08/16] feat(moq-net): ask for an IETF track's properties once per copy From draft-20, a SUBSCRIBE resuming a lingering copy and every group FETCH set INCLUDE_PROPERTIES=0: the copy already learned the track from SUBSCRIBE_OK or TRACK_STATUS_OK. A resumed copy keeps the timescale it learned. FETCH_OK's properties were already ignored once the track was accepted, so the code building info from them goes. Deletes quest/m1/ietf-properties-opt-out.md, which this covers. Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 4 +- quest/m1/README.md | 1 - quest/m1/fetch-ok-properties.md | 17 +- quest/m1/ietf-properties-opt-out.md | 39 ----- rs/moq-net/src/fuzz.rs | 1 + rs/moq-net/src/ietf/fetch.rs | 209 ++++++++++++++---------- rs/moq-net/src/ietf/publisher.rs | 6 + rs/moq-net/src/ietf/subscriber.rs | 242 ++++++++++++++++------------ rs/moq-net/src/ietf/version.rs | 1 + rs/moq-net/tests/rejoin.rs | 55 +++++++ 10 files changed, 335 insertions(+), 240 deletions(-) delete mode 100644 quest/m1/ietf-properties-opt-out.md diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 3919f383866..2a3d33dab77 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -69,7 +69,9 @@ publisher that refuses `TRACK_STATUS` with `NOT_SUPPORTED` gets the Location on every draft, and the Track Properties from draft-18, when the answer gained them. A `SUBSCRIBE` that sets `INCLUDE_PROPERTIES` to 0 gets an empty properties block, but its objects keep their Timestamps, whose units -`TRACK_STATUS` reports. +`TRACK_STATUS` reports. From draft-20, a Rust subscriber sets +`INCLUDE_PROPERTIES` to 0 on every later `SUBSCRIBE` or `FETCH` for a track +it already learned, instead of receiving the same properties again. On drafts 14–19, the Rust publisher also serves relative and absolute joining `FETCH` requests for `NextObject` subscriptions, for the subscription group's diff --git a/quest/m1/README.md b/quest/m1/README.md index a9d0e2bad75..f62eb398e93 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -81,7 +81,6 @@ blocks. The quests that gated m0 lines moved under them. - [JS data consumer timestamps](/quest/m1/js-data-consumer-timestamps.md) - @moq/json and @moq/flate consumers return each value's timestamp, with snapshot `next()` and `latest()` - [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) - a draft-20+ FETCH within one group is served from its LOCATION_FILTER, as older drafts are - [FETCH_OK properties](/quest/m1/fetch-ok-properties.md) - our FETCH_OK carries the track properties SUBSCRIBE_OK does, as draft 16+ requires -- [Properties opt-out](/quest/m1/ietf-properties-opt-out.md) - our IETF subscriber learns track properties once, from TRACK_STATUS, once the opt-out stops costing released publishers their properties and timestamps - [IETF object gaps](/quest/m1/ietf-object-gaps.md) - a gapped object ID is refused loudly like a subgroup in Rust and JS, and an overflowing one closes the session in Rust - [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - `@moq/net` sends and receives datagram groups over moq-transport, like Rust - [Datagrams are live-only](/quest/m1/datagram-unfetchable.md) - no FETCH, replay to a new subscriber, or cache fill ever returns a datagram group diff --git a/quest/m1/fetch-ok-properties.md b/quest/m1/fetch-ok-properties.md index 9eed24f5e5d..6068736f2a6 100644 --- a/quest/m1/fetch-ok-properties.md +++ b/quest/m1/fetch-ok-properties.md @@ -9,16 +9,14 @@ meets draft 16+'s rule to include all of a track's properties. ## Plan -Found (2026-10-01) in #4647: a group FETCH with no -prior SUBSCRIBE_OK now reads its max age from FETCH_OK, but our publisher -sends an empty block. +Found (2026-10-01) in #4647: our publisher's FETCH_OK sends an empty +properties block. Our own subscriber no longer reads it (#4974): it learns a +track from SUBSCRIBE_OK or TRACK_STATUS_OK before any FETCH, and from draft 20 +opts out with INCLUDE_PROPERTIES = 0. Other fetch-only readers still need it. - `rs/moq-net/src/ietf/publisher.rs` sends FETCH_OK with `properties: Default::default()`. SUBSCRIBE_OK fills them in the block near `publisher.rs:655`. Share that code so the two can't diverge. -- Our group-fetch accept path reads only the max age from FETCH_OK and - hardcodes a microsecond timescale. Apply the same properties SUBSCRIBE_OK - does (timescale, priority, group order) there too. - Drafts 14-15 allow the omission: MAX_CACHE_DURATION is a MAY there, and FETCH_OK has no properties block. From draft 16, a relay MUST include all Extension Headers / Properties associated with a track in FETCH_OK (d16 @@ -36,8 +34,8 @@ sends an empty block. - Honour INCLUDE_PROPERTIES (0x35) on FETCH from draft 20, once [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) serves draft-20 FETCH at all (today our publisher refuses every one). It defaults to sending the - properties; at 0 the block is present but empty. `fetch.rs` decodes it on - draft 20+ and drops it, so it is not honoured today. + properties; at 0 the block is present but empty. `fetch.rs` decodes it into + `Fetch::properties_wanted`, which the publisher does not honour yet. - Test per draft range: FETCH_OK round-trips the properties SUBSCRIBE_OK would carry, a standalone FETCH's objects arrive stamped, and INCLUDE_PROPERTIES = 0 empties the block. @@ -49,6 +47,3 @@ the block. Interop: run `just test interop --all`. - [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) - draft-20 FETCH is served at all, and edits the same `run_fetch_stream` -## Related - -- [Properties opt-out](/quest/m1/ietf-properties-opt-out.md) - our subscriber opting out of these properties, to learn them from TRACK_STATUS instead diff --git a/quest/m1/ietf-properties-opt-out.md b/quest/m1/ietf-properties-opt-out.md deleted file mode 100644 index b4a7dedd0e4..00000000000 --- a/quest/m1/ietf-properties-opt-out.md +++ /dev/null @@ -1,39 +0,0 @@ -# [S] IETF subscriber opts out of duplicate properties - -## Goal - -Our IETF subscriber learns a track's properties once, from TRACK_STATUS, and -from draft 20 opts out of the copy in SUBSCRIBE_OK and FETCH_OK with -INCLUDE_PROPERTIES = 0, without losing properties or object timestamps -against a publisher that refuses TRACK_STATUS. - -## Plan - -Decided (2026-10-01), carried over from fetch-only demand (#4974): send -TRACK_STATUS in parallel with every SUBSCRIBE or FETCH, set -INCLUDE_PROPERTIES to 0 on those from draft 20, and ignore the duplicate -before draft 20. The publisher half already landed there: our publisher -serves TRACK_STATUS, and the opt-out no longer strips object timestamps. -Fetch-only demand already asks TRACK_STATUS instead of subscribing. - -Open, for the maintainer (found 2026-10-07): every released moq-rs -publisher, drafts 20-22 included, refuses TRACK_STATUS with NOT_SUPPORTED, -and on INCLUDE_PROPERTIES = 0 strips the Timestamp from every object. A -subscriber that opts out against one loses max cache duration, priority, -group order, and timestamps on a published version. Each SUBSCRIBE also -gains a request, which spends request-ID credit. Options: - -- Opt out only once this session's peer has answered a TRACK_STATUS, so the - first request on a session pays the duplicate. -- Opt out unconditionally, once releases that serve TRACK_STATUS are old - enough to drop. -- Drop the opt-out: SUBSCRIBE_OK and FETCH_OK keep carrying the properties, - and TRACK_STATUS stays for fetch-only demand. Then delete this quest. - -Test TRACK_STATUS plus the opt-out with source timestamps that differ from -arrival, and against a publisher that refuses TRACK_STATUS. - -## Related - -- [FETCH_OK properties](/quest/m1/fetch-ok-properties.md) - the FETCH_OK half of the same properties -- [Untimed model](/quest/m1/untimed-model.md) - decides what a track without declared units looks like to the subscriber diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 03fb24810e9..3e4b70eeb56 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -878,6 +878,7 @@ pub fn seeds() -> Vec { fetch_type, range_filters: false, fill_timeout: false, + properties_wanted: true, }; fetch.encode_bytes(*version).ok() }) else { diff --git a/rs/moq-net/src/ietf/fetch.rs b/rs/moq-net/src/ietf/fetch.rs index 0a50802c150..002c4e94058 100644 --- a/rs/moq-net/src/ietf/fetch.rs +++ b/rs/moq-net/src/ietf/fetch.rs @@ -133,6 +133,9 @@ pub struct Fetch<'a> { /// that ends in Timed-Out gaps we cannot write, so the request is refused rather than /// left to wait without it. Never encoded; we send no fill timeout. pub fill_timeout: bool, + /// Whether the requester wants Track Properties in FETCH_OK (INCLUDE_PROPERTIES, + /// draft-20+). Always `true` before draft-20, which has no way to opt out. + pub properties_wanted: bool, } impl Message for Fetch<'_> { @@ -174,10 +177,13 @@ impl Message for Fetch<'_> { }; encode_namespace(w, namespace)?; w.string(track)?; + // INCLUDE_PROPERTIES defaults to 1, so only the opt-out is worth bytes. + let include_properties = (!self.properties_wanted).then_some(false); encode_params!(w, version, 0x20 => self.subscriber_priority, 0x21 => *filter, 0x22 => group_order, + 0x35 => include_properties, ); } } @@ -190,92 +196,102 @@ impl Message for Fetch<'_> { let _required_request_id_delta = buf.varint()?; } - // The token is ignored: the session's grant is what authorizes the request, and - // INCLUDE_PROPERTIES only shapes a FETCH_OK we don't send on draft-20. - let (fetch_type, subscriber_priority, group_order, range_filters, fill_timeout) = match version { - Version::Draft14 => { - let subscriber_priority = buf.u8()?; - let group_order = GroupOrder::decode(buf, version)?; - let fetch_type = FetchType::decode(buf, version)?; - Parameters::skip(buf)?; - (fetch_type, Some(subscriber_priority), Some(group_order), false, false) - } - Version::Draft15 | Version::Draft16 | Version::Draft17 | Version::Draft18 | Version::Draft19 => { - let fetch_type = FetchType::decode(buf, version)?; - decode_params!(buf, version, - 0x03 => _authorization_token: Vec, - 0x0A => fill_timeout: Option, - 0x20 => subscriber_priority: Option, - 0x22 => group_order: Option, - 0x25 => subgroup_filter: Vec, - 0x26 => object_id_filter: Vec, - 0x27 => priority_filter: Vec, - 0x28 => object_property_filter: Vec, - ); - let range_filters = [ - subgroup_filter, - object_id_filter, - priority_filter, - object_property_filter, - ] - .iter() - .any(|filter| !filter.is_empty()); - - // An unknown message parameter is a protocol violation, so each stays rejected - // on the drafts that predate it: FILL_TIMEOUT arrived in draft-18. - let has_fill_timeout = !matches!(version, Version::Draft15 | Version::Draft16 | Version::Draft17); - if (fill_timeout.is_some() && !has_fill_timeout) || (range_filters && !has_range_filters(version)) { - return Err(DecodeError::InvalidValue); + // The token is ignored: the session's grant is what authorizes the request. + let (fetch_type, subscriber_priority, group_order, range_filters, fill_timeout, properties_wanted) = + match version { + Version::Draft14 => { + let subscriber_priority = buf.u8()?; + let group_order = GroupOrder::decode(buf, version)?; + let fetch_type = FetchType::decode(buf, version)?; + Parameters::skip(buf)?; + ( + fetch_type, + Some(subscriber_priority), + Some(group_order), + false, + false, + true, + ) } - - ( - fetch_type, - subscriber_priority, - group_order, - range_filters, - fill_timeout.is_some(), - ) - } - // Draft-20 names the track up front and moves the range into LOCATION_FILTER. - _ => { - let namespace = decode_namespace(buf)?; - let track = Cow::Owned(buf.string()?); - decode_params!(buf, version, - 0x03 => _authorization_token: Vec, - 0x0A => fill_timeout: Option, - 0x20 => subscriber_priority: Option, - 0x21 => filter: Option, - 0x22 => group_order: Option, - 0x25 => subgroup_filter: Vec, - 0x26 => object_id_filter: Vec, - 0x27 => priority_filter: Vec, - 0x28 => object_property_filter: Vec, - 0x35 => _include_properties: Option, - ); - let range_filters = [ - subgroup_filter, - object_id_filter, - priority_filter, - object_property_filter, - ] - .iter() - .any(|filter| !filter.is_empty()); - - let fetch_type = FetchType::Filtered { - namespace, - track, - // An absent LOCATION_FILTER fetches the whole track. - filter: filter.unwrap_or(Filter::Unfiltered), - }; - ( - fetch_type, - subscriber_priority, - group_order, - range_filters, - fill_timeout.is_some(), - ) - } - }; + Version::Draft15 | Version::Draft16 | Version::Draft17 | Version::Draft18 | Version::Draft19 => { + let fetch_type = FetchType::decode(buf, version)?; + decode_params!(buf, version, + 0x03 => _authorization_token: Vec, + 0x0A => fill_timeout: Option, + 0x20 => subscriber_priority: Option, + 0x22 => group_order: Option, + 0x25 => subgroup_filter: Vec, + 0x26 => object_id_filter: Vec, + 0x27 => priority_filter: Vec, + 0x28 => object_property_filter: Vec, + ); + let range_filters = [ + subgroup_filter, + object_id_filter, + priority_filter, + object_property_filter, + ] + .iter() + .any(|filter| !filter.is_empty()); + + // An unknown message parameter is a protocol violation, so each stays rejected + // on the drafts that predate it: FILL_TIMEOUT arrived in draft-18. + let has_fill_timeout = !matches!(version, Version::Draft15 | Version::Draft16 | Version::Draft17); + if (fill_timeout.is_some() && !has_fill_timeout) || (range_filters && !has_range_filters(version)) { + return Err(DecodeError::InvalidValue); + } + + ( + fetch_type, + subscriber_priority, + group_order, + range_filters, + fill_timeout.is_some(), + true, + ) + } + // Draft-20 names the track up front and moves the range into LOCATION_FILTER. + _ => { + let namespace = decode_namespace(buf)?; + let track = Cow::Owned(buf.string()?); + decode_params!(buf, version, + 0x03 => _authorization_token: Vec, + 0x0A => fill_timeout: Option, + 0x20 => subscriber_priority: Option, + 0x21 => filter: Option, + 0x22 => group_order: Option, + 0x25 => subgroup_filter: Vec, + 0x26 => object_id_filter: Vec, + 0x27 => priority_filter: Vec, + 0x28 => object_property_filter: Vec, + 0x35 => include_properties: Option, + ); + let range_filters = [ + subgroup_filter, + object_id_filter, + priority_filter, + object_property_filter, + ] + .iter() + .any(|filter| !filter.is_empty()); + + let fetch_type = FetchType::Filtered { + namespace, + track, + // An absent LOCATION_FILTER fetches the whole track. + filter: filter.unwrap_or(Filter::Unfiltered), + }; + ( + fetch_type, + subscriber_priority, + group_order, + range_filters, + fill_timeout.is_some(), + // Defaults to 1, so an absent parameter means the requester wants them. + include_properties.unwrap_or(true), + ) + } + }; Ok(Self { request_id, @@ -285,6 +301,7 @@ impl Message for Fetch<'_> { fetch_type, range_filters, fill_timeout, + properties_wanted, }) } } @@ -681,6 +698,7 @@ mod tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft14); @@ -704,6 +722,7 @@ mod tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft15); @@ -745,6 +764,7 @@ mod tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft16); @@ -768,6 +788,7 @@ mod tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft17); @@ -845,6 +866,7 @@ mod tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }; let encoded = encode_message(&msg, Version::Draft18); @@ -924,6 +946,7 @@ mod tests { assert_eq!(fetch.group_order, GroupOrder::Ascending); assert!(fetch.range_filters, "{version}"); assert!(fetch.fill_timeout, "{version}"); + assert!(!fetch.properties_wanted, "{version}"); assert_eq!( fetch.fetch_type, FetchType::Filtered { @@ -954,6 +977,7 @@ mod tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }; #[rustfmt::skip] @@ -965,6 +989,21 @@ mod tests { assert_eq!(encode_message(&fetch, Version::Draft20), expected); assert_eq!(decode_message::(&expected, Version::Draft20).unwrap(), fetch); + // Only the opt-out is written, as a uint8 with no Length. + let opted_out = Fetch { + properties_wanted: false, + ..fetch.clone() + }; + #[rustfmt::skip] + let expected = [FETCH_HEAD, &[ + 0x03, // Number of Parameters + 0x20, 0x80, // SUBSCRIBER_PRIORITY = 128 + 0x02, 0x02, // GROUP_ORDER (0x22) = Descending + 0x13, 0x00, // INCLUDE_PROPERTIES (0x35) = 0 + ]].concat(); + assert_eq!(encode_message(&opted_out, Version::Draft20), expected); + assert_eq!(decode_message::(&expected, Version::Draft20).unwrap(), opted_out); + // The tagged forms have no spelling from draft-20 on, and this one none before it. let mut buf = Vec::new(); assert!( diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 9cd0ce5f921..cee9b9c1d9a 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -3543,6 +3543,7 @@ mod serve_tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, } .encode_msg(&mut Encoder::new(&mut fetch_body, version.into()), version) .unwrap(); @@ -4175,6 +4176,7 @@ mod serve_tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }, ) .await?; @@ -4653,6 +4655,7 @@ mod serve_tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }, ) .await @@ -4893,6 +4896,7 @@ mod serve_tests { fetch_type, range_filters: false, fill_timeout: false, + properties_wanted: true, }, ) .await @@ -4936,6 +4940,7 @@ mod serve_tests { }, range_filters: false, fill_timeout: false, + properties_wanted: true, }, ) .await @@ -6729,6 +6734,7 @@ mod tests { fetch_type, range_filters: false, fill_timeout: false, + properties_wanted: true, }, ) .await diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 512bc6c84ce..63cffec8e52 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1860,10 +1860,27 @@ where // Write Subscribe message. The aggregate is read now: a subscriber can join while // the request ID and stream were awaited, and nothing updates the priority after. let priority = target.subscription().map(|s| s.priority).unwrap_or(0); - if let Err(err) = self - .write_subscribe(&mut stream, request_id, broadcast_path, track_name, priority, join) - .await - { + let written = async { + stream.writer.varint(ietf::Subscribe::ID).await?; + stream + .writer + .encode(&ietf::Subscribe { + request_id, + track_namespace: broadcast_path.to_owned(), + track_name: track_name.into(), + subscriber_priority: super::priority::to_wire(priority), + group_order: GroupOrder::Descending, + filter: join.filter, + fill: join.fill, + // A resumed copy already knows the track, from the answer that accepted it. + properties_wanted: matches!(target, Target::Request(_)), + forward: true, + range_filters: false, + }) + .await + } + .await; + if let Err(err) = written { tracing::debug!(%err, "failed to write subscribe"); self.remove_subscribe(request_id); target.fail(err); @@ -2010,6 +2027,11 @@ where largest, } = accepted; let info = accepted_info(max_age, priority); + // A resumed copy opted out of the properties, so it keeps the units it learned. + let timescale = match &resumed { + Some(idle) => idle.timescale, + None => timescale, + }; // The copy already holds the track: it is current again up to the answer's Largest // Location once the join's head lands or a newer group does, since leaving cut the // group it was in. @@ -2381,34 +2403,6 @@ where Ok(()) } - async fn write_subscribe( - &self, - stream: &mut Stream, - request_id: RequestId, - broadcast: &Path<'_>, - track_name: &str, - priority: u8, - join: Join, - ) -> Result<(), Error> { - stream.writer.varint(ietf::Subscribe::ID).await?; - stream - .writer - .encode(&ietf::Subscribe { - request_id, - track_namespace: broadcast.to_owned(), - track_name: track_name.into(), - subscriber_priority: super::priority::to_wire(priority), - group_order: GroupOrder::Descending, - filter: join.filter, - fill: join.fill, - properties_wanted: true, - forward: true, - range_filters: false, - }) - .await?; - Ok(()) - } - /// Send the joining FETCH that follows a pre-draft-20 SUBSCRIBE, and keep reading its /// answer until the subscription ends. /// @@ -2480,6 +2474,7 @@ where fetch_type, range_filters: false, fill_timeout: false, + properties_wanted: true, }) .await?; Ok::<(), Error>(()) @@ -3584,6 +3579,9 @@ where }, range_filters: false, fill_timeout: false, + // The copy learned the track before any fetch, from SUBSCRIBE_OK or + // TRACK_STATUS_OK, so a FETCH_OK repeating it is waste. + properties_wanted: false, }) .await?; Ok::<_, Error>(()) @@ -3642,15 +3640,10 @@ where // the stream owes every object up to it. One past the group covers it whole. let end = (end.group == sequence).then_some(end.object); - // Only a track nothing has subscribed to yet takes this info, as SUBSCRIBE_OK would - // have set it. FETCH_OK carries the same Track Properties, so the retention window - // comes from it. - let info = track::Info::default() - .with_timescale(Timescale::MICRO) - .with_max_age(ok.properties.max_cache_duration); // Joined fetches still count until they pick the accepted group up from the cache. let joined = request.result.clone(); - let mut producer = match request.accept(info) { + // The track was accepted before any fetch was asked, so it keeps that info. + let mut producer = match request.accept(None) { Ok(producer) => producer, // Already served by a concurrent fetch, or the track closed. Err(err) => { @@ -5448,6 +5441,113 @@ mod tests { } } + /// A copy that already knows its track, here from TRACK_STATUS_OK, opts out of the + /// properties when a subscriber returns, and keeps the units it learned rather than + /// taking the empty block the opted-out SUBSCRIBE_OK carries. + #[moq_net_sim::test] + async fn a_resumed_subscribe_opts_out_of_known_properties() { + use crate::coding::Encode as _; + fn message(id: u64, msg: &M, version: Version) -> Vec { + let mut buf = Vec::new(); + crate::coding::Encoder::new(&mut buf, version.into()) + .varint(id) + .unwrap(); + msg.encode(&mut crate::coding::Encoder::new(&mut buf, version.into()), version) + .unwrap(); + buf + } + + let declared = Timescale::new(90_000).unwrap(); + for version in [Version::Draft20, Version::Draft22] { + let status_ok = message( + ietf::TrackStatusOk::ID, + &ietf::TrackStatusOk { + request_id: None, + largest: None, + properties: ietf::Properties { + timescale: Some(declared), + ..Default::default() + }, + }, + version, + ); + let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![ + status_ok, + // Opted out, so its block is empty. + message( + ietf::SubscribeOk::ID, + &ietf::SubscribeOk { + request_id: None, + track_alias: 7, + largest: None, + properties: Default::default(), + }, + version, + ), + ]); + let log = session.log.clone(); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let mut subscriber = Subscriber::new( + crate::time::Clock::sim(), + session, + crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + Control::new(None, false), + None, + peer::PeerSetup::default(), + crate::Hop::new(1).unwrap(), + None, + version, + tasks, + Default::default(), + ); + + let producer = crate::broadcast::Info::default().produce(); + let mut dynamic = producer.dynamic(); + let consumer = producer.consume(); + let track = consumer.track("video").unwrap(); + // Demand with nobody subscribing learns the track from TRACK_STATUS. + let mut query = std::pin::pin!(track.query()); + assert!(futures::poll!(query.as_mut()).is_pending()); + let request = dynamic.requested_track().await.expect("no track requested"); + + let probe = subscriber.clone(); + let serving = moq_net_sim::spawn(async move { + subscriber.run_subscribe(Path::new("broadcast"), dynamic, request).await; + }); + settle().await; + let _subscription = track.subscribe(None); + settle().await; + + let writes = log.writes.lock().unwrap().clone(); + let mut buf = Decoder::new(&writes, version.into()); + let mut subscribes = Vec::new(); + while !buf.is_empty() { + let type_id = buf.varint().unwrap(); + let size = buf.u16().unwrap(); + let mut body = Decoder::new(buf.slice(size as usize).unwrap(), version.into()); + if type_id == ietf::Subscribe::ID { + subscribes.push(ietf::Subscribe::decode_msg(&mut body, version).unwrap()); + } + } + assert_eq!(subscribes.len(), 1, "{version}: the returning subscriber subscribes"); + assert!( + !subscribes[0].properties_wanted, + "{version}: TRACK_STATUS_OK already gave the properties" + ); + + let timescale = probe + .state + .lock() + .subscribes + .values() + .next() + .expect("the subscription is registered") + .timescale; + assert_eq!(timescale, Some(declared), "{version}: the learned units are kept"); + serving.abort(); + } + } + /// The publisher opens no fetch stream for an empty range, so a fill against a track /// with no content is owed nothing. Leaving it outstanding would withhold every later /// group behind a head that is never coming. @@ -8973,70 +9073,6 @@ mod joining_fetch_tests { assert!(fetch.await.is_err(), "the accepted group was aborted"); } - /// A group FETCH for a track nothing subscribed to takes its retention window from - /// FETCH_OK's properties, as a subscription would from SUBSCRIBE_OK's. - #[moq_net_sim::test] - async fn a_group_fetch_takes_fetch_ok_max_age() { - const GROUP: u64 = 4; - const MAX_AGE: Duration = Duration::from_secs(12); - - for version in [Version::Draft17, Version::Draft19] { - let ok = message_bytes( - ietf::FetchOk::ID, - &ietf::FetchOk { - request_id: None, - group_order: GroupOrder::Ascending, - end_of_track: false, - end_location: ietf::Location { - group: GROUP + 1, - object: 0, - }, - properties: ietf::Properties { - max_cache_duration: Some(MAX_AGE), - ..Default::default() - }, - }, - version, - ); - - let session = ScriptedSession::per_stream_reset(vec![ok]); - let (tasks, _task_set) = crate::util::TaskSet::new(); - let subscriber = Subscriber::new( - crate::time::Clock::sim(), - session.clone(), - crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), - Control::new(None, false), - None, - peer::PeerSetup::default(), - crate::Hop::new(1).unwrap(), - None, - version, - tasks, - Default::default(), - ); - - // A track the publisher has not accepted, so the fetch supplies its info. - let broadcast = crate::broadcast::Info::new().produce(); - let reserved = broadcast.reserve_track("video").unwrap(); - let dynamic = reserved.dynamic(); - let track = broadcast.consume().track("video").unwrap(); - let mut fetch = std::pin::pin!(track.fetch_group(GROUP, None)); - assert!(futures::poll!(fetch.as_mut()).is_pending()); - let request = dynamic.requested_group().await.expect("no group requested"); - - subscriber - .clone() - .run_group_fetch(Path::new("broadcast").to_owned(), "video".into(), request, None) - .await; - - let info = moq_net_sim::timeout(Duration::from_secs(1), track.query()) - .await - .expect("the fetch installed the track info") - .expect("info"); - assert_eq!(info.max_age, Some(MAX_AGE), "{version}"); - } - } - /// A cache miss for a group's tail asks upstream from the frame the reader wants and /// numbers what arrives from there, so a publisher that evicted the prefix can answer. #[moq_net_sim::test] diff --git a/rs/moq-net/src/ietf/version.rs b/rs/moq-net/src/ietf/version.rs index 7f0e1b44d91..4e2af0633ad 100644 --- a/rs/moq-net/src/ietf/version.rs +++ b/rs/moq-net/src/ietf/version.rs @@ -137,6 +137,7 @@ mod tests { }, range_filters: false, fill_timeout: false, + properties_wanted: false, }; let group = GroupHeader { diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index f8621f9cd67..ec2221b6893 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -462,3 +462,58 @@ async fn leaving_after_the_cache_window_keeps_the_latest_group() { .unwrap_or_else(|_| panic!("{version}: timed out")); } } + +/// A rejoin keeps the source's timestamps. +/// +/// The returning subscription opts out of the track's properties from draft-20, since the +/// copy already learned them, so its units must come from that first answer rather than +/// the empty block the second one carries. +#[moq_net_sim::test] +async fn rejoin_keeps_source_timestamps() { + for version in ["moq-transport-19", "moq-transport-22"] { + moq_net_sim::timeout(TEST_TIMEOUT, async { + let publisher = produce_origin(1); + let client = produce_origin(2); + + let broadcast = publisher.create_broadcast("bench").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(client.clone()); + let _pair = connect_mock(options).await; + + let consumer = client.consume(); + consumer.routed("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench").await.unwrap(); + // Far from any arrival time, so a frame stamped on arrival cannot pass. + let ts = |ms| Timestamp::from_millis(ms).unwrap(); + + for (sequence, ms) in [(0, 123_456), (1, 654_321)] { + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + track.demand().used().await.unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(ts(ms), b"frame".as_ref()).unwrap(); + group.finish().unwrap(); + + let mut group = loop { + let group = sub.recv_group().await.unwrap().unwrap(); + if group.sequence == sequence { + break group; + } + }; + let frame = group.read_frame().await.unwrap().unwrap(); + assert_eq!( + frame.timestamp.as_millis(), + u128::from(ms), + "{version} group {sequence}" + ); + drop((group, sub)); + track.demand().unused().await.unwrap(); + } + }) + .await + .unwrap_or_else(|_| panic!("{version}: timed out")); + } +} From 793ae1b2be859e5e3313101006148a8e01214e2e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 11:28:28 -0700 Subject: [PATCH 09/16] feat(moq-net): a refused TRACK_STATUS refuses the fetch-only request Supported or refused: drop the NOT_SUPPORTED fallback to SUBSCRIBE. Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 3 +- rs/moq-net/src/ietf/subscriber.rs | 135 ++++++------------------------ 2 files changed, 25 insertions(+), 113 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 2a3d33dab77..f0369d32b23 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -63,8 +63,7 @@ carries no timestamps, since no `SUBSCRIBE_OK` declared a timescale for it. A Rust subscriber learns a track that is only fetched, never subscribed, from `TRACK_STATUS` instead of a `SUBSCRIBE`, so a finished track stays fetchable and a relay puts no live subscription upstream to race its fetches. A -publisher that refuses `TRACK_STATUS` with `NOT_SUPPORTED` gets the -`SUBSCRIBE` instead. The Rust publisher answers `TRACK_STATUS` with what its +publisher that refuses `TRACK_STATUS` refuses the fetch. The Rust publisher answers `TRACK_STATUS` with what its `SUBSCRIBE_OK` would carry, as far as the draft's answer has room: the Largest Location on every draft, and the Track Properties from draft-18, when the answer gained them. A `SUBSCRIBE` that sets `INCLUDE_PROPERTIES` to 0 gets an diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index a2b371ef643..5bef9356004 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -455,16 +455,6 @@ fn accepted_info(max_age: Option, priority: Option) -> .with_priority(super::priority::from_wire(priority.unwrap_or(128))) } -/// What a TRACK_STATUS for a track nobody subscribes to came to. -enum Status { - /// Accepted as a copy with no subscription. - Known(Idle), - /// The publisher does not implement TRACK_STATUS, so a SUBSCRIBE has to learn the track. - Unsupported(track::Request), - /// The request was answered: refused, or abandoned. - Ended, -} - /// A copy with no subscription upstream, kept for fetches and a returning subscriber. struct Idle { track: track::Producer, @@ -1762,17 +1752,18 @@ where // one made only to learn the track races the fetches it was for. let mut target = match request.subscription() { Some(_) => Target::Request(request), - None => match self.track_status(&broadcast_path, &track_name, request).await { - Status::Known(idle) => match self + None => { + let Some(idle) = self.track_status(&broadcast_path, &track_name, request).await else { + return; + }; + let Some(next) = self .linger(&broadcast_path, &track_name, idle, &mut group_fetches) .await - { - Some(next) => Target::Resume(next), - None => return, - }, - Status::Unsupported(request) => Target::Request(request), - Status::Ended => return, - }, + else { + return; + }; + Target::Resume(next) + } }; loop { let Some(idle) = self @@ -2576,20 +2567,26 @@ where /// Learn a track nobody subscribes to from TRACK_STATUS, and accept it as a copy with no /// subscription, which [`Self::linger`] serves fetches from and subscribes once someone - /// does. - async fn track_status(&mut self, broadcast_path: &Path<'_>, track_name: &str, request: track::Request) -> Status { + /// does. `None` once the request was refused or abandoned: a publisher that refuses + /// TRACK_STATUS refuses the fetch-only request too. + async fn track_status( + &mut self, + broadcast_path: &Path<'_>, + track_name: &str, + request: track::Request, + ) -> Option { let request_id = match self.control.next_request_id(&self.runtime).await { Ok(id) => id, Err(err) => { request.reject(err); - return Status::Ended; + return None; } }; let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { Ok(stream) => stream, Err(err) => { request.reject(err); - return Status::Ended; + return None; } }; let written = async { @@ -2607,7 +2604,7 @@ where .await; if let Err(err) = written { request.reject(err); - return Status::Ended; + return None; } // Nobody wanting the track any more ends the wait on a peer that may never answer, @@ -2640,19 +2637,17 @@ where let Some(response) = response else { stream.reader.abort(&Error::Cancel); stream.writer.abort(&Error::Cancel); - return Status::Ended; + return None; }; // The publisher has read the request, since it answered, and FINs after its answer: // FIN our side too and let the stream drop. let _ = stream.writer.finish(); let ok = match response { Ok(ok) => ok, - // A publisher that does not implement TRACK_STATUS still answers a SUBSCRIBE. - Err(Error::Unsupported) => return Status::Unsupported(request), Err(err) => { tracing::debug!(%err, broadcast = %self.origin.absolute(broadcast_path), track = %track_name, "track status refused"); request.reject(err); - return Status::Ended; + return None; } }; @@ -2661,7 +2656,7 @@ where let mut track = request.accept(accepted_info(ok.properties.max_cache_duration, ok.properties.priority)); // Nothing feeds this copy live: what it caches is fetched, never the live edge. track.set_idle(); - Status::Known(Idle { + Some(Idle { demand: track.demand(), track, dynamic, @@ -5359,88 +5354,6 @@ mod tests { outstanding } - /// A publisher that does not implement TRACK_STATUS (an older release, or another - /// implementation) refuses it NOT_SUPPORTED. Fetch-only demand then learns the track - /// from a SUBSCRIBE, as it did before TRACK_STATUS was asked. - #[moq_net_sim::test] - async fn an_unsupported_track_status_falls_back_to_subscribe() { - for version in [ - Version::Draft14, - Version::Draft15, - Version::Draft17, - Version::Draft18, - Version::Draft22, - ] { - let log = crate::lite::test_transport::Log::default(); - let mut writer = - crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); - let reason = "TRACK_STATUS is not supported"; - match version { - Version::Draft14 => { - writer.varint(ietf::TRACK_STATUS_ERROR_14).await.unwrap(); - writer - .encode(&ietf::SubscribeError { - request_id: RequestId(0), - error_code: 0x3, - reason_phrase: reason.into(), - }) - .await - .unwrap(); - } - _ => { - writer.varint(ietf::RequestError::ID).await.unwrap(); - writer - .encode(&ietf::RequestError { - request_id: matches!(version, Version::Draft15 | Version::Draft16).then_some(RequestId(0)), - error_code: 0x3, - reason_phrase: reason.into(), - retry_interval: 0, - }) - .await - .unwrap(); - } - } - let refusal = log.writes.lock().unwrap().clone(); - - let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![refusal]); - let log = session.log.clone(); - let (tasks, _task_set) = crate::util::TaskSet::new(); - let mut subscriber = Subscriber::new( - crate::time::Clock::sim(), - session, - crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), - Control::new(None, false), - None, - peer::PeerSetup::default(), - crate::Hop::new(1).unwrap(), - None, - version, - tasks, - Default::default(), - ); - - let producer = crate::broadcast::Info::default().produce(); - let mut dynamic = producer.dynamic(); - let consumer = producer.consume(); - // Fetch-only demand: nobody subscribes. - let _fetching = consumer.track("video").unwrap().fetch_group(0, None); - let request = dynamic.requested_track().await.expect("no track requested"); - assert!(request.subscription().is_none()); - - let serving = moq_net_sim::spawn(async move { - subscriber.run_subscribe(Path::new("broadcast"), dynamic, request).await; - }); - settle().await; - - assert_eq!( - control_message_types(&log, version), - [ietf::TrackStatus::ID, ietf::Subscribe::ID], - "{version}: a refused TRACK_STATUS must fall back to SUBSCRIBE" - ); - serving.abort(); - } - } - /// A copy that already knows its track, here from TRACK_STATUS_OK, opts out of the /// properties when a subscriber returns, and keeps the units it learned rather than /// taking the empty block the opted-out SUBSCRIBE_OK carries. From ba732bbe4ad5c307efc93f12a9ecc9cf0dea8d0a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 12:08:19 -0700 Subject: [PATCH 10/16] fix(moq-net): a resumed draft-17 copy takes SUBSCRIBE_OK's timescale Draft-17's TRACK_STATUS_OK has no properties block and its SUBSCRIBE cannot opt out, so the answer to a resumed SUBSCRIBE is the first to declare the units. Keep the learned units only when the answer declares none. Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 9 ++++---- rs/moq-net/src/ietf/subscriber.rs | 35 +++++++++++++++++-------------- 2 files changed, 24 insertions(+), 20 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index f0369d32b23..b3863b9617d 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -63,10 +63,11 @@ carries no timestamps, since no `SUBSCRIBE_OK` declared a timescale for it. A Rust subscriber learns a track that is only fetched, never subscribed, from `TRACK_STATUS` instead of a `SUBSCRIBE`, so a finished track stays fetchable and a relay puts no live subscription upstream to race its fetches. A -publisher that refuses `TRACK_STATUS` refuses the fetch. The Rust publisher answers `TRACK_STATUS` with what its -`SUBSCRIBE_OK` would carry, as far as the draft's answer has room: the Largest -Location on every draft, and the Track Properties from draft-18, when the -answer gained them. A `SUBSCRIBE` that sets `INCLUDE_PROPERTIES` to 0 gets an +publisher that refuses `TRACK_STATUS` refuses the fetch. The Rust publisher +answers `TRACK_STATUS` with what its `SUBSCRIBE_OK` would carry, as far as the +draft's answer has room: the Largest Location on every draft, and the Track +Properties from draft-18, when the answer gained them. A relay with no +subscription upstream reports the Largest Location it has cached. A `SUBSCRIBE` that sets `INCLUDE_PROPERTIES` to 0 gets an empty properties block, but its objects keep their Timestamps, whose units `TRACK_STATUS` reports. From draft-20, a Rust subscriber sets `INCLUDE_PROPERTIES` to 0 on every later `SUBSCRIBE` or `FETCH` for a track diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 5bef9356004..e3b359388bc 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2018,11 +2018,10 @@ where largest, } = accepted; let info = accepted_info(max_age, priority); - // A resumed copy opted out of the properties, so it keeps the units it learned. - let timescale = match &resumed { - Some(idle) => idle.timescale, - None => timescale, - }; + // A resumed copy opts out of the properties from draft-20, so it keeps the units it + // learned. Before that the answer still declares them, and may be the first to: draft-17's + // TRACK_STATUS_OK has no properties block. + let timescale = timescale.or(resumed.as_ref().and_then(|idle| idle.timescale)); // The copy already holds the track: it is current again up to the answer's Largest // Location once the join's head lands or a newer group does, since leaving cut the // group it was in. @@ -5356,7 +5355,9 @@ mod tests { /// A copy that already knows its track, here from TRACK_STATUS_OK, opts out of the /// properties when a subscriber returns, and keeps the units it learned rather than - /// taking the empty block the opted-out SUBSCRIBE_OK carries. + /// taking the empty block the opted-out SUBSCRIBE_OK carries. On draft-17, where + /// TRACK_STATUS_OK has no properties block and SUBSCRIBE cannot opt out, the copy takes + /// the units SUBSCRIBE_OK declares. #[moq_net_sim::test] async fn a_resumed_subscribe_opts_out_of_known_properties() { use crate::coding::Encode as _; @@ -5371,29 +5372,31 @@ mod tests { } let declared = Timescale::new(90_000).unwrap(); - for version in [Version::Draft20, Version::Draft22] { + for version in [Version::Draft17, Version::Draft20, Version::Draft22] { + let opts_out = ietf::Filter::is_draft20(version); + let declares = |yes: bool| ietf::Properties { + timescale: yes.then_some(declared), + ..Default::default() + }; let status_ok = message( ietf::TrackStatusOk::ID, &ietf::TrackStatusOk { request_id: None, largest: None, - properties: ietf::Properties { - timescale: Some(declared), - ..Default::default() - }, + properties: declares(opts_out), }, version, ); let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![ status_ok, - // Opted out, so its block is empty. + // Empty when opted out. Draft-17 cannot opt out, so it declares the units. message( ietf::SubscribeOk::ID, &ietf::SubscribeOk { request_id: None, track_alias: 7, largest: None, - properties: Default::default(), + properties: declares(!opts_out), }, version, ), @@ -5443,8 +5446,8 @@ mod tests { } } assert_eq!(subscribes.len(), 1, "{version}: the returning subscriber subscribes"); - assert!( - !subscribes[0].properties_wanted, + assert_eq!( + subscribes[0].properties_wanted, !opts_out, "{version}: TRACK_STATUS_OK already gave the properties" ); @@ -5456,7 +5459,7 @@ mod tests { .next() .expect("the subscription is registered") .timescale; - assert_eq!(timescale, Some(declared), "{version}: the learned units are kept"); + assert_eq!(timescale, Some(declared), "{version}: the copy keeps the units it learned"); serving.abort(); } } From 3ad859e3b53eae3ed38f85f41044c6d425559502 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 12:18:41 -0700 Subject: [PATCH 11/16] docs: rewrap the TRACK_STATUS paragraph Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index b3863b9617d..67fa849fd40 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -67,9 +67,10 @@ publisher that refuses `TRACK_STATUS` refuses the fetch. The Rust publisher answers `TRACK_STATUS` with what its `SUBSCRIBE_OK` would carry, as far as the draft's answer has room: the Largest Location on every draft, and the Track Properties from draft-18, when the answer gained them. A relay with no -subscription upstream reports the Largest Location it has cached. A `SUBSCRIBE` that sets `INCLUDE_PROPERTIES` to 0 gets an -empty properties block, but its objects keep their Timestamps, whose units -`TRACK_STATUS` reports. From draft-20, a Rust subscriber sets +subscription upstream reports the Largest Location it has cached. A +`SUBSCRIBE` that sets `INCLUDE_PROPERTIES` to 0 gets an empty properties +block, but its objects keep their Timestamps, whose units `TRACK_STATUS` +reports. From draft-20, a Rust subscriber sets `INCLUDE_PROPERTIES` to 0 on every later `SUBSCRIBE` or `FETCH` for a track it already learned, instead of receiving the same properties again. From ac1e723adf80a03326fc31208110f894b6c5495b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 16:18:36 -0700 Subject: [PATCH 12/16] style: cargo fmt the timescale assertion Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/subscriber.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index e3b359388bc..c49d415dde6 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -5459,7 +5459,11 @@ mod tests { .next() .expect("the subscription is registered") .timescale; - assert_eq!(timescale, Some(declared), "{version}: the copy keeps the units it learned"); + assert_eq!( + timescale, + Some(declared), + "{version}: the copy keeps the units it learned" + ); serving.abort(); } } From e0a40d2da9ae25e0f94a44855c962951f8412da4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 21:27:15 -0700 Subject: [PATCH 13/16] fix(moq-net): adapt TRACK_STATUS and tests to main's epoch and untimed frames Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 2 +- rs/moq-net/tests/fetch_only.rs | 2 +- rs/moq-net/tests/rejoin.rs | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 7c154919c2f..8249681589f 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1647,7 +1647,7 @@ where let broadcast = match self .serving_origin() .await - .request_broadcast(&msg.track_namespace) + .request_broadcast(&msg.track_namespace, None) .await { Ok(broadcast) => broadcast, diff --git a/rs/moq-net/tests/fetch_only.rs b/rs/moq-net/tests/fetch_only.rs index dc531089d99..db44f94c0d1 100644 --- a/rs/moq-net/tests/fetch_only.rs +++ b/rs/moq-net/tests/fetch_only.rs @@ -27,7 +27,7 @@ async fn resolve(origin: &moq_net::origin::Producer) -> moq_net::broadcast::Cons .await .expect("announce timeout") .expect("routed"); - moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) .await .expect("resolve timeout") .expect("broadcast resolves") diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index 46afaaa78ae..50b95b55224 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -496,7 +496,7 @@ async fn rejoin_keeps_source_timestamps() { let consumer = client.consume(); consumer.routed("bench").await.unwrap(); - let remote = consumer.request_broadcast("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench", None).await.unwrap(); // Far from any arrival time, so a frame stamped on arrival cannot pass. let ts = |ms| Timestamp::from_millis(ms).unwrap(); @@ -515,8 +515,8 @@ async fn rejoin_keeps_source_timestamps() { }; let frame = group.read_frame().await.unwrap().unwrap(); assert_eq!( - frame.timestamp.as_millis(), - u128::from(ms), + frame.timestamp.map(Timestamp::as_millis), + Some(u128::from(ms)), "{version} group {sequence}" ); drop((group, sub)); From 0a5e3e19c46c60c2ff2192c5033bf7ddee2cffe3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 21:54:04 -0700 Subject: [PATCH 14/16] feat(moq-net): draft-17 fetch-only demand still subscribes Draft-17's REQUEST_OK answer to TRACK_STATUS has no properties block, so a copy learned from it cannot know whether the track is timed, and a later SUBSCRIBE_OK declaring TIMESCALE would feed stamped frames into an untimed track. Draft-17 is not an interop target, so it keeps SUBSCRIBE. Also tests that a SUBSCRIBE after TRACK_STATUS keeps timestamps on draft-18 and later. Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 4 ++- rs/moq-net/src/ietf/subscriber.rs | 21 +++++++------- rs/moq-net/src/ietf/track.rs | 7 +++++ rs/moq-net/tests/fetch_only.rs | 48 +++++++++++++++++++++++++++++-- 4 files changed, 66 insertions(+), 14 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 94cfedc2223..4be7308b7c1 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -75,7 +75,9 @@ so every track there is untimed. A Rust subscriber learns a track that is only fetched, never subscribed, from `TRACK_STATUS` instead of a `SUBSCRIBE`, so a finished track stays fetchable and a relay puts no live subscription upstream to race its fetches. A -publisher that refuses `TRACK_STATUS` refuses the fetch. The Rust publisher +publisher that refuses `TRACK_STATUS` refuses the fetch. Draft-17 is the +exception and still subscribes: its `TRACK_STATUS` answer has no properties +block, so it can't say whether the track is timed. The Rust publisher answers `TRACK_STATUS` with what its `SUBSCRIBE_OK` would carry, as far as the draft's answer has room: the Largest Location on every draft, and the Track Properties from draft-18, when the answer gained them. A relay with no diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 07ae61e4c2c..160b873023b 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1759,10 +1759,10 @@ where // Demand with nobody subscribing (fetches, or a TRACK_STATUS downstream) learns the // track from TRACK_STATUS rather than a SUBSCRIBE: a finished track refuses one, and - // one made only to learn the track races the fetches it was for. + // one made only to learn the track races the fetches it was for. Draft-17 still + // SUBSCRIBEs: its TRACK_STATUS answer cannot say whether the track is timed. let mut target = match request.subscription() { - Some(_) => Target::Request(request), - None => { + None if ietf::TrackStatusOk::describes_track(self.version) => { let Some(idle) = self.track_status(&broadcast_path, &track_name, request).await else { return; }; @@ -1774,6 +1774,7 @@ where }; Target::Resume(next) } + _ => Target::Request(request), }; loop { let Some(idle) = self @@ -2028,8 +2029,7 @@ where largest, } = accepted; // A resumed copy opts out of the properties from draft-20, so it keeps the units it - // learned. Before that the answer still declares them, and may be the first to: draft-17's - // TRACK_STATUS_OK has no properties block. + // learned. Before that the answer declares them again. let timescale = timescale.or(resumed.as_ref().and_then(|idle| idle.timescale)); let info = accepted_info(timescale, max_age, priority); // The copy already holds the track: it is current again up to the answer's Largest @@ -5489,9 +5489,7 @@ mod tests { /// A copy that already knows its track, here from TRACK_STATUS_OK, opts out of the /// properties when a subscriber returns, and keeps the units it learned rather than - /// taking the empty block the opted-out SUBSCRIBE_OK carries. On draft-17, where - /// TRACK_STATUS_OK has no properties block and SUBSCRIBE cannot opt out, the copy takes - /// the units SUBSCRIBE_OK declares. + /// taking the empty block the opted-out SUBSCRIBE_OK carries. #[moq_net_sim::test] async fn a_resumed_subscribe_opts_out_of_known_properties() { use crate::coding::Encode as _; @@ -5506,7 +5504,7 @@ mod tests { } let declared = Timescale::new(90_000).unwrap(); - for version in [Version::Draft17, Version::Draft20, Version::Draft22] { + for version in [Version::Draft18, Version::Draft20, Version::Draft22] { let opts_out = ietf::Filter::is_draft20(version); let declares = |yes: bool| ietf::Properties { timescale: yes.then_some(declared), @@ -5517,13 +5515,14 @@ mod tests { &ietf::TrackStatusOk { request_id: None, largest: None, - properties: declares(opts_out), + properties: declares(true), }, version, ); let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![ status_ok, - // Empty when opted out. Draft-17 cannot opt out, so it declares the units. + // Empty when opted out. Before draft-20 SUBSCRIBE cannot opt out, so it declares + // the units again. message( ietf::SubscribeOk::ID, &ietf::SubscribeOk { diff --git a/rs/moq-net/src/ietf/track.rs b/rs/moq-net/src/ietf/track.rs index 5d82d976d1f..984b2488e19 100644 --- a/rs/moq-net/src/ietf/track.rs +++ b/rs/moq-net/src/ietf/track.rs @@ -104,6 +104,13 @@ impl TrackStatusOk { } } + /// Whether this answer describes the track as fully as SUBSCRIBE_OK on `version`. + /// Draft-17 is the gap: SUBSCRIBE_OK declares TIMESCALE, but REQUEST_OK has no + /// properties block to say whether the track is timed. + pub(crate) fn describes_track(version: Version) -> bool { + Self::has_properties(version) || !super::Properties::sends_timescale(version) + } + /// Whether the answer carries a Track Properties block on `version`. fn has_properties(version: Version) -> bool { !matches!( diff --git a/rs/moq-net/tests/fetch_only.rs b/rs/moq-net/tests/fetch_only.rs index db44f94c0d1..60c991f483b 100644 --- a/rs/moq-net/tests/fetch_only.rs +++ b/rs/moq-net/tests/fetch_only.rs @@ -82,12 +82,12 @@ fn frames(sequence: u64) -> Vec> { } /// Drafts whose FETCH we serve, and the newest, whose FETCH we refuse until draft-20 -/// FETCH lands: it must still not subscribe. +/// FETCH lands: it must still not subscribe. Draft-17 is absent: its TRACK_STATUS answer +/// cannot say whether the track is timed, so it still SUBSCRIBEs to learn the track. const VERSIONS: &[(&str, bool)] = &[ ("moq-transport-14", true), ("moq-transport-15", true), ("moq-transport-16", true), - ("moq-transport-17", true), ("moq-transport-18", true), ("moq-transport-19", true), ("moq-transport-22", false), @@ -160,6 +160,50 @@ async fn relayed(version: &str) { drop((remote, downstream, upstream, broadcast, client, relay, publisher)); } +/// A subscriber arriving at a copy learned from TRACK_STATUS still gets timestamped +/// frames: the copy is timed from TRACK_STATUS_OK, and the resumed SUBSCRIBE keeps it so. +#[moq_net_sim::test] +async fn a_subscribe_after_track_status_keeps_timestamps() { + for version in ["moq-transport-18", "moq-transport-19", "moq-transport-22"] { + moq_net_sim::timeout(TIMEOUT, async { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let client = produce_origin(2); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(client.clone()); + let _pair = connect_mock(options).await; + + let remote = resolve(&client).await; + let copy = remote.track("video").unwrap(); + let info = copy.query().await.expect("TRACK_STATUS resolves the track"); + assert!(info.timescale.is_some(), "{version}: the copy is untimed"); + assert!(track.subscription().is_none(), "{version}: the query subscribed"); + + let mut sub = copy.subscribe(None).await.unwrap(); + track.demand().used().await.unwrap(); + // Far from any arrival time, so a frame stamped on arrival cannot pass. + let stamp = Timestamp::from_millis(123_456).unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(stamp, b"frame".as_ref()).unwrap(); + group.finish().unwrap(); + + let mut group = sub.recv_group().await.unwrap().unwrap(); + let frame = group.read_frame().await.unwrap().unwrap(); + assert_eq!( + frame.timestamp.map(Timestamp::as_millis), + Some(stamp.as_millis()), + "{version}" + ); + }) + .await + .unwrap_or_else(|_| panic!("{version}: timed out")); + } +} + /// A relay serves a downstream fetch with a fetch upstream, never a subscription. #[moq_net_sim::test] async fn a_relay_fetches_without_subscribing_upstream() { From 90cada6166831ccabda487badbde6e5a01c3a99d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 22:15:00 -0700 Subject: [PATCH 15/16] quest: drop the link to the finished ietf-fetch-only quest Co-Authored-By: Claude Opus 5.5 --- quest/m1/ietf-cold-largest.md | 1 - 1 file changed, 1 deletion(-) diff --git a/quest/m1/ietf-cold-largest.md b/quest/m1/ietf-cold-largest.md index 0b84d0bca42..e9ebe920cf4 100644 --- a/quest/m1/ietf-cold-largest.md +++ b/quest/m1/ietf-cold-largest.md @@ -58,4 +58,3 @@ Public API: none expected. Wire: none new. ## Related - [moq-transport ranges](/quest/m1/subscribe-ranges/ietf.md) - the same upstream fill path, for multi-group ranges -- [Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md) - the other FETCH gap at a relay From 5a47135004a2ba993d50d7e565af535c964d7fff Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 8 Oct 2026 08:54:05 -0700 Subject: [PATCH 16/16] fix(moq-net): gate TRACK_STATUS_OK's parameters per draft like SUBSCRIBE_OK Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/track.rs | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/ietf/track.rs b/rs/moq-net/src/ietf/track.rs index 984b2488e19..243f06037b4 100644 --- a/rs/moq-net/src/ietf/track.rs +++ b/rs/moq-net/src/ietf/track.rs @@ -178,10 +178,10 @@ impl Message for TrackStatusOk { // Read as SUBSCRIBE_OK's are: EXPIRES is ignored, and MAX_CACHE_DURATION is a // parameter only on draft-15. decode_params!(r, version, - 0x04 => max_cache_duration: Option, + 0x04 => max_cache_duration: Option where version == Version::Draft15, 0x08 => _expires: Option, 0x09 => largest: Option, - 0x22 => group_order: Option, + 0x22 => group_order: Option where version == Version::Draft15, ); let mut properties = match Self::has_properties(version) { true => Properties::decode(r, version)?, @@ -233,6 +233,22 @@ mod tests { buf.to_vec() } + /// GROUP_ORDER and MAX_CACHE_DURATION are TRACK_STATUS_OK parameters only on draft-15, + /// as on SUBSCRIBE_OK. Draft-16 ignores a GROUP_ORDER it defines elsewhere, even with a + /// value no field could hold, and rejects 0x04, which it does not define at all. Draft-18 + /// rejects a known parameter on a message that does not define it. + #[test] + fn track_status_ok_parameters_follow_the_draft() { + // Request 1, then GROUP_ORDER as the varint 256. + let ok = decode_message::(&[0x01, 0x01, 0x22, 0x41, 0x00], Version::Draft16) + .expect("draft-16 ignores GROUP_ORDER"); + assert_eq!(ok.properties.group_order, None); + // Request 1, then MAX_CACHE_DURATION of 5. + assert!(decode_message::(&[0x01, 0x01, 0x04, 0x05], Version::Draft16).is_err()); + // GROUP_ORDER = Ascending, then an empty properties block. + assert!(decode_message::(&[0x01, 0x22, 0x01], Version::Draft18).is_err()); + } + fn decode_message(bytes: &[u8], version: Version) -> Result { let mut buf = bytes::Bytes::from(bytes.to_vec()); crate::coding::decode_buf(&mut buf, version, M::decode_msg)