car-inference 0.63.0

//! Pacing, back-off and page reading for requests to DuckDuckGo, the keyless
//! search provider.
//!
//! DuckDuckGo's result page throttles a client by address, not by request:
//! after a burst of searches it answers every endpoint — the HTML page, the
//! lite page, the instant-answer API — with a 202 and a challenge page, and
//! keeps doing so until the client has been quiet for most of a minute.
//! Measured from one Mac on 2026-10-22: of form searches sent back to back
//! (about one every 0.85 s), the first ten were answered and every request
//! after was refused; the first answer came back 46 s after the burst, with
//! requests every 15 s still arriving. Started 2.35 s apart, and again 3.8 s
//! apart, fourteen searches in a row were all answered.
//!
//! So a [`SearchGate`] does three things for one provider. It spaces requests
//! at least [`MIN_INTERVAL`] apart (a caller waits for its turn, and
//! concurrent callers queue in arrival order); once the provider has
//! throttled a request it refuses turns for [`BACKOFF`], because asking only
//! extends the throttle — checked again after a caller's wait, so a caller
//! queued behind the throttled request does not send; and it refuses a turn
//! that would start after the caller's deadline instead of sleeping to it.
//!
//! Every web search in the process that reaches DuckDuckGo shares
//! [`SearchGate::duckduckgo`], because the provider counts the address, so a
//! daemon's sessions and the `search` capability draw on the same budget.
//!
//! Pacing and back-off are per process. Two CAR processes on one machine (the
//! daemon and a command-line `car do`, say) each keep their own gate, and
//! DuckDuckGo counts both; nothing is shared between processes or kept
//! across a restart.
//!
//! The helpers below ([`is_challenge_page`], [`is_throttle_status`],
//! [`THROTTLE_HINT`], [`is_ad_link`]) are the one copy of how a DuckDuckGo
//! answer is read, shared by both of CAR's DuckDuckGo readers.

use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use tokio::time::Instant;

/// The least time between the starts of two requests to one provider: a
/// little over the 2.35 s spacing measured to be answered (module docs).
pub const MIN_INTERVAL: Duration = Duration::from_millis(2500);

/// How long after a throttled request the provider is left alone.
pub const BACKOFF: Duration = Duration::from_secs(60);

/// The whole budget of one search: its wait for turns and every request it
/// makes. A search can make up to three requests (a browser tab, the result
/// page, the instant-answer API), each allowed 30 s on its own, so without a
/// total a search could run for well over a minute and a half plus its queue.
pub const SEARCH_DEADLINE: Duration = Duration::from_secs(60);

/// Why the gate refused a turn.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Refusal {
    /// The provider throttled a request; this much of the back-off is left.
    BackingOff(Duration),
    /// The turn would start after the caller's deadline.
    PastDeadline,
}

/// Pacing and back-off state for one search provider.
pub struct SearchGate {
    interval: Duration,
    backoff: Duration,
    state: Mutex<GateState>,
}

#[derive(Default)]
struct GateState {
    /// When the next request may go out; `None` before the first one.
    next_turn: Option<Instant>,
    /// Until when the provider is left alone after throttling a request.
    backoff_until: Option<Instant>,
}

impl GateState {
    fn backoff_left(&self, now: Instant) -> Option<Duration> {
        let left = self.backoff_until?.saturating_duration_since(now);
        (!left.is_zero()).then_some(left)
    }
}

impl SearchGate {
    /// A gate spacing requests `interval` apart and backing off for
    /// `backoff` after a throttled request.
    pub fn new(interval: Duration, backoff: Duration) -> Self {
        Self {
            interval,
            backoff,
            state: Mutex::new(GateState::default()),
        }
    }

    /// The process-wide gate for DuckDuckGo.
    pub fn duckduckgo() -> Arc<SearchGate> {
        static GATE: OnceLock<Arc<SearchGate>> = OnceLock::new();
        GATE.get_or_init(|| Arc::new(SearchGate::new(MIN_INTERVAL, BACKOFF)))
            .clone()
    }

    /// Wait for a turn to send one request to the provider, and take it.
    ///
    /// Refused at once while the provider is backing off, or when the turn
    /// would start after `deadline` (nothing is reserved then). Otherwise the
    /// turn is reserved before waiting, so callers arriving together are
    /// spaced one interval apart each, in arrival order — and the back-off is
    /// checked again once the wait is over, so a caller that queued behind a
    /// request the provider then throttled does not send.
    pub async fn take_turn(&self, deadline: Instant) -> Result<(), Refusal> {
        let turn = {
            let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
            let now = Instant::now();
            if let Some(left) = state.backoff_left(now) {
                return Err(Refusal::BackingOff(left));
            }
            let turn = state.next_turn.map_or(now, |next| next.max(now));
            if turn > deadline {
                return Err(Refusal::PastDeadline);
            }
            state.next_turn = Some(turn + self.interval);
            turn
        };
        tokio::time::sleep_until(turn).await;
        match self.backoff_left() {
            Some(left) => Err(Refusal::BackingOff(left)),
            None => Ok(()),
        }
    }

    /// Record that the provider throttled a request: it is left alone for
    /// the back-off window from now.
    pub fn throttled(&self) {
        let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
        state.backoff_until = Some(Instant::now() + self.backoff);
    }

    /// How much of the back-off window is left, or `None` when the provider
    /// may be asked.
    pub fn backoff_left(&self) -> Option<Duration> {
        let state = self.state.lock().unwrap_or_else(|p| p.into_inner());
        state.backoff_left(Instant::now())
    }
}

/// The time a request may take: its own `limit`, or what is left before
/// `deadline` when that is less.
pub fn time_left(deadline: Instant, limit: Duration) -> Duration {
    deadline
        .saturating_duration_since(Instant::now())
        .min(limit)
}

/// Why the provider answers with its challenge page or a 202, 403 or 429.
pub const THROTTLE_HINT: &str =
    ", which it does when it throttles or blocks automated searches; try again later";

/// Whether an HTTP status is one the provider answers a client it is
/// throttling or blocking with: 202 (with its challenge page), 403 or 429.
pub fn is_throttle_status(status: u16) -> bool {
    matches!(status, 202 | 403 | 429)
}

/// Whether a result page that held no results is the provider's challenge
/// page (its anomaly modal and challenge form) rather than a page of no
/// results.
pub fn is_challenge_page(html: &str) -> bool {
    html.contains("anomaly-modal") || html.contains("challenge-form")
}

/// Whether a result's address is one of the provider's ad links (an `https`
/// link to its own `/y.js` click tracker) rather than a result.
pub fn is_ad_link(url: &str) -> bool {
    reqwest::Url::parse(url).is_ok_and(|url| {
        url.scheme() == "https"
            && url
                .host_str()
                .is_some_and(|host| host == "duckduckgo.com" || host.ends_with(".duckduckgo.com"))
            && url.path() == "/y.js"
    })
}

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

    /// A deadline no test reaches.
    fn far() -> Instant {
        Instant::now() + Duration::from_secs(3600)
    }

    /// Requests taken back to back are spaced one interval apart each,
    /// whether they come one after another or all at once.
    #[tokio::test(start_paused = true)]
    async fn requests_are_spaced_one_interval_apart() {
        let gate = Arc::new(SearchGate::new(Duration::from_secs(2), BACKOFF));
        let start = Instant::now();
        gate.take_turn(far()).await.unwrap();
        gate.take_turn(far()).await.unwrap();
        assert_eq!(start.elapsed(), Duration::from_secs(2));

        let start = Instant::now();
        let waits = (0..3).map(|_| {
            let gate = gate.clone();
            tokio::spawn(async move {
                gate.take_turn(far()).await.unwrap();
                start.elapsed()
            })
        });
        let mut done = Vec::new();
        for wait in waits {
            done.push(wait.await.unwrap());
        }
        done.sort();
        assert_eq!(
            done,
            [2, 4, 6].map(Duration::from_secs),
            "the first of these waits for the interval after the last request"
        );
    }

    /// A burst of ten searches arriving together through the production
    /// gate goes out spaced at least [`MIN_INTERVAL`] apart — the spacing
    /// that keeps a run of searches answered, where ten sent back to back
    /// were the most the provider answered before throttling.
    #[tokio::test(start_paused = true)]
    async fn a_burst_of_searches_goes_out_spaced_by_the_gate() {
        let gate = Arc::new(SearchGate::new(MIN_INTERVAL, BACKOFF));
        let sends = (0..10).map(|_| {
            let gate = gate.clone();
            tokio::spawn(async move {
                gate.take_turn(far()).await.unwrap();
                Instant::now()
            })
        });
        let mut sent = Vec::new();
        for send in sends {
            sent.push(send.await.unwrap());
        }
        sent.sort();
        for pair in sent.windows(2) {
            assert!(pair[1] - pair[0] >= MIN_INTERVAL, "{:?}", pair[1] - pair[0]);
        }
    }

    /// A request after a quiet spell goes out at once.
    #[tokio::test(start_paused = true)]
    async fn a_request_after_a_quiet_spell_is_not_delayed() {
        let gate = SearchGate::new(Duration::from_secs(2), BACKOFF);
        gate.take_turn(far()).await.unwrap();
        tokio::time::sleep(Duration::from_secs(10)).await;
        let start = Instant::now();
        gate.take_turn(far()).await.unwrap();
        assert_eq!(start.elapsed(), Duration::ZERO);
    }

    /// After a throttled request the provider is left alone for the
    /// back-off window, and may be asked again once it has passed.
    #[tokio::test(start_paused = true)]
    async fn a_throttled_request_starts_the_backoff_window() {
        let gate = SearchGate::new(MIN_INTERVAL, Duration::from_secs(60));
        assert_eq!(gate.backoff_left(), None);
        gate.throttled();
        assert_eq!(gate.backoff_left(), Some(Duration::from_secs(60)));
        assert_eq!(
            gate.take_turn(far()).await,
            Err(Refusal::BackingOff(Duration::from_secs(60)))
        );
        tokio::time::sleep(Duration::from_secs(59)).await;
        assert_eq!(gate.backoff_left(), Some(Duration::from_secs(1)));
        tokio::time::sleep(Duration::from_secs(1)).await;
        assert_eq!(gate.backoff_left(), None);
        assert_eq!(gate.take_turn(far()).await, Ok(()));
    }

    /// Searches queued behind one the provider then throttles do not send:
    /// the back-off is checked again after each caller's wait.
    #[tokio::test(start_paused = true)]
    async fn callers_queued_behind_a_throttled_request_do_not_send() {
        let gate = Arc::new(SearchGate::new(Duration::from_secs(2), BACKOFF));
        let turns: Vec<_> = (0..3)
            .map(|_| {
                let gate = gate.clone();
                tokio::spawn(async move { gate.take_turn(far()).await })
            })
            .collect();
        tokio::task::yield_now().await;
        let mut turns = turns.into_iter();
        assert_eq!(turns.next().unwrap().await.unwrap(), Ok(()));
        // The first request went out and the provider throttled it.
        gate.throttled();
        for queued in turns {
            assert!(
                matches!(queued.await.unwrap(), Err(Refusal::BackingOff(_))),
                "a queued caller sent during the back-off"
            );
        }
    }

    /// A turn that would start after the caller's deadline is refused at
    /// once rather than slept to, and reserves nothing.
    #[tokio::test(start_paused = true)]
    async fn a_turn_past_the_deadline_is_refused_at_once() {
        let gate = SearchGate::new(Duration::from_secs(5), BACKOFF);
        gate.take_turn(far()).await.unwrap();
        let start = Instant::now();
        // short-wait-ok: paused clock; a deadline the gate must refuse at once
        let deadline = start + Duration::from_secs(4);
        assert_eq!(gate.take_turn(deadline).await, Err(Refusal::PastDeadline));
        assert_eq!(start.elapsed(), Duration::ZERO);
        // Nothing was reserved: the next caller's turn is still 5 s after
        // the first request.
        gate.take_turn(far()).await.unwrap();
        assert_eq!(start.elapsed(), Duration::from_secs(5));
    }

    /// A request is allowed its own limit or what is left of the deadline,
    /// whichever is less.
    #[tokio::test(start_paused = true)]
    async fn a_request_gets_what_is_left_of_the_deadline() {
        // short-wait-ok: paused clock; deadline arithmetic only
        let deadline = Instant::now() + Duration::from_secs(10);
        assert_eq!(
            time_left(deadline, Duration::from_secs(30)),
            Duration::from_secs(10)
        );
        assert_eq!(
            time_left(deadline, Duration::from_secs(4)),
            Duration::from_secs(4)
        );
        tokio::time::sleep(Duration::from_secs(11)).await;
        assert_eq!(time_left(deadline, Duration::from_secs(30)), Duration::ZERO);
    }

    /// Only the provider's own `/y.js` click tracker is an ad link.
    #[test]
    fn ad_links_are_the_providers_click_tracker() {
        for ad in [
            "https://duckduckgo.com/y.js?ad_domain=abebooks.com&ad_provider=bingv7aa",
            "https://www.duckduckgo.com/y.js?u3=x",
        ] {
            assert!(is_ad_link(ad), "{ad}");
        }
        for result in [
            "https://en.wikipedia.org/wiki/John_Adams",
            "https://duckduckgo.com/about",
            "https://example.com/y.js",
            "https://duckduckgo.com.evil.example/y.js",
            "http://duckduckgo.com/y.js",
            "not a url",
        ] {
            assert!(!is_ad_link(result), "{result}");
        }
    }
}