car-browser 0.51.0

//! Broadcast-capable CDP screencast pump.
//!
//! A single `Page.startScreencast` connection, fanned out to any number of
//! consumers — disk recording ([`crate::recorder`]) is one; a future
//! `car-server-core` live-preview stream is another. Two properties carry
//! over unchanged from the single-consumer pump this replaces:
//!
//! - **Every frame is ACKed exactly once**, regardless of how many
//!   consumers are subscribed. Chrome keeps at most a small number of
//!   un-acked frames in flight; miss the ack and the stream simply stops,
//!   silently. The ack happens once per incoming CDP event, in [`run`],
//!   before subscriber count is even consulted — consumers never see or
//!   influence it.
//! - **Frames arrive only when the page CHANGES.** A screencast is not a
//!   fixed-rate capture, so `captured_at` on each [`ScreencastFrame`] is a
//!   real wall-clock timestamp, not an assumed frame interval.
//!
//! A third property is new here: **zero consumers means zero frames.** CDP
//! capture does not start until the first [`ScreencastPump::subscribe`]
//! call, so a pump nobody has subscribed to generates no CDP traffic at
//! all — attaching alone (which only pins the viewport) is free.
//!
//! Fan-out is centralized in one loop ([`run`]), which is also the single
//! point a future privacy-blackout gate would suppress delivery to every
//! consumer at once — no consumer implements its own policy.
//!
//! A pump is re-attachable in the sense that matters for this crate: each
//! [`ScreencastPump::attach`] call creates an independent pump tied to
//! whichever page is passed, so stopping one pump on page A and attaching a
//! fresh one to page B is the normal way to move a recording (or a live
//! view) between pages.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use base64::engine::general_purpose::STANDARD as BASE64;
use base64::Engine as _;
use chromiumoxide::cdp::browser_protocol::emulation::{
    ClearDeviceMetricsOverrideParams, SetDeviceMetricsOverrideParams,
};
use chromiumoxide::cdp::browser_protocol::page::{
    EventScreencastFrame, ScreencastFrameAckParams, StartScreencastFormat, StartScreencastParams,
    StopScreencastParams,
};
use chromiumoxide::Page;
use futures::{Stream, StreamExt};
use tokio::sync::{mpsc, Mutex};

use crate::backend::BrowserError;
use crate::models::Viewport;

/// How long a teardown CDP round trip may take before the pump stops waiting
/// on Chrome and falls through to aborting its loop.
///
/// Matches the 2s budget `chromium.rs` already gives `page.close()` and
/// `fetch_nav_state`: a target that has not answered a `stopScreencast` or a
/// `clearDeviceMetricsOverride` in two seconds is wedged, and this path holds
/// the lock every other pump operation needs.
const CDP_TEARDOWN_TIMEOUT: Duration = Duration::from_secs(2);

/// One captured screencast frame: decoded JPEG bytes plus the viewport it
/// was captured at, so a consumer can interpret pixel coordinates without
/// re-querying the page.
#[derive(Debug, Clone)]
pub struct ScreencastFrame {
    /// Decoded JPEG bytes. The raw CDP payload is base64 text; this is
    /// already decoded (`page.screenshot()` decodes for you, the raw
    /// screencast event does not). Shared so cloning a frame for N fan-out
    /// subscribers copies only this pointer, not the full viewport JPEG.
    pub jpeg: Arc<[u8]>,
    /// Viewport pinned for the lifetime of this pump attachment.
    pub viewport: Viewport,
    /// Wall-clock seconds since capture started — real arrival time, since
    /// the screencast is change-driven, not fixed-rate.
    pub captured_at: f64,
}

/// A live subscription — yields frames published after
/// [`ScreencastPump::subscribe`] was called, up to [`FRAME_CHANNEL_CAP`] of
/// them buffered.
pub type FrameReceiver = mpsc::Receiver<ScreencastFrame>;

/// Per-consumer frame buffer, in whole full-viewport JPEGs.
///
/// **Bounded on purpose, and the reason is a behaviour change this pump
/// introduced.** Before the broadcast pump existed, the recorder acked a CDP
/// frame and wrote it to disk in the SAME loop iteration, so the next ack
/// could not be issued until the previous frame had landed — and because
/// Chrome caps un-acked screencast frames in flight, the writer's speed
/// throttled Chrome's emission rate. That flow control was accidental but
/// real, and decoupling the ack from the fan-out removed it: an unbounded
/// channel in front of a consumer slower than Chrome (a blocking disk write,
/// a stalled socket, a contended worker) grows without limit, and each slot
/// is a full-viewport JPEG.
///
/// The policy here is BOUND AND DROP rather than bound-and-await, for both
/// audiences. Awaiting would restore Chrome-level throttling, but it also
/// lets the slowest consumer stall every other one — a recording on a network
/// mount would freeze the live drawer. Dropping costs nothing the outcome
/// notices: a dropped preview frame is invisible, and a dropped RECORDED
/// frame simply extends the previous frame's duration, because the manifest
/// measures durations from real arrival timestamps rather than assuming a
/// fixed rate. A slow disk therefore yields a lower-frame-rate recording —
/// exactly what Chrome throttling produced — with bounded memory either way.
pub const FRAME_CHANNEL_CAP: usize = 16;

/// Acknowledges one CDP screencast frame. Abstracted so the pump's ACK
/// discipline — exactly one ack per incoming frame, independent of how many
/// consumers are subscribed — can be unit tested without a live CDP
/// connection. [`Page`] is the only production implementation.
#[async_trait]
trait FrameAck: Send + Sync {
    async fn ack(&self, session_id: i64);
}

#[async_trait]
impl FrameAck for Page {
    async fn ack(&self, session_id: i64) {
        // ACK FIRST — Chrome stalls the stream on an un-acked frame, and it
        // does so silently. Best-effort: a failed ack surfaces as the
        // stream stalling, which is the same failure mode as today.
        let _ = self
            .execute(ScreencastFrameAckParams::new(session_id))
            .await;
    }
}

/// One raw incoming screencast event: the session id (for the ack) plus the
/// base64 JPEG payload. Decoupled from chromiumoxide's CDP event type so
/// [`run`] can be driven by synthetic input in tests.
struct RawFrame {
    session_id: i64,
    data_base64: String,
}

/// A CDP screencast attached to one page, fanned out to any number of
/// consumers.
///
/// Attaching pins the viewport but does not start capture — capture begins
/// on the first [`Self::subscribe`] and ends only when [`Self::stop`] is
/// called, which stops delivery to every consumer at once. There is
/// deliberately no per-consumer unsubscribe: consumers are meant to be
/// centrally gated (a future privacy blackout stops the whole pump, not
/// each consumer individually).
pub struct ScreencastPump {
    page: Page,
    viewport: Viewport,
    quality: i64,
    every_nth_frame: i64,
    subscribers: Arc<Mutex<Vec<mpsc::Sender<ScreencastFrame>>>>,
    running: Mutex<Option<RunningCapture>>,
    /// Whether `attach` applied `Emulation.setDeviceMetricsOverride` to the
    /// page, and it has not been handed back yet.
    ///
    /// Tracked separately from `running` because attaching MUTATES the page
    /// before anything subscribes: a pump attached and then dropped without a
    /// subscriber still owes the page its metrics back, and keying that
    /// cleanup off "was there a capture" left the live page pinned.
    metrics_overridden: AtomicBool,
}

struct RunningCapture {
    stop: Arc<AtomicBool>,
    task: tokio::task::JoinHandle<()>,
    started: Instant,
}

/// Decide whether an existing capture can be reused as-is. Returns the
/// instant it started if so; `None` if there is no capture yet, OR the
/// previous one's task has already finished on its own — the CDP event
/// stream ended (page navigation, tab close, Chrome crash) without
/// `ScreencastPump::stop()` being called, so `running.is_some()` alone is
/// not enough to tell "still capturing" from "capture died silently".
/// `JoinHandle::is_finished()` is what tells the two apart, and it's a pure
/// check — no CDP connection needed — which is what makes this function
/// unit-testable without a live `Page` (see `tests::reusable_start_*`
/// below).
fn reusable_start(running: &Option<RunningCapture>) -> Option<Instant> {
    running
        .as_ref()
        .filter(|r| !r.task.is_finished())
        .map(|r| r.started)
}

impl ScreencastPump {
    /// Attach to `page` and pin its viewport. Does not start capturing —
    /// call [`Self::subscribe`] to begin.
    ///
    /// `every_nth_frame` throttles Chrome's emission at the source —
    /// cheaper than capturing everything and dropping frames later.
    pub async fn attach(
        page: &Page,
        quality: i64,
        every_nth_frame: i64,
        max_width: u32,
        max_height: u32,
    ) -> Result<Self, BrowserError> {
        // Force the PAGE VIEWPORT to the target size before capturing. A
        // headed Chromium's window size is not its viewport size (browser
        // chrome, OS decorations, DPI scaling all eat into it), so
        // launching at 1920x1080 still rendered the page into a fraction
        // of the frame with dead margins around it. Overriding device
        // metrics is what makes the capture full-bleed and deterministic
        // across machines — the same thing Playwright does for its
        // `viewport` option.
        page.execute(
            SetDeviceMetricsOverrideParams::builder()
                .width(max_width as i64)
                .height(max_height as i64)
                .device_scale_factor(1.0)
                .mobile(false)
                .build()
                .map_err(|e| {
                    BrowserError::ScreenshotFailed(format!("device metrics params: {e}"))
                })?,
        )
        .await
        .map_err(|e| BrowserError::ScreenshotFailed(format!("setDeviceMetricsOverride: {e}")))?;

        Ok(Self {
            page: page.clone(),
            viewport: Viewport {
                width: max_width,
                height: max_height,
                device_pixel_ratio: 1.0,
            },
            quality: quality.clamp(1, 100),
            every_nth_frame: every_nth_frame.max(1),
            subscribers: Arc::new(Mutex::new(Vec::new())),
            running: Mutex::new(None),
            // `attach` has just applied it, above.
            metrics_overridden: AtomicBool::new(true),
        })
    }

    /// Register a consumer. The first subscription starts CDP capture; a
    /// pump already capturing just gains another fan-out target — the
    /// running capture is untouched. Returns the receiver plus the instant
    /// capture started, so a consumer can compute frame durations on the
    /// same clock the frames themselves are timestamped against.
    pub async fn subscribe(&self) -> Result<(FrameReceiver, Instant), BrowserError> {
        let (tx, rx) = mpsc::channel(FRAME_CHANNEL_CAP);
        // Register the sender BEFORE (possibly) starting capture, so a
        // frame can never be published while this subscriber is still
        // missing from the fan-out list.
        self.subscribers.lock().await.push(tx);
        let started = match self.ensure_running().await {
            Ok(started) => started,
            Err(e) => {
                // The registration above outlives a FAILED start otherwise,
                // and `ensure_running` fails for real reasons — the event
                // listener and `Page.startScreencast` both error against a
                // target that is navigating or has just closed. `run`'s
                // per-frame prune is the only other collector, and a pump that
                // never started emits no frame to trigger it, so the dead
                // sender sat in the fan-out permanently: `subscribers` never
                // empties, and every "is anybody watching" question above
                // answers yes for a consumer that does not exist.
                //
                // Dropping the receiver FIRST is what makes the sender look
                // closed to the retain — the caller never gets it back.
                drop(rx);
                self.subscribers.lock().await.retain(|s| !s.is_closed());
                return Err(e);
            }
        };
        Ok((rx, started))
    }

    /// Start CDP capture if it isn't already running. Returns the instant
    /// capture started (existing or newly begun).
    async fn ensure_running(&self) -> Result<Instant, BrowserError> {
        let mut running = self.running.lock().await;
        if let Some(started) = reusable_start(&running) {
            return Ok(started);
        }
        // Either nothing has ever subscribed, or the previous capture's
        // task already exited on its own — the CDP event stream ended
        // (page navigation, tab close, Chrome crash) without
        // `ScreencastPump::stop()` being called. Either way, fall through
        // and (re)start rather than handing back a receiver that will
        // never see another frame.

        let events = self
            .page
            .event_listener::<EventScreencastFrame>()
            .await
            .map_err(|e| BrowserError::ScreenshotFailed(format!("screencast listener: {e}")))?
            .map(|frame| RawFrame {
                session_id: frame.session_id,
                data_base64: AsRef::<str>::as_ref(&frame.data).to_string(),
            })
            .boxed();

        // Pin the frame size to the viewport. Omitting max_width/max_height
        // lets Chrome choose, and it letterboxes the page into a
        // differently-shaped frame — a 1920x1080 viewport came back as
        // 1600x1200 with dead margins to the right and below, which is
        // unusable as product footage.
        self.page
            .execute(
                StartScreencastParams::builder()
                    .format(StartScreencastFormat::Jpeg)
                    .quality(self.quality)
                    .every_nth_frame(self.every_nth_frame)
                    .max_width(self.viewport.width as i64)
                    .max_height(self.viewport.height as i64)
                    .build(),
            )
            .await
            .map_err(|e| BrowserError::ScreenshotFailed(format!("startScreencast: {e}")))?;

        let stop = Arc::new(AtomicBool::new(false));
        let started = Instant::now();
        let task = {
            let stop = Arc::clone(&stop);
            let subscribers = Arc::clone(&self.subscribers);
            let page = self.page.clone();
            let viewport = self.viewport;
            tokio::spawn(async move {
                run(events, &page, &stop, started, viewport, &subscribers).await;
            })
        };

        *running = Some(RunningCapture {
            stop,
            task,
            started,
        });
        Ok(started)
    }

    /// Stop CDP capture for every consumer at once.
    ///
    /// Safe to call even if nobody ever subscribed — but NOT a no-op then:
    /// `attach` already pinned the page's device metrics, so the override has
    /// to come back regardless of whether a capture ever ran.
    pub async fn stop(&self) {
        // Unconditional, and first. `attach` mutates the page BEFORE anything
        // subscribes, so "nobody ever subscribed" — the supervisor bailing
        // between attach and `subscribe`, a teardown in that window — is
        // exactly the case the early return below used to skip, leaving the
        // live page pinned at 1920x1080 / dsf 1.0 for the process's life.
        // `supervise` re-attaches on every active-tab change, so that is once
        // per tab visited.
        self.clear_metrics_override().await;
        // The guard is held across the whole teardown, INCLUDING the
        // `stopScreencast` round trip.
        //
        // Taking the capture out in a `let ... else` released the lock at the
        // end of that statement, so the two operations that actually stop
        // Chrome ran unlocked — and both `stop(&self)` and `subscribe(&self)`
        // are `&self` on a pump the module header describes as fanned out to N
        // consumers. Interleaved: A takes the capture and releases; B
        // subscribes, sees `None`, sends `startScreencast`, spawns a fresh
        // task and stores it; A then sends `stopScreencast`. Chrome stops
        // emitting while the slot holds a live task, so `reusable_start`
        // reports the capture healthy forever and every later `subscribe`
        // short-circuits onto a receiver that never yields a frame — with no
        // error raised anywhere. Holding the guard makes `ensure_running`
        // block until the stop has actually reached Chrome.
        let mut running_guard = self.running.lock().await;
        let Some(running) = running_guard.take() else {
            return;
        };
        running.stop.store(true, Ordering::SeqCst);
        // Bounded, like every other CDP call this crate makes from a teardown
        // path (`chromium.rs` gives `page.close()` 2s and nav state
        // `NAV_STATE_TIMEOUT`). Holding the guard across the round trip is
        // deliberate — it is what stops a concurrent `subscribe` starting a
        // fresh screencast into a page this one is still stopping — but an
        // unbounded round trip under it means a wedged renderer parks every
        // later `subscribe`, `set_quality` and `stop` on this pump forever.
        // The abort below is the fallback: past the deadline, killing the loop
        // is strictly better than holding the lock waiting for Chrome.
        let _ = tokio::time::timeout(
            CDP_TEARDOWN_TIMEOUT,
            self.page.execute(StopScreencastParams::default()),
        )
        .await;
        // The pump loop only rechecks the stop flag between frames, and a
        // still page may never send one — so don't await the task, just
        // kill it.
        running.task.abort();
    }

    /// Hand the page's device metrics back, exactly once.
    async fn clear_metrics_override(&self) {
        if !self.metrics_overridden.swap(false, Ordering::SeqCst) {
            return;
        }
        // Bounded for the same reason as the stop below, and it matters more
        // here: this one runs BEFORE the lock is taken, so an unbounded wait
        // delays the teardown that follows it as well.
        let _ = tokio::time::timeout(
            CDP_TEARDOWN_TIMEOUT,
            self.page
                .execute(ClearDeviceMetricsOverrideParams::default()),
        )
        .await;
    }
}

impl Drop for ScreencastPump {
    /// Dropping a pump must end its capture, not orphan it.
    ///
    /// Dropping a tokio `JoinHandle` **detaches** its task rather than
    /// aborting it, so without this the pump loop survived its own pump:
    /// it kept acking and fanning out frames, kept its consumers'
    /// senders alive (so "zero consumers" never tripped anywhere above),
    /// and `Page.stopScreencast` was never sent — leaving Chrome
    /// capturing for the life of the process. Two live paths reach this:
    /// a `RecordingHandle` dropped without `stop()`, and
    /// `FrameFanout::set_quality`, which `abort()`s the supervisor task
    /// that owns the pump as a task local.
    fn drop(&mut self) {
        let overridden = self.metrics_overridden.swap(false, Ordering::SeqCst);
        let running = self.running.get_mut().take();
        if running.is_none() && !overridden {
            return;
        }
        if let Some(running) = &running {
            running.stop.store(true, Ordering::SeqCst);
            running.task.abort();
        }
        // Telling Chrome to stop is async and `drop` is not, so it goes
        // out best-effort on a spawned task — and only when there is a
        // runtime to spawn onto. During a runtime shutdown there is not,
        // and the whole CDP connection is going away regardless.
        let page = self.page.clone();
        let had_capture = running.is_some();
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            handle.spawn(async move {
                if had_capture {
                    let _ = page.execute(StopScreencastParams::default()).await;
                }
                if overridden {
                    let _ = page
                        .execute(ClearDeviceMetricsOverrideParams::default())
                        .await;
                }
            });
        }
    }
}

/// Read raw screencast events, ack + fan each one out to every subscriber.
/// The single ack call per iteration — before subscriber count is even
/// consulted — is what keeps the ACK discipline independent of how many
/// consumers are listening. This loop is also the single point a future
/// privacy-blackout gate would suppress delivery to every consumer at once.
async fn run<A: FrameAck>(
    mut events: impl Stream<Item = RawFrame> + Unpin,
    acker: &A,
    stop: &AtomicBool,
    started: Instant,
    viewport: Viewport,
    subscribers: &Mutex<Vec<mpsc::Sender<ScreencastFrame>>>,
) {
    while let Some(raw) = events.next().await {
        if stop.load(Ordering::SeqCst) {
            break;
        }
        acker.ack(raw.session_id).await;

        // `Binary` wraps the BASE64 TEXT — its `AsRef<[u8]>` hands back the
        // bytes of that text, not the image, so decode explicitly.
        let Ok(bytes) = BASE64.decode(&raw.data_base64) else {
            continue;
        };
        publish(
            subscribers,
            ScreencastFrame {
                jpeg: bytes.into(),
                viewport,
                captured_at: started.elapsed().as_secs_f64(),
            },
        )
        .await;
    }
}

/// Fan `frame` out to every live subscriber, dropping senders whose
/// receiver has gone away. With zero subscribers this is a no-op — nothing
/// is buffered for a future subscriber that hasn't arrived yet.
async fn publish(subscribers: &Mutex<Vec<mpsc::Sender<ScreencastFrame>>>, frame: ScreencastFrame) {
    let mut subs = subscribers.lock().await;
    // `try_send`, never `send().await`: a consumer that has fallen behind
    // loses the frame instead of stalling the pump loop — which would delay
    // the ACK for every OTHER consumer too, and the ack discipline above is
    // what keeps Chrome emitting at all. Only a closed channel deregisters a
    // subscriber; a full one is alive and just behind. See
    // [`FRAME_CHANNEL_CAP`] for why dropping is the right policy for both
    // audiences.
    subs.retain(|tx| match tx.try_send(frame.clone()) {
        Ok(()) => true,
        Err(mpsc::error::TrySendError::Full(_)) => true,
        Err(mpsc::error::TrySendError::Closed(_)) => false,
    });
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::stream;
    use std::sync::atomic::AtomicUsize;

    /// Fake [`FrameAck`] that counts invocations instead of talking to a
    /// live CDP connection — lets the "ack exactly once per frame,
    /// independent of subscriber count" discipline be tested in isolation.
    struct CountingAck(AtomicUsize);

    #[async_trait]
    impl FrameAck for CountingAck {
        async fn ack(&self, _session_id: i64) {
            self.0.fetch_add(1, Ordering::SeqCst);
        }
    }

    fn raw_frames(n: usize) -> Vec<RawFrame> {
        (0..n)
            .map(|i| RawFrame {
                session_id: i as i64,
                data_base64: BASE64.encode(format!("frame-{i}")),
            })
            .collect()
    }

    fn test_viewport() -> Viewport {
        Viewport {
            width: 1920,
            height: 1080,
            device_pixel_ratio: 1.0,
        }
    }

    async fn drain(mut rx: FrameReceiver) -> Vec<ScreencastFrame> {
        let mut out = Vec::new();
        while let Ok(frame) = rx.try_recv() {
            out.push(frame);
        }
        out
    }

    /// The pump used to hand every frame to an UNBOUNDED channel, which
    /// removed the only flow control the recorder ever had: before the
    /// broadcast pump, the ack and the disk write shared a loop iteration, so
    /// a slow write throttled Chrome's emission through the un-acked-frame
    /// cap. Decoupled and unbounded, a consumer slower than Chrome grew a
    /// full-viewport JPEG per frame without limit.
    ///
    /// The policy is bound-and-drop: memory is capped, the pump never stalls
    /// (which would delay the ACK for every OTHER consumer), and a consumer
    /// that is merely behind keeps its subscription.
    #[tokio::test]
    async fn a_consumer_that_falls_behind_loses_frames_but_keeps_its_subscription() {
        let subscribers: Mutex<Vec<mpsc::Sender<ScreencastFrame>>> = Mutex::new(Vec::new());
        let (tx, mut rx) = mpsc::channel(FRAME_CHANNEL_CAP);
        subscribers.lock().await.push(tx);

        // Twice the buffer, published while nothing drains.
        for i in 0..(FRAME_CHANNEL_CAP * 2) {
            publish(
                &subscribers,
                ScreencastFrame {
                    jpeg: vec![i as u8].into(),
                    viewport: test_viewport(),
                    captured_at: i as f64,
                },
            )
            .await;
        }

        assert_eq!(
            subscribers.lock().await.len(),
            1,
            "a consumer that is behind is still a consumer — only a CLOSED channel deregisters"
        );
        let mut buffered = 0usize;
        while rx.try_recv().is_ok() {
            buffered += 1;
        }
        assert_eq!(
            buffered, FRAME_CHANNEL_CAP,
            "memory is bounded by the cap, not by how long the consumer stays slow"
        );
    }

    /// Every subscriber gets every frame — the point of fan-out.
    #[tokio::test]
    async fn every_subscriber_receives_every_frame() {
        let acker = CountingAck(AtomicUsize::new(0));
        let stop = AtomicBool::new(false);
        let subscribers = Mutex::new(Vec::new());

        let (tx_a, rx_a) = mpsc::channel(FRAME_CHANNEL_CAP);
        let (tx_b, rx_b) = mpsc::channel(FRAME_CHANNEL_CAP);
        subscribers.lock().await.push(tx_a);
        subscribers.lock().await.push(tx_b);

        let frames = raw_frames(3);
        run(
            stream::iter(frames),
            &acker,
            &stop,
            Instant::now(),
            test_viewport(),
            &subscribers,
        )
        .await;

        let a = drain(rx_a).await;
        let b = drain(rx_b).await;
        assert_eq!(a.len(), 3, "subscriber A should see all 3 frames");
        assert_eq!(b.len(), 3, "subscriber B should see all 3 frames");
        for (got, i) in a.iter().zip(0..) {
            assert_eq!(
                got.jpeg,
                Arc::<[u8]>::from(format!("frame-{i}").into_bytes())
            );
            assert_eq!(got.viewport.width, 1920);
        }
        assert_eq!(
            a.iter().map(|f| &f.jpeg).collect::<Vec<_>>(),
            b.iter().map(|f| &f.jpeg).collect::<Vec<_>>(),
            "both subscribers must see identical frame content"
        );
        for (left, right) in a.iter().zip(&b) {
            assert!(
                Arc::ptr_eq(&left.jpeg, &right.jpeg),
                "fan-out must share the JPEG allocation instead of deep-cloning it"
            );
        }
    }

    /// The ack count must equal the number of incoming frames — exactly
    /// once each — no matter how many subscribers are attached. A bug that
    /// moved the ack call inside the per-subscriber fan-out loop would
    /// multiply this count; a bug that skipped acking on an empty
    /// subscriber list would zero it.
    #[tokio::test]
    async fn ack_count_is_independent_of_subscriber_count() {
        for subscriber_count in [0usize, 1, 3] {
            let acker = CountingAck(AtomicUsize::new(0));
            let stop = AtomicBool::new(false);
            let subscribers = Mutex::new(Vec::new());
            let mut receivers = Vec::new();
            for _ in 0..subscriber_count {
                let (tx, rx) = mpsc::channel(FRAME_CHANNEL_CAP);
                subscribers.lock().await.push(tx);
                receivers.push(rx);
            }

            run(
                stream::iter(raw_frames(5)),
                &acker,
                &stop,
                Instant::now(),
                test_viewport(),
                &subscribers,
            )
            .await;

            assert_eq!(
                acker.0.load(Ordering::SeqCst),
                5,
                "5 incoming frames must ack exactly 5 times with {subscriber_count} subscribers"
            );
            for rx in receivers {
                assert_eq!(drain(rx).await.len(), 5);
            }
        }
    }

    /// Zero subscribers ⇒ zero frames delivered anywhere, and nothing is
    /// buffered waiting for a future subscriber. Frames are still acked
    /// (that discipline is independent of delivery — see the test above);
    /// what must be zero is what reaches consumers.
    #[tokio::test]
    async fn zero_subscribers_means_zero_frames_delivered() {
        let acker = CountingAck(AtomicUsize::new(0));
        let stop = AtomicBool::new(false);
        let subscribers: Mutex<Vec<mpsc::Sender<ScreencastFrame>>> = Mutex::new(Vec::new());

        run(
            stream::iter(raw_frames(4)),
            &acker,
            &stop,
            Instant::now(),
            test_viewport(),
            &subscribers,
        )
        .await;

        assert!(
            subscribers.lock().await.is_empty(),
            "no subscriber ever registered, so none should exist after the run"
        );
        // No panic, no leaked state: publish() is a safe no-op with no
        // subscribers, which is the whole of what "zero consumers ⇒ zero
        // frames pumped" means at the fan-out layer.
    }

    /// A subscriber that drops its receiver mid-stream is pruned rather
    /// than causing future publishes to fail or panic.
    #[tokio::test]
    async fn dropped_subscriber_is_pruned_not_fatal() {
        let acker = CountingAck(AtomicUsize::new(0));
        let stop = AtomicBool::new(false);
        let subscribers = Mutex::new(Vec::new());

        let (tx_survivor, rx_survivor) = mpsc::channel(FRAME_CHANNEL_CAP);
        let (tx_dropped, rx_dropped) = mpsc::channel(FRAME_CHANNEL_CAP);
        subscribers.lock().await.push(tx_survivor);
        subscribers.lock().await.push(tx_dropped);
        drop(rx_dropped);

        run(
            stream::iter(raw_frames(2)),
            &acker,
            &stop,
            Instant::now(),
            test_viewport(),
            &subscribers,
        )
        .await;

        assert_eq!(drain(rx_survivor).await.len(), 2);
        assert_eq!(
            subscribers.lock().await.len(),
            1,
            "the dropped receiver's sender should have been pruned"
        );
    }

    /// The stop flag halts the loop before processing (and acking) any
    /// further frames — checked at the top of each iteration, matching the
    /// original single-consumer pump's behavior.
    #[tokio::test]
    async fn stop_flag_halts_processing() {
        let acker = CountingAck(AtomicUsize::new(0));
        let stop = AtomicBool::new(true);
        let subscribers = Mutex::new(Vec::new());

        run(
            stream::iter(raw_frames(3)),
            &acker,
            &stop,
            Instant::now(),
            test_viewport(),
            &subscribers,
        )
        .await;

        assert_eq!(acker.0.load(Ordering::SeqCst), 0);
    }

    /// A `RunningCapture` whose task has already finished — no CDP
    /// connection needed, just a dummy tokio task run to completion. This
    /// is the exact stale state `ensure_running` must detect: a capture
    /// task that exited on its own (the CDP event stream ended — page
    /// navigation, tab close, Chrome crash) without `ScreencastPump::stop`
    /// being called.
    async fn dead_running_capture() -> RunningCapture {
        let task = tokio::spawn(async {});
        // The task body does nothing, so it completes on its very first
        // poll — yield until the executor has actually run it, rather than
        // racing on a sleep.
        for _ in 0..1000 {
            if task.is_finished() {
                break;
            }
            tokio::task::yield_now().await;
        }
        assert!(task.is_finished(), "dummy task never finished");
        RunningCapture {
            stop: Arc::new(AtomicBool::new(false)),
            task,
            started: Instant::now(),
        }
    }

    /// A `RunningCapture` whose task is still alive (never completes) — the
    /// normal in-progress-capture state.
    fn alive_running_capture() -> RunningCapture {
        RunningCapture {
            stop: Arc::new(AtomicBool::new(false)),
            task: tokio::spawn(std::future::pending::<()>()),
            started: Instant::now(),
        }
    }

    #[tokio::test]
    async fn reusable_start_is_none_with_no_capture_yet() {
        assert!(reusable_start(&None).is_none());
    }

    #[tokio::test]
    async fn reusable_start_reuses_a_live_capture() {
        let capture = alive_running_capture();
        let started = capture.started;
        let running = Some(capture);

        assert_eq!(reusable_start(&running), Some(started));

        // The dummy task never completes on its own; abort it rather than
        // leaking it past this test.
        if let Some(r) = running {
            r.task.abort();
        }
    }

    /// The bug this fix closes: `ensure_running` used to treat any
    /// `Some(RunningCapture)` as still running, even after its task had
    /// already exited on its own. A later `subscribe()` then returned a
    /// receiver that would never see a frame, with no error surfaced
    /// anywhere. `reusable_start` is the decision `ensure_running` now
    /// makes before reusing anything — it must discard a dead capture
    /// rather than hand back its stale `started` instant.
    #[tokio::test]
    async fn reusable_start_discards_a_dead_capture() {
        let running = Some(dead_running_capture().await);
        assert_eq!(
            reusable_start(&running),
            None,
            "a finished capture task must not be treated as still running"
        );
    }
}