media-pp 0.2.0

//! Pause, Resume, Stop, Flush, Seek, and Finish — and the channel they travel
//! through.
//!
//! Control follows the same pad-to-pad path as data but on a dedicated
//! channel, because unlike [`Eos`](crate::buffer::MediaBuffer::Eos) it has to
//! reach elements mid-stream and, at a [`Queue`](crate::queue::Queue), jump
//! ahead of whatever is already backed up instead of queueing behind it.
//!
//! A [`SourceElement`](crate::element::SourceElement) loop stays responsive by
//! calling [`drain_control`] every iteration. The returned [`ControlOutcome`]
//! is not only a "should I stop" flag: a source that schedules against the
//! wall clock must add `paused_for` back into its own timing, or resuming will
//! look like a burst of catch-up work owed all at once.

use std::{
    collections::HashSet,
    sync::{Arc, Condvar, Mutex},
    time::{Duration, Instant},
};

use thiserror::Error;

use crate::pp_log::pp_trace;
use crossbeam_channel::{Receiver, Sender, unbounded};

use crate::{
    bus::{Bus, BusEvent},
    element::{ElementType, SourceElement},
    error::Result,
    graph::ElementId,
};

/// A command that can be sent down a running [`crate::pipeline::Pipeline`]
/// — travels the same pad-to-pad path `MediaBuffer` does (see
/// [`crate::element::Sink::control`]), but through a dedicated channel
/// instead of riding along as data: unlike `Eos`, it has to be able to
/// reach every element even mid-stream, and (for `Queue`) jump ahead of
/// whatever data is already backed up rather than wait in line behind it.
#[derive(Debug, Clone)]
pub enum ControlMsg {
    /// Freeze in place. Every [`crate::queue::Queue`] downstream stops
    /// pulling from its data channel until `Resume`/`Stop` — which also
    /// backpressures anything feeding it, since a full queue blocks the
    /// sender. Pairs with [`crate::clock::Clock::pause`], which
    /// [`crate::pipeline::Pipeline::pause`] calls at the same time so
    /// paced elements don't see a jump once resumed.
    Pause,
    /// Undoes `Pause`.
    Resume,
    /// Abandon immediately rather than draining to a natural `Eos` —
    /// whatever's in flight is dropped, not flushed. The pipeline isn't
    /// reusable afterward; build a new one for the next run.
    Stop,
    /// Discard buffered data and reset state that belongs to the current
    /// timeline without changing the source position. Pipelines issue this
    /// before `Seek`; keeping the two controls separate lets paused preroll
    /// and future timeline operations compose the same flush boundary.
    Flush,
    /// Ask every reachable element whether it can participate in a seek.
    /// Rejections are collected without mutating playback state; the caller
    /// inspects the shared context after the synchronous cascade returns.
    CheckSeek(Arc<SeekCheckContext>),
    /// Temporarily lets paused source and queue workers process data until
    /// every expected terminal reports its first new-timeline sample.
    Preroll(Arc<PrerollContext>),
    /// Jump to an absolute position from the start of the media.
    /// The source repositions via [`crate::element::SourceElement::seek`]
    /// before this is forwarded downstream. Timeline state is discarded by
    /// the preceding `Flush`, not implicitly by this message.
    Seek(Duration),
}

impl PartialEq for ControlMsg {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Self::Pause, Self::Pause)
            | (Self::Resume, Self::Resume)
            | (Self::Stop, Self::Stop)
            | (Self::Flush, Self::Flush) => true,
            (Self::Seek(left), Self::Seek(right)) => left == right,
            (Self::CheckSeek(left), Self::CheckSeek(right)) => Arc::ptr_eq(left, right),
            (Self::Preroll(left), Self::Preroll(right)) => Arc::ptr_eq(left, right),
            _ => false,
        }
    }
}

impl Eq for ControlMsg {}

/// Why one element refused a pipeline-wide seek check.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SeekRejectReason {
    /// The source follows an external timeline that cannot be repositioned.
    LiveSource,
    /// The source cannot reposition its input timeline.
    SourceNotSeekable,
    /// A downstream element cannot preserve its contract across a seek.
    ElementNotSeekable,
}

/// One element that prevents a pipeline-wide seek.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SeekRejection {
    /// Kind of the element that rejected the check.
    pub element_type: ElementType,
    /// Caller-selected instance name of the rejecting element.
    pub name: Arc<str>,
    /// Capability that made this element incompatible with seeking.
    pub reason: SeekRejectReason,
}

/// Shared result accumulator carried by [`ControlMsg::CheckSeek`].
#[derive(Debug, Default)]
pub struct SeekCheckContext {
    rejections: Mutex<Vec<SeekRejection>>,
}

impl SeekCheckContext {
    /// Creates an empty accumulator for one synchronous seek check cascade.
    pub fn new() -> Self {
        Self::default()
    }

    /// Records an element refusal. Repeated visits to the same graph path do
    /// not make the public result noisy with identical entries.
    pub fn reject(&self, element_type: ElementType, name: Arc<str>, reason: SeekRejectReason) {
        let rejection = SeekRejection {
            element_type,
            name,
            reason,
        };
        let mut rejections = self
            .rejections
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if !rejections.contains(&rejection) {
            rejections.push(rejection);
        }
    }

    /// Returns a snapshot of every distinct rejection collected so far.
    pub fn rejections(&self) -> Vec<SeekRejection> {
        self.rejections
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }

    /// Succeeds when every visited element accepted the seek check.
    pub fn result(&self) -> std::result::Result<(), SeekError> {
        let rejections = self.rejections();
        if rejections.is_empty() {
            Ok(())
        } else {
            Err(SeekError { rejections })
        }
    }
}

/// A pipeline-wide seek check found at least one incompatible element.
#[derive(Debug, Error)]
#[error("pipeline seek rejected by {rejections:?}")]
pub struct SeekError {
    rejections: Vec<SeekRejection>,
}

impl SeekError {
    /// Elements that rejected the attempted pipeline seek.
    pub fn rejections(&self) -> &[SeekRejection] {
        &self.rejections
    }
}

#[derive(Debug, Default)]
struct PrerollState {
    ready: HashSet<ElementId>,
    cancelled: bool,
}

/// Shared completion state for one preroll pass.
#[derive(Debug)]
pub struct PrerollContext {
    expected: HashSet<ElementId>,
    target: Option<Duration>,
    state: Mutex<PrerollState>,
    changed: Condvar,
}

impl PrerollContext {
    /// Creates a preroll that completes once every supplied terminal ID is
    /// marked ready (or EOS-equivalent).
    pub fn new(terminals: impl IntoIterator<Item = ElementId>) -> Self {
        Self {
            expected: terminals.into_iter().collect(),
            target: None,
            state: Mutex::new(PrerollState::default()),
            changed: Condvar::new(),
        }
    }

    /// Creates a seek preroll whose decoded timing gates should discard
    /// samples before `target` while decoding forward from the landed
    /// keyframe.
    pub fn for_seek(terminals: impl IntoIterator<Item = ElementId>, target: Duration) -> Self {
        Self {
            expected: terminals.into_iter().collect(),
            target: Some(target),
            state: Mutex::new(PrerollState::default()),
            changed: Condvar::new(),
        }
    }

    /// Exact requested position for seek preroll, or `None` for ordinary
    /// first-sample preroll.
    pub fn target(&self) -> Option<Duration> {
        self.target
    }

    /// Marks one expected terminal's first valid sample as ready.
    pub fn mark_ready(&self, terminal: ElementId) {
        if !self.expected.contains(&terminal) {
            return;
        }
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.ready.insert(terminal) {
            self.changed.notify_all();
        }
    }

    /// EOS means this terminal cannot produce a sample and therefore must not
    /// leave the whole preroll waiting forever.
    pub fn mark_eos(&self, terminal: ElementId) {
        self.mark_ready(terminal);
    }

    /// Stops expecting a terminal that has left the graph.
    ///
    /// The expected set is fixed when the seek starts, but the topology is
    /// not: detaching a `Tee` branch mid-seek removes its terminal without
    /// removing the obligation to hear from it, and the wait would run to its
    /// timeout for a sample nobody is left to produce. Like EOS, this is
    /// "cannot produce one", not "produced one".
    pub fn mark_departed(&self, terminal: ElementId) {
        self.mark_ready(terminal);
    }

    /// Whether this one terminal has already taken its preroll sample.
    ///
    /// A terminal stops accepting as soon as *it* is ready, not when the whole
    /// preroll is. Waiting for the others would let a branch that reached the
    /// target first keep consuming for as long as the slowest branch takes —
    /// which is how the two streams end up at different positions when preroll
    /// finally completes.
    pub fn is_ready(&self, terminal: ElementId) -> bool {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        !state.cancelled && state.ready.contains(&terminal)
    }

    /// Whether every terminal in one downstream branch has completed.
    pub(crate) fn are_ready(&self, terminals: &[ElementId]) -> bool {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        !state.cancelled && terminals.iter().all(|id| state.ready.contains(id))
    }

    /// Returns whether every expected terminal has completed this preroll.
    pub fn is_complete(&self) -> bool {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        !state.cancelled && self.expected.is_subset(&state.ready)
    }

    /// Cancels a pending wait, used by stop and failed seek recovery.
    pub fn cancel(&self) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.cancelled = true;
        self.changed.notify_all();
    }

    /// Waits until every expected terminal is ready, cancellation is
    /// requested, or `timeout` expires.
    pub fn wait(&self, timeout: Duration) -> std::result::Result<(), PrerollError> {
        let deadline = Instant::now() + timeout;
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        loop {
            if state.cancelled {
                return Err(PrerollError::Cancelled);
            }
            let mut pending: Vec<_> = self.expected.difference(&state.ready).copied().collect();
            pending.sort_unstable();
            if pending.is_empty() {
                return Ok(());
            }
            let now = Instant::now();
            if now >= deadline {
                return Err(PrerollError::TimedOut { pending });
            }
            let remaining = deadline.saturating_duration_since(now);
            let (next, _) = self
                .changed
                .wait_timeout(state, remaining)
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            state = next;
        }
    }
}

/// Failure while waiting for terminal preroll completion.
#[derive(Debug, Error, PartialEq, Eq)]
pub enum PrerollError {
    /// Stop or recovery cancelled the in-flight preroll.
    #[error("preroll was cancelled")]
    Cancelled,
    /// At least one terminal did not receive a first sample in time.
    #[error("preroll timed out with pending terminals {pending:?}")]
    TimedOut { pending: Vec<ElementId> },
}

/// A request carried by a control channel. Ordinary controls cascade through
/// the graph immediately; `Finish` is source-only because graceful completion
/// must enter the graph as an ordered [`crate::buffer::MediaBuffer::Eos`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum RequestKind {
    Control(ControlMsg),
    Finish,
}

/// One in-flight control request: the message plus a rendezvous channel
/// the receiver acks once it (and everything it cascaded into downstream)
/// has finished handling it — this is what makes
/// [`ControlSender::send`] synchronous. Fields are `pub(crate)` so
/// [`crate::queue::Queue`]'s worker loop can match on one directly out of
/// a `crossbeam_channel::select!` arm (which needs the raw `Receiver`,
/// not the [`ControlReceiver::try_recv`]/[`ControlReceiver::recv`]
/// wrappers used everywhere else).
pub(crate) struct Request {
    pub(crate) kind: RequestKind,
    pub(crate) ack: Sender<()>,
}

/// The sending half of a control channel — cloneable, cheap, `Send +
/// Sync`. [`crate::pipeline::Pipeline`] holds one to reach its source;
/// [`crate::queue::Queue`] holds one internally to reach its worker
/// thread across the thread boundary it owns.
#[derive(Clone)]
pub struct ControlSender {
    tx: Sender<Request>,
}

/// The receiving half — not `Clone` in spirit (only one thing should be
/// driving a given control channel at a time) but crossbeam's
/// `Receiver<T>` is a cheap shared handle under the hood, which is
/// exactly what [`crate::pipeline::Pipeline::run`] needs: it clones this
/// into a fresh worker thread on every call.
#[derive(Clone)]
pub struct ControlReceiver {
    pub(crate) rx: Receiver<Request>,
}

/// Creates a control channel.
///
/// The channel is unbounded, because a control request must never be blocked by
/// backpressure on the data path — that is the whole reason control does not
/// travel as data. [`Pipeline`](crate::pipeline::Pipeline) creates one per
/// source; [`Queue`](crate::queue::Queue) creates one to reach its own worker.
pub fn channel() -> (ControlSender, ControlReceiver) {
    let (tx, rx) = unbounded();
    (ControlSender { tx }, ControlReceiver { rx })
}

impl ControlSender {
    /// Sends `msg` and blocks until the receiver — and, transitively,
    /// everything downstream of it — has finished handling it. A no-op
    /// (returns immediately) if nothing is on the other end to receive it
    /// (e.g. the pipeline already finished).
    pub fn send(&self, msg: ControlMsg) {
        self.send_request(RequestKind::Control(msg));
    }

    /// Requests source-originated EOS without exposing `Finish` as a
    /// downstream [`ControlMsg`]. Used only by [`crate::pipeline::Pipeline`].
    pub(crate) fn finish(&self) {
        self.send_request(RequestKind::Finish);
    }

    fn send_request(&self, kind: RequestKind) {
        let (ack_tx, ack_rx) = crossbeam_channel::bounded(0);
        if self.tx.send(Request { kind, ack: ack_tx }).is_ok() {
            let _ = ack_rx.recv();
        }
    }
}

impl ControlReceiver {
    pub(crate) fn try_recv(&self) -> Option<(RequestKind, Sender<()>)> {
        self.rx.try_recv().ok().map(|r| (r.kind, r.ack))
    }

    pub(crate) fn recv(&self) -> Option<(RequestKind, Sender<()>)> {
        self.rx.recv().ok().map(|r| (r.kind, r.ack))
    }
}

/// What draining pending source requests actually did — whether `Stop` or
/// source-only `Finish` ended it, and how long (if any) was spent frozen
/// inside a `Pause`/`Resume` pair. A source built on wall-clock scheduling (an elapsed-time
/// budget like [`crate::elements::TestAudioSource`]/
/// [`crate::elements::AudioMixer`], or an absolute next-tick deadline like
/// [`crate::elements::TestVideoSource`]/`DxgiCaptureSource`)
/// has to fold `paused_for` back into its own schedule after every
/// [`drain_control`] call — real (`Instant`) time keeps moving during a
/// `Pause`, but the media timeline must not, or `Resume` would look like a
/// burst of catch-up work owed all at once.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ControlOutcome {
    /// `true` if either `Stop` or source-only `Finish` was seen: the caller
    /// should return `Ok(())` immediately. `Stop` abandons without EOS;
    /// `Finish` has already pushed ordered EOS from the source boundary.
    /// Keeping this terminal flag true for both also makes existing custom
    /// source loops honor the new graceful request without continuing to emit
    /// after EOS.
    pub stopped: bool,
    /// Wall-clock time from starting the synchronous downstream `Pause`
    /// cascade through finishing the matching `Resume` (or terminating
    /// `Stop`) cascade during this call — `Duration::ZERO` if no `Pause`
    /// was seen. Still meaningful
    /// even when `stopped` is `true` (the sender simply going away while
    /// paused is treated the same as `Stop`, see `wait_out_pause`), so a
    /// caller that also tracks its own paused-time total can fold this in
    /// unconditionally rather than only on the non-stopped path.
    pub paused_for: Duration,
}

/// Call once per loop iteration in a [`SourceElement::run`] implementation,
/// right before pulling the next unit of work — mirrors how a natural
/// `Eos` is pushed into the source's own pads at the end of that same
/// loop, just for externally-triggered control instead.
///
/// Drains every pending message (see `apply_one` for what "handling
/// one" means, including `Pause`'s blocking wait). Non-blocking if
/// nothing's pending — a [`SourceElement::run`] whose own "next unit of
/// work" can't be waited on via `control`'s own channel (e.g.
/// [`crate::elements::FileDemuxer`]'s blocking file read) calls this once
/// before that blocking step; one that *can* (e.g.
/// [`crate::elements::AppSource`]'s channel receive) selects on both
/// instead, calling `apply_one`/`wait_out_pause` directly so a
/// pending `Stop`/`Finish` is never left waiting behind a slow/absent producer —
/// same reason `WasapiCaptureSource` also drives the
/// raw receiver directly, to bracket the wait with resetting/restarting
/// its capture device rather than leaving it running unread through the
/// whole pause.
///
/// See [`ControlOutcome`] for what the return value means.
pub fn drain_control<S: SourceElement>(
    control: &ControlReceiver,
    source: &mut S,
    bus: &Bus,
) -> Result<ControlOutcome> {
    let mut paused_for = Duration::ZERO;
    while let Some((request, ack)) = control.try_recv() {
        let RequestKind::Control(msg) = request else {
            apply_finish(source, bus, &ack);
            return Ok(ControlOutcome {
                stopped: true,
                paused_for,
            });
        };
        if msg == ControlMsg::Pause {
            // Start measuring before forwarding Pause. `apply_one` is a
            // synchronous cascade and may itself spend substantial time
            // waiting for a busy Queue/Sink to become paused; the source
            // produces no media during that time, so it belongs to the
            // frozen interval just as much as the later wait for Resume.
            let pause_start = Instant::now();
            apply_one(source, bus, &msg, &ack)?;
            let stopped = wait_out_pause(control, source, bus)?;
            paused_for += pause_start.elapsed();
            if stopped {
                return Ok(ControlOutcome {
                    stopped: true,
                    paused_for,
                });
            }
            continue;
        }
        if apply_one(source, bus, &msg, &ack)? {
            return Ok(ControlOutcome {
                stopped: true,
                paused_for,
            });
        }
    }
    Ok(ControlOutcome {
        stopped: false,
        paused_for,
    })
}

/// Applies one source-only graceful completion request. Unlike
/// [`apply_one`], this never calls `Sink::control`: EOS has to sit behind every
/// already-produced buffer in each data path so queues and stateful elements
/// drain in order.
pub(crate) fn apply_finish<S: SourceElement>(source: &mut S, bus: &Bus, ack: &Sender<()>) {
    pp_trace!(
        pp_log: source.pp_log(),
        "event=finish phase=received"
    );
    let pp_log = source.pp_log().clone();
    let element_type = source.element_type();
    let name = source.name();
    for pad in source.src_pads() {
        if let Err(error) = pad.push_eos(&pp_log) {
            bus.post(
                &pp_log,
                BusEvent::Error {
                    element_type,
                    name: name.clone(),
                    error,
                },
            );
        }
    }
    let _ = ack.send(());
    pp_trace!(
        pp_log: source.pp_log(),
        "event=finish phase=completed outcome=ok"
    );
}

/// Applies one already-received control message to `source`: repositions
/// it first on `Seek` (see [`apply_seek`]), then forwards `msg` to every
/// one of `source`'s pads (so it cascades through the graph exactly like
/// a data buffer would), then acks. Returns `true` for `Stop` — same
/// meaning as [`drain_control`]'s own return.
pub(crate) fn apply_one<S: SourceElement>(
    source: &mut S,
    bus: &Bus,
    msg: &ControlMsg,
    ack: &Sender<()>,
) -> Result<bool> {
    let is_stop = apply_one_unacked(source, bus, msg)?;
    let _ = ack.send(());
    Ok(is_stop)
}

/// The forwarding half of [`apply_one`], split out for a source that must
/// finish source-local state changes before the synchronous request is
/// acknowledged. [`crate::elements::WasapiCaptureSource`] uses this for
/// `Resume`: downstream is resumed first, then its capture device is
/// restarted, and only then may the caller observe the request as done.
pub(crate) fn apply_one_unacked<S: SourceElement>(
    source: &mut S,
    bus: &Bus,
    msg: &ControlMsg,
) -> Result<bool> {
    pp_trace!(
        pp_log: source.pp_log(),
        "event=control control={msg:?} phase=received"
    );
    let result: Result<bool> = (|| {
        apply_seek_check(source, msg);
        source.on_control(msg);
        apply_seek(source, bus, msg)?;
        for pad in source.src_pads() {
            pad.control(msg.clone())?;
        }
        Ok(*msg == ControlMsg::Stop)
    })();
    match &result {
        Ok(_) => pp_trace!(
            pp_log: source.pp_log(),
            "event=control control={msg:?} phase=completed outcome=ok"
        ),
        Err(error) => pp_trace!(
            pp_log: source.pp_log(),
            "event=control control={msg:?} phase=completed outcome=error error={error}"
        ),
    }
    result
}

/// Blocks on `control` alone — not whatever `source.run()` itself is
/// otherwise waiting on — until `Resume`, `Stop`, or `Finish`, applying (and
/// acking) every request seen in between. Returns `true` if `Stop`/`Finish`
/// ended it (including the sender simply going away, treated the same as
/// `Stop`); `false` once `Resume` or `Preroll` arrives.
pub(crate) fn wait_out_pause<S: SourceElement>(
    control: &ControlReceiver,
    source: &mut S,
    bus: &Bus,
) -> Result<bool> {
    loop {
        let Some((request, ack)) = control.recv() else {
            return Ok(true); // sender gone — treat like Stop
        };
        let RequestKind::Control(msg) = request else {
            apply_finish(source, bus, &ack);
            return Ok(true);
        };
        if apply_one(source, bus, &msg, &ack)? {
            return Ok(true);
        }
        if matches!(msg, ControlMsg::Resume | ControlMsg::Preroll(_)) {
            return Ok(false);
        }
        // Another Pause while already paused: already forwarded above
        // (harmless no-op downstream), just keep waiting.
    }
}

/// `Seek`'s source-specific half of `drain_control` — repositions
/// `source` (see [`SourceElement::seek`]) and reports where it actually
/// landed via [`BusEvent::Seeked`], since that can differ from what was
/// requested. No-op for every other [`ControlMsg`].
fn apply_seek_check<S: SourceElement>(source: &S, msg: &ControlMsg) {
    let ControlMsg::CheckSeek(context) = msg else {
        return;
    };
    let reason = if source.is_live() {
        Some(SeekRejectReason::LiveSource)
    } else if !source.is_seekable() {
        Some(SeekRejectReason::SourceNotSeekable)
    } else {
        None
    };
    if let Some(reason) = reason {
        context.reject(source.element_type(), source.name(), reason);
    }
}

fn apply_seek<S: SourceElement>(source: &mut S, bus: &Bus, msg: &ControlMsg) -> Result<()> {
    if let ControlMsg::Seek(target) = msg {
        let landed = source.seek(*target)?;
        bus.post(
            source.pp_log(),
            BusEvent::Seeked {
                element_type: source.element_type(),
                name: source.name(),
                requested: *target,
                landed,
            },
        );
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::{sync::Arc, thread};

    use crate::pp_log::PpLog;

    use super::*;
    use crate::{
        buffer::MediaBuffer,
        element::{Element, ElementType, Sink, Source, element_pp_log},
        pad::SrcPad,
    };

    /// A `SourceElement` with no real I/O — just enough surface for
    /// `drain_control`/`wait_out_pause` to drive, since this module's own
    /// logic doesn't care what the source actually produces.
    struct DummySource {
        pp_log: PpLog,
        pad: SrcPad,
        flushes: usize,
    }

    impl DummySource {
        fn new() -> Self {
            Self {
                flushes: 0,
                pp_log: element_pp_log(ElementType::Other, "dummy", None),
                pad: SrcPad::new("dummy_src"),
            }
        }
    }

    impl Element for DummySource {
        fn name(&self) -> Arc<str> {
            "dummy".into()
        }

        fn element_type(&self) -> ElementType {
            ElementType::Other
        }

        fn pp_log(&self) -> &PpLog {
            &self.pp_log
        }

        fn pp_log_mut(&mut self) -> &mut PpLog {
            &mut self.pp_log
        }
    }

    impl Source for DummySource {
        fn src_pads(&mut self) -> &mut [SrcPad] {
            std::slice::from_mut(&mut self.pad)
        }
    }

    impl SourceElement for DummySource {
        fn is_live(&self) -> bool {
            false
        }

        fn is_seekable(&self) -> bool {
            false
        }

        fn run(&mut self, _control: &ControlReceiver, _bus: &Bus) -> Result<()> {
            unreachable!("not exercised by these tests")
        }

        fn on_control(&mut self, msg: &ControlMsg) {
            if *msg == ControlMsg::Flush {
                self.flushes += 1;
            }
        }

        fn seek(&mut self, target: Duration) -> Result<Duration> {
            Ok(target)
        }
    }

    /// A source that holds data of its own — `FileDemuxer` parks packets for a
    /// pad that cannot accept one yet — has to discard it on the same boundary
    /// every downstream element does. Nothing else can: the packets exist only
    /// there, so releasing them after the reposition is the one way old media
    /// reaches a decoder that has already reset for the new timeline.
    #[test]
    fn flush_reaches_the_source_itself_and_nothing_else_does() {
        let (bus, _bus_rx) = Bus::new();
        let mut source = DummySource::new();

        for msg in [
            ControlMsg::Pause,
            ControlMsg::Resume,
            ControlMsg::Seek(Duration::from_secs(1)),
            ControlMsg::CheckSeek(Arc::new(SeekCheckContext::new())),
        ] {
            apply_one_unacked(&mut source, &bus, &msg).expect("control applies");
        }
        assert_eq!(source.flushes, 0, "only Flush may discard source-held data");

        apply_one_unacked(&mut source, &bus, &ControlMsg::Flush).expect("flush applies");
        assert_eq!(source.flushes, 1);
    }

    struct SlowPauseSink {
        pp_log: PpLog,
        pause_delay: Duration,
    }

    impl Element for SlowPauseSink {
        fn name(&self) -> Arc<str> {
            "slow-pause".into()
        }

        fn element_type(&self) -> ElementType {
            ElementType::Other
        }

        fn pp_log(&self) -> &PpLog {
            &self.pp_log
        }

        fn pp_log_mut(&mut self) -> &mut PpLog {
            &mut self.pp_log
        }
    }

    impl Sink for SlowPauseSink {
        fn consume(&mut self, _buf: MediaBuffer) -> Result<()> {
            Ok(())
        }

        fn control(&mut self, msg: ControlMsg) -> Result<()> {
            if msg == ControlMsg::Pause {
                thread::sleep(self.pause_delay);
            }
            Ok(())
        }
    }

    /// The edge case called out in `wait_out_pause`'s own docs: the
    /// `ControlSender` going away entirely (e.g. the owning `Pipeline`
    /// dropped) while paused has to be treated the same as an explicit
    /// `Stop`, not left blocking forever on a channel nothing will ever
    /// send on again.
    #[test]
    fn wait_out_pause_treats_a_dropped_sender_as_stop() {
        let (tx, rx) = channel();
        drop(tx);

        let (bus, _bus_rx) = Bus::new();
        let mut source = DummySource::new();

        let stopped = wait_out_pause(&rx, &mut source, &bus)
            .expect("no real seek/push happens on this path, so this can't fail");
        assert!(
            stopped,
            "a dropped ControlSender must be treated the same as an explicit Stop"
        );
    }

    #[test]
    fn seek_check_collects_a_non_seekable_source_without_mutating_it() {
        let context = Arc::new(SeekCheckContext::new());
        let (ack, _ack_rx) = crossbeam_channel::bounded(1);
        let (bus, _bus_rx) = Bus::new();
        let mut source = DummySource::new();

        apply_one(
            &mut source,
            &bus,
            &ControlMsg::CheckSeek(Arc::clone(&context)),
            &ack,
        )
        .expect("capability checks must not fail the control cascade");

        let error = context.result().expect_err("dummy source is not seekable");
        assert_eq!(
            error.rejections(),
            [SeekRejection {
                element_type: ElementType::Other,
                name: "dummy".into(),
                reason: SeekRejectReason::SourceNotSeekable,
            }]
        );
    }

    #[test]
    fn preroll_waits_for_every_terminal_and_reports_pending_ids() {
        let first = ElementId::for_test(1);
        let second = ElementId::for_test(2);
        let context = PrerollContext::new([first, second]);

        context.mark_ready(first);
        assert_eq!(
            context.wait(Duration::ZERO),
            Err(PrerollError::TimedOut {
                pending: vec![second]
            })
        );

        context.mark_eos(second);
        assert_eq!(context.wait(Duration::ZERO), Ok(()));
    }

    #[test]
    fn preroll_wait_can_be_cancelled() {
        let context = PrerollContext::new([ElementId::for_test(1)]);
        context.cancel();
        assert_eq!(
            context.wait(Duration::from_secs(1)),
            Err(PrerollError::Cancelled)
        );
    }

    #[test]
    fn wait_out_pause_returns_when_preroll_arrives() {
        let (tx, rx) = channel();
        let (bus, _bus_rx) = Bus::new();
        let mut source = DummySource::new();
        let context = Arc::new(PrerollContext::new([]));

        let worker = thread::spawn(move || wait_out_pause(&rx, &mut source, &bus));
        tx.send(ControlMsg::Preroll(context));

        assert!(!worker.join().unwrap().unwrap());
    }

    /// `wait_out_pause` blocks past any number of redundant `Pause`s and
    /// only returns (`Ok(false)`, meaning "keep running") once `Resume`
    /// actually arrives.
    #[test]
    fn wait_out_pause_blocks_until_resume_then_returns_false() {
        let (tx, rx) = channel();
        let (bus, _bus_rx) = Bus::new();
        let mut source = DummySource::new();

        let worker = thread::spawn(move || wait_out_pause(&rx, &mut source, &bus));

        // A redundant Pause while already paused: per `wait_out_pause`'s
        // own docs, forwarded (harmless no-op downstream) and then it
        // keeps waiting rather than returning.
        tx.send(ControlMsg::Pause);
        tx.send(ControlMsg::Resume);

        let stopped = worker
            .join()
            .expect("worker must not panic")
            .expect("no real seek/push happens on this path, so this can't fail");
        assert!(
            !stopped,
            "Resume must unblock wait_out_pause with Ok(false)"
        );
    }

    /// `paused_for` starts when the source begins forwarding Pause, not
    /// only after every downstream element has finally acknowledged it.
    /// Otherwise a slow control cascade is miscounted as playable media
    /// time and an elapsed-time source catches that interval up as a burst.
    #[test]
    fn drain_control_counts_the_pause_cascade_as_paused_time() {
        let pause_delay = Duration::from_millis(80);
        let (tx, rx) = channel();
        let controller = thread::spawn(move || {
            tx.send(ControlMsg::Pause);
            tx.send(ControlMsg::Resume);
        });

        let (bus, _bus_rx) = Bus::new();
        let mut source = DummySource::new();
        source.pad.link(Box::new(SlowPauseSink {
            pause_delay,
            pp_log: element_pp_log(ElementType::Other, "slow-pause", None),
        }));

        let outcome = loop {
            let outcome = drain_control(&rx, &mut source, &bus)
                .expect("the synthetic control cascade cannot fail");
            if outcome.paused_for > Duration::ZERO {
                break outcome;
            }
            thread::yield_now();
        };
        controller.join().expect("controller must not panic");

        assert!(!outcome.stopped);
        assert!(
            outcome.paused_for >= Duration::from_millis(60),
            "the {:?} Pause cascade was omitted from paused_for: {:?}",
            pause_delay,
            outcome.paused_for
        );
    }
}