arcly_stream/bus/

handle.rs

//! The live stream handle — a lock-free, zero-copy broadcast fan-out bus.

use crate::observe::{NoopObserver, Observer};
use crate::{frame::FrameFlags, AppName, MediaFrame, Result, StreamId, StreamKey};
use arc_swap::{ArcSwap, ArcSwapOption};
use std::net::SocketAddr;
use std::ops::ControlFlow;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex, OnceLock};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::{broadcast, Notify, RwLock};
use tokio_util::sync::CancellationToken;

/// A per-frame egress sink driven by [`StreamHandle::drive_to`].
///
/// Implementors send one frame to a single downstream peer (an RTSP player, an
/// upstream RTMP server, …). Returning [`ControlFlow::Break`] stops the drive
/// cleanly — typically because the peer disconnected — while `Err` aborts it.
#[async_trait::async_trait]
pub trait FrameSink: Send {
    /// Forward one frame; `Break` ends the session gracefully.
    async fn send(&mut self, frame: Arc<MediaFrame>) -> Result<ControlFlow<()>>;
}

/// Current wall-clock time in Unix milliseconds (saturating to 0 pre-epoch).
///
/// Use this only for *displayed* timestamps (e.g. `started_at_ms`); it is
/// subject to NTP steps and operator clock changes, so it must never drive
/// elapsed-time decisions. For those, use [`mono_ms`].
pub(crate) fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// Milliseconds elapsed on a process-local **monotonic** clock since the first
/// call. Unlike [`now_ms`], this never jumps backward (or forward) when the wall
/// clock is adjusted, so it is the correct basis for QoS windows and the idle
/// reaper: a leap-second or NTP correction can't spuriously reap a live stream
/// or distort a measured bitrate.
pub(crate) fn mono_ms() -> u64 {
    static EPOCH: OnceLock<Instant> = OnceLock::new();
    EPOCH.get_or_init(Instant::now).elapsed().as_millis() as u64
}

/// Current lifecycle state of a stream.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StreamState {
    /// No publisher yet.
    Idle,
    /// A publisher is connected and sending data.
    Publishing,
    /// The stream is being transcoded into one or more renditions.
    Transcoding,
    /// The stream is being recorded.
    Recording,
    /// The publisher has disconnected; the stream has ended.
    Ended,
}

impl std::fmt::Display for StreamState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            StreamState::Idle => "idle",
            StreamState::Publishing => "publishing",
            StreamState::Transcoding => "transcoding",
            StreamState::Recording => "recording",
            StreamState::Ended => "ended",
        })
    }
}

/// Runtime metadata about a stream, updated continuously while publishing.
///
/// Resolution and the ingest protocol are set by the protocol handler (e.g. via
/// [`StreamHandle::update_metadata`] after parsing the codec config), while the
/// `fps` and `*_bitrate_bps` fields are overlaid live from measured throughput
/// by [`StreamHandle::metadata_snapshot`].
#[derive(Debug, Clone)]
pub struct StreamMetadata {
    /// The `(app, stream_id)` this metadata describes.
    pub key: StreamKey,
    /// Publisher remote address.
    pub publisher_addr: Option<SocketAddr>,
    /// Video width in pixels (0 = unknown).
    pub width: u32,
    /// Video height in pixels (0 = unknown).
    pub height: u32,
    /// Video frames per second (0 = unknown). Overlaid from measured throughput.
    pub fps: f64,
    /// Measured video ingest bitrate in bits-per-second.
    pub video_bitrate_bps: u64,
    /// Measured audio ingest bitrate in bits-per-second.
    pub audio_bitrate_bps: u64,
    /// Timestamp of the first frame received (Unix ms).
    pub started_at_ms: u64,
    /// Protocol used for ingest (e.g. `"rtmp"`).
    pub ingest_protocol: String,
}

impl StreamMetadata {
    /// Create zeroed metadata for `(app, stream_id)`.
    pub fn new(app: AppName, stream_id: StreamId) -> Self {
        Self {
            key: StreamKey::new(app, stream_id),
            publisher_addr: None,
            width: 0,
            height: 0,
            fps: 0.0,
            video_bitrate_bps: 0,
            audio_bitrate_bps: 0,
            started_at_ms: 0,
            ingest_protocol: String::new(),
        }
    }
}

/// A point-in-time snapshot of a stream's measured quality of service.
#[derive(Debug, Clone, Copy, Default)]
pub struct Qos {
    /// Video bitrate over the last ~1s window (bits/sec).
    pub video_bitrate_bps: u64,
    /// Audio bitrate over the last ~1s window (bits/sec).
    pub audio_bitrate_bps: u64,
    /// Video frames per second over the last ~1s window.
    pub fps: f64,
    /// Cumulative frames published on this stream.
    pub total_frames: u64,
    /// Cumulative payload bytes published on this stream.
    pub total_bytes: u64,
}

/// Lock-free throughput counters folded into [`Qos`] / [`StreamMetadata`].
///
/// A stream has a single publisher (enforced by `start_publish`), so the
/// read-modify-write here is effectively single-writer and `Relaxed` is sound.
#[derive(Default)]
struct QosCounters {
    total_frames: AtomicU64,
    total_bytes: AtomicU64,
    window_start_ms: AtomicU64,
    window_video_bytes: AtomicU64,
    window_audio_bytes: AtomicU64,
    window_video_frames: AtomicU64,
    cur_video_bitrate: AtomicU64,
    cur_audio_bitrate: AtomicU64,
    cur_fps_milli: AtomicU64, // fps × 1000, integer-encoded
    last_frame_ms: AtomicU64,
}

/// Keyframe-anchored replay buffer for instant playback start.
struct GopBuffer {
    /// Frames since (and including) the most recent keyframe.
    frames: Vec<Arc<MediaFrame>>,
    /// Hard cap on buffered frames (frame-count bound between keyframes).
    capacity: usize,
    /// Hard cap on buffered payload **bytes** between keyframes (0 = unbounded).
    /// Frame count alone does not bound memory — one 4K keyframe can be megabytes
    /// — so a high-bitrate stream can be capped by size independently of count.
    byte_capacity: usize,
    /// Payload bytes currently buffered in `frames` (kept in sync on push/clear).
    bytes: usize,
    /// Whether the current GOP already overflowed `capacity`/`byte_capacity` (so
    /// the truncation signal fires once per GOP, not once per dropped frame).
    overflowed: bool,
}

/// A live handle to a single active stream.
///
/// Multiple subscribers (HLS packager, DASH packager, WebRTC SFU, recorders …)
/// call [`StreamHandle::subscribe_resilient`] to receive every [`MediaFrame`]
/// cheaply via a `broadcast` channel (zero-copy `Bytes` cloning).
///
/// Each broadcast slot holds one `Arc<MediaFrame>` pointer (8 bytes), so e.g.
/// 4096 slots ≈ 32 KB per stream.
///
/// # Backpressure model
///
/// Fan-out uses a **single, fixed-capacity ring buffer per stream** (a
/// `tokio::broadcast` channel sized by `AppSpec::broadcast_capacity`). This is a
/// deliberate design choice, with consequences worth understanding:
///
/// - **The publisher never blocks on a slow subscriber.** Publishing is a
///   non-awaiting pointer write; one subscriber falling behind can never apply
///   backpressure to the publisher or to its peers. This is what keeps the hot
///   path lock-free and the fast publisher isolated from the slow viewer.
/// - **Backpressure is resolved by dropping, not stalling.** A subscriber that
///   can't keep up overruns the ring and observes lag.
///   [`Subscription::recv`] resynchronizes to the oldest still-buffered frame
///   and reports the gap via [`Observer::on_subscriber_lagged`]; with
///   [`Subscription::max_lag`] a chronically slow consumer is evicted
///   ([`Observer::on_subscriber_evicted`]) rather than churning forever.
/// - **Capacity is the tuning knob**, traded per stream: larger capacity
///   tolerates burstier consumers at higher per-stream memory, smaller capacity
///   sheds laggards sooner. There is intentionally no per-subscriber queue —
///   that would reintroduce unbounded memory growth and per-consumer locking,
///   the very things this design avoids.
///
/// In short: a slow subscriber degrades only *its own* view (lag, then
/// eviction), never the publisher's or another subscriber's. Wire an
/// [`Observer`] to see lag and eviction as they happen.
#[derive(Clone)]
pub struct StreamHandle {
    metadata: Arc<RwLock<StreamMetadata>>,
    state: Arc<RwLock<StreamState>>,
    key: StreamKey,
    /// The frame-bus sender, held *indirectly* so its lifetime is owned by the
    /// stream's lifecycle — not by how many `StreamHandle` clones happen to be
    /// alive. Cloning a handle shares this cell; it does not mint a new sender.
    ///
    /// This is the structural fix for a sharp edge: previously the handle stored
    /// the `broadcast::Sender` by value, so any consumer that merely retained a
    /// handle (to subscribe, read metadata, request a keyframe) silently pinned
    /// the channel open and defeated the `Closed` shutdown signal. Now
    /// [`close`](Self::close) — called by the registry when a publish ends — empties
    /// this cell, dropping the sole sender and closing the channel regardless of
    /// any lingering handle clones.
    tx: Arc<ArcSwapOption<broadcast::Sender<Arc<MediaFrame>>>>,
    /// Latest video CONFIG (AVCDecoderConfigurationRecord) frame, if seen.
    /// Uses `ArcSwap` for lock-free reads from multiple subscriber tasks.
    video_config: Arc<ArcSwap<Option<Arc<MediaFrame>>>>,
    /// Latest audio CONFIG (AudioSpecificConfig) frame, if seen.
    audio_config: Arc<ArcSwap<Option<Arc<MediaFrame>>>>,
    /// Rolling GOP buffer for instant-start (empty when `gop_capacity == 0`).
    gop: Arc<StdMutex<GopBuffer>>,
    gop_capacity: usize,
    /// Default cumulative-lag budget applied to [`subscribe_resilient`]
    /// subscriptions: a consumer that drops more than this many frames is evicted.
    /// `None` = no default bound (a caller may still set one per subscription).
    default_max_lag: Option<u64>,
    /// Live throughput counters.
    qos: Arc<QosCounters>,
    /// Injected telemetry hook (no-op by default).
    observer: Arc<dyn Observer>,
    /// Upstream keyframe-request signal, shared across this handle's clones. A
    /// playback consumer (e.g. a WHEP viewer whose decoder needs an IDR) calls
    /// [`request_keyframe`](Self::request_keyframe); the *publisher's* ingest loop
    /// awaits [`keyframe_requested`](Self::keyframe_requested) and asks its source
    /// for a fresh keyframe (e.g. an RTCP PLI to a WHIP publisher).
    keyframe_request: Arc<Notify>,
}

impl StreamHandle {
    /// Create a handle with the no-op observer and no GOP cache.
    pub fn new(app: AppName, stream_id: StreamId, capacity: usize) -> Self {
        Self::with_observer(app, stream_id, capacity, 0, Arc::new(NoopObserver))
    }

    /// Create a handle wired to a host-supplied observer.
    ///
    /// `gop_capacity` bounds the keyframe-anchored replay buffer (0 disables it).
    pub fn with_observer(
        app: AppName,
        stream_id: StreamId,
        capacity: usize,
        gop_capacity: usize,
        observer: Arc<dyn Observer>,
    ) -> Self {
        Self::with_config(app, stream_id, capacity, gop_capacity, 0, None, observer)
    }

    /// Create a handle with the full per-stream policy: an optional GOP byte cap
    /// (`gop_byte_capacity`, 0 = unbounded) and an optional default subscriber
    /// lag budget (`default_max_lag`). The engine threads these from
    /// [`AppSpec`](crate::AppSpec); [`with_observer`](Self::with_observer) is the
    /// shorthand with both disabled.
    #[allow(clippy::too_many_arguments)]
    pub fn with_config(
        app: AppName,
        stream_id: StreamId,
        capacity: usize,
        gop_capacity: usize,
        gop_byte_capacity: usize,
        default_max_lag: Option<u64>,
        observer: Arc<dyn Observer>,
    ) -> Self {
        let (tx, _) = broadcast::channel(capacity);
        let qos = QosCounters::default();
        // Treat creation as the last activity so a just-claimed stream is not
        // instantly considered idle before its first frame arrives. Monotonic, to
        // match the idle reaper (see `mono_ms`).
        qos.last_frame_ms.store(mono_ms(), Ordering::Relaxed);
        Self {
            metadata: Arc::new(RwLock::new(StreamMetadata::new(
                app.clone(),
                stream_id.clone(),
            ))),
            state: Arc::new(RwLock::new(StreamState::Idle)),
            key: StreamKey::new(app, stream_id),
            tx: Arc::new(ArcSwapOption::new(Some(Arc::new(tx)))),
            video_config: Arc::new(ArcSwap::new(Arc::new(None))),
            audio_config: Arc::new(ArcSwap::new(Arc::new(None))),
            gop: Arc::new(StdMutex::new(GopBuffer {
                frames: Vec::new(),
                capacity: gop_capacity,
                byte_capacity: gop_byte_capacity,
                bytes: 0,
                overflowed: false,
            })),
            gop_capacity,
            default_max_lag,
            qos: Arc::new(qos),
            observer,
            keyframe_request: Arc::new(Notify::new()),
        }
    }

    /// Request a fresh keyframe from the stream's publisher.
    ///
    /// Wakes the publisher's ingest loop (see [`keyframe_requested`]); a no-op if
    /// nothing is currently awaiting. Used by playback consumers — e.g. a WHEP
    /// viewer relaying a browser PLI — to recover when the GOP cache holds no
    /// keyframe a new subscriber can start from.
    ///
    /// [`keyframe_requested`]: Self::keyframe_requested
    pub fn request_keyframe(&self) {
        self.keyframe_request.notify_one();
    }

    /// Await the next keyframe request raised by [`request_keyframe`](Self::request_keyframe).
    ///
    /// The publisher's ingest loop selects on this and, when it fires, asks its
    /// source for a fresh keyframe (e.g. an RTCP PLI to a WHIP publisher).
    pub async fn keyframe_requested(&self) {
        self.keyframe_request.notified().await;
    }

    /// The `(app, stream_id)` this handle belongs to.
    pub fn key(&self) -> &StreamKey {
        &self.key
    }

    /// Publish a frame to all current subscribers.  Returns the number of
    /// active receivers; returns `Ok(0)` when there are no subscribers.
    pub fn publish_frame(&self, frame: MediaFrame) -> crate::Result<usize> {
        self.observer.on_frame(&self.key, &frame);
        let len = frame.data.len() as u64;
        let is_audio = frame.is_audio();
        let is_key = frame.is_keyframe();
        let is_config = frame.flags.contains(FrameFlags::CONFIG);
        let arc = Arc::new(frame);

        // Cache the latest CONFIG frame for late-joining subscribers.
        if is_config {
            if is_audio {
                self.audio_config.store(Arc::new(Some(Arc::clone(&arc))));
            } else {
                self.video_config.store(Arc::new(Some(Arc::clone(&arc))));
            }
        }

        // Maintain the keyframe-anchored GOP replay buffer.
        let mut gop_truncated = false;
        if self.gop_capacity > 0 {
            if let Ok(mut g) = self.gop.lock() {
                if is_key {
                    g.frames.clear();
                    g.bytes = len as usize;
                    g.overflowed = false;
                    g.frames.push(Arc::clone(&arc));
                } else if !is_config && !g.frames.is_empty() {
                    // Only buffer once a keyframe anchors the GOP; CONFIG frames
                    // are replayed separately via `cached_configs`.
                    //
                    // Admit the frame only if it stays within *both* the frame
                    // count and the byte budget (0 = byte cap disabled). Frame
                    // count alone does not bound memory, so the byte cap keeps a
                    // high-bitrate GOP from growing without limit.
                    let fits_count = g.frames.len() < g.capacity;
                    let fits_bytes =
                        g.byte_capacity == 0 || g.bytes + len as usize <= g.byte_capacity;
                    if fits_count && fits_bytes {
                        g.bytes += len as usize;
                        g.frames.push(Arc::clone(&arc));
                    } else if !g.overflowed {
                        // First overflow of this GOP: the replay will be truncated.
                        // Latch it so the signal fires once, not per dropped frame.
                        g.overflowed = true;
                        gop_truncated = true;
                    }
                }
            }
        }
        if gop_truncated {
            self.observer.on_gop_truncated(&self.key, self.gop_capacity);
        }

        self.record_qos(len, is_audio, is_key);

        // The sender is gone once the stream is closed; publishing then is a
        // no-op (Ok(0)) rather than an error, so a publisher racing teardown
        // winds down cleanly.
        let count = match self.tx.load_full() {
            Some(tx) => tx.send(arc).unwrap_or(0),
            None => 0,
        };
        Ok(count)
    }

    /// Fold one frame into the rolling throughput window.
    fn record_qos(&self, len: u64, is_audio: bool, _is_key: bool) {
        let q = &self.qos;
        let now = mono_ms();
        q.total_frames.fetch_add(1, Ordering::Relaxed);
        q.total_bytes.fetch_add(len, Ordering::Relaxed);
        q.last_frame_ms.store(now, Ordering::Relaxed);
        if is_audio {
            q.window_audio_bytes.fetch_add(len, Ordering::Relaxed);
        } else {
            q.window_video_bytes.fetch_add(len, Ordering::Relaxed);
            q.window_video_frames.fetch_add(1, Ordering::Relaxed);
        }

        let ws = q.window_start_ms.load(Ordering::Relaxed);
        if ws == 0 {
            q.window_start_ms.store(now, Ordering::Relaxed);
        } else if now.saturating_sub(ws) >= 1000 {
            let elapsed = (now - ws) as f64 / 1000.0;
            let vbytes = q.window_video_bytes.swap(0, Ordering::Relaxed);
            let abytes = q.window_audio_bytes.swap(0, Ordering::Relaxed);
            let vframes = q.window_video_frames.swap(0, Ordering::Relaxed);
            q.cur_video_bitrate
                .store((vbytes as f64 * 8.0 / elapsed) as u64, Ordering::Relaxed);
            q.cur_audio_bitrate
                .store((abytes as f64 * 8.0 / elapsed) as u64, Ordering::Relaxed);
            q.cur_fps_milli.store(
                (vframes as f64 / elapsed * 1000.0) as u64,
                Ordering::Relaxed,
            );
            q.window_start_ms.store(now, Ordering::Relaxed);
        }
    }

    /// A snapshot of measured throughput (bitrate, fps, totals).
    pub fn qos(&self) -> Qos {
        let q = &self.qos;
        Qos {
            video_bitrate_bps: q.cur_video_bitrate.load(Ordering::Relaxed),
            audio_bitrate_bps: q.cur_audio_bitrate.load(Ordering::Relaxed),
            fps: q.cur_fps_milli.load(Ordering::Relaxed) as f64 / 1000.0,
            total_frames: q.total_frames.load(Ordering::Relaxed),
            total_bytes: q.total_bytes.load(Ordering::Relaxed),
        }
    }

    /// Monotonic-clock timestamp (process-local milliseconds) of the most
    /// recently published frame (or stream creation if none yet). Used by the
    /// engine's idle reaper; this is elapsed monotonic time, not wall-clock time,
    /// so compare it only against other readings of the same monotonic clock.
    pub fn last_frame_ms(&self) -> u64 {
        self.qos.last_frame_ms.load(Ordering::Relaxed)
    }

    /// Returns the most recently seen video and audio CONFIG frames,
    /// for replaying to late-joining subscribers.
    pub fn cached_configs(&self) -> (Option<Arc<MediaFrame>>, Option<Arc<MediaFrame>>) {
        let video = (**self.video_config.load()).clone();
        let audio = (**self.audio_config.load()).clone();
        (video, audio)
    }

    /// The frames a late joiner should be handed before going live: cached
    /// decoder configs followed by the current GOP (keyframe + trailing deltas).
    ///
    /// Replaying these lets a new subscriber start decoding immediately rather
    /// than waiting for the next keyframe — sub-second join times at scale.
    /// Requires the app to have enabled a GOP cache; otherwise only the cached
    /// configs are returned.
    pub fn replay_buffer(&self) -> Vec<Arc<MediaFrame>> {
        let (vcfg, acfg) = self.cached_configs();
        let mut out = Vec::new();
        out.extend(vcfg.clone());
        out.extend(acfg.clone());
        if self.gop_capacity > 0 {
            if let Ok(g) = self.gop.lock() {
                out.reserve(g.frames.len());
                for f in &g.frames {
                    // Avoid duplicating a config frame already pushed above. Only
                    // the (≤2) cached configs can collide with a GOP frame — GOP
                    // frames are themselves distinct — so compare against just
                    // those, keeping this O(n) rather than O(n²) over the GOP.
                    let dup = vcfg.as_ref().is_some_and(|c| Arc::ptr_eq(c, f))
                        || acfg.as_ref().is_some_and(|c| Arc::ptr_eq(c, f));
                    if !dup {
                        out.push(Arc::clone(f));
                    }
                }
            }
        }
        out
    }

    /// Subscribe to this stream's frame bus.
    ///
    /// The returned raw [`broadcast::Receiver`] surfaces [`RecvError::Lagged`]
    /// when a slow consumer falls behind the channel capacity — callers that
    /// `while let Ok(_) = rx.recv().await` will silently terminate on the first
    /// lag. Prefer [`subscribe_resilient`](Self::subscribe_resilient) unless you
    /// are deliberately handling lag yourself.
    pub fn subscribe(&self) -> broadcast::Receiver<Arc<MediaFrame>> {
        match self.tx.load_full() {
            Some(tx) => tx.subscribe(),
            // The stream has already closed: hand back a receiver on a spent
            // channel so the caller's `recv` loop terminates immediately with
            // `Closed` rather than blocking forever.
            None => {
                let (_, rx) = broadcast::channel(1);
                rx
            }
        }
    }

    /// Subscribe with a [`Subscription`] that resynchronizes after lag instead
    /// of terminating, reporting each gap to the installed [`Observer`] via
    /// [`Observer::on_subscriber_lagged`].
    ///
    /// If the app configured [`AppSpec::subscriber_max_lag`](crate::AppSpec::subscriber_max_lag),
    /// that bound is applied by default, so a chronically slow consumer is evicted
    /// automatically; a caller may still tighten it with [`Subscription::max_lag`].
    pub fn subscribe_resilient(&self) -> Subscription {
        Subscription {
            rx: self.subscribe(),
            key: self.key.clone(),
            observer: Arc::clone(&self.observer),
            max_lag: self.default_max_lag,
            skipped: 0,
        }
    }

    /// Drive this stream to a [`FrameSink`]: replay the instant-start buffer
    /// (cached configs + cached GOP) so the consumer can decode immediately,
    /// then forward live frames until the stream ends or `shutdown` fires.
    ///
    /// This is the playback loop every egress that pulls from a single stream
    /// shares (RTSP server, RTMP relay, …); each supplies only its per-frame
    /// [`FrameSink::send`].
    pub async fn drive_to(
        &self,
        shutdown: &CancellationToken,
        sink: &mut dyn FrameSink,
    ) -> Result<()> {
        for frame in self.replay_buffer() {
            if sink.send(frame).await?.is_break() {
                return Ok(());
            }
        }
        let mut sub = self.subscribe_resilient();
        loop {
            tokio::select! {
                _ = shutdown.cancelled() => break,
                next = sub.recv() => match next {
                    Some(frame) => {
                        if sink.send(frame).await?.is_break() {
                            break;
                        }
                    }
                    None => break, // stream ended
                }
            }
        }
        Ok(())
    }

    /// Drive this stream into a [`Packager`](crate::packager::Packager) (HLS
    /// segmenter, DASH packager, …): replay the instant-start buffer — **which
    /// carries the cached CONFIG access unit (SPS/PPS)** — then forward live
    /// frames until the stream ends or `shutdown` fires, and `finish`.
    ///
    /// Priming with [`replay_buffer`](Self::replay_buffer) is essential, not just
    /// an optimization: the CONFIG frame is published once at stream start, so a
    /// packager that only `subscribe_resilient`s (subscribing *after* the event
    /// that spawned it) never sees it — and then a fragmented-MP4 muxer never
    /// builds its `init.m4s` and the master playlist never learns its codec
    /// string. Replaying configs first makes both deterministic.
    #[cfg(feature = "hls")]
    pub async fn package_to(
        &self,
        shutdown: &CancellationToken,
        packager: &mut dyn crate::packager::Packager,
    ) -> Result<()> {
        for frame in self.replay_buffer() {
            packager.push(&frame).await?;
        }
        let mut sub = self.subscribe_resilient();
        loop {
            tokio::select! {
                _ = shutdown.cancelled() => break,
                next = sub.recv() => match next {
                    Some(frame) => packager.push(&frame).await?,
                    None => break, // stream ended
                }
            }
        }
        packager.finish().await
    }

    /// Number of active subscribers (0 once the stream is closed).
    pub fn subscriber_count(&self) -> usize {
        self.tx
            .load_full()
            .map(|tx| tx.receiver_count())
            .unwrap_or(0)
    }

    /// Close the frame bus: drop the sole sender so every subscriber's `recv`
    /// observes `Closed` and terminates, *regardless* of how many `StreamHandle`
    /// clones are still alive.
    ///
    /// Called by the registry when a publish ends (see
    /// `Application::end_publish`). Idempotent. This is what makes the channel's
    /// lifetime track the stream's lifecycle rather than handle reachability.
    pub fn close(&self) {
        self.tx.store(None);
    }

    /// Transition to a new state.
    pub async fn set_state(&self, state: StreamState) {
        let mut guard = self.state.write().await;
        *guard = state;
    }

    /// The current lifecycle state.
    pub async fn current_state(&self) -> StreamState {
        self.state.read().await.clone()
    }

    /// A consistent point-in-time copy of this stream's [`StreamMetadata`], with
    /// the live measured `fps`/bitrate overlaid from [`qos`](Self::qos).
    ///
    /// Cloning the snapshot releases the lock immediately, so callers never hold
    /// the metadata `RwLock` across an `.await`.
    pub async fn metadata_snapshot(&self) -> StreamMetadata {
        let mut m = self.metadata.read().await.clone();
        let q = self.qos();
        m.video_bitrate_bps = q.video_bitrate_bps;
        m.audio_bitrate_bps = q.audio_bitrate_bps;
        if q.fps > 0.0 {
            m.fps = q.fps;
        }
        m
    }

    /// Mutate this stream's [`StreamMetadata`] under the write lock.
    ///
    /// Ingest handlers call this as they parse the stream — e.g. on the first
    /// keyframe to record resolution from the codec config, or to set the
    /// publisher address — so the metadata exposed to operators and the control
    /// plane stays live rather than frozen at its zeroed defaults.
    ///
    /// ```no_run
    /// # use arcly_stream::StreamHandle;
    /// # async fn demo(handle: &StreamHandle, addr: std::net::SocketAddr) {
    /// handle
    ///     .update_metadata(|m| {
    ///         m.publisher_addr = Some(addr);
    ///         m.width = 1920;
    ///         m.height = 1080;
    ///         m.ingest_protocol = "rtmp".to_string();
    ///     })
    ///     .await;
    /// # }
    /// ```
    pub async fn update_metadata(&self, f: impl FnOnce(&mut StreamMetadata)) {
        let mut guard = self.metadata.write().await;
        f(&mut guard);
    }
}

impl std::fmt::Debug for StreamHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StreamHandle")
            .field("key", &self.key)
            .field("subscribers", &self.subscriber_count())
            .finish()
    }
}

/// A lag-tolerant subscription to a stream's frame bus.
///
/// Returned by [`StreamHandle::subscribe_resilient`]. Unlike a raw
/// [`broadcast::Receiver`], [`recv`](Self::recv) does not terminate when the
/// consumer falls behind: the dropped span is reported to the [`Observer`] as
/// [`on_subscriber_lagged`](Observer::on_subscriber_lagged) and reception
/// continues from the oldest still-buffered frame. This is the recommended
/// consumer loop for packagers, recorders, and SFUs.
///
/// An optional [`max_lag`](Self::max_lag) bound turns chronic lag into
/// eviction: once cumulative dropped frames exceed the bound, `recv` returns
/// `None` (after an [`on_subscriber_evicted`](Observer::on_subscriber_evicted)
/// notification) so a hopelessly slow consumer is shed rather than wasting
/// buffer churn forever.
pub struct Subscription {
    rx: broadcast::Receiver<Arc<MediaFrame>>,
    key: StreamKey,
    observer: Arc<dyn Observer>,
    max_lag: Option<u64>,
    skipped: u64,
}

impl Subscription {
    /// Evict this subscriber once cumulative dropped frames exceed `max`.
    ///
    /// ```no_run
    /// # use arcly_stream::StreamHandle;
    /// # fn demo(handle: &StreamHandle) {
    /// let sub = handle.subscribe_resilient().max_lag(10_000);
    /// # let _ = sub;
    /// # }
    /// ```
    pub fn max_lag(mut self, max: u64) -> Self {
        self.max_lag = Some(max);
        self
    }

    /// Total frames dropped from this subscriber's view so far.
    pub fn dropped(&self) -> u64 {
        self.skipped
    }

    /// Receive the next frame, resynchronizing past any lag.
    ///
    /// Returns `None` when the stream's sender is dropped (the publisher ended)
    /// or when the `max_lag` eviction threshold is crossed:
    ///
    /// ```no_run
    /// # async fn run(sub: &mut arcly_stream::bus::Subscription) {
    /// while let Some(frame) = sub.recv().await {
    ///     // packetize `frame` …
    /// }
    /// # }
    /// ```
    pub async fn recv(&mut self) -> Option<Arc<MediaFrame>> {
        loop {
            match self.rx.recv().await {
                Ok(frame) => return Some(frame),
                Err(RecvError::Lagged(skipped)) => {
                    self.skipped = self.skipped.saturating_add(skipped);
                    self.observer.on_subscriber_lagged(&self.key, skipped);
                    if let Some(max) = self.max_lag {
                        if self.skipped > max {
                            self.observer.on_subscriber_evicted(&self.key);
                            return None;
                        }
                    }
                    continue;
                }
                Err(RecvError::Closed) => return None,
            }
        }
    }

    /// Borrow the underlying raw receiver, for callers that need
    /// [`broadcast::Receiver`] APIs directly.
    pub fn raw(&mut self) -> &mut broadcast::Receiver<Arc<MediaFrame>> {
        &mut self.rx
    }
}

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

    fn video(pts: i64, key: bool) -> MediaFrame {
        MediaFrame::new_video(
            pts,
            pts,
            bytes::Bytes::from_static(&[0, 0, 0, 1, 0x65]),
            CodecId::H264,
            key,
        )
    }

    /// Regression guard for the sender-lifetime fix: `close()` must terminate a
    /// subscriber's `recv` even while a `StreamHandle` clone is still held — the
    /// exact shape that previously hung WHEP egress forever.
    #[tokio::test]
    async fn close_terminates_recv_while_a_handle_clone_is_held() {
        let handle = StreamHandle::new("live".into(), "show".into(), 16);
        let mut sub = handle.subscribe_resilient();

        // A retained clone (e.g. an egress pump) must NOT pin the channel open.
        let retained = handle.clone();
        handle.publish_frame(video(0, true)).unwrap();
        assert!(sub.recv().await.is_some(), "frame delivered before close");

        retained.close();

        // Without the registry-owned sender this would block forever; bound it so
        // a regression fails loudly instead of hanging the suite.
        let got = tokio::time::timeout(std::time::Duration::from_secs(5), sub.recv())
            .await
            .expect("recv resolved after close (no hang)");
        assert!(got.is_none(), "recv returns None once the stream is closed");

        // Publishing post-close is a clean no-op, not a panic.
        assert_eq!(retained.publish_frame(video(1, false)).unwrap(), 0);
        assert_eq!(retained.subscriber_count(), 0);
    }

    /// `package_to` must replay the cached CONFIG access unit even though it was
    /// published *before* the packager subscribed — the regression behind DASH's
    /// missing `init.m4s` and the absent master playlist.
    #[cfg(feature = "hls")]
    #[tokio::test]
    async fn package_to_replays_config_published_before_subscribe() {
        use crate::packager::Packager;
        use tokio_util::sync::CancellationToken;

        // A trivial packager that records the flags of every frame it is pushed.
        struct RecordingPackager(std::sync::Arc<std::sync::Mutex<Vec<FrameFlags>>>);
        #[async_trait::async_trait]
        impl Packager for RecordingPackager {
            async fn push(&mut self, frame: &MediaFrame) -> crate::Result<()> {
                self.0.lock().unwrap().push(frame.flags);
                Ok(())
            }
            async fn finish(&mut self) -> crate::Result<()> {
                Ok(())
            }
        }

        let handle = StreamHandle::new("live".into(), "cam".into(), 16);

        // The one-shot CONFIG frame arrives BEFORE any packager subscribes.
        let mut cfg = video(0, true);
        cfg.flags |= FrameFlags::CONFIG;
        handle.publish_frame(cfg).unwrap();

        let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let mut pkg = RecordingPackager(seen.clone());
        let h = handle.clone();
        let shutdown = CancellationToken::new();
        let task = tokio::spawn(async move { h.package_to(&shutdown, &mut pkg).await });

        // Give the replay + subscribe a moment, then end the stream.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        handle.publish_frame(video(1, false)).unwrap();
        handle.close();
        task.await.unwrap().unwrap();

        assert!(
            seen.lock()
                .unwrap()
                .iter()
                .any(|f| f.contains(FrameFlags::CONFIG)),
            "package_to must replay the cached CONFIG frame to the packager"
        );
    }

    /// The GOP-overflow signal fires once per GOP (not per dropped frame) and
    /// resets on the next keyframe so a fresh GOP can report again.
    #[test]
    fn gop_truncation_signals_once_per_gop() {
        use crate::StreamKey;
        use std::sync::atomic::{AtomicUsize, Ordering};

        #[derive(Default)]
        struct CountingObserver {
            truncations: AtomicUsize,
        }
        impl Observer for CountingObserver {
            fn on_gop_truncated(&self, _key: &StreamKey, _capacity: usize) {
                self.truncations.fetch_add(1, Ordering::Relaxed);
            }
        }

        let obs = Arc::new(CountingObserver::default());
        // Capacity 2: keyframe + 1 delta fit; the 2nd delta onward overflows.
        let handle = StreamHandle::with_observer(
            "live".into(),
            "g".into(),
            64,
            2,
            Arc::clone(&obs) as Arc<dyn Observer>,
        );

        handle.publish_frame(video(0, true)).unwrap(); // keyframe anchors GOP
        handle.publish_frame(video(1, false)).unwrap(); // fills to capacity
        handle.publish_frame(video(2, false)).unwrap(); // first overflow → signal
        handle.publish_frame(video(3, false)).unwrap(); // still overflowing → silent
        assert_eq!(
            obs.truncations.load(Ordering::Relaxed),
            1,
            "one signal per GOP"
        );

        // A new keyframe resets the latch; overflowing again signals once more.
        handle.publish_frame(video(4, true)).unwrap();
        handle.publish_frame(video(5, false)).unwrap();
        handle.publish_frame(video(6, false)).unwrap();
        handle.publish_frame(video(7, false)).unwrap();
        assert_eq!(
            obs.truncations.load(Ordering::Relaxed),
            2,
            "next GOP signals again"
        );
    }

    /// The GOP byte cap truncates a high-bitrate GOP independently of the frame
    /// count: a buffer with a generous frame count but a tight byte budget stops
    /// admitting frames once the budget is reached.
    #[test]
    fn gop_byte_cap_bounds_replay_buffer_memory() {
        use crate::StreamKey;
        use std::sync::atomic::{AtomicUsize, Ordering};

        #[derive(Default)]
        struct CountingObserver {
            truncations: AtomicUsize,
        }
        impl Observer for CountingObserver {
            fn on_gop_truncated(&self, _key: &StreamKey, _capacity: usize) {
                self.truncations.fetch_add(1, Ordering::Relaxed);
            }
        }

        fn frame(pts: i64, key: bool, len: usize) -> MediaFrame {
            MediaFrame::new_video(
                pts,
                pts,
                bytes::Bytes::from(vec![0u8; len]),
                CodecId::H264,
                key,
            )
        }

        let obs = Arc::new(CountingObserver::default());
        // Frame count is generous (100) but the byte budget admits only ~250 B.
        let handle = StreamHandle::with_config(
            "live".into(),
            "hb".into(),
            64,
            100,
            250,
            None,
            Arc::clone(&obs) as Arc<dyn Observer>,
        );

        handle.publish_frame(frame(0, true, 100)).unwrap(); // keyframe: 100 B
        handle.publish_frame(frame(1, false, 100)).unwrap(); // +100 → 200 B (fits)
        handle.publish_frame(frame(2, false, 100)).unwrap(); // +100 → 300 B > 250 → drop
        handle.publish_frame(frame(3, false, 100)).unwrap(); // still over → silent

        // Only keyframe + one delta were buffered, despite the 100-frame count cap.
        assert_eq!(handle.replay_buffer().len(), 2);
        assert_eq!(
            obs.truncations.load(Ordering::Relaxed),
            1,
            "byte-cap overflow signals once per GOP"
        );

        // A new keyframe resets the byte accounting so the next GOP buffers again.
        handle.publish_frame(frame(4, true, 100)).unwrap();
        handle.publish_frame(frame(5, false, 100)).unwrap();
        assert_eq!(handle.replay_buffer().len(), 2, "fresh GOP buffers anew");
    }

    /// A default `max_lag` configured on the handle is applied to every resilient
    /// subscription, so a chronically slow consumer is evicted without the caller
    /// setting a per-subscription bound.
    #[tokio::test]
    async fn default_max_lag_evicts_slow_subscriber() {
        // Capacity 4, default lag budget 4 frames.
        let handle = StreamHandle::with_config(
            "live".into(),
            "slow".into(),
            4,
            0,
            0,
            Some(4),
            Arc::new(NoopObserver),
        );
        let mut sub = handle.subscribe_resilient();

        // Overrun the ring far past the budget without ever receiving.
        for i in 0..50 {
            handle.publish_frame(video(i, i == 0)).unwrap();
        }
        // The first recv observes lag beyond the default budget → eviction (None).
        assert!(
            sub.recv().await.is_none(),
            "subscriber evicted by the app default max_lag"
        );
    }

    /// `mono_ms` is monotonic and drives `last_frame_ms`, so the idle reaper's
    /// elapsed-time math is independent of the wall clock.
    #[test]
    fn mono_clock_is_monotonic_and_drives_last_frame() {
        let a = mono_ms();
        let b = mono_ms();
        assert!(b >= a, "monotonic clock never goes backward");

        let handle = StreamHandle::new("live".into(), "m".into(), 4);
        let before = mono_ms();
        handle.publish_frame(video(0, true)).unwrap();
        assert!(
            handle.last_frame_ms() >= before,
            "last_frame_ms advances on the monotonic clock"
        );
    }
}