moq-hls 0.4.10

//! Export: serve a MoQ broadcast as HLS, fetching media on demand.
//!
//! A [`Broadcaster`] subscribes to one broadcast's catalog and, per rendition, to its
//! timeline track (see [`hang::timeline`]). That is the *only* standing traffic: playlists
//! are rendered from timeline records (each record maps a group to its start timestamp), and
//! media bytes move only when an HTTP client requests a segment, which FETCHes exactly the
//! groups that segment covers from the relay cache and transmuxes them to CMAF. Renditions
//! whose catalog entry advertises no timeline can't be served this way and are skipped.
//!
//! The same machinery serves two kinds of consumer:
//!
//! * the HTTP serve path (pull): [`Broadcaster::rendition`] / [`Broadcaster::master_playlist`]
//!   and the crate-internal `Rendition::playlist` / `Rendition::segment`, rendered/fetched per
//!   request (that pull surface is gated behind the `server` feature); and
//! * a recorder (push): the [`renditions::Consumer`] and [`segments::Consumer`] cursors, which
//!   yield every rendition and every finalized segment in order, for mirroring a broadcast to
//!   storage.

mod master;
mod playlist;
mod rendition;
mod upstream;

pub mod renditions;
pub mod segments;

use std::sync::{Arc, Mutex};
use std::time::Duration;

use moq_mux::catalog::{self, CatalogFormat, Stream};
use rand::RngExt;

pub(crate) use playlist::render_media;
pub use rendition::{Kind, Rendition};
pub(crate) use upstream::Upstream;

/// Backoff bounds for the initial catalog subscription.
///
/// The usual failure is a publisher that has announced its broadcast but not yet written its
/// catalog track, so this waits for external state rather than repeating a failed request.
/// Escalating for that reason: a source that stays silent for an hour must not be polled four times
/// a second for an hour.
///
/// Deliberately no give-up budget, and no attempt to judge which failures are worth waiting on. The
/// broadcast closing is what ends the wait, and a relay-side broadcast outlives its publisher's
/// session, so any deadline here is a window in which a publisher outage leaves the broadcaster
/// permanently empty with nothing to recover it.
const CATALOG_RETRY_MIN: Duration = Duration::from_millis(250);
const CATALOG_RETRY_MAX: Duration = Duration::from_secs(5);

/// Export tuning shared across renditions.
///
/// Construct via [`Config::default`] and set the fields you need, so new options
/// stay additive.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct Config {
	/// Minimum duration of media listed in each rendition's playlist window. Older timeline
	/// records are evicted once the remaining segments still cover this span; keep it within
	/// the relay's group-cache retention, since segments are fetched from there on request.
	pub window: Duration,
}

impl Default for Config {
	fn default() -> Self {
		Self {
			window: Duration::from_secs(16),
		}
	}
}

/// All renditions of one broadcast, kept in sync with its catalog.
pub struct Broadcaster {
	// Read only by the serve path's `is_closed`/`closed` (server-gated); a recording consumer
	// observes close through its `renditions()` cursor ending instead.
	#[cfg_attr(not(feature = "server"), allow(dead_code))]
	broadcast: moq_net::broadcast::Consumer,
	/// The current rendition set, reconciled from the catalog by the watcher task.
	renditions: renditions::Producer,
	/// Aborts the catalog watcher when the broadcaster is dropped; `Drop` then retires the
	/// renditions themselves (see [`renditions::Producer::clear`]). Set once, right after
	/// construction.
	watcher: Mutex<Option<tokio::task::JoinHandle<()>>>,
}

impl Broadcaster {
	/// Resolve `source`'s catalog broadcast and start tracking its renditions.
	pub async fn new(source: moq_mux::Source, config: Config) -> crate::Result<Arc<Self>> {
		let broadcast = source.broadcast().await?;
		let upstream = Upstream {
			source,
			broadcast: broadcast.clone(),
		};
		let renditions = renditions::Producer::new();
		let broadcaster = Arc::new(Self {
			broadcast,
			renditions: renditions.clone(),
			watcher: Mutex::new(None),
		});
		// The watcher owns its own producer clone; the `Broadcaster`'s `Drop` aborts it so the
		// standing catalog subscription stops when nobody's serving from this broadcaster.
		let watcher = tokio::spawn(watch_catalog(upstream, config, renditions));
		*broadcaster.watcher.lock().unwrap() = Some(watcher);
		Ok(broadcaster)
	}

	/// Whether the source broadcast has closed (ended or dropped).
	#[cfg(feature = "server")]
	pub(crate) fn is_closed(&self) -> bool {
		self.broadcast.is_closed()
	}

	/// Resolve once the source broadcast closes, so the server can evict a dead broadcaster.
	#[cfg(feature = "server")]
	pub(crate) async fn closed(&self) {
		self.broadcast.closed().await;
	}

	/// Look up a rendition by kind and name. Video and audio are separate axes, so a
	/// video and an audio rendition may share a name without colliding.
	pub fn rendition(&self, kind: Kind, name: &str) -> Option<Arc<Rendition>> {
		self.renditions.get(kind, name)
	}

	/// A cursor over rendition add/remove events, for a consumer that mirrors or records the
	/// whole broadcast. It replays the current renditions, then yields changes, and returns
	/// `None` once the source closes.
	pub fn renditions(&self) -> renditions::Consumer {
		self.renditions.subscribe()
	}

	/// Resolve once at least one rendition has been discovered. How long to wait is the
	/// caller's policy: wrap this in a timeout (or select against it) as needed.
	pub async fn ready(&self) {
		self.renditions.ready().await;
	}

	/// Render the multivariant (master) playlist from the current renditions.
	///
	/// `query` is an optional query string (without the leading `?`, e.g. `jwt=<token>`)
	/// appended to every child media-playlist URL, so a credential the master was fetched
	/// with propagates to the rendition playlists a stock player loads next. It is an
	/// argument rather than a [`Config`] field because one broadcaster fans out to viewers
	/// holding different tokens; a credential in `Config` would embed one viewer's in
	/// another's playlist.
	pub fn master_playlist(&self, query: Option<&str>) -> String {
		let mut video = Vec::new();
		let mut audio = Vec::new();
		for rendition in self.renditions.snapshot() {
			match rendition.kind {
				Kind::Video => video.push(master::VideoVariant {
					name: rendition.name.clone(),
					bandwidth: rendition.bandwidth,
					width: rendition.width,
					height: rendition.height,
					codec: rendition.codec.clone(),
				}),
				Kind::Audio => audio.push(master::AudioVariant {
					name: rendition.name.clone(),
					bandwidth: rendition.bandwidth,
					codec: rendition.codec.clone(),
				}),
			}
		}
		master::render_master(&video, &audio, query)
	}

	/// Whether the current catalog contains no servable renditions (serve path).
	#[cfg_attr(not(feature = "server"), allow(dead_code))]
	pub(crate) fn is_empty(&self) -> bool {
		self.renditions.is_empty()
	}
}

impl Drop for Broadcaster {
	fn drop(&mut self) {
		if let Some(watcher) = self.watcher.lock().unwrap().take() {
			watcher.abort();
		}
		// The rendition map lives in shared state, so unlike an owned map it does NOT free its
		// renditions when this handle goes away: a surviving `renditions::Consumer` would pin
		// every `Rendition`, and with it the timeline watcher and source subscription it holds.
		// Release them here so teardown can't leak a standing subscription.
		self.renditions.clear();
	}
}

async fn watch_catalog(upstream: Upstream, config: Config, renditions: renditions::Producer) {
	let broadcast = &upstream.broadcast;
	let mut delay = CATALOG_RETRY_MIN;

	let mut consumer = loop {
		match catalog::Consumer::<()>::new(broadcast, CatalogFormat::Hang).await {
			Ok(consumer) => break consumer,
			Err(err) => {
				tracing::warn!(%err, "failed to subscribe to broadcast catalog, retrying");
				let wait = delay.mul_f64(0.5 + rand::rng().random::<f64>() / 2.0);
				delay = (delay * 2).min(CATALOG_RETRY_MAX);
				tokio::select! {
					_ = tokio::time::sleep(wait) => {}
					_ = kio::wait(|waiter| broadcast.poll_closed(waiter)) => {
						renditions.close();
						return;
					}
				}
			}
		}
	};

	loop {
		match kio::wait(|waiter| consumer.poll_next(waiter)).await {
			Ok(Some(catalog)) => renditions.sync(&upstream, &config, &catalog),
			Ok(None) => break,
			Err(err) => {
				tracing::warn!(%err, "broadcast catalog stream ended with error");
				break;
			}
		}
	}

	// The source is done (or errored): let recording cursors finish.
	renditions.close();
}

#[cfg(test)]
mod tests {
	use super::*;

	fn frame(micros: u64, keyframe: bool) -> moq_mux::container::Frame {
		payload_frame(micros, keyframe, &[0xDE, 0xAD, 0xBE, 0xEF])
	}

	fn payload_frame(micros: u64, keyframe: bool, payload: &'static [u8]) -> moq_mux::container::Frame {
		moq_mux::container::Frame {
			timestamp: moq_net::Timestamp::from_micros(micros).unwrap(),
			payload: bytes::Bytes::from_static(payload),
			keyframe,
			duration: None,
		}
	}

	fn vp8_frame(micros: u64, keyframe: bool) -> moq_mux::container::Frame {
		let payload = if keyframe {
			// Key frame tag, start code, and 320x240 geometry.
			&[0x10, 0x00, 0x00, 0x9d, 0x01, 0x2a, 0x40, 0x01, 0xf0, 0x00][..]
		} else {
			&[0x31, 0x00, 0x00][..]
		};
		moq_mux::container::Frame {
			timestamp: moq_net::Timestamp::from_micros(micros).unwrap(),
			payload: bytes::Bytes::copy_from_slice(payload),
			keyframe,
			duration: None,
		}
	}

	fn video_config(catalog: &moq_mux::catalog::Producer) -> hang::catalog::VideoConfig {
		let mut config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8);
		config.coded_width = Some(320);
		config.coded_height = Some(240);
		config.framerate = Some(30.0);
		config.timeline = Some(catalog.timeline("video0").unwrap().section());
		config
	}

	// Let the origin's spawned attach task run so a created broadcast is routable.
	async fn settle() {
		for _ in 0..10 {
			tokio::task::yield_now().await;
		}
	}

	#[tokio::test]
	async fn escaping_broadcast_reference_is_not_advertised() {
		let origin = moq_net::Origin::random().produce();
		let _broadcast = origin
			.create_broadcast("a/pub", moq_net::broadcast::Route::new().with_announce(true))
			.expect("publish allowed");
		settle().await;
		let source = moq_mux::Source::new(origin.consume(), "a/pub");
		let upstream = Upstream {
			broadcast: source.broadcast().await.unwrap(),
			source,
		};
		let mut config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8);
		config.broadcast = Some(moq_net::PathRelative::new("../../source").to_owned());
		config.timeline = Some(hang::catalog::Timeline::new("video.timeline"));
		let mut catalog = moq_mux::catalog::hang::Catalog::default();
		catalog.video.renditions.insert("video".to_string(), config);

		let renditions = renditions::Producer::new();
		renditions.sync(&upstream, &Config::default(), &catalog);
		assert!(renditions.get(Kind::Video, "video").is_none());
	}

	// The whole fetch-on-demand path in process: a broadcast publishes media through the
	// catalog (which records the timeline), the Broadcaster renders playlists from the
	// timeline alone, and a segment request fetches and transmuxes exactly its groups.
	#[tokio::test]
	async fn serves_playlist_and_segments_from_the_timeline() {
		let origin = moq_net::Origin::random().produce();
		let mut broadcast = origin
			.create_broadcast("live", moq_net::broadcast::Route::new().with_announce(true))
			.expect("publish allowed");
		settle().await;
		let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();

		let reserved = catalog.reserve();
		let mut registration = reserved.video("video0");
		let mut config = video_config(&catalog);
		config.coded_width = None;
		config.coded_height = None;
		registration.set(config);
		drop(reserved);

		// Three GOPs, 2s apart: groups 0 and 1 are complete, group 2 is the live edge.
		let track = broadcast.create_track("video0", None).unwrap();
		let mut media = catalog
			.media_producer(track, moq_mux::catalog::hang::Container::Legacy)
			.unwrap();
		media.write(vp8_frame(0, true)).unwrap();
		media.write(vp8_frame(1_000_000, false)).unwrap();
		media.write(vp8_frame(2_000_000, true)).unwrap();
		media.write(vp8_frame(3_000_000, false)).unwrap();
		media.write(vp8_frame(4_000_000, true)).unwrap();

		let source = moq_mux::Source::new(origin.consume(), "live");
		let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap();
		let _ = tokio::time::timeout(Duration::from_secs(5), broadcaster.ready()).await;
		let rendition = broadcaster
			.rendition(Kind::Video, "video0")
			.expect("rendition discovered from the catalog");
		let _ = tokio::time::timeout(Duration::from_secs(5), rendition.playable()).await;

		let master = broadcaster.master_playlist(None);
		assert!(master.contains("video/video0/media.m3u8"), "master lists the rendition");

		let playlist = rendition.playlist();
		assert_eq!(playlist.segments.len(), 2, "the live-edge group is not listed");
		assert_eq!(playlist.segments[0].group, 0);
		assert_eq!(playlist.segments[0].duration, 2.0);
		assert_eq!(playlist.segments[1].group, 1);
		assert_eq!(playlist.target_duration, 2, "ceil of the longest timeline gap");
		assert!(!playlist.finished);

		let rendered = rendition.media_playlist(None).expect("playable");
		assert!(rendered.contains("#EXT-X-MAP:URI=\"init.mp4\"\n"));
		assert!(rendered.contains("seg/0.m4s\n"));
		assert!(rendered.contains("seg/1.m4s\n"));

		// The same render, but carrying a credential into every child URL.
		let signed = rendition.media_playlist(Some("jwt=abc.def")).expect("playable");
		assert!(signed.contains("#EXT-X-MAP:URI=\"init.mp4?jwt=abc.def\"\n"));
		assert!(signed.contains("seg/0.m4s?jwt=abc.def\n"));

		let init = rendition.init().await.unwrap().expect("init segment");
		assert_eq!(&init[4..8], b"ftyp");

		let segment = rendition.segment(0).await.unwrap().expect("segment fetched on demand");
		assert_eq!(&segment[4..8], b"moof", "a fetched group transmuxes to moof+mdat");

		// The live-edge group isn't a segment yet, and unknown groups miss.
		assert!(rendition.segment(2).await.unwrap().is_none());
		assert!(rendition.segment(99).await.unwrap().is_none());

		// Keep the publisher alive for the whole test.
		drop((media, registration, broadcast));
	}

	// Dropping the broadcaster must not truncate a recording in progress. A cursor holds its
	// rendition alive, so the rendition's own watcher still ends the timeline with `end()` --
	// which is what promotes the live-edge record into the final segment. Force-closing the
	// rendition here instead would race that (`end()` no-ops on a closed channel) and silently
	// drop the last segment of every recording.
	#[tokio::test]
	async fn dropping_the_broadcaster_keeps_a_cursor_drainable() {
		let origin = moq_net::Origin::random().produce();
		let mut broadcast = origin
			.create_broadcast("live", moq_net::broadcast::Route::new().with_announce(true))
			.expect("publish allowed");
		settle().await;
		let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();

		let reserved = catalog.reserve();
		let mut registration = reserved.video("video0");
		let config = video_config(&catalog);
		registration.set(config);
		drop(reserved);

		// Two GOPs: group 0 is complete, while group 1 stays at the live edge until the
		// publisher finishes.
		let track = broadcast.create_track("video0", None).unwrap();
		let mut media = catalog
			.media_producer(track, moq_mux::catalog::hang::Container::Legacy)
			.unwrap();
		media.write(frame(0, true)).unwrap();
		media.write(frame(2_000_000, true)).unwrap();

		let source = moq_mux::Source::new(origin.consume(), "live");
		let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap();
		let _ = tokio::time::timeout(Duration::from_secs(5), broadcaster.ready()).await;
		let rendition = broadcaster
			.rendition(Kind::Video, "video0")
			.expect("rendition discovered");
		let mut segments = rendition.segments();

		// The recorder drops the broadcaster before finalizing its uploads.
		drop(broadcaster);

		// Finish the source after teardown, forcing the broadcaster's drop to land before the
		// rendition watcher's clean-end path.
		media.finish().unwrap();
		catalog.finish().unwrap();

		let mut groups = Vec::new();
		while let Some(segment) = tokio::time::timeout(Duration::from_secs(5), segments.next())
			.await
			.expect("the cursor drains without parking")
			.unwrap()
		{
			groups.push(segment.group);
		}
		assert!(
			groups.contains(&1),
			"the final segment survives dropping the broadcaster, got {groups:?}"
		);

		drop((catalog, media, registration, broadcast, rendition));
	}

	// A same-path republish takes the origin leaf over with a brand new broadcast (an ordinary
	// publisher is `Origin::UNKNOWN`, which never counts as the same publisher, so even a plain
	// reconnect qualifies). Renditions derived from the old broadcast's catalog must not serve the
	// replacement's media: its group numbering restarts, so those bytes would be served under the
	// replaced broadcast's segment number, duration, and PROGRAM-DATE-TIME.
	#[tokio::test]
	async fn a_replacement_publisher_is_not_served_under_the_replaced_catalog() {
		const OLD: &[u8] = b"OLDOLDOLDOLDOLDO";
		const NEW: &[u8] = b"NEWNEWNEWNEWNEWN";

		// Publish broadcast "live": a `video0` rendition with a timeline, plus three keyframe
		// groups whose frames carry `payload` so the two epochs' media differs byte for byte.
		// The returned handle keeps the publisher alive until it is dropped.
		fn publish(
			origin: &moq_net::origin::Producer,
			payload: &'static [u8],
		) -> (Box<dyn std::any::Any>, hang::catalog::VideoConfig) {
			let mut broadcast = origin
				.create_broadcast("live", moq_net::broadcast::Route::new().with_announce(true))
				.expect("publish allowed");
			let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();

			let reserved = catalog.reserve();
			let mut registration = reserved.video("video0");
			let config = video_config(&catalog);
			registration.set(config.clone());
			drop(reserved);

			let track = broadcast.create_track("video0", None).unwrap();
			let mut media = catalog
				.media_producer(track, moq_mux::catalog::hang::Container::Legacy)
				.unwrap();
			media.write(payload_frame(0, true, payload)).unwrap();
			media.write(payload_frame(2_000_000, true, payload)).unwrap();
			media.write(payload_frame(4_000_000, true, payload)).unwrap();

			(Box::new((broadcast, catalog, registration, media)), config)
		}

		fn contains(haystack: &bytes::Bytes, needle: &[u8]) -> bool {
			haystack.windows(needle.len()).any(|window| window == needle)
		}

		let origin = moq_net::Origin::random().produce();
		let (old, config) = publish(&origin, OLD);
		settle().await;

		// The export binds to the broadcast it reads the catalog from.
		let source = moq_mux::Source::new(origin.consume(), "live");
		let upstream = Upstream {
			broadcast: source.broadcast().await.unwrap(),
			source,
		};
		let mut catalog = moq_mux::catalog::hang::Catalog::default();
		catalog.video.renditions.insert("video0".to_string(), config);

		// The publisher reconnects before the catalog snapshot is reconciled. That gap is wide in
		// practice: `watch_catalog` retries a not-yet-written catalog track with backoff, so
		// seconds can pass between resolving the broadcast and syncing its first snapshot.
		drop(old);
		let (new, _) = publish(&origin, NEW);
		settle().await;

		// Reconcile by hand, which is what the catalog watcher does with each snapshot.
		let renditions = renditions::Producer::new();
		renditions.sync(&upstream, &Config::default(), &catalog);
		let rendition = renditions.get(Kind::Video, "video0").expect("rendition synced");

		let _ = tokio::time::timeout(Duration::from_secs(1), rendition.playable()).await;
		let served = rendition.segment(0).await.unwrap();
		if let Some(served) = &served {
			assert!(
				!contains(served, NEW),
				"the replacement publisher's media was served under the replaced broadcast's catalog"
			);
		}
		assert!(
			served.is_none(),
			"the replaced broadcast is closed, so it serves nothing"
		);

		// Control: an export started after the takeover does serve the new publisher's media, so
		// the assertion above is about which epoch a rendition is bound to, not about a broadcast
		// that happens to be unreadable.
		let source = moq_mux::Source::new(origin.consume(), "live");
		let fresh = Upstream {
			broadcast: source.broadcast().await.unwrap(),
			source,
		};
		let renditions = renditions::Producer::new();
		renditions.sync(&fresh, &Config::default(), &catalog);
		let rendition = renditions.get(Kind::Video, "video0").expect("rendition synced");
		tokio::time::timeout(Duration::from_secs(5), rendition.playable())
			.await
			.expect("the new publisher's timeline arrives");
		let served = rendition
			.segment(0)
			.await
			.unwrap()
			.expect("the new publisher is servable on its own catalog");
		assert!(
			contains(&served, NEW),
			"the new export serves the new publisher's media"
		);

		drop(new);
	}

	// A rendition the catalog drops must end its segment cursors. The cursor holds an
	// `Arc<Rendition>`, so without an explicit close the rendition (and its timeline
	// subscription) would stay alive and the cursor would park at the live edge forever.
	#[tokio::test]
	async fn removing_a_rendition_ends_its_segment_cursor() {
		let origin = moq_net::Origin::random().produce();
		let broadcast = origin
			.create_broadcast("live", moq_net::broadcast::Route::new().with_announce(true))
			.expect("publish allowed");
		let source = moq_mux::Source::new(origin.consume(), "live");
		settle().await;
		let upstream = Upstream {
			broadcast: source.broadcast().await.unwrap(),
			source,
		};

		// Drive the producer directly so the catalog can be reconciled synchronously.
		let renditions = renditions::Producer::new();
		let mut media = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8);
		media.timeline = Some(hang::catalog::Timeline::new("video.timeline"));
		let mut catalog = moq_mux::catalog::hang::Catalog::default();
		catalog.video.renditions.insert("video".to_string(), media);
		renditions.sync(&upstream, &Config::default(), &catalog);

		let rendition = renditions.get(Kind::Video, "video").expect("rendition synced");
		let mut segments = rendition.segments();

		// The catalog drops the rendition: its cursor must run dry rather than park.
		renditions.sync(
			&upstream,
			&Config::default(),
			&moq_mux::catalog::hang::Catalog::default(),
		);
		let ended = tokio::time::timeout(Duration::from_secs(5), segments.next())
			.await
			.expect("a removed rendition's cursor ends instead of parking")
			.unwrap();
		assert!(ended.is_none(), "no segments remained to drain");

		drop((broadcast, rendition));
	}

	// Dropping the broadcaster must release its renditions even when a cursor (and an
	// `Arc<Rendition>` it was handed) outlives it. The map lives in shared state, so without an
	// explicit teardown those would pin every rendition's timeline watcher -- and the standing
	// source subscription it holds -- for as long as the consumer lived.
	#[tokio::test]
	async fn dropping_the_broadcaster_releases_its_renditions() {
		let origin = moq_net::Origin::random().produce();
		let mut broadcast = origin
			.create_broadcast("live", moq_net::broadcast::Route::new().with_announce(true))
			.expect("publish allowed");
		settle().await;
		let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();

		let reserved = catalog.reserve();
		let mut registration = reserved.video("video0");
		let config = video_config(&catalog);
		registration.set(config);
		drop(reserved);

		let track = broadcast.create_track("video0", None).unwrap();
		let mut media = catalog
			.media_producer(track, moq_mux::catalog::hang::Container::Legacy)
			.unwrap();
		media.write(frame(0, true)).unwrap();
		media.write(frame(2_000_000, true)).unwrap();

		let source = moq_mux::Source::new(origin.consume(), "live");
		let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap();

		let mut renditions = broadcaster.renditions();
		let added = tokio::time::timeout(Duration::from_secs(5), renditions.next())
			.await
			.expect("a rendition is discovered");
		let Some(renditions::Event::Added(_rendition)) = added else {
			panic!("expected an Added event");
		};

		// The cursor and the handed-out rendition both outlive the broadcaster.
		drop(broadcaster);

		let next = tokio::time::timeout(Duration::from_secs(5), renditions.next())
			.await
			.expect("the cursor resolves after the broadcaster drops");
		assert!(
			matches!(next, Some(renditions::Event::Removed { .. })),
			"dropping the broadcaster releases its renditions to a live cursor"
		);

		// ...and the cursor then reaches a terminal state rather than parking. Releasing the
		// renditions is only safe if consumers actually finish.
		let ended = tokio::time::timeout(Duration::from_secs(5), renditions.next())
			.await
			.expect("the cursor terminates after the broadcaster drops");
		assert!(ended.is_none(), "the cursor ends once the broadcaster is gone");

		drop((catalog, media, registration, broadcast));
	}

	// The record path: the renditions cursor yields the video rendition, its segment cursor
	// yields each finalized segment WITH its media, and closing the publisher drains the final
	// segment and then ends both cursors.
	#[tokio::test]
	async fn record_cursors_yield_renditions_and_segments() {
		let origin = moq_net::Origin::random().produce();
		let mut broadcast = origin
			.create_broadcast("live", moq_net::broadcast::Route::new().with_announce(true))
			.expect("publish allowed");
		settle().await;
		let catalog = moq_mux::catalog::Producer::new(&mut broadcast).unwrap();

		let reserved = catalog.reserve();
		let mut registration = reserved.video("video0");
		let config = video_config(&catalog);
		registration.set(config);
		drop(reserved);

		// Groups 0 and 1 are complete; group 2 is the live edge until the publisher drops.
		let track = broadcast.create_track("video0", None).unwrap();
		let mut media = catalog
			.media_producer(track, moq_mux::catalog::hang::Container::Legacy)
			.unwrap();
		media.write(frame(0, true)).unwrap();
		media.write(frame(2_000_000, true)).unwrap();
		media.write(frame(4_000_000, true)).unwrap();

		let source = moq_mux::Source::new(origin.consume(), "live");
		let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap();

		let mut renditions = broadcaster.renditions();
		let rendition = match tokio::time::timeout(Duration::from_secs(5), renditions.next())
			.await
			.expect("a rendition is discovered")
		{
			Some(renditions::Event::Added(rendition)) => rendition,
			other => panic!("expected an Added event, got {:?}", other.is_some()),
		};
		assert_eq!(rendition.kind, Kind::Video);
		assert_eq!(rendition.name, "video0");

		let mut segments = rendition.segments();
		assert!(segments.init().await.unwrap().is_some(), "init is buildable");

		let first = tokio::time::timeout(Duration::from_secs(5), segments.next())
			.await
			.expect("first segment finalizes")
			.unwrap()
			.expect("a segment, not end");
		assert_eq!(first.group, 0);
		assert_eq!(&first.media[4..8], b"moof", "the segment carries its transmuxed media");
		assert_eq!(first.duration, 2.0);
		assert!(!first.discontinuity, "a clean start is not a discontinuity");

		let second = segments.next().await.unwrap().expect("second segment");
		assert_eq!(second.group, 1);
		assert!(!second.discontinuity, "consecutive segments are continuous");

		// Tear down the publisher mid-group. The track ends abruptly (the cursor drains the
		// segments it already saw and ends; the still-open live-edge group is NOT finalized,
		// since a reset can't vouch that its media is complete), while finishing the broadcast
		// ends it promptly instead of lingering for a reconnect. (Clean-end finalization of the
		// live edge is covered by segments::tests::next_after_walks_finalized_segments.)
		drop((catalog, media, registration));
		broadcast.finish();

		let end = tokio::time::timeout(Duration::from_secs(5), segments.next())
			.await
			.expect("segment cursor resolves after the source is lost")
			.unwrap();
		assert!(
			end.is_none(),
			"an abrupt end does not finalize the open live-edge group"
		);

		// The renditions cursor also ends once the source closes.
		let ended = tokio::time::timeout(Duration::from_secs(5), renditions.next())
			.await
			.expect("renditions cursor resolves");
		assert!(ended.is_none(), "renditions cursor ends when the broadcast closes");
	}
}