media-pp 0.2.0

use std::{
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    },
    thread::{self, JoinHandle},
    time::{Duration, Instant},
};

use crate::pp_log::{PpLog, pp_info, pp_trace, pp_warn};

use crate::{
    bus::{Bus, BusEvent, BusReceiver},
    clock::Clock,
    control::{
        ControlMsg, ControlReceiver, ControlSender, PrerollContext, PrerollError, SeekCheckContext,
    },
    element::{Context, SourceElement},
    error::{Result, ThreadSpawnError},
    graph::{GraphSnapshot, NodeInfo, PipelineGraph, log_topology},
    playback_clock::PlaybackClock,
};

use super::{PipelineBuilder, builder::SourceEntry};

/// How [`Pipeline::seek`] chooses the sample shown at the requested position.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SeekMode {
    /// Land at the preceding keyframe and preview the first decodable sample.
    Keyframe,
    /// Decode forward from the preceding keyframe and preview the sample that
    /// covers the requested timestamp.
    Accurate,
}

/// Top-level pipeline: one or more sources (see [`PipelineBuilder`], with
/// everything reachable from each source's own src pads already linked)
/// plus the bus every source reports events on and the [`Clock`] every
/// [`crate::elements::Pacer`] in it shares.
///
/// `run()` is asynchronous: it starts every source on its own background
/// thread and returns immediately, rather than blocking the caller for the
/// whole play-through. It returns a
/// [`ThreadSpawnError`](crate::error::ThreadSpawnError) if a source worker
/// cannot be created; any workers already created for that call are stopped
/// and joined before the error is returned. The one-shot pipeline is not
/// reusable after that failure. Returned as `Arc<Pipeline>` (that's what
/// [`Pipeline::new`]/[`PipelineBuilder::build`] return) — the background
/// threads deliberately do not retain an owning handle, so dropping the
/// last external `Arc` can stop them. The `Arc` also lets [`Pipeline::pause`]/
/// [`Pipeline::resume`]/[`Pipeline::stop`] be called from another thread
/// while it's running.
///
/// There's no separate "is it done yet" query or callback: watch
/// [`Pipeline::bus`] instead. [`BusReceiver::iter`]/
/// [`BusReceiver::log_events`] block until every [`Bus`] sender has been
/// dropped. Under the normal ownership path that happens once every
/// source's background thread (and everything reachable from it) has
/// fully finished, so draining the bus doubles as "wait for completion" —
/// with more than one source, that means waiting for *all* of them, not
/// just the first to reach `Eos`. A caller that clones the [`Context`]
/// supplied to a source's own `wire` closure also retains its `Bus`
/// sender; in that case bus draining intentionally remains blocked until
/// that extra context is dropped. A source-level failure (returned from
/// [`crate::element::SourceElement::run`] itself, as opposed to one
/// reported from inside a `Queue`) shows up there too, as a
/// [`BusEvent::Error`] under that source's own name, since there's no
/// synchronous return path left to carry it.
///
/// A `Pipeline` isn't reusable once `run()` has been called (whether it
/// finished via every source's natural `Eos`, [`Pipeline::finish`], or
/// [`Pipeline::stop`]) — a
/// second `run()` call is a no-op; build a fresh `Pipeline` for another
/// play-through.
pub struct Pipeline {
    /// This pipeline's own id — passed to [`Pipeline::new`]/
    /// [`PipelineBuilder::new`], stamped onto every source's own `pp_log`
    /// there and onto every element that passes through a [`super::ChainBuilder`]
    /// built with it (see [`Pipeline::id`]).
    pub(super) id: Arc<str>,
    /// Logging identity for pipeline-level topology records.
    pub(super) pp_log: PpLog,
    pub(super) sources: Mutex<Option<Vec<SourceEntry>>>,
    /// Taken (leaving `None` behind) the moment `run()` starts, and cloned
    /// once per source into that source's own background thread — so once
    /// a pipeline is running, `Pipeline` itself no longer holds a `Bus`
    /// sender directly. If it did, [`BusReceiver::iter`] could never
    /// observe every sender dropped (one would always still be sitting
    /// right here), and would block forever instead of unblocking once
    /// every source actually finishes.
    pub(super) bus: Mutex<Option<Bus>>,
    /// One [`ControlSender`] per source, in the same order
    /// [`PipelineBuilder::add_source`] was called — [`Pipeline::finish`]/
    /// `stop`/`pause`/`resume`/`seek` send to every one of these in turn (each
    /// `send` is its own synchronous rendezvous with that source's own
    /// control cascade — see [`crate::control::ControlSender::send`] — so
    /// this serializes across sources rather than fanning out in
    /// parallel; fine for the handful of sources this is meant for).
    pub(super) control_txs: Vec<ControlSender>,
    /// Taken (leaving `None` behind) the moment `run()` starts, and moved
    /// one per thread — same reasoning as `bus` above. If `Pipeline` kept
    /// its own clone of each alive for its whole lifetime instead, that
    /// control channel's receiver side would never fully disconnect even
    /// after its thread has long since exited, so a
    /// [`Pipeline::stop`]/`pause`/`resume` racing that thread's own
    /// natural end (e.g. called right as it finishes on its own) could
    /// enqueue a `Request` nobody will ever read *or drop* — leaving
    /// [`crate::control::ControlSender::send`]'s rendezvous ack blocked
    /// forever instead of unblocked by the disconnect, the way it is the
    /// moment the *last* `ControlReceiver` clone actually goes away.
    pub(super) control_rxs: Mutex<Option<Vec<ControlReceiver>>>,
    pub(super) clock: Arc<Clock>,
    pub(super) playback_clock: Arc<PlaybackClock>,
    pub(super) bus_rx: BusReceiver,
    /// How many source threads are still running — `0` before `run()` and
    /// again once every source's thread has finished. `AtomicUsize` rather
    /// than a per-source flag: every call site (`pause`/`resume`/`stop`/
    /// `seek`) only ever needs "is anything still running at all", never
    /// which specific source.
    pub(super) running: Arc<AtomicUsize>,
    /// Tracks whether `Pipeline::pause` has completed without a matching
    /// resume. This cannot be inferred from `Clock`: pausing before the first
    /// media timestamp leaves an unset clock unchanged while downstream
    /// queues are nevertheless paused.
    pub(super) paused: AtomicBool,
    /// Serializes public lifecycle/timeline operations. `paused` remains the
    /// caller-requested state while seek temporarily pauses the runtime.
    pub(super) operation: Arc<Mutex<()>>,
    /// The preroll a [`Pipeline::seek`] is currently waiting on, and whether
    /// the pipeline has since been abandoned.
    ///
    /// Held here rather than only inside `seek` so a caller that wants the
    /// pipeline to end can reach it *without* the operation lock. `Stop` would
    /// otherwise have to queue behind that wait, which is the one thing it
    /// promises not to do — and the cancellation the terminals already forward
    /// on `Stop` cannot arrive either, because sending it needs the same lock.
    pub(super) preroll_slot: Mutex<PrerollSlot>,
    /// Handles for every source thread started by [`Pipeline::run`]. They
    /// are retained so dropping the pipeline can synchronously stop and
    /// join live sources instead of leaving detached work behind.
    pub(super) workers: Mutex<Vec<JoinHandle<()>>>,
    /// Live node/edge graph backing snapshots and topology rendering.
    pub(super) graph: PipelineGraph,
}

impl Pipeline {
    /// `id` names this pipeline — stamped into the source's own `pp_log` as
    /// its `pipeline_id` right away, and folded into the [`Context`] handed
    /// to `wire` (see [`super::ChainBuilder`]'s own docs).
    ///
    /// `wire` is called once with the freshly created source and a
    /// [`Context`] bundling this pipeline's `Bus`, `id`, [`PipelineGraph`]
    /// (already seeded with the source itself), and `Clock` (share it with
    /// every [`crate::elements::Pacer`] via `Clock::clone` — one clock per
    /// pipeline, so every paced branch agrees on the same t=0 and the same
    /// pause/resume timeline) — everything a [`super::ChainBuilder`]/
    /// [`crate::elements::Tee`] needs, in one `Arc` clone instead of four
    /// separate arguments. `wire` creates detached chains and attaches
    /// them through [`Context::attach`]. Pads left unattached drop data.
    ///
    /// The single-source special case of [`PipelineBuilder`] — see its own
    /// docs for combining more than one live source (e.g. a video capture
    /// and an audio capture) into one `Pipeline`.
    pub fn new<S: SourceElement + 'static>(
        id: impl Into<String>,
        source: S,
        wire: impl FnOnce(&mut S, &Arc<Context>) -> Result<()>,
    ) -> Result<Arc<Self>> {
        Ok(PipelineBuilder::new(id).add_source(source, wire)?.build())
    }

    /// This pipeline's own id, as passed to [`Pipeline::new`].
    pub fn id(&self) -> &str {
        &self.id
    }

    /// Returns the receiver for asynchronous element and thread-boundary
    /// events produced by this pipeline.
    ///
    /// Calling [`BusReceiver::iter`](crate::bus::BusReceiver::iter) blocks
    /// until every sender has dropped, which normally coincides with all source
    /// and queue workers finishing. A custom element that retains a cloned
    /// [`Context`] can intentionally keep the receiver connected longer.
    pub fn bus(&self) -> &BusReceiver {
        &self.bus_rx
    }

    /// Returns a consistent node/edge snapshot of the live graph. Detached
    /// branches do not appear; a successful attach or detach increments its
    /// revision exactly once.
    pub fn graph(&self) -> GraphSnapshot {
        self.graph.snapshot()
    }

    /// Returns the nodes in a consistent snapshot of the currently attached
    /// graph.
    ///
    /// The returned values are owned copies and do not hold graph locks.
    /// Detached branch plans are absent until attachment succeeds, and removed
    /// branches disappear from later calls.
    pub fn elements(&self) -> Vec<NodeInfo> {
        self.graph().nodes
    }

    /// Human-readable rundown of [`Pipeline::elements`]: one line per
    /// branch — each element nothing else in the graph feeds into (a
    /// terminal sink, or an empty [`crate::elements::Tee`] with no sinks
    /// attached yet) — formatted `Type(name) - Type(name) - ...` by
    /// walking that element's `upstream` chain back to the source.
    /// Multiple branches (fan-out across more than one src pad, or a
    /// `Tee`) are joined by newlines.
    pub fn topology(&self) -> String {
        self.graph().topology()
    }

    /// The clock every `Pacer` in this pipeline paces against — see
    /// [`Pipeline::pause`] for why callers don't usually need to touch
    /// this directly.
    pub fn clock(&self) -> &Arc<Clock> {
        &self.clock
    }

    /// Media-position clock shared by audio output and video scheduling.
    pub fn playback_clock(&self) -> &Arc<PlaybackClock> {
        &self.playback_clock
    }

    /// Whether any source of this pipeline is still on a thread of its own.
    ///
    /// `false` before [`Pipeline::run`], and `true` from then until every
    /// source has finished — by its own `Eos`, by [`Pipeline::stop`] or
    /// [`Pipeline::finish`], or by returning an error it could not continue
    /// past. Those endings look different on the bus and identical here,
    /// which is what a caller wanting only "is this still producing?" is
    /// asking: a live capture whose target went away has to be noticed by
    /// whoever might reopen it, and that caller has no reason to care which
    /// way it ended.
    ///
    /// Draining the bus stays the way to learn *why* — see this type's own
    /// docs — and remains the only way to wait for the end rather than poll
    /// for it.
    pub fn is_running(&self) -> bool {
        self.running.load(Ordering::Acquire) > 0
    }

    /// Starts driving the source on a background thread and returns
    /// immediately — see the type-level docs for how to learn when it's
    /// actually done. A no-op if this `Pipeline` is already running or
    /// has already finished a previous run — this type has no "reset"
    /// path; build a fresh `Pipeline` for another play-through.
    /// If a source worker cannot be created, any source workers already
    /// started by this call are stopped and joined before the error returns.
    /// The one-shot pipeline is not reusable after that failure.
    pub fn run(&self) -> Result<()> {
        self.run_with_spawner(|thread_name, task| {
            thread::Builder::new().name(thread_name).spawn(task)
        })
    }

    pub(super) fn run_with_spawner(
        &self,
        mut spawn: impl FnMut(
            String,
            Box<dyn FnOnce() + Send + 'static>,
        ) -> std::io::Result<JoinHandle<()>>,
    ) -> Result<()> {
        let Some(sources) = self.sources.lock().unwrap().take() else {
            return Ok(());
        };
        // Always `Some` in lockstep with `sources` above — all three taken
        // exactly once, on whichever `run()` call actually wins the
        // `sources` guard.
        let Some(bus) = self.bus.lock().unwrap().take() else {
            return Ok(());
        };
        let Some(control_rxs) = self.control_rxs.lock().unwrap().take() else {
            return Ok(());
        };

        if crate::log::enabled(crate::log::Level::Info) {
            log_topology(&self.pp_log, "run", &self.graph());
        }
        let source_count = sources.len();
        self.running.store(source_count, Ordering::Release);
        for (index, ((source_id, source), control_rx)) in
            sources.into_iter().zip(control_rxs).enumerate()
        {
            let bus = bus.for_element(source_id);
            let running = Arc::clone(&self.running);
            let thread_name = "pipeline:source".to_owned();
            let spawn_result = spawn(
                thread_name.clone(),
                Box::new(move || {
                    // Keep these as locals in this order. During unwinding the
                    // guard is dropped first, then the receiver, then the
                    // source. That makes a Pipeline indirectly retained by a
                    // custom source safe to drop from this worker thread.
                    let mut source = source;
                    let control_rx = control_rx;
                    let _running = RunningSourceGuard::new(running);

                    let source_name = source.name();
                    let source_type = source.element_type();
                    // `source.run()` itself already reports non-fatal,
                    // per-buffer failures to `bus` as it goes (see
                    // `SourceElement::run`'s docs) — a returned `Err` here
                    // means something genuinely ended this source, e.g.
                    // a `Seek` that failed outright.
                    let outcome = if let Err(error) = source.run(&control_rx, &bus) {
                        bus.post(
                            source.pp_log(),
                            BusEvent::Error {
                                element_type: source_type,
                                name: source_name.clone(),
                                error,
                            },
                        );
                        // Nothing has told this source's branch that it is
                        // over: `run` returned instead of being stopped, and
                        // dropping it merely tears the elements down. A muxer
                        // waiting on this track would then never write its
                        // trailer, leaving an unplayable file — so cascade the
                        // same `Stop` a deliberate shutdown would have sent.
                        // `Stop` rather than `Eos` because the source failed:
                        // there is no complete stream to drain, only state to
                        // finalize.
                        // The log identity is cloned first: `src_pads` borrows
                        // the source mutably for the whole loop.
                        let source_log = source.pp_log().clone();
                        for pad in source.src_pads() {
                            if let Err(error) = pad.control(ControlMsg::Stop) {
                                pp_warn!(
                                    pp_log: &source_log,
                                    "failed to stop the branch after a source error: {error}"
                                );
                            }
                        }
                        "error"
                    } else {
                        "ok"
                    };
                    pp_info!(pp_log: source.pp_log(), "finished outcome={outcome}");
                }),
            );
            match spawn_result {
                Ok(handle) => self.workers.lock().unwrap().push(handle),
                Err(source) => {
                    // This pipeline is one-shot, so sources already started
                    // cannot be reconstructed for a retry. Stop and join them,
                    // then account for the failed and not-yet-started sources
                    // so callers never observe a half-running pipeline after
                    // this error returns.
                    self.running
                        .fetch_sub(source_count - index, Ordering::AcqRel);
                    self.clock.interrupt();
                    for control_tx in self.control_txs.iter().take(index) {
                        control_tx.send(ControlMsg::Stop);
                    }
                    self.join_workers();
                    return Err(ThreadSpawnError::new(thread_name, source).into());
                }
            }
        }
        Ok(())
    }

    /// Blocks until every element downstream of every source has paused —
    /// see [`crate::control::drain_control`] (source side) and
    /// [`crate::queue::Queue`]'s worker loop (each thread boundary). Also
    /// pauses this pipeline's `Clock` before that synchronous cascade
    /// starts, so time spent waiting for a busy downstream element to
    /// acknowledge `Pause` is frozen too and a `Pacer` doesn't see a jump
    /// once resumed. No-op if `run()` isn't currently in progress on
    /// another thread.
    pub fn pause(&self) {
        let _operation = self
            .operation
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if self.running.load(Ordering::Acquire) == 0 {
            return;
        }
        self.paused.store(true, Ordering::Release);
        self.pause_runtime();
    }

    fn pause_runtime(&self) {
        let msg = ControlMsg::Pause;
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=requested"
        );
        self.clock.interrupt();
        self.clock.pause();
        for control_tx in &self.control_txs {
            control_tx.send(msg.clone());
        }
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=completed outcome=ok"
        );
    }

    /// Undoes [`Pipeline::pause`]. Resumes the `Clock` first, so it's
    /// already shifted forward by the time `Pacer`s start receiving
    /// frames again.
    pub fn resume(&self) {
        let _operation = self
            .operation
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if self.running.load(Ordering::Acquire) == 0 {
            return;
        }
        self.paused.store(false, Ordering::Release);
        self.resume_runtime();
    }

    fn resume_runtime(&self) {
        let msg = ControlMsg::Resume;
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=requested"
        );
        self.clock.resume();
        for control_tx in &self.control_txs {
            control_tx.send(msg.clone());
        }
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=completed outcome=ok"
        );
    }

    /// Performs an early, full stop — abandons buffered work rather than
    /// draining to a natural `Eos`. This call is synchronous: it sends
    /// [`ControlMsg::Stop`] to every source in turn and waits for each
    /// one's own cascade to finish before moving to the next — sequential,
    /// not parallel, across sources (fine for the handful of sources this
    /// is meant for). It therefore cannot preempt an arbitrary
    /// source read or `Sink::consume` call already blocked inside user or
    /// external-library code; the call returns only after that work gives
    /// the control cascade a turn. After it returns, watch [`Pipeline::bus`]
    /// for every source's background thread to finish. Not reusable
    /// afterward — build a new `Pipeline` for the next play-through.
    pub fn stop(&self) {
        // Before the operation lock, not after: a seek holds that lock for as
        // long as its preroll wait, and an abandoning caller must not be made
        // to sit through it. Cancelling first turns that wait into an
        // immediate return, so this only waits for the seek's own control
        // cascade to unwind.
        self.abandon_preroll();
        let _operation = self
            .operation
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if self.running.load(Ordering::Acquire) == 0 {
            return;
        }
        self.stop_runtime();
    }

    /// Announces the preroll a seek is about to wait on, cancelling it
    /// immediately if the pipeline has already been abandoned.
    ///
    /// That second case is not hypothetical: `stop` runs before the operation
    /// lock, so it can arrive in the window between the seek repositioning its
    /// sources and reaching this call. One mutex covers both sides — either
    /// `stop` finds the preroll here, or this finds `stop`'s flag.
    fn publish_preroll(&self, preroll: &Arc<PrerollContext>) {
        let mut slot = self
            .preroll_slot
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if slot.abandoned {
            preroll.cancel();
            return;
        }
        slot.active = Some(Arc::clone(preroll));
    }

    fn retire_preroll(&self) {
        self.preroll_slot
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .active = None;
    }

    /// Waits for `preroll`, rechecking the topology whenever it has not
    /// finished yet.
    ///
    /// The expected terminals are fixed when the seek starts; the graph is
    /// not. Detaching a `Tee` branch mid-seek removes its terminal without
    /// removing the obligation to hear from it, and nothing is left to report
    /// a sample for it. Rather than lock topology changes out for the whole
    /// wait, this simply stops expecting whoever has since left — which also
    /// covers any other way a terminal can disappear, not just that one.
    ///
    /// The graph snapshot only happens on a poll that found work still
    /// pending, so a preroll that completes promptly never takes one.
    fn await_preroll(
        &self,
        preroll: &PrerollContext,
        timeout: Duration,
    ) -> std::result::Result<(), PrerollError> {
        const TOPOLOGY_POLL_INTERVAL: Duration = Duration::from_millis(50);

        let deadline = Instant::now() + timeout;
        loop {
            let remaining = deadline.saturating_duration_since(Instant::now());
            let slice = remaining.min(TOPOLOGY_POLL_INTERVAL);
            match preroll.wait(slice) {
                Err(PrerollError::TimedOut { pending }) => {
                    let live = self.graph().terminal_ids();
                    for terminal in pending
                        .iter()
                        .filter(|terminal| !live.contains(terminal))
                        .copied()
                    {
                        pp_trace!(
                            pp_log: &self.pp_log,
                            "event=control control=Preroll phase=pending \
                             outcome=departed terminal={terminal:?}"
                        );
                        preroll.mark_departed(terminal);
                    }
                    if remaining <= TOPOLOGY_POLL_INTERVAL {
                        // Deadline reached; report what is still owed, minus
                        // anything the prune above just resolved.
                        return preroll.wait(Duration::ZERO);
                    }
                }
                outcome => return outcome,
            }
        }
    }

    /// Ends an in-flight seek's preroll wait and refuses any that starts
    /// afterwards. Safe with none in flight, and deliberately takes no other
    /// lock: the whole point is to run *before* the operation lock a seek is
    /// holding.
    fn abandon_preroll(&self) {
        let preroll = {
            let mut slot = self
                .preroll_slot
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            slot.abandoned = true;
            slot.active.take()
        };
        if let Some(preroll) = preroll {
            preroll.cancel();
        }
    }

    fn stop_runtime(&self) {
        let msg = ControlMsg::Stop;
        self.paused.store(false, Ordering::Release);
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=requested"
        );
        self.clock.interrupt();
        for control_tx in &self.control_txs {
            control_tx.send(msg.clone());
        }
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=completed outcome=ok"
        );
    }

    /// Gracefully completes every source and waits for the whole graph to
    /// drain. Each source stops producing and places `MediaBuffer::Eos` behind
    /// its already-produced data; queues preserve that order, stateful codecs
    /// flush delayed output, and muxers finalize only after their EOS arrives.
    ///
    /// Unlike [`Pipeline::stop`], this does not abandon queued work. If the
    /// pipeline is paused, it resumes the control cascade first so a full
    /// paused queue cannot prevent its ordered EOS from being enqueued. The
    /// call returns only after every source thread (and the Queue workers each
    /// source owns) has finished. The pipeline is not reusable afterward.
    pub fn finish(&self) {
        // Same reasoning as `Pipeline::stop`: an in-flight seek's preroll wait
        // is about to be discarded by this completion, so there is nothing to
        // gain by sitting through it first.
        self.abandon_preroll();
        let _operation = self
            .operation
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if self.running.load(Ordering::Acquire) == 0 {
            self.join_workers();
            return;
        }

        pp_trace!(
            pp_log: &self.pp_log,
            "event=finish phase=requested"
        );
        self.clock.interrupt();
        if self.paused.load(Ordering::Acquire) {
            self.paused.store(false, Ordering::Release);
            self.resume_runtime();
        }
        for control_tx in &self.control_txs {
            control_tx.finish();
        }
        self.join_workers();
        pp_trace!(
            pp_log: &self.pp_log,
            "event=finish phase=completed outcome=ok"
        );
    }

    fn join_workers(&self) {
        let current_thread = thread::current().id();
        let mut workers = self
            .workers
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        for worker in workers.drain(..) {
            if worker.thread().id() != current_thread {
                let _ = worker.join();
            }
        }
    }

    /// Jumps to an absolute position from the start of the media. The whole
    /// operation is serialized against lifecycle controls and internally runs
    /// `Pause -> Flush -> Seek -> Preroll`. Every source repositions (see
    /// [`crate::element::SourceElement::seek`]) and every downstream element
    /// reacts before preroll begins. Once every terminal in the starting
    /// topology snapshot has
    /// accepted a first sample (or EOS), a paused pipeline remains paused and
    /// a playing pipeline resumes. No-op if `run()` isn't currently active.
    ///
    /// Signals the clock's interrupt epoch before starting the synchronous
    /// cascade so a `Pacer` in a long wait can return its worker promptly.
    /// The clock's playback anchor is still reset later, inside
    /// [`Sink::control`](crate::element::Sink::control) on `Pacer`, after
    /// that in-flight frame is
    /// out of the way.
    ///
    /// Before changing anything, a synchronous `CheckSeek` cascade verifies
    /// every source and branch. A live/non-seekable source or recording muxer
    /// returns [`crate::control::SeekError`] without flushing the current
    /// timeline.
    ///
    /// `mode` chooses whether decoding stops at the preceding keyframe or
    /// advances to the sample covering `target`.
    ///
    /// Completion means every terminal accepted its first new-timeline sample
    /// according to [`Sink::consume`](crate::element::Sink::consume). For a
    /// video renderer that includes installing or submitting the preview
    /// frame, but not waiting for physical display scanout.
    pub fn seek(&self, target: Duration, mode: SeekMode) -> Result<()> {
        const PREROLL_TIMEOUT: Duration = Duration::from_secs(5);

        let _operation = self
            .operation
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if self.running.load(Ordering::Acquire) == 0 {
            return Ok(());
        }
        let check = Arc::new(SeekCheckContext::new());
        for control_tx in &self.control_txs {
            control_tx.send(ControlMsg::CheckSeek(Arc::clone(&check)));
        }
        check.result()?;
        let restore_paused = self.paused.load(Ordering::Acquire);
        if !restore_paused {
            self.pause_runtime();
        }
        let msg = ControlMsg::Seek(target);
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=requested"
        );
        self.clock.interrupt();
        self.playback_clock.reset_for_seek();
        for control_tx in &self.control_txs {
            control_tx.send(ControlMsg::Flush);
        }
        for control_tx in &self.control_txs {
            control_tx.send(msg.clone());
        }
        let terminals = self.graph().terminal_ids();
        let preroll = Arc::new(match mode {
            SeekMode::Keyframe => PrerollContext::new(terminals),
            SeekMode::Accurate => PrerollContext::for_seek(terminals, target),
        });
        // Publish before waiting, so `stop` can end this wait rather than
        // queue behind it. Cleared on every exit below, including the error
        // one, so no later `stop` cancels a preroll that already finished.
        self.publish_preroll(&preroll);
        for control_tx in &self.control_txs {
            control_tx.send(ControlMsg::Preroll(Arc::clone(&preroll)));
        }
        let preroll_result = self.await_preroll(&preroll, PREROLL_TIMEOUT);
        self.retire_preroll();
        if restore_paused {
            self.pause_runtime();
        } else {
            self.resume_runtime();
        }
        preroll_result?;
        pp_trace!(
            pp_log: &self.pp_log,
            "event=control control={msg:?} phase=completed outcome=ok"
        );
        Ok(())
    }
}

/// Decrements the live-source count even if a source panics while running.
struct RunningSourceGuard {
    running: Arc<AtomicUsize>,
}

impl RunningSourceGuard {
    fn new(running: Arc<AtomicUsize>) -> Self {
        Self { running }
    }
}

impl Drop for RunningSourceGuard {
    fn drop(&mut self) {
        self.running.fetch_sub(1, Ordering::AcqRel);
    }
}

impl Drop for Pipeline {
    fn drop(&mut self) {
        // Send Stop while every sender is still alive. Merely dropping the
        // senders would not wake a source polling an empty control channel.
        self.stop();

        self.join_workers();
    }
}

/// Seek's preroll wait, reachable without the operation lock.
///
/// `abandoned` is sticky because the calls that set it — `stop` and `finish` —
/// both end the pipeline for good. Once set, a seek that has not yet published
/// its preroll cancels it on arrival instead of waiting out a timeout nobody
/// is going to collect.
#[derive(Default)]
pub(super) struct PrerollSlot {
    active: Option<Arc<PrerollContext>>,
    abandoned: bool,
}