Fix CI: await async subscribe_track and clippy/fmt cleanups

subscribe_track is now async; await it in the moq-ffi Rust tests and the
moq-rs Python test. Also drop a useless .into() and a needless borrow
flagged by clippy in moq-mux, plus rustfmt fixes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: kixelated

py/moq-rs/tests/test_local.py

@@ -243,7 +243,7 @@ async def test_dynamic_track_request():
     broadcast = moq.BroadcastProducer()
     dynamic = broadcast.dynamic()
     consumer = broadcast.consume()
-    track_consumer = consumer.subscribe_track("events")
+    track_consumer = await consumer.subscribe_track("events")
 
     track = await asyncio.wait_for(dynamic.requested_track(), timeout=5.0)
     assert track.name == "events"

rs/moq-ffi/src/test.rs

@@ -76,7 +76,7 @@ async fn dynamic_track_request() {
 	let broadcast = MoqBroadcastProducer::new().unwrap();
 	let dynamic = broadcast.dynamic().unwrap();
 	let consumer = broadcast.consume().unwrap();
-	let track_consumer = consumer.subscribe_track("events".into()).unwrap();
+	let track_consumer = consumer.subscribe_track("events".into()).await.unwrap();
 
 	let track = tokio::time::timeout(TIMEOUT, dynamic.requested_track())
 		.await
@@ -103,7 +103,7 @@ async fn dynamic_track_request_can_abort() {
 	let broadcast = MoqBroadcastProducer::new().unwrap();
 	let dynamic = broadcast.dynamic().unwrap();
 	let consumer = broadcast.consume().unwrap();
-	let _track_consumer = consumer.subscribe_track("unknown".into()).unwrap();
+	let _track_consumer = consumer.subscribe_track("unknown".into()).await.unwrap();
 
 	let track = tokio::time::timeout(TIMEOUT, dynamic.requested_track())
 		.await

rs/moq-gst/src/source/imp.rs

@@ -417,8 +417,7 @@ async fn reconcile(
 		let id = NEXT_PAD_ID.fetch_add(1, Ordering::Relaxed);
 
 		let track_subscriber = broadcast.track(&name)?.subscribe(None)?.await?;
-		let track =
-			moq_mux::container::Consumer::new(track_subscriber, container).with_latency(Duration::from_secs(1));
+		let track = moq_mux::container::Consumer::new(track_subscriber, container).with_latency(Duration::from_secs(1));
 
 		let descriptor = TrackDescriptor {
 			kind: d.kind,

rs/moq-mux/src/catalog/consumer.rs

@@ -49,7 +49,7 @@ impl<E: CatalogExt> Stream for Consumer<E> {
 				// MSF carries only the media sections, so the extension defaults.
 				let media = match ready!(c.poll_next(waiter)) {
 					Ok(media) => media,
-					Err(err) => return Poll::Ready(Err(err.into())),
+					Err(err) => return Poll::Ready(Err(err)),
 				};
 				Poll::Ready(Ok(media.map(|m| Catalog::<E> {
 					video: m.video,

rs/moq-mux/src/catalog/filter.rs

@@ -147,7 +147,11 @@ impl<S: Stream> Stream for Filter<S> {
 
 /// Apply the active video / audio filters to a raw snapshot, dropping
 /// renditions that don't match. Axes with no filter pass through unchanged.
-fn apply<E: CatalogExt>(mut catalog: Catalog<E>, video: Option<&FilterVideo>, audio: Option<&FilterAudio>) -> Catalog<E> {
+fn apply<E: CatalogExt>(
+	mut catalog: Catalog<E>,
+	video: Option<&FilterVideo>,
+	audio: Option<&FilterAudio>,
+) -> Catalog<E> {
 	if let Some(filter) = video {
 		catalog.video.renditions.retain(|name, config| {
 			if let Some(want) = &filter.name

rs/moq-mux/src/catalog/target.rs

@@ -162,7 +162,11 @@ impl<S: Stream> Stream for Target<S> {
 
 /// Apply the active video / audio targets to a raw snapshot, narrowing each
 /// axis to at most one rendition. Axes with no target pass through unchanged.
-fn apply<E: CatalogExt>(mut catalog: Catalog<E>, video: Option<&TargetVideo>, audio: Option<&TargetAudio>) -> Catalog<E> {
+fn apply<E: CatalogExt>(
+	mut catalog: Catalog<E>,
+	video: Option<&TargetVideo>,
+	audio: Option<&TargetAudio>,
+) -> Catalog<E> {
 	if let Some(target) = video {
 		if let Some(name) = select_video(&catalog.video.renditions, target) {
 			let mut kept = BTreeMap::new();

rs/moq-mux/src/container/ts/export_test.rs

@@ -323,7 +323,12 @@ async fn export_scte35_roundtrip() {
 	// Create and write the .scte35 cue track BEFORE moving `broadcast` into
 	// `Import` (which consumes it); the producer stays alive so the exporter can
 	// subscribe to the retained track.
-	let scte = broadcast.unique_track(".scte35", moq_net::TrackInfo::default().with_timescale(hang::container::TIMESCALE)).unwrap();
+	let scte = broadcast
+		.unique_track(
+			".scte35",
+			moq_net::TrackInfo::default().with_timescale(hang::container::TIMESCALE),
+		)
+		.unwrap();
 	let scte_name = scte.name().to_string();
 	{
 		let mut cfg = scte35::Config::new();
@@ -349,7 +354,12 @@ async fn export_scte35_roundtrip() {
 
 	// `import`, `catalog`, and `scte_producer` stay alive: retained tracks. The
 	// exporter must carry the extension to see the scte35 section.
-	let ts = drain_with(Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang).await.unwrap()).await;
+	let ts = drain_with(
+		Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang)
+			.await
+			.unwrap(),
+	)
+	.await;
 	assert_packet_aligned(&ts);
 
 	// The first PMT advertises the SCTE-35 ES (0x86) and the CUEI descriptor.
@@ -390,7 +400,7 @@ async fn export_scte35_roundtrip() {
 	assert_eq!(snapshot.scte35.renditions.len(), 1, "round-trip lost the SCTE-35 track");
 	let name = snapshot.scte35.renditions.keys().next().unwrap();
 
-	let track = consumer2.track(&name).unwrap().subscribe(None).unwrap().await.unwrap();
+	let track = consumer2.track(name).unwrap().subscribe(None).unwrap().await.unwrap();
 	let mut scte_reader = crate::container::Consumer::new(track, HangContainer::Legacy);
 	let frame = scte_reader
 		.read()
@@ -415,7 +425,12 @@ async fn scte35_without_video_export_is_rejected() {
 			.unwrap();
 
 	// A scte35 cue track and nothing else.
-	let scte = broadcast.unique_track(".scte35", moq_net::TrackInfo::default().with_timescale(hang::container::TIMESCALE)).unwrap();
+	let scte = broadcast
+		.unique_track(
+			".scte35",
+			moq_net::TrackInfo::default().with_timescale(hang::container::TIMESCALE),
+		)
+		.unwrap();
 	let scte_name = scte.name().to_string();
 	{
 		let mut cfg = scte35::Config::new();
@@ -434,7 +449,9 @@ async fn scte35_without_video_export_is_rejected() {
 	producer.finish_group().unwrap();
 	producer.finish().unwrap();
 
-	let mut exporter = Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang).await.unwrap();
+	let mut exporter = Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang)
+		.await
+		.unwrap();
 	let err = loop {
 		match tokio::time::timeout(std::time::Duration::from_secs(1), exporter.next()).await {
 			Ok(Ok(Some(_))) => continue,
@@ -563,7 +580,12 @@ async fn scte35_fixtures_survive_roundtrip() {
 		);
 
 		// Export and re-ingest.
-		let ts = drain_with(Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang).await.unwrap()).await;
+		let ts = drain_with(
+			Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang)
+				.await
+				.unwrap(),
+		)
+		.await;
 		assert_packet_aligned(&ts);
 
 		let mut broadcast2 = moq_net::BroadcastInfo::new().produce();

rs/moq-mux/src/container/ts/import.rs

@@ -437,8 +437,10 @@ impl<E: scte35::Catalog> ScteStream<E> {
 		// Cues ride the legacy container, which normalizes the per-frame timestamp to
 		// microseconds on the wire (see `hang::container::Frame::encode`), so the track
 		// declares that timescale to match.
-		let track = broadcast
-			.unique_track(".scte35", moq_net::TrackInfo::default().with_timescale(hang::container::TIMESCALE))?;
+		let track = broadcast.unique_track(
+			".scte35",
+			moq_net::TrackInfo::default().with_timescale(hang::container::TIMESCALE),
+		)?;
 		let mut config = scte35::Config::new();
 		config.container = hang::catalog::Container::Legacy;
 		scte35.renditions.insert(track.name().to_string(), config);
@@ -1320,8 +1322,8 @@ mod test {
 	#[tokio::test(start_paused = true)]
 	async fn scte35_cue_stamped_with_video_pts() {
 		use crate::catalog::hang::{Catalog, Container};
-		use crate::container::ts::scte35;
 		use crate::container::Consumer;
+		use crate::container::ts::scte35;
 		use moq_net::Timestamp;
 
 		const VIDEO_PID: u16 = 0x0050;

rs/moq-native/src/tls.rs

@@ -556,7 +556,8 @@ pub(crate) fn server_config(config: &Server, alpn: Vec<Vec<u8>>) -> Result<Arc<r
 	let certs = Arc::new(certs);
 
 	// TCP can negotiate TLS 1.2 as well as 1.3, unlike QUIC which is 1.3-only.
-	let builder = rustls::ServerConfig::builder_with_provider(provider.clone()).with_safe_default_protocol_versions()?;
+	let builder =
+		rustls::ServerConfig::builder_with_provider(provider.clone()).with_safe_default_protocol_versions()?;
 
 	let mut tls = if config.root.is_empty() {
 		builder.with_no_client_auth().with_cert_resolver(certs)