fix(net): an IETF TRACK_STATUS refused as NOT_SUPPORTED falls back to SUBSCRIBE

doc/concept/standard.md

@@ -26,7 +26,7 @@ maps everything else to "not supported" or a harmless equivalent. The
 subset drops. What a peer actually observes against this implementation:
 
 - **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined; announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited, and we also ask for every prefix we may discover. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We never send `PUBLISH`: on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session.
-- **History is one group.** A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed.
+- **History is one group.** A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed, and so does a reader whose publisher refuses `TRACK_STATUS` as `NOT_SUPPORTED`, as JavaScript does.
 - **Timing.** A track whose `SUBSCRIBE_OK` declares no `TIMESCALE` is untimed, as is every track on drafts 14 through 16. A Rust reader that never subscribed takes the units from `TRACK_STATUS` on draft 18 and later.
 - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. On draft 17 and later, a `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. Message parameters follow the negotiated draft's lists: drafts 14 and 15 ignore an unknown or misplaced parameter, draft 16 ignores one defined only for another message but closes the session on an unknown one, and draft 17 and later close it on both.
 - **One credential per session**, carried in `SETUP` and forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. A token attached to an individual request is otherwise decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. `SUBSCRIBE_TRACKS` is refused unread, so its token is not decoded.

rs/moq-net/src/ietf/subscriber.rs

@@ -1997,16 +1997,19 @@ where
 		// SUBSCRIBEs: its TRACK_STATUS answer cannot say whether the track is timed.
 		let mut target = match request.subscription() {
 			None if ietf::TrackStatusOk::describes_track(self.version) => {
-				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, &mut gate)
-					.await
-				else {
-					return;
-				};
-				Target::Resume(next)
+				match self.track_status(&broadcast_path, &track_name, request).await {
+					Some(Target::Resume(idle)) => {
+						let Some(next) = self
+							.linger(&broadcast_path, &track_name, idle, &mut group_fetches, &mut gate)
+							.await
+						else {
+							return;
+						};
+						Target::Resume(next)
+					}
+					Some(target) => target,
+					None => return,
+				}
 			}
 			_ => Target::Request(request),
 		};
@@ -2853,15 +2856,17 @@ 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. `None` once the request was refused or abandoned: a publisher that refuses
-	/// TRACK_STATUS refuses the fetch-only request too.
+	/// subscription ([`Target::Resume`]), which [`Self::linger`] serves fetches from and
+	/// subscribes once someone does. A publisher that does not implement TRACK_STATUS hands
+	/// the request back ([`Target::Request`]) to be learned from SUBSCRIBE. `None` once the
+	/// request was refused or abandoned: a publisher that refuses the track's status refuses
+	/// the fetch-only request too.
 	async fn track_status(
 		&mut self,
 		broadcast_path: &Path<'_>,
 		track_name: &str,
 		request: track::Request,
-	) -> Option<Idle> {
+	) -> Option<Target> {
 		let validity = self.validity.start();
 		// An allocated ID is always sent, as in `subscribe_once`.
 		let request_id = match validity.run(self.control.next_request_id(&self.runtime)).await {
@@ -2936,6 +2941,12 @@ where
 		let _ = stream.writer.finish();
 		let ok = match response {
 			Ok(ok) => ok,
+			// NOT_SUPPORTED refuses the method, not the track. The origin queries a track before
+			// its first reader's subscription lands, so this is often a subscriber's request.
+			Err(Error::Unsupported) => {
+				tracing::debug!(broadcast = %self.origin.absolute(broadcast_path), track = %track_name, "track status not supported, subscribing");
+				return Some(Target::Request(request));
+			}
 			Err(err) => {
 				tracing::debug!(%err, broadcast = %self.origin.absolute(broadcast_path), track = %track_name, "track status refused");
 				request.reject(err);
@@ -2952,12 +2963,12 @@ where
 		));
 		// Nothing feeds this copy live: what it caches is fetched, never the live edge.
 		track.set_idle();
-		Some(Idle {
+		Some(Target::Resume(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
@@ -5961,22 +5972,91 @@ mod tests {
 		outstanding
 	}
 
+	/// A control message as the peer writes it: type id, then the length-prefixed body.
+	fn message<M: Message>(id: u64, msg: &M, version: Version) -> Vec<u8> {
+		use crate::coding::Encode as _;
+		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
+	}
+
+	/// A publisher without TRACK_STATUS still serves the track: its NOT_SUPPORTED refuses
+	/// the method, so the request is learned from SUBSCRIBE instead. The origin queries a new
+	/// track before its first reader's subscription lands, so a reader reaching such a
+	/// publisher through a relay takes this path.
+	#[moq_net_sim::test]
+	async fn an_unsupported_track_status_subscribes_instead() {
+		for version in [Version::Draft18, Version::Draft19, Version::Draft22] {
+			let refused = message(
+				ietf::RequestError::ID,
+				&ietf::RequestError {
+					request_id: None,
+					error_code: request::to_code(&Error::Unsupported, request::Kind::TrackStatus, version),
+					reason_phrase: "TRACK_STATUS is not supported".into(),
+					retry_interval: 0,
+				},
+				version,
+			);
+			let subscribe_ok = message(
+				ietf::SubscribeOk::ID,
+				&ietf::SubscribeOk {
+					request_id: None,
+					track_alias: 7,
+					largest: None,
+					properties: Default::default(),
+				},
+				version,
+			);
+			let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![refused, subscribe_ok]);
+			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();
+			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 serving = moq_net_sim::spawn(async move {
+				subscriber.run_subscribe(Path::new("broadcast"), dynamic, request).await;
+			});
+			settle().await;
+
+			assert!(query.await.is_ok(), "{version}: NOT_SUPPORTED refused the track");
+			assert!(track.subscribe(None).await.is_ok(), "{version}: the reader was refused");
+			assert_eq!(
+				control_message_types(&log, version)[..2],
+				[ietf::TrackStatus::ID, ietf::Subscribe::ID],
+				"{version}"
+			);
+			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.
 	#[moq_net_sim::test]
 	async fn a_resumed_subscribe_opts_out_of_known_properties() {
-		use crate::coding::Encode as _;
-		fn message<M: Message>(id: u64, msg: &M, version: Version) -> Vec<u8> {
-			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::Draft18, Version::Draft20, Version::Draft22] {
 			let opts_out = ietf::Filter::is_draft20(version);