lvqr-transcode 1.1.0

//! [`TranscodeRunner`] + [`TranscodeRunnerHandle`] + [`TranscoderStats`].
//!
//! Wires registered [`crate::TranscoderFactory`] instances into a
//! shared [`lvqr_fragment::FragmentBroadcasterRegistry`] and drives
//! one tokio drain task per `(transcoder, rendition, broadcast,
//! track)` instance. Mirrors [`lvqr_agent::AgentRunner`] one-for-
//! one, with `(factory_name, rendition_name, broadcast, track)` as
//! the four-tuple stats key so metrics distinguish renditions of
//! the same factory.

use std::panic::AssertUnwindSafe;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use dashmap::DashMap;
use dashmap::mapref::entry::Entry;
use lvqr_fragment::{BroadcasterStream, FragmentBroadcaster, FragmentBroadcasterRegistry, FragmentStream};
use parking_lot::RwLock;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
use tracing::{info, warn};

use crate::transcoder::{Transcoder, TranscoderContext, TranscoderFactory};

/// Per-`(transcoder, rendition, broadcast, track)` outcome
/// counters.
#[derive(Debug, Default)]
pub struct TranscoderStats {
    /// Total fragments handed to [`Transcoder::on_fragment`]
    /// (regardless of panic outcome).
    pub fragments_seen: AtomicU64,

    /// Count of caught panics across `on_start`, `on_fragment`,
    /// and `on_stop` for this key.
    pub panics: AtomicU64,
}

/// Stats key: `(transcoder_name, rendition_name, broadcast, track)`.
/// Two factories of the same name targeting different renditions
/// live under separate keys so metrics distinguish them.
type StatsKey = (String, String, String, String);

/// Shared runner state held jointly by the registry `on_entry_created`
/// callback and the [`TranscodeRunnerHandle`]. The factory set is mutable
/// (behind an `RwLock`) so renditions can be added / removed at runtime;
/// `tasks` is keyed per drain instance so a removed rendition can abort just
/// its tasks; `registry` is a (cheap, shared) clone so a runtime-added
/// rendition can retroactively spawn drain tasks against already-live
/// sources.
struct RunnerInner {
    registry: FragmentBroadcasterRegistry,
    factories: RwLock<Vec<Arc<dyn TranscoderFactory>>>,
    tasks: DashMap<StatsKey, JoinHandle<()>>,
    stats: DashMap<StatsKey, Arc<TranscoderStats>>,
}

impl RunnerInner {
    /// Build + spawn a drain task for `factory` against the source
    /// broadcaster `bc`, unless an identical instance is already running or
    /// the factory opts out of this `(broadcast, track)`. Returns true iff a
    /// task was spawned. Idempotent on the `StatsKey` so retroactive spawning
    /// (runtime add) cannot race the `on_entry_created` callback into a
    /// duplicate drain (which would double-produce output fragments).
    fn spawn_for(
        &self,
        broadcast: &str,
        track: &str,
        bc: &Arc<FragmentBroadcaster>,
        factory: &Arc<dyn TranscoderFactory>,
    ) -> bool {
        let rendition = factory.rendition().clone();
        let key: StatsKey = (
            factory.name().to_string(),
            rendition.name.clone(),
            broadcast.to_string(),
            track.to_string(),
        );
        // Cheap pre-check before the (potentially slow) factory build.
        if self.tasks.contains_key(&key) {
            return false;
        }
        // Never transcode a transcode OUTPUT. Outputs are named
        // `<source>/<rendition>`, so the broadcast's final path segment is a
        // rendition name. Checking against the LIVE ladder (not a per-factory
        // frozen skip list) is what makes runtime `add_rendition` safe: a
        // newly added rendition must not transcode another rendition's output
        // (which would recurse), and the pre-existing factories' frozen skip
        // lists do not know the new rendition's name.
        let last_seg = broadcast.rsplit('/').next().unwrap_or(broadcast);
        if self.factories.read().iter().any(|f| f.rendition().name == last_seg) {
            return false;
        }
        let ctx = TranscoderContext {
            broadcast: broadcast.to_string(),
            track: track.to_string(),
            meta: bc.meta(),
            rendition: rendition.clone(),
        };
        let Some(transcoder) = factory.build(&ctx) else {
            return false;
        };
        let handle = match Handle::try_current() {
            Ok(h) => h,
            Err(_) => {
                warn!(
                    broadcast = %broadcast,
                    track = %track,
                    "TranscodeRunner: no tokio runtime; no drain spawned",
                );
                return false;
            }
        };
        // Reserve the key under the shard lock so a concurrent caller cannot
        // also spawn this instance. Build happened above (outside the lock);
        // if we lost the race the built transcoder is dropped here.
        match self.tasks.entry(key.clone()) {
            Entry::Occupied(_) => false,
            Entry::Vacant(slot) => {
                let sub = bc.subscribe();
                let stat = Arc::clone(
                    self.stats
                        .entry(key.clone())
                        .or_insert_with(|| Arc::new(TranscoderStats::default()))
                        .value(),
                );
                let task = handle.spawn(drive(transcoder, key.0.clone(), ctx, sub, stat));
                slot.insert(task);
                true
            }
        }
    }
}

/// Cheaply-cloneable handle returned by
/// [`TranscodeRunner::install`].
///
/// Holds the spawned per-transcoder drain tasks alive for the
/// server lifetime; tests and admin consumers read per-
/// `(transcoder, rendition, broadcast, track)` counters off this
/// handle. Mid-stride aborts (drop / [`Self::remove_rendition`]) do NOT
/// call [`Transcoder::on_stop`], matching the
/// [`lvqr_agent::AgentRunnerHandle`] shutdown shape.
///
/// The ladder is mutable at runtime via [`Self::add_rendition`] /
/// [`Self::remove_rendition`]; [`Self::renditions`] reports the current set
/// so introspection stays consistent with edits.
#[derive(Clone)]
pub struct TranscodeRunnerHandle {
    inner: Arc<RunnerInner>,
}

impl std::fmt::Debug for TranscodeRunnerHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TranscodeRunnerHandle")
            .field("tracked_keys", &self.inner.stats.len())
            .field("renditions", &self.inner.factories.read().len())
            .finish()
    }
}

impl TranscodeRunnerHandle {
    /// Total fragments observed by `transcoder` producing
    /// `rendition` from `(broadcast, track)`. Returns 0 if no
    /// transcoder under that key has fired yet.
    pub fn fragments_seen(&self, transcoder: &str, rendition: &str, broadcast: &str, track: &str) -> u64 {
        self.stat(transcoder, rendition, broadcast, track)
            .map(|s| s.fragments_seen.load(Ordering::Relaxed))
            .unwrap_or(0)
    }

    /// Caught-panic count for `transcoder` producing `rendition`
    /// from `(broadcast, track)`. Aggregates `on_start`,
    /// `on_fragment`, and `on_stop` panics under one counter.
    pub fn panics(&self, transcoder: &str, rendition: &str, broadcast: &str, track: &str) -> u64 {
        self.stat(transcoder, rendition, broadcast, track)
            .map(|s| s.panics.load(Ordering::Relaxed))
            .unwrap_or(0)
    }

    /// Snapshot of every `(transcoder, rendition, broadcast, track)`
    /// quadruple the runner has spawned a drain task for.
    pub fn tracked(&self) -> Vec<StatsKey> {
        self.inner.stats.iter().map(|e| e.key().clone()).collect()
    }

    /// The current ladder's rendition specs, in registration order. Reflects
    /// runtime [`Self::add_rendition`] / [`Self::remove_rendition`] edits, so
    /// the admin introspection route stays consistent with the live ladder.
    pub fn renditions(&self) -> Vec<crate::RenditionSpec> {
        self.inner
            .factories
            .read()
            .iter()
            .map(|f| f.rendition().clone())
            .collect()
    }

    /// Add a rendition factory to the live ladder. New broadcasts pick it up
    /// via the `on_entry_created` callback; already-live sources get a drain
    /// task spawned retroactively. No-op returning `false` when a factory with
    /// the same `(name, rendition)` is already registered (the caller should
    /// map that to a 409). Note a single rendition legitimately carries
    /// multiple factories of different names -- e.g. a `"software"` video
    /// encoder plus an `"audio-passthrough"` -- so the guard keys on the
    /// `(factory name, rendition)` pair, not the rendition name alone. Must be
    /// called from within a tokio runtime.
    pub fn add_rendition(&self, factory: Arc<dyn TranscoderFactory>) -> bool {
        {
            let mut factories = self.inner.factories.write();
            if factories
                .iter()
                .any(|f| f.name() == factory.name() && f.rendition().name == factory.rendition().name)
            {
                return false;
            }
            factories.push(Arc::clone(&factory));
        }
        // Retroactively spawn against every already-live source. The factory
        // opts out of non-source / wrong-kind tracks via `build()`.
        for (broadcast, track) in self.inner.registry.keys() {
            if let Some(bc) = self.inner.registry.get(&broadcast, &track) {
                self.inner.spawn_for(&broadcast, &track, &bc, &factory);
            }
        }
        true
    }

    /// Remove every factory + drain task for `rendition`. Returns the number
    /// of drain tasks aborted. Aborting skips `on_stop`; the rendition's
    /// already-published output broadcasters drain to their subscribers and
    /// close when the source ends. Returns 0 when no such rendition exists.
    pub fn remove_rendition(&self, rendition: &str) -> usize {
        self.inner.factories.write().retain(|f| f.rendition().name != rendition);
        let mut aborted = 0usize;
        self.inner.tasks.retain(|key, task| {
            if key.1 == rendition {
                task.abort();
                aborted += 1;
                false
            } else {
                true
            }
        });
        aborted
    }

    fn stat(&self, transcoder: &str, rendition: &str, broadcast: &str, track: &str) -> Option<Arc<TranscoderStats>> {
        self.inner
            .stats
            .get(&(
                transcoder.to_string(),
                rendition.to_string(),
                broadcast.to_string(),
                track.to_string(),
            ))
            .map(|e| Arc::clone(e.value()))
    }
}

/// Builder that collects [`TranscoderFactory`] registrations and
/// installs them onto a [`FragmentBroadcasterRegistry`]. Typical
/// usage -- three rungs of the default ladder:
///
/// ```no_run
/// # use lvqr_transcode::{PassthroughTranscoderFactory, RenditionSpec, TranscodeRunner};
/// # use lvqr_fragment::FragmentBroadcasterRegistry;
/// let registry = FragmentBroadcasterRegistry::new();
/// let _handle = TranscodeRunner::new()
///     .with_ladder(RenditionSpec::default_ladder(), |spec| {
///         PassthroughTranscoderFactory::new(spec)
///     })
///     .install(&registry);
/// // hold _handle for the server lifetime
/// ```
#[derive(Default)]
pub struct TranscodeRunner {
    factories: Vec<Arc<dyn TranscoderFactory>>,
}

impl TranscodeRunner {
    /// Construct an empty runner.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a transcoder factory by value.
    pub fn with_factory<F: TranscoderFactory>(mut self, factory: F) -> Self {
        self.factories.push(Arc::new(factory));
        self
    }

    /// Register a pre-arc'd factory. Useful when the caller
    /// already shares an `Arc<dyn TranscoderFactory>` with other
    /// server-side state.
    pub fn with_factory_arc(mut self, factory: Arc<dyn TranscoderFactory>) -> Self {
        self.factories.push(factory);
        self
    }

    /// Convenience: register one factory per rendition in the
    /// supplied ladder, building each factory from its rendition
    /// via `build`. Mirrors the `RenditionSpec::default_ladder()`
    /// -> three `PassthroughTranscoderFactory` pattern without
    /// forcing the caller to unroll it.
    pub fn with_ladder<F, Fn_>(mut self, ladder: Vec<crate::RenditionSpec>, build: Fn_) -> Self
    where
        F: TranscoderFactory,
        Fn_: Fn(crate::RenditionSpec) -> F,
    {
        for spec in ladder {
            self.factories.push(Arc::new(build(spec)));
        }
        self
    }

    /// How many factories are currently registered. Useful for
    /// `Default`-instantiated runners that want to gate their own
    /// install calls.
    pub fn factory_count(&self) -> usize {
        self.factories.len()
    }

    /// Wire an `on_entry_created` callback on `registry` so every
    /// new `(broadcast, track)` pair gets one drain task per
    /// transcoder the registered factories opt into. Returns a
    /// handle the caller MUST hold for the server lifetime;
    /// dropping it aborts every spawned task.
    ///
    /// Callback semantics mirror [`lvqr_agent::AgentRunner::install`]:
    /// the callback runs on the thread that wins the
    /// `get_or_create` insertion race, subscribes synchronously
    /// so no emit can race ahead of the drain loop, and spawns
    /// the per-transcoder drain task on the current tokio
    /// runtime. If no tokio runtime is available the warn logs
    /// and no task spawns.
    pub fn install(self, registry: &FragmentBroadcasterRegistry) -> TranscodeRunnerHandle {
        let inner = Arc::new(RunnerInner {
            registry: registry.clone(),
            factories: RwLock::new(self.factories),
            tasks: DashMap::new(),
            stats: DashMap::new(),
        });

        let inner_cb = Arc::clone(&inner);
        registry.on_entry_created(move |broadcast, track, bc| {
            // Snapshot the current factory Arcs so a concurrent add/remove
            // does not hold the read lock across the spawn loop.
            let factories: Vec<Arc<dyn TranscoderFactory>> = inner_cb.factories.read().clone();
            for factory in &factories {
                inner_cb.spawn_for(broadcast, track, bc, factory);
            }
        });

        info!(
            renditions = inner.factories.read().len(),
            "TranscodeRunner installed on FragmentBroadcasterRegistry",
        );

        TranscodeRunnerHandle { inner }
    }
}

/// Per-transcoder drain task. Runs until the broadcaster closes.
/// All trait dispatch is wrapped in `catch_unwind` so a panic in
/// any of `on_start` / `on_fragment` / `on_stop` is logged +
/// counted but does not propagate to the spawning runtime.
async fn drive(
    mut transcoder: Box<dyn Transcoder>,
    transcoder_name: String,
    ctx: TranscoderContext,
    mut sub: BroadcasterStream,
    stats: Arc<TranscoderStats>,
) {
    let rendition_name = ctx.rendition.name.clone();

    // Refresh the meta snapshot before `on_start`. The
    // `on_entry_created` callback fires synchronously inside
    // `FragmentBroadcasterRegistry::get_or_create`, *before* the
    // ingest side calls `set_init_segment`. A transcoder that
    // reads `ctx.meta.init_segment` at on_start time would miss
    // the header bytes -- which is a silent break for the
    // software pipeline (qtdemux finds no playable streams). The
    // refresh below catches the late init without changing the
    // trait surface. Tier 4 item 4.6 session 106 C fix.
    sub.refresh_meta();
    let ctx = TranscoderContext {
        broadcast: ctx.broadcast,
        track: ctx.track,
        meta: sub.meta().clone(),
        rendition: ctx.rendition,
    };

    // on_start: a panic here means we abort the drain loop.
    // Handing fragments to a transcoder whose setup panicked
    // would amplify the fault, not contain it.
    let started = std::panic::catch_unwind(AssertUnwindSafe(|| transcoder.on_start(&ctx)));
    if started.is_err() {
        stats.panics.fetch_add(1, Ordering::Relaxed);
        metrics::counter!(
            "lvqr_transcode_panics_total",
            "transcoder" => transcoder_name.clone(),
            "rendition" => rendition_name.clone(),
            "phase" => "start",
        )
        .increment(1);
        warn!(
            transcoder = %transcoder_name,
            rendition = %rendition_name,
            broadcast = %ctx.broadcast,
            track = %ctx.track,
            "Transcoder::on_start panicked; skipping drain loop",
        );
        return;
    }

    while let Some(frag) = sub.next_fragment().await {
        stats.fragments_seen.fetch_add(1, Ordering::Relaxed);
        metrics::counter!(
            "lvqr_transcode_fragments_total",
            "transcoder" => transcoder_name.clone(),
            "rendition" => rendition_name.clone(),
        )
        .increment(1);
        let result = std::panic::catch_unwind(AssertUnwindSafe(|| transcoder.on_fragment(&frag)));
        if result.is_err() {
            stats.panics.fetch_add(1, Ordering::Relaxed);
            metrics::counter!(
                "lvqr_transcode_panics_total",
                "transcoder" => transcoder_name.clone(),
                "rendition" => rendition_name.clone(),
                "phase" => "fragment",
            )
            .increment(1);
            warn!(
                transcoder = %transcoder_name,
                rendition = %rendition_name,
                broadcast = %ctx.broadcast,
                track = %ctx.track,
                group_id = frag.group_id,
                object_id = frag.object_id,
                "Transcoder::on_fragment panicked; skipping fragment and continuing",
            );
        }
    }

    let stopped = std::panic::catch_unwind(AssertUnwindSafe(|| transcoder.on_stop()));
    if stopped.is_err() {
        stats.panics.fetch_add(1, Ordering::Relaxed);
        metrics::counter!(
            "lvqr_transcode_panics_total",
            "transcoder" => transcoder_name.clone(),
            "rendition" => rendition_name.clone(),
            "phase" => "stop",
        )
        .increment(1);
        warn!(
            transcoder = %transcoder_name,
            rendition = %rendition_name,
            broadcast = %ctx.broadcast,
            track = %ctx.track,
            "Transcoder::on_stop panicked",
        );
    }

    info!(
        transcoder = %transcoder_name,
        rendition = %rendition_name,
        broadcast = %ctx.broadcast,
        track = %ctx.track,
        seen = stats.fragments_seen.load(Ordering::Relaxed),
        panics = stats.panics.load(Ordering::Relaxed),
        "TranscodeRunner: drain terminated",
    );
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::passthrough::PassthroughTranscoderFactory;
    use crate::rendition::RenditionSpec;
    use bytes::Bytes;
    use lvqr_fragment::{Fragment, FragmentFlags, FragmentMeta};
    use parking_lot::Mutex as PMutex;
    use std::time::Duration;

    fn meta() -> FragmentMeta {
        FragmentMeta::new("avc1.640028", 90_000)
    }

    fn frag(idx: u64) -> Fragment {
        Fragment::new(
            "0.mp4",
            idx,
            0,
            0,
            idx * 1000,
            idx * 1000,
            1000,
            FragmentFlags::DELTA,
            Bytes::from(vec![0xAB; 16]),
        )
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn passthrough_sees_every_fragment_and_stops() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_720p()))
            .install(&registry);

        let bc = registry.get_or_create("live/demo", "0.mp4", meta());
        for i in 0..5 {
            bc.emit(frag(i));
        }
        drop(bc);
        registry.remove("live/demo", "0.mp4");
        tokio::time::sleep(Duration::from_millis(150)).await;

        assert_eq!(handle.fragments_seen("passthrough", "720p", "live/demo", "0.mp4"), 5);
        assert_eq!(handle.panics("passthrough", "720p", "live/demo", "0.mp4"), 0);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn default_ladder_spawns_one_task_per_rendition() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_ladder(RenditionSpec::default_ladder(), PassthroughTranscoderFactory::new)
            .install(&registry);

        let bc = registry.get_or_create("live/ladder", "0.mp4", meta());
        bc.emit(frag(0));
        bc.emit(frag(1));
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Three renditions, each observing both fragments.
        let mut tracked = handle.tracked();
        tracked.sort();
        assert_eq!(tracked.len(), 3, "one drain task per rendition");
        for (_transcoder, rendition, _broadcast, _track) in &tracked {
            let seen = handle.fragments_seen("passthrough", rendition, "live/ladder", "0.mp4");
            assert_eq!(seen, 2, "rendition {rendition} saw both fragments");
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn factory_opt_out_skips_non_video_tracks() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_720p()))
            .install(&registry);

        let bc_audio = registry.get_or_create("live/demo", "1.mp4", FragmentMeta::new("mp4a.40.2", 48_000));
        bc_audio.emit(frag(0));
        tokio::time::sleep(Duration::from_millis(80)).await;

        // Passthrough factory opts out of non-video tracks; no
        // drain task spawns for the audio track.
        assert!(handle.tracked().is_empty());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn panic_in_on_fragment_is_caught_and_counted() {
        struct PanicAtTwo;
        impl Transcoder for PanicAtTwo {
            fn on_fragment(&mut self, fragment: &Fragment) {
                if fragment.group_id == 2 {
                    panic!("simulated encoder fault at group 2");
                }
            }
        }
        struct PanicAtTwoFactory {
            rendition: RenditionSpec,
        }
        impl TranscoderFactory for PanicAtTwoFactory {
            fn name(&self) -> &str {
                "panicky"
            }
            fn rendition(&self) -> &RenditionSpec {
                &self.rendition
            }
            fn build(&self, _ctx: &TranscoderContext) -> Option<Box<dyn Transcoder>> {
                Some(Box::new(PanicAtTwo))
            }
        }

        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PanicAtTwoFactory {
                rendition: RenditionSpec::preset_720p(),
            })
            .install(&registry);

        let bc = registry.get_or_create("live/panic", "0.mp4", meta());
        for i in 0..5 {
            bc.emit(frag(i));
        }
        tokio::time::sleep(Duration::from_millis(120)).await;

        assert_eq!(handle.fragments_seen("panicky", "720p", "live/panic", "0.mp4"), 5);
        assert_eq!(handle.panics("panicky", "720p", "live/panic", "0.mp4"), 1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn panic_in_on_start_skips_drain_loop() {
        struct PanicStart;
        impl Transcoder for PanicStart {
            fn on_start(&mut self, _ctx: &TranscoderContext) {
                panic!("simulated start failure");
            }
            fn on_fragment(&mut self, _fragment: &Fragment) {
                unreachable!("on_fragment must not run after on_start panics");
            }
        }
        struct PanicStartFactory {
            rendition: RenditionSpec,
        }
        impl TranscoderFactory for PanicStartFactory {
            fn name(&self) -> &str {
                "bad_start"
            }
            fn rendition(&self) -> &RenditionSpec {
                &self.rendition
            }
            fn build(&self, _ctx: &TranscoderContext) -> Option<Box<dyn Transcoder>> {
                Some(Box::new(PanicStart))
            }
        }

        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PanicStartFactory {
                rendition: RenditionSpec::preset_480p(),
            })
            .install(&registry);

        let bc = registry.get_or_create("live/panic-start", "0.mp4", meta());
        bc.emit(frag(0));
        bc.emit(frag(1));
        tokio::time::sleep(Duration::from_millis(100)).await;

        assert_eq!(
            handle.fragments_seen("bad_start", "480p", "live/panic-start", "0.mp4"),
            0
        );
        assert_eq!(handle.panics("bad_start", "480p", "live/panic-start", "0.mp4"), 1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn empty_runner_installs_callback_but_spawns_nothing() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new().install(&registry);

        let bc = registry.get_or_create("live/empty", "0.mp4", meta());
        bc.emit(frag(0));
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert!(handle.tracked().is_empty());
    }

    #[test]
    fn runner_default_is_empty() {
        let r = TranscodeRunner::default();
        assert_eq!(r.factory_count(), 0);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn add_rendition_spawns_for_existing_live_source() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_720p()))
            .install(&registry);

        let bc = registry.get_or_create("live/x", "0.mp4", meta());
        bc.emit(frag(0));
        tokio::time::sleep(Duration::from_millis(60)).await;

        // Add 480p at runtime; it must spawn a drain task against the
        // already-live source.
        assert!(handle.add_rendition(Arc::new(
            PassthroughTranscoderFactory::new(RenditionSpec::preset_480p())
        )));
        tokio::time::sleep(Duration::from_millis(60)).await;

        bc.emit(frag(1));
        bc.emit(frag(2));
        tokio::time::sleep(Duration::from_millis(120)).await;

        assert_eq!(handle.fragments_seen("passthrough", "720p", "live/x", "0.mp4"), 3);
        let s480 = handle.fragments_seen("passthrough", "480p", "live/x", "0.mp4");
        assert!(s480 >= 2, "runtime-added 480p must see post-add fragments; saw {s480}");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn add_rendition_rejects_duplicate_name() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_720p()))
            .install(&registry);

        // Same rendition name -> rejected.
        assert!(!handle.add_rendition(Arc::new(
            PassthroughTranscoderFactory::new(RenditionSpec::preset_720p())
        )));
        // Distinct name -> accepted.
        assert!(handle.add_rendition(Arc::new(
            PassthroughTranscoderFactory::new(RenditionSpec::preset_240p())
        )));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn remove_rendition_aborts_its_drain_and_leaves_others() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_ladder(RenditionSpec::default_ladder(), PassthroughTranscoderFactory::new)
            .install(&registry);

        let bc = registry.get_or_create("live/r", "0.mp4", meta());
        bc.emit(frag(0));
        tokio::time::sleep(Duration::from_millis(80)).await;

        let aborted = handle.remove_rendition("480p");
        assert_eq!(aborted, 1, "exactly the 480p drain task aborts");
        tokio::time::sleep(Duration::from_millis(40)).await;

        bc.emit(frag(1));
        bc.emit(frag(2));
        tokio::time::sleep(Duration::from_millis(120)).await;

        let s720 = handle.fragments_seen("passthrough", "720p", "live/r", "0.mp4");
        let s480 = handle.fragments_seen("passthrough", "480p", "live/r", "0.mp4");
        assert!(s720 >= 3, "surviving 720p keeps draining; saw {s720}");
        assert!(s480 < s720, "removed 480p stopped draining ({s480}) vs 720p ({s720})");

        let names: Vec<String> = handle.renditions().iter().map(|r| r.name.clone()).collect();
        assert!(!names.contains(&"480p".to_string()), "480p gone from ladder: {names:?}");
        assert!(names.contains(&"720p".to_string()));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn does_not_transcode_a_rendition_output_broadcast() {
        // A broadcast whose final segment matches a rendition name is a
        // transcode output and must never be (re-)transcoded, or runtime adds
        // would recurse. The source "live/x" is transcoded; the output-shaped
        // "live/x/720p" is skipped.
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_720p()))
            .install(&registry);

        let src = registry.get_or_create("live/x", "0.mp4", meta());
        let output = registry.get_or_create("live/x/720p", "0.mp4", meta());
        src.emit(frag(0));
        output.emit(frag(0));
        tokio::time::sleep(Duration::from_millis(100)).await;

        assert_eq!(handle.fragments_seen("passthrough", "720p", "live/x", "0.mp4"), 1);
        assert_eq!(
            handle.fragments_seen("passthrough", "720p", "live/x/720p", "0.mp4"),
            0,
            "output-shaped broadcast must not be transcoded"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn two_factories_share_a_rendition_name() {
        // A rendition carries both a video encoder and an audio passthrough
        // (different factory names, same rendition). The dup guard keys on
        // (factory name, rendition), so both must register; removing the
        // rendition drops both.
        struct AltFactory {
            rendition: RenditionSpec,
        }
        impl TranscoderFactory for AltFactory {
            fn name(&self) -> &str {
                "alt"
            }
            fn rendition(&self) -> &RenditionSpec {
                &self.rendition
            }
            fn build(&self, _ctx: &TranscoderContext) -> Option<Box<dyn Transcoder>> {
                None
            }
        }

        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_720p()))
            .install(&registry);

        // Same rendition "720p", different factory name "alt" -> accepted.
        assert!(handle.add_rendition(Arc::new(AltFactory {
            rendition: RenditionSpec::preset_720p(),
        })));
        assert_eq!(handle.renditions().len(), 2, "two factories, both for 720p");

        // Removing the rendition drops both factories.
        handle.remove_rendition("720p");
        assert!(handle.renditions().is_empty());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn renditions_reflects_runtime_edits() {
        let registry = FragmentBroadcasterRegistry::new();
        let handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_720p()))
            .install(&registry);

        let names =
            |h: &TranscodeRunnerHandle| -> Vec<String> { h.renditions().iter().map(|r| r.name.clone()).collect() };
        assert_eq!(names(&handle), vec!["720p".to_string()]);

        assert!(handle.add_rendition(Arc::new(
            PassthroughTranscoderFactory::new(RenditionSpec::preset_480p())
        )));
        let mut after_add = names(&handle);
        after_add.sort();
        assert_eq!(after_add, vec!["480p".to_string(), "720p".to_string()]);

        assert_eq!(
            handle.remove_rendition("720p"),
            0,
            "no live source, so no task to abort"
        );
        assert_eq!(names(&handle), vec!["480p".to_string()]);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn downstream_subscriber_still_sees_every_fragment() {
        // A downstream consumer of the source broadcaster (e.g.
        // the LL-HLS bridge) must not be perturbed by transcoder
        // drain tasks. Assert the fan-out by subscribing
        // independently and reading every fragment.
        let registry = FragmentBroadcasterRegistry::new();
        let _handle = TranscodeRunner::new()
            .with_factory(PassthroughTranscoderFactory::new(RenditionSpec::preset_240p()))
            .install(&registry);

        let bc = registry.get_or_create("live/fanout", "0.mp4", meta());
        let mut downstream = bc.subscribe();
        let emitted = PMutex::new(Vec::<u64>::new());
        for i in 0..4 {
            bc.emit(frag(i));
            emitted.lock().push(i);
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
        for expected in 0..4u64 {
            let f = downstream.next_fragment().await.expect("downstream frag");
            assert_eq!(f.group_id, expected);
        }
    }
}