Mangalord/backend/src/crawler/nav.rs

//! Page navigation helpers — wrap `chromiumoxide` `wait_for_navigation`
//! with a timeout so a hung TLS handshake or a page that never fires
//! `load` cannot wedge a worker (or the cron metadata pass) forever.
//!
//! [`NAV_TIMEOUT`] is the global budget. Callers in the crawler use
//! [`wait_for_nav`] to get back a typed error so transient timeouts can
//! be reported separately from underlying CDP errors.

use std::time::Duration;

use chromiumoxide::error::CdpError;
use chromiumoxide::Page;
use thiserror::Error;

/// Maximum wall-clock time we'll wait for a single page navigation. A
/// healthy Chromium reaches `load` in well under a second on the target
/// site; a 30-second cap is generous enough for slow TLS handshakes on
/// the first request after a fresh process while still catching real
/// hangs before they wedge the daemon.
pub const NAV_TIMEOUT: Duration = Duration::from_secs(30);

/// Outcome of a timed-out navigation. `Timeout` is the transient signal
/// callers translate into a retry-friendly error
/// ([`crate::crawler::detect::PageError::Transient`] in the source path,
/// a context'd anyhow elsewhere). `Cdp` carries the underlying
/// chromiumoxide error unchanged.
#[derive(Debug, Error)]
pub enum NavError {
    #[error("navigation timed out after {0:?}")]
    Timeout(Duration),
    #[error(transparent)]
    Cdp(#[from] CdpError),
}

/// Wait for the page's next navigation to complete, capped at
/// [`NAV_TIMEOUT`]. Replaces bare `page.wait_for_navigation().await`
/// throughout the crawler.
pub async fn wait_for_nav(page: &Page) -> Result<(), NavError> {
    match tokio::time::timeout(NAV_TIMEOUT, page.wait_for_navigation()).await {
        Err(_elapsed) => Err(NavError::Timeout(NAV_TIMEOUT)),
        Ok(Err(e)) => Err(NavError::Cdp(e)),
        Ok(Ok(_)) => Ok(()),
    }
}

/// Poll interval for [`wait_for_selector`]. 100ms is fast enough that a
/// page rendering in 200ms isn't held back noticeably, and slow enough
/// not to spam CDP with `find_element` calls on a page that's actually
/// taking its time.
const SELECTOR_POLL_INTERVAL: Duration = Duration::from_millis(100);

/// Wait until `selector` matches at least one element on `page`, or
/// `timeout` elapses. Used after a navigation to confirm a page-type-
/// specific marker is in the DOM before parsing — replaces the fixed
/// post-nav sleep that previously masked partial-render races.
///
/// chromiumoxide 0.7.0 has no built-in `wait_for_selector`, so we poll
/// `find_element` at [`SELECTOR_POLL_INTERVAL`] until success or budget
/// exhaustion. A failed `find_element` is *not* an error here — it just
/// means "not yet" — we only surface an error once the overall
/// `timeout` is up.
pub async fn wait_for_selector(
    page: &Page,
    selector: &str,
    timeout: Duration,
) -> Result<(), NavError> {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        if page.find_element(selector).await.is_ok() {
            return Ok(());
        }
        if tokio::time::Instant::now() >= deadline {
            return Err(NavError::Timeout(timeout));
        }
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        let sleep_for = SELECTOR_POLL_INTERVAL.min(remaining);
        tokio::time::sleep(sleep_for).await;
    }
}

/// Per-page-type budget for [`wait_for_selector`]. Shorter than
/// [`NAV_TIMEOUT`] because by the time we're waiting on a selector, the
/// page has already responded — we're only absorbing post-load JS
/// finishing its row injection, which on a healthy site takes well
/// under a second.
pub const SELECTOR_TIMEOUT: Duration = Duration::from_secs(10);

/// Run `body` to completion, then run `close` — on *every* exit path,
/// including an early `?` / `return` inside `body`. chromiumoxide's `Page`
/// does **not** close its CDP target on drop, so a fetch helper that opens a
/// page with `browser.new_page(...)` and then bails via `?` on a nav /
/// content-read error would leak a browser tab for the process's lifetime.
/// Wrapping the fallible work in `close_after` — with `close` built from a
/// clone of the page — guarantees the tab is closed regardless of how `body`
/// returns.
///
/// Both arguments are pre-built futures, so this stays generic over the
/// body's return type (the anyhow and `PageError` fetch paths both use it)
/// and references no browser types — which also lets it be unit-tested
/// without standing up a real `Page`.
pub(crate) async fn close_after<R>(
    close: impl std::future::Future<Output = ()>,
    body: impl std::future::Future<Output = R>,
) -> R {
    let result = body.await;
    close.await;
    result
}

impl NavError {
    /// Does this navigation error indicate the underlying Chromium
    /// process has died or its CDP connection has dropped? Used by the
    /// dispatcher to decide whether to invalidate the
    /// [`crate::crawler::browser_manager::BrowserManager`] handle so
    /// the next acquire re-launches.
    ///
    /// Both variants count: a `Timeout` past [`NAV_TIMEOUT`] is in
    /// practice always either a hung CDP transport or a wedged page
    /// the browser can't recover from on its own, and a `Cdp` error
    /// surfacing at the navigation layer means the chromium-facing
    /// channel is the failing layer.
    pub fn is_likely_browser_dead(&self) -> bool {
        match self {
            Self::Timeout(_) => true,
            Self::Cdp(_) => true,
        }
    }
}

/// Walk an `anyhow::Error` chain looking for typed evidence that the
/// chromium-facing layer is the failing one. Two markers count:
///
/// 1. A wrapped [`NavError`] flagged by [`NavError::is_likely_browser_dead`].
/// 2. A wrapped [`CdpError`] (via `anyhow::Error::from(CdpError)` at a
///    `Browser::new_page` call site, or any other direct CDP boundary).
///
/// Earlier versions also substring-matched the chain for "connection",
/// "closed", "channel", etc. as a fallback. That was too broad —
/// reqwest TCP-reset errors during CDN image downloads, sqlx
/// connection-pool errors, and similar non-browser failures contain
/// those words and triggered spurious chromium relaunches. The typed
/// downcasts cover every place we hand a chromium error to anyhow,
/// so the fallback is unnecessary.
pub fn anyhow_looks_browser_dead(err: &anyhow::Error) -> bool {
    for cause in err.chain() {
        if let Some(nav) = cause.downcast_ref::<NavError>() {
            if nav.is_likely_browser_dead() {
                return true;
            }
        }
        if cause.downcast_ref::<CdpError>().is_some() {
            return true;
        }
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::future::pending;

    /// Sanity-check the timeout pattern used by [`wait_for_nav`]: a
    /// future that never resolves must yield `Elapsed` within the
    /// configured budget. We can't easily stand up a real `Page` in a
    /// unit test, so we assert the underlying primitive behaves the way
    /// the helper depends on.
    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn timeout_elapses_on_a_future_that_never_resolves() {
        let result =
            tokio::time::timeout(Duration::from_millis(50), pending::<()>()).await;
        assert!(result.is_err(), "expected Elapsed on a hung future");
    }

    #[tokio::test]
    async fn close_after_runs_close_and_returns_value_on_ok() {
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::Arc;
        let closed = Arc::new(AtomicBool::new(false));
        let c = closed.clone();
        let out: anyhow::Result<i32> = close_after(
            async move { c.store(true, Ordering::SeqCst) },
            async { Ok(7) },
        )
        .await;
        assert_eq!(out.unwrap(), 7);
        assert!(closed.load(Ordering::SeqCst), "close must run on the happy path");
    }

    #[tokio::test]
    async fn close_after_runs_close_even_when_body_errs() {
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::Arc;
        let closed = Arc::new(AtomicBool::new(false));
        let c = closed.clone();
        // This is the leak the fix targets: the body bails before it could
        // close the page itself, and `close_after` must still close it.
        let out: anyhow::Result<()> =
            close_after(async move { c.store(true, Ordering::SeqCst) }, async {
                anyhow::bail!("nav failed")
            })
            .await;
        assert!(out.is_err());
        assert!(
            closed.load(Ordering::SeqCst),
            "the page must be closed even when the body returns Err"
        );
    }

    #[test]
    fn nav_error_timeout_message_includes_duration() {
        let e = NavError::Timeout(Duration::from_secs(30));
        assert_eq!(e.to_string(), "navigation timed out after 30s");
    }

    #[test]
    fn timeout_is_treated_as_likely_browser_dead() {
        let e = NavError::Timeout(NAV_TIMEOUT);
        assert!(e.is_likely_browser_dead());
    }

    #[test]
    fn anyhow_with_nav_timeout_in_chain_is_flagged() {
        let outer = NavError::Timeout(NAV_TIMEOUT);
        let wrapped: anyhow::Error =
            anyhow::Error::new(outer).context("wait for chapter nav");
        assert!(anyhow_looks_browser_dead(&wrapped));
    }

    #[test]
    fn anyhow_with_cdp_error_in_chain_is_flagged() {
        // `Browser::new_page` errors get wrapped via
        // `anyhow::Error::from(CdpError)` at the navigate / dispatch
        // call sites. Walking the chain and downcasting to CdpError is
        // what catches that path. Any CdpError variant counts; the
        // Serde variant is the easiest to construct in a unit test.
        let serde_err: serde_json::Error =
            serde_json::from_str::<i32>("not a number").unwrap_err();
        let cdp = CdpError::Serde(serde_err);
        let wrapped: anyhow::Error =
            anyhow::Error::from(cdp).context("open chapter page");
        assert!(anyhow_looks_browser_dead(&wrapped));
    }

    #[test]
    fn anyhow_with_innocuous_parse_error_is_not_flagged() {
        let e: anyhow::Error =
            anyhow::anyhow!("parse manga detail: chapter row regex did not match");
        assert!(!anyhow_looks_browser_dead(&e));
    }

    #[test]
    fn anyhow_with_reqwest_style_connection_message_is_not_flagged() {
        // Regression: the earlier substring fallback flagged any error
        // whose message contained "connection" or "closed" as browser-
        // dead. A TCP reset from a CDN during image download, or a
        // sqlx pool-connection error, would burn a chromium relaunch
        // even though the browser is fine. Typed downcasts only —
        // these untyped strings must pass through.
        for msg in [
            "error sending request: connection reset by peer",
            "PoolTimedOut: timed out waiting for a connection",
            "request to https://cdn/x.jpg: connection closed before message completed",
            "transport error during image fetch",
        ] {
            let e: anyhow::Error = anyhow::anyhow!("{msg}");
            assert!(
                !anyhow_looks_browser_dead(&e),
                "must not flag non-browser error: {msg}"
            );
        }
    }

    /// Same sanity check as [`timeout_elapses_on_a_future_that_never_resolves`],
    /// but for the [`wait_for_selector`] polling pattern: the loop must
    /// surrender on `Elapsed` rather than spinning past the deadline.
    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn selector_polling_pattern_surrenders_at_deadline() {
        let timeout = Duration::from_millis(300);
        let start = tokio::time::Instant::now();
        let deadline = start + timeout;
        // Simulate find_element forever returning "not found".
        let mut polls = 0u32;
        let result: Result<(), NavError> = loop {
            polls += 1;
            if tokio::time::Instant::now() >= deadline {
                break Err(NavError::Timeout(timeout));
            }
            tokio::time::sleep(SELECTOR_POLL_INTERVAL).await;
        };
        assert!(matches!(result, Err(NavError::Timeout(_))));
        // 300ms / 100ms poll interval ≈ 3 iterations plus the final check
        // that breaks out. Allow some slack since the first poll happens
        // before any sleep.
        assert!(polls >= 3, "expected at least 3 poll iterations, got {polls}");
    }
}