moq-net: reshape consume_track() to return TrackPending (step 1 of the redesign)

First step of the model API redesign. consume_track() now returns a TrackPending
handle (Clone) exposing subscribe()/info()/ok(); a new resolved TrackConsumer
(from ok()) carries the producer for immediate subscribe/info. The previous
subscribe-future TrackPending is renamed SubscribePending (one moq-mux ref).

Every existing caller (consume_track(x).subscribe(None).await / .info().await)
keeps the same signature. This step is API-only; the demand-side rewrite
(TrackState shared by TrackRequest/TrackProducer, async subscription(), accept
upgrading the request into the producer) is the follow-up that removes the
current InfoRequest-era plumbing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: kixelated

rs/moq-mux/src/container/source.rs

@@ -49,7 +49,7 @@ impl VideoTransform {
 /// A subscription that resolves on first poll, then the live consumer.
 enum SourceState {
 	/// Waiting for the subscription to resolve (blocks on the publisher's SUBSCRIBE_OK).
-	Subscribing(moq_net::TrackPending),
+	Subscribing(moq_net::SubscribePending),
 	/// The resolved consumer, reading frames.
 	Active(Consumer<HangContainer>),
 }

rs/moq-net/src/model/broadcast.rs

@@ -413,7 +413,7 @@ impl Drop for TrackRequest {
 	}
 }
 
-/// A pending info lookup returned by [`TrackConsumer::info`].
+/// A pending info lookup returned by [`TrackPending::info`].
 ///
 /// Resolves once the track's immutable [`Track`] is known: synchronously if a
 /// producer already exists or the info is cached, otherwise once the dynamic
@@ -478,46 +478,46 @@ impl std::future::Future for InfoPending {
 	}
 }
 
-/// A pending subscription returned by [`TrackConsumer::subscribe`].
+/// A pending subscription returned by [`TrackPending::subscribe`].
 ///
 /// The subscription isn't live until the publisher accepts it (for the wire,
 /// SUBSCRIBE_OK). It implements [`Future`], so `.await` it to get the
 /// [`TrackSubscriber`] (or an error). Poll-based callers can instead drive it
 /// with [`Self::poll_ok`] inside a `kio` poll loop.
-pub struct TrackPending {
-	inner: TrackPendingInner,
+pub struct SubscribePending {
+	inner: SubscribePendingInner,
 	/// Kept alive between `Future::poll` calls so its registration in the
 	/// resolver channel stays valid until the next poll replaces it.
 	waiter: Option<kio::Waiter>,
 }
 
-enum TrackPendingInner {
+enum SubscribePendingInner {
 	/// Resolved synchronously: the track already existed, or it failed immediately.
 	Ready(Result<TrackSubscriber, Error>),
 	/// Waiting for the publisher to accept or deny via the dynamic handler.
 	Waiting(kio::Consumer<PendingSlot>),
 }
 
-impl TrackPending {
+impl SubscribePending {
 	fn ready(result: Result<TrackSubscriber, Error>) -> Self {
 		Self {
-			inner: TrackPendingInner::Ready(result),
+			inner: SubscribePendingInner::Ready(result),
 			waiter: None,
 		}
 	}
 
 	fn waiting(consumer: kio::Consumer<PendingSlot>) -> Self {
 		Self {
-			inner: TrackPendingInner::Waiting(consumer),
+			inner: SubscribePendingInner::Waiting(consumer),
 			waiter: None,
 		}
 	}
 
 	/// Poll for the resolved [`TrackSubscriber`], without blocking.
 	pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<TrackSubscriber, Error>> {
 		match &self.inner {
-			TrackPendingInner::Ready(result) => Poll::Ready(result.clone()),
-			TrackPendingInner::Waiting(consumer) => match consumer.poll(waiter, |slot| match &**slot {
+			SubscribePendingInner::Ready(result) => Poll::Ready(result.clone()),
+			SubscribePendingInner::Waiting(consumer) => match consumer.poll(waiter, |slot| match &**slot {
 				Some(result) => Poll::Ready(result.clone()),
 				None => Poll::Pending,
 			}) {
@@ -533,7 +533,7 @@ impl TrackPending {
 	}
 }
 
-impl std::future::Future for TrackPending {
+impl std::future::Future for SubscribePending {
 	type Output = Result<TrackSubscriber, Error>;
 
 	fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
@@ -725,40 +725,40 @@ impl Deref for BroadcastConsumer {
 impl BroadcastConsumer {
 	/// Get a handle to a track on this broadcast.
 	///
-	/// This is a cheap, synchronous lookup that returns a [`TrackConsumer`] bound
+	/// This is a cheap, synchronous lookup that returns a [`TrackPending`] bound
 	/// to `name`. Nothing is sent to the publisher yet: call
-	/// [`TrackConsumer::subscribe`] to open a live subscription (blocking on
+	/// [`TrackPending::subscribe`] to open a live subscription (blocking on
 	/// SUBSCRIBE_OK), or hold the handle and subscribe later.
-	pub fn consume_track(&self, name: &str) -> TrackConsumer {
-		TrackConsumer {
+	pub fn consume_track(&self, name: &str) -> TrackPending {
+		TrackPending {
 			broadcast: self.clone(),
 			name: name.to_string(),
 		}
 	}
 
-	/// Register a subscription for `name` and return a [`TrackPending`] that
+	/// Register a subscription for `name` and return a [`SubscribePending`] that
 	/// resolves once the publisher accepts it.
 	///
 	/// Reuses a live producer if one is already publishing the track (the pending
 	/// resolves right away), otherwise queues a dynamic request served via
 	/// [`BroadcastDynamic::requested_track`] and [`TrackRequest::accept`] (for the
 	/// wire this is SUBSCRIBE_OK). Resolves to [`Error::NotFound`] if no dynamic
 	/// producer exists to handle the request.
-	fn request_subscribe(&self, name: &str, subscription: Subscription) -> TrackPending {
+	fn request_subscribe(&self, name: &str, subscription: Subscription) -> SubscribePending {
 		// Upgrade to a temporary producer so we can modify the state.
 		let Some(producer) = self.state.produce() else {
 			let err = self.state.read().abort.clone().unwrap_or(Error::Dropped);
-			return TrackPending::ready(Err(err));
+			return SubscribePending::ready(Err(err));
 		};
 		let mut state = match modify(&producer) {
 			Ok(state) => state,
-			Err(err) => return TrackPending::ready(Err(err)),
+			Err(err) => return SubscribePending::ready(Err(err)),
 		};
 
 		// Reuse a live producer if one is already publishing the track.
 		if let Some(weak) = state.tracks.get(name) {
 			if !weak.is_closed() {
-				return TrackPending::ready(Ok(weak.subscribe(subscription)));
+				return SubscribePending::ready(Ok(weak.subscribe(subscription)));
 			}
 			// Drop the stale entry and fall through to a fresh request.
 			state.tracks.remove(name);
@@ -771,7 +771,7 @@ impl BroadcastConsumer {
 			// Coalesce onto an in-flight request for the same name.
 			pending.resolvers.push((subscription, slot));
 		} else if state.dynamic == 0 {
-			return TrackPending::ready(Err(Error::NotFound));
+			return SubscribePending::ready(Err(Error::NotFound));
 		} else {
 			state.requests.insert(
 				name.to_string(),
@@ -783,7 +783,7 @@ impl BroadcastConsumer {
 			state.request_order.push_back(name.to_string());
 		}
 
-		TrackPending::waiting(consumer)
+		SubscribePending::waiting(consumer)
 	}
 
 	/// Resolve a track's immutable info, returning an [`InfoPending`].
@@ -841,6 +841,16 @@ impl BroadcastConsumer {
 		InfoPending::waiting(consumer)
 	}
 
+	/// The live producer's weak handle for `name`, if one is currently publishing.
+	pub(crate) fn track_weak(&self, name: &str) -> Option<TrackWeak> {
+		self.state
+			.read()
+			.tracks
+			.get(name)
+			.filter(|weak| !weak.is_closed())
+			.cloned()
+	}
+
 	/// Block until the broadcast is closed and return the cause.
 	///
 	/// Returns [`Error::Dropped`] if every producer was dropped without an
@@ -878,29 +888,29 @@ impl BroadcastConsumer {
 /// to multiple times, and clones are cheap.
 // TODO: add `fetch` for one-shot retrieval of a past group range.
 #[derive(Clone)]
-pub struct TrackConsumer {
+pub struct TrackPending {
 	broadcast: BroadcastConsumer,
 	name: String,
 }
 
-impl TrackConsumer {
+impl TrackPending {
 	/// The track name this handle is bound to.
 	pub fn name(&self) -> &str {
 		&self.name
 	}
 
 	/// Open a live subscription.
 	///
-	/// Returns a [`TrackPending`] that resolves once the publisher accepts the
+	/// Returns a [`SubscribePending`] that resolves once the publisher accepts the
 	/// subscription (SUBSCRIBE_OK on the wire). `.await` it for the
 	/// [`TrackSubscriber`], which carries the publisher's [`Track`] and reads its
-	/// groups; or drive it with [`TrackPending::poll_ok`] from a poll loop.
+	/// groups; or drive it with [`SubscribePending::poll_ok`] from a poll loop.
 	///
 	/// `subscription` is this subscriber's preferences and feeds the producer's
 	/// [`TrackProducer::subscription`] aggregate; pass `None` for
 	/// [`Subscription::default`]. Concurrent subscribers to the same name coalesce
 	/// onto one request.
-	pub fn subscribe(&self, subscription: impl Into<Option<Subscription>>) -> TrackPending {
+	pub fn subscribe(&self, subscription: impl Into<Option<Subscription>>) -> SubscribePending {
 		self.broadcast
 			.request_subscribe(&self.name, subscription.into().unwrap_or_default())
 	}
@@ -914,6 +924,43 @@ impl TrackConsumer {
 	pub fn info(&self) -> InfoPending {
 		self.broadcast.request_info(&self.name)
 	}
+
+	/// Resolve the track, returning a [`TrackConsumer`] once its producer exists.
+	///
+	/// Elicits the track (like [`Self::info`]) and waits for the publisher to
+	/// accept it. `await`ing the [`TrackPending`] itself is equivalent.
+	pub async fn ok(&self) -> Result<TrackConsumer, Error> {
+		// Resolving info means the producer was accepted (and its weak inserted).
+		self.info().await?;
+		self.broadcast
+			.track_weak(&self.name)
+			.map(|weak| TrackConsumer { weak })
+			.ok_or(Error::Cancel)
+	}
+}
+
+/// A resolved track: its producer exists and its immutable [`Track`] is known, so
+/// subscribing and reading info are immediate. Obtained from [`TrackPending::ok`].
+#[derive(Clone)]
+pub struct TrackConsumer {
+	weak: TrackWeak,
+}
+
+impl TrackConsumer {
+	/// The track name.
+	pub fn name(&self) -> &str {
+		&self.weak.info.name
+	}
+
+	/// The track's immutable publisher properties.
+	pub fn info(&self) -> Track {
+		self.weak.info.clone()
+	}
+
+	/// Open a live subscription. Pass `None` for [`Subscription::default`].
+	pub fn subscribe(&self, subscription: impl Into<Option<Subscription>>) -> TrackSubscriber {
+		self.weak.subscribe(subscription.into().unwrap_or_default())
+	}
 }
 
 #[cfg(test)]
@@ -932,7 +979,7 @@ mod test {
 	use super::*;
 
 	/// Subscribe and assert the result hasn't resolved yet (it stays pending until
-	/// a publisher accepts). Returns the [`TrackPending`] to resolve after accepting.
+	/// a publisher accepts). Returns the [`SubscribePending`] to resolve after accepting.
 	macro_rules! subscribe_pending {
 		($consumer:expr, $name:expr) => {{
 			let pending = $consumer.consume_track($name).subscribe(None);

rs/moq-net/src/model/track.rs

@@ -754,7 +754,7 @@ impl TrackWeak {
 
 /// A live subscription to a track, used to read its groups.
 ///
-/// Created via [`TrackConsumer::subscribe`](crate::TrackConsumer::subscribe), or
+/// Created via [`TrackPending::subscribe`](crate::TrackPending::subscribe), or
 /// directly from a [`TrackProducer`] for an in-process track. Carries this
 /// subscriber's [`Subscription`] preferences, which feed the producer's aggregate.
 #[derive(Clone)]