millipede_html/
kind.rs

1use std::{
2    fmt,
3    sync::{Arc, Mutex, MutexGuard},
4    time::Duration,
5};
6
7use futures_util::future::BoxFuture;
8use millipede_core::{
9    antibot::AntiBotDetector,
10    crawler::{
11        AttemptObservation, Crawler, CrawlerEnv, CrawlerHandle, CrawlerKind, RequestEnv,
12        RequestOutcome, RequestPrep,
13    },
14    enqueue::EnqueueLinker,
15    errors::CrawlError,
16    http_client::{HttpClient, HttpClientError, HttpResponse},
17    proxy::{ProxyBuckets, ProxyConfiguration, ProxyInfo, ProxyStrategy},
18    request::Request,
19    router::HasRequest,
20    session::{Session, SessionPool, SessionPoolOptions},
21    storage::StorageHandle,
22};
23use millipede_http::{HttpContext, HttpKind, HttpKindBuilder, HttpPostHookCtx, HttpPreHookCtx};
24
25use crate::HtmlLinkExtractor;
26
27/// A parsed HTML document with the synchronization required for shared handler access.
28///
29/// Scraper 0.24's `atomic` feature makes [`scraper::Html`] `Send`, but not `Sync`:
30/// `scraper::node::Element` populates `std::cell::OnceCell` caches through shared references, and
31/// atomic tendrils implement `Send` but not `Sync`. Consequently, `Arc<scraper::Html>` is not
32/// `Send` and cannot be stored directly in a crawler context that moves between Tokio workers.
33///
34/// This type owns the required mutex rather than exposing a guard to handlers. Its query methods
35/// require owned results, so the lock is always released before a handler can reach an `.await`
36/// point. Adding an unsafe `Sync` implementation for direct shared access would be unsound (and
37/// the workspace forbids unsafe code in any case).
38///
39/// `Arc<scraper::Html>` must not accidentally be treated as sendable shared state. This
40/// compile-fail guard complements the positive assertions for `SynchronizedHtml`:
41///
42/// ```compile_fail
43/// fn assert_send_sync<T: Send + Sync>() {}
44/// assert_send_sync::<std::sync::Arc<scraper::Html>>();
45/// ```
46pub struct SynchronizedHtml {
47    html: Mutex<scraper::Html>,
48}
49
50impl SynchronizedHtml {
51    pub(crate) fn from_html(html: scraper::Html) -> Self {
52        Self {
53            html: Mutex::new(html),
54        }
55    }
56
57    /// Runs a synchronous query against the parsed document and returns its owned result.
58    ///
59    /// The callback cannot return data borrowed from the document. Complete the query before an
60    /// `.await`, then move the returned value into subsequent asynchronous work.
61    ///
62    /// # Panics
63    ///
64    /// Panics if an earlier query callback panicked while holding the document lock.
65    pub fn with_html<R>(&self, query: impl FnOnce(&scraper::Html) -> R) -> R {
66        let html = self.lock();
67        query(&html)
68    }
69
70    /// Locks the document and exposes the complete [`scraper::Html`] API through dereferencing.
71    ///
72    /// Prefer [`Self::with_html`], [`Self::select`], or [`Self::select_first`] when an owned result
73    /// is sufficient. A guard must be dropped before an `.await` point so another handler does not
74    /// block an executor thread waiting for the synchronous mutex.
75    ///
76    /// # Panics
77    ///
78    /// Panics if an earlier query panicked while holding the document lock.
79    pub fn lock(&self) -> MutexGuard<'_, scraper::Html> {
80        self.html.lock().expect("HTML document mutex poisoned")
81    }
82
83    /// Maps every element matching `selector` to an owned value.
84    ///
85    /// The document lock is released before the returned vector is available to the caller.
86    pub fn select<T, F>(&self, selector: &scraper::Selector, mut map: F) -> Vec<T>
87    where
88        F: for<'a> FnMut(scraper::ElementRef<'a>) -> T,
89    {
90        self.with_html(|html| html.select(selector).map(&mut map).collect())
91    }
92
93    /// Maps the first element matching `selector` to an owned value.
94    ///
95    /// The document lock is released before the returned option is available to the caller.
96    pub fn select_first<T, F>(&self, selector: &scraper::Selector, map: F) -> Option<T>
97    where
98        F: for<'a> FnOnce(scraper::ElementRef<'a>) -> T,
99    {
100        self.with_html(|html| html.select(selector).next().map(map))
101    }
102}
103
104impl fmt::Debug for SynchronizedHtml {
105    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
106        formatter
107            .debug_tuple("SynchronizedHtml")
108            .field(&"<scraper::Html>")
109            .finish()
110    }
111}
112
113/// Per-request context produced by [`HtmlKind`].
114///
115/// This intentionally differs from the design sketch in INTERFACE §4.2 in two ways. The response is an
116/// `Arc<HttpResponse>` because [`CrawlerKind::Context`] must be a cheap aliasing clone and the
117/// engine clones a context for each attempt. The HTML field uses [`SynchronizedHtml`] because
118/// `scraper::Html` is not `Sync`, as detailed on that type. The proposed `log: Log` field is
119/// omitted because no `Log` type exists yet, logging is not scheduled by the roadmap, and Phase
120/// 2's `BasicContext` likewise omits it; use `tracing` macros in the meantime.
121#[derive(Clone)]
122#[non_exhaustive]
123pub struct HtmlContext {
124    /// Crawl request that produced this context.
125    pub request: Arc<Request>,
126    /// Final buffered response, shared cheaply across context clones.
127    pub response: Arc<HttpResponse>,
128    /// Parsed HTML document, shared cheaply and synchronized across context clones.
129    ///
130    /// The owned query helpers cover common operations, while [`SynchronizedHtml::lock`] exposes
131    /// the complete [`scraper::Html`] API behind a dereferencing guard.
132    pub html: Arc<SynchronizedHtml>,
133    /// Session used for this attempt, if sessions are enabled.
134    pub session: Option<Arc<Session>>,
135    /// Proxy selected for this attempt.
136    pub proxy_info: Option<ProxyInfo>,
137    /// URL enqueue helper linked to the running crawler.
138    pub enqueue: EnqueueLinker,
139    /// Open default storage resources. For example, `ctx.storage.dataset().push(&item)` works
140    /// when [`millipede_core::storage::DatasetExt`] is in scope.
141    pub storage: StorageHandle,
142    /// Weak handle back to the running crawler.
143    pub crawler: CrawlerHandle,
144    http: HttpContext,
145}
146
147impl HasRequest for HtmlContext {
148    fn request(&self) -> &Request {
149        &self.request
150    }
151}
152
153impl fmt::Debug for HtmlContext {
154    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
155        formatter
156            .debug_struct("HtmlContext")
157            .field("request", &self.request)
158            .field("response_url", &self.response.url)
159            .field("response_status", &self.response.status)
160            .field("response_headers", &self.response.headers)
161            .field("response_body_bytes", &self.response.body.len())
162            .field("redirect_chain", &self.response.redirect_chain)
163            .field("html", &"<scraper::Html>")
164            .field("session", &self.session)
165            .field("proxy_info", &self.proxy_info)
166            .field("enqueue", &self.enqueue)
167            .field("storage", &self.storage)
168            .field("crawler", &self.crawler)
169            .finish()
170    }
171}
172
173/// Errors specific to HTML response processing.
174#[derive(Debug, thiserror::Error)]
175#[non_exhaustive]
176pub enum HtmlError {
177    /// The response declares a media type that cannot be parsed as HTML.
178    #[error("unsupported content type for HTML parsing: {content_type}")]
179    UnsupportedContentType {
180        /// Media type declared by the response.
181        content_type: String,
182    },
183}
184
185/// HTML fetching behavior that delegates transport concerns to [`HttpKind`].
186pub struct HtmlKind {
187    http: HttpKind,
188}
189
190impl HtmlKind {
191    /// Starts configuring an HTML crawler kind.
192    pub fn builder() -> HtmlKindBuilder {
193        HtmlKindBuilder::default()
194    }
195
196    /// Creates an HTML kind with all defaults.
197    pub fn new() -> Result<Self, HttpClientError> {
198        Self::builder().build()
199    }
200
201    /// Wraps an already configured HTTP kind.
202    pub fn from_http(http: HttpKind) -> Self {
203        Self { http }
204    }
205}
206
207/// Configures [`HtmlKind`] by delegating HTTP settings to [`HttpKindBuilder`].
208#[must_use = "builders do nothing unless consumed by build"]
209pub struct HtmlKindBuilder {
210    http: HttpKindBuilder,
211}
212
213impl Default for HtmlKindBuilder {
214    fn default() -> Self {
215        Self {
216            http: HttpKind::builder(),
217        }
218    }
219}
220
221impl HtmlKindBuilder {
222    /// Injects the HTTP transport.
223    pub fn http_client(mut self, client: Arc<dyn HttpClient>) -> Self {
224        self.http = self.http.http_client(client);
225        self
226    }
227
228    /// Enables or disables optional in-flight request coalescing. It is disabled by default.
229    pub fn coalesce_in_flight(mut self, enabled: bool) -> Self {
230        self.http = self.http.coalesce_in_flight(enabled);
231        self
232    }
233
234    /// Enables sessions with the supplied pool options.
235    pub fn session_pool(mut self, options: SessionPoolOptions) -> Self {
236        self.http = self.http.session_pool(options);
237        self
238    }
239
240    /// Uses an existing session pool without managing its persistence lifecycle.
241    pub fn shared_session_pool(mut self, pool: Arc<SessionPool>) -> Self {
242        self.http = self.http.shared_session_pool(pool);
243        self
244    }
245
246    /// Disables session selection, cookie persistence, and session rotation.
247    pub fn disable_sessions(mut self) -> Self {
248        self.http = self.http.disable_sessions();
249        self
250    }
251
252    /// Sets the default proxy bucket.
253    pub fn proxy(mut self, proxy: ProxyConfiguration) -> Self {
254        self.http = self.http.proxy(proxy);
255        self
256    }
257
258    /// Replaces all logical proxy buckets.
259    pub fn proxy_buckets(mut self, proxies: ProxyBuckets) -> Self {
260        self.http = self.http.proxy_buckets(proxies);
261        self
262    }
263
264    /// Sets the policy that selects a proxy bucket per attempt.
265    pub fn proxy_strategy<S: ProxyStrategy>(mut self, strategy: S) -> Self {
266        self.http = self.http.proxy_strategy(strategy);
267        self
268    }
269
270    /// Replaces the rotating User-Agent set.
271    pub fn user_agents<I, U>(mut self, user_agents: I) -> Self
272    where
273        I: IntoIterator<Item = U>,
274        U: Into<String>,
275    {
276        self.http = self.http.user_agents(user_agents);
277        self
278    }
279
280    /// Replaces the exact status codes classified as ordinary retries.
281    pub fn retry_status_codes(mut self, codes: impl IntoIterator<Item = u16>) -> Self {
282        self.http = self.http.retry_status_codes(codes);
283        self
284    }
285
286    /// Controls whether every 5xx response is retried.
287    pub fn retry_server_errors(mut self, enabled: bool) -> Self {
288        self.http = self.http.retry_server_errors(enabled);
289        self
290    }
291
292    /// Replaces statuses that trigger the `retry_on_blocked` session-rotation behavior.
293    pub fn session_status_codes(mut self, codes: impl IntoIterator<Item = u16>) -> Self {
294        self.http = self.http.session_status_codes(codes);
295        self
296    }
297
298    /// Sets the HTTP request deadline.
299    pub fn request_timeout(mut self, timeout: Duration) -> Self {
300        self.http = self.http.request_timeout(timeout);
301        self
302    }
303
304    /// Sets the maximum number of redirects followed for one request.
305    pub fn max_redirects(mut self, maximum: u32) -> Self {
306        self.http = self.http.max_redirects(maximum);
307        self
308    }
309
310    /// Enables response inspection with a caller-supplied anti-bot detector.
311    pub fn detect_anti_bot(mut self, detector: Arc<dyn AntiBotDetector>) -> Self {
312        self.http = self.http.detect_anti_bot(detector);
313        self
314    }
315
316    /// Opts into the default anti-bot detector.
317    pub fn detect_anti_bot_default(mut self) -> Self {
318        self.http = self.http.detect_anti_bot_default();
319        self
320    }
321
322    /// Enables or disables deterministic browser-like request headers.
323    pub fn header_generator(mut self, enabled: bool) -> Self {
324        self.http = self.http.header_generator(enabled);
325        self
326    }
327
328    /// Enables or disables response-body snapshots when handlers fail.
329    pub fn snapshot_errors_on_failure(mut self, enabled: bool) -> Self {
330        self.http = self.http.snapshot_errors_on_failure(enabled);
331        self
332    }
333
334    /// Registers an HTTP hook that runs immediately before navigation.
335    pub fn pre_navigation_hook<F>(mut self, hook: F) -> Self
336    where
337        F: for<'a> Fn(
338                HttpPreHookCtx<'a>,
339            ) -> futures_util::future::BoxFuture<'a, Result<(), CrawlError>>
340            + Send
341            + Sync
342            + 'static,
343    {
344        self.http = self.http.pre_navigation_hook(hook);
345        self
346    }
347
348    /// Registers an HTTP hook that runs after navigation.
349    pub fn post_navigation_hook<F>(mut self, hook: F) -> Self
350    where
351        F: for<'a> Fn(
352                HttpPostHookCtx<'a>,
353            ) -> futures_util::future::BoxFuture<'a, Result<(), CrawlError>>
354            + Send
355            + Sync
356            + 'static,
357    {
358        self.http = self.http.post_navigation_hook(hook);
359        self
360    }
361
362    /// Builds the kind, constructing the default HTTP client when none was injected.
363    pub fn build(self) -> Result<HtmlKind, HttpClientError> {
364        self.http.build().map(HtmlKind::from_http)
365    }
366}
367
368/// A crawler using [`HtmlKind`] to fetch and parse HTML documents.
369pub type HtmlCrawler = Crawler<HtmlKind>;
370
371impl CrawlerKind for HtmlKind {
372    type Context = HtmlContext;
373
374    fn start<'a>(&'a self, env: &'a CrawlerEnv) -> BoxFuture<'a, Result<(), CrawlError>> {
375        self.http.start(env)
376    }
377
378    fn before_request<'a>(
379        &'a self,
380        prep: &'a mut RequestPrep,
381    ) -> BoxFuture<'a, Result<(), CrawlError>> {
382        self.http.before_request(prep)
383    }
384
385    fn execute<'a>(
386        &'a self,
387        env: RequestEnv<'a>,
388    ) -> BoxFuture<'a, Result<Self::Context, CrawlError>> {
389        Box::pin(async move {
390            let http_ctx = self.http.execute(env).await?;
391            if let Some(value) = http_ctx.response.headers.get(http::header::CONTENT_TYPE) {
392                let content_type = String::from_utf8_lossy(value.as_bytes()).into_owned();
393                let media_type = content_type.split(';').next().unwrap_or_default().trim();
394                if !media_type.eq_ignore_ascii_case("text/html")
395                    && !media_type.eq_ignore_ascii_case("application/xhtml+xml")
396                {
397                    return Err(CrawlError::non_retryable(
398                        HtmlError::UnsupportedContentType { content_type },
399                    ));
400                }
401            }
402
403            let html = Arc::new(SynchronizedHtml::from_html(scraper::Html::parse_document(
404                &http_ctx.response.text(),
405            )));
406            let enqueue = EnqueueLinker::with_extractor(
407                http_ctx.crawler.clone(),
408                &http_ctx.request,
409                Arc::new(HtmlLinkExtractor::from_synchronized(
410                    Arc::clone(&html),
411                    http_ctx.response.url.clone(),
412                )),
413            );
414            Ok(HtmlContext {
415                request: http_ctx.request.clone(),
416                response: http_ctx.response.clone(),
417                html,
418                session: http_ctx.session.clone(),
419                proxy_info: http_ctx.proxy_info.clone(),
420                enqueue,
421                storage: http_ctx.storage.clone(),
422                crawler: http_ctx.crawler.clone(),
423                http: http_ctx,
424            })
425        })
426    }
427
428    fn observe(&self, ctx: &Self::Context) -> AttemptObservation {
429        self.http.observe(&ctx.http)
430    }
431
432    fn after_success<'a>(
433        &'a self,
434        ctx: &'a mut Self::Context,
435    ) -> BoxFuture<'a, Result<(), CrawlError>> {
436        self.http.after_success(&mut ctx.http)
437    }
438
439    fn cleanup(
440        &self,
441        outcome: RequestOutcome<Self::Context>,
442    ) -> BoxFuture<'_, Result<(), CrawlError>> {
443        let outcome = match outcome {
444            RequestOutcome::Handled(ctx) => RequestOutcome::Handled(ctx.http),
445            RequestOutcome::HandlerFailed { ctx, error } => RequestOutcome::HandlerFailed {
446                ctx: ctx.http,
447                error,
448            },
449            RequestOutcome::ExecuteFailed { request, error } => {
450                RequestOutcome::ExecuteFailed { request, error }
451            }
452        };
453        self.http.cleanup(outcome)
454    }
455
456    fn stop<'a>(&'a self, env: &'a CrawlerEnv) -> BoxFuture<'a, Result<(), CrawlError>> {
457        self.http.stop(env)
458    }
459}
460
461#[cfg(test)]
462mod tests {
463    use super::*;
464
465    fn assert_send<T: Send>() {}
466
467    fn assert_send_sync<T: Send + Sync>() {}
468
469    fn assert_ctx<T: Send + Clone + 'static>() {}
470
471    #[test]
472    fn context_types_satisfy_engine_bounds() {
473        assert_send::<scraper::Html>();
474        assert_send_sync::<SynchronizedHtml>();
475        assert_send_sync::<Arc<SynchronizedHtml>>();
476        assert_ctx::<HtmlContext>();
477    }
478
479    #[test]
480    fn lock_guard_exposes_the_complete_scraper_api() {
481        let html =
482            SynchronizedHtml::from_html(scraper::Html::parse_document("<title>Phase 5</title>"));
483        let selector = scraper::Selector::parse("title").expect("valid selector");
484        let guard = html.lock();
485        let title = guard
486            .select(&selector)
487            .next()
488            .map(|element| element.text().collect::<String>());
489
490        assert_eq!(title.as_deref(), Some("Phase 5"));
491    }
492}
millipede_html/kind.rs

millipede_html/
kind.rs
