spire_core/crawler/event/
mod.rs

1//! Typed [`Event`]s a crawl reports as it runs, see [`Crawler::events`].
2//!
3//! [`Crawler::events`]: crate::crawler::Crawler::events
4
5use std::fmt;
6use std::pin::Pin;
7use std::sync::Arc;
8use std::sync::atomic::{AtomicUsize, Ordering};
9use std::task::{Context, Poll};
10use std::time::Duration;
11
12use ::http::{Method, StatusCode, Uri};
13use futures::Stream;
14use tokio::sync::broadcast;
15use tokio::sync::broadcast::error::RecvError;
16
17use crate::context::Fetch;
18use crate::crawler::{Attempt, Depth, Stats, Tag};
19use crate::frontier::seed::Seed;
20use crate::worker::Outcome;
21use crate::{Error, ErrorKind};
22
23#[cfg(test)]
24mod tests;
25
26/// Events kept for a subscriber that falls behind, before the oldest are
27/// dropped for it.
28pub(crate) const CAPACITY: usize = 1024;
29
30/// Something that happened in a crawl, as [`Crawler::events`] reports it.
31///
32/// Every request reports [`RequestStarted`](Self::RequestStarted), then
33/// [`Fetched`](Self::Fetched) for each fetch its handlers made, then
34/// [`Handled`](Self::Handled) once they returned. After that, it is
35/// [`Retried`](Self::Retried) or [`Failed`](Self::Failed) if it did not
36/// succeed. A request still running when the shutdown grace period ends is
37/// [`Cancelled`](Self::Cancelled) instead of handled.
38///
39/// [`Crawler::events`]: crate::crawler::Crawler::events
40#[derive(Debug, Clone)]
41#[non_exhaustive]
42pub enum Event {
43    /// The run started.
44    Started,
45    /// A request started: its client is connecting.
46    RequestStarted {
47        /// The request.
48        request: RequestInfo,
49    },
50    /// A request was fetched, when its handlers first asked for the
51    /// response. A failed fetch is reported too, and fetched again if a
52    /// handler asks again.
53    Fetched {
54        /// The request.
55        request: RequestInfo,
56        /// The status the server answered with, also when the backend
57        /// failed the fetch for it, e.g. a `404`; `None` if no response
58        /// came, e.g. the connection failed.
59        status: Option<StatusCode>,
60        /// How long the fetch took.
61        latency: Duration,
62    },
63    /// A request's handlers returned.
64    Handled {
65        /// The request.
66        request: RequestInfo,
67        /// What the handlers returned.
68        outcome: OutcomeKind,
69        /// Why the request failed, for [`OutcomeKind::Fail`] and
70        /// [`OutcomeKind::Fatal`].
71        error: Option<Failure>,
72        /// How long the request took since it started, connecting and
73        /// fetching included.
74        elapsed: Duration,
75    },
76    /// The frontier took a request back for another attempt.
77    Retried {
78        /// The attempt that failed.
79        request: RequestInfo,
80        /// How long the next attempt waits at least, if the request said.
81        after: Option<Duration>,
82    },
83    /// A request failed for good: not retryable, out of retries, or refused
84    /// a retry by the frontier.
85    Failed {
86        /// The request.
87        request: RequestInfo,
88    },
89    /// A request was not crawled on purpose.
90    Skipped {
91        /// The request.
92        request: RequestInfo,
93        /// Why it was skipped.
94        reason: SkipReason,
95    },
96    /// A request was still in flight when the shutdown grace period ended.
97    Cancelled {
98        /// The request.
99        request: RequestInfo,
100    },
101    /// A host was throttled, and the crawl slowed down on it.
102    HostSlowed {
103        /// The host's authority, in lowercase.
104        host: String,
105        /// How the host was slowed down.
106        slowdown: Slowdown,
107    },
108    /// A host asked to be left alone, with `Retry-After`, and nothing
109    /// starts on it for a while.
110    HostPaused {
111        /// The host's authority, in lowercase.
112        host: String,
113        /// How long the host is paused.
114        duration: Duration,
115    },
116    /// An origin's robots.txt was fetched, or could not be.
117    RobotsFetched {
118        /// The origin: its scheme, host and port.
119        origin: String,
120        /// Why the fetch failed, if it did.
121        error: Option<Failure>,
122    },
123    /// The run is stopping: nothing new starts, and requests in flight
124    /// finish.
125    Stopping {
126        /// Why the run stops.
127        reason: StopReason,
128    },
129    /// The run finished, with its totals.
130    Finished {
131        /// The totals of the run.
132        stats: Stats,
133    },
134    /// This subscriber fell behind and missed events, which were dropped
135    /// for it; the crawl is never slowed down for a subscriber.
136    Lagged {
137        /// How many events were missed.
138        missed: u64,
139    },
140}
141
142/// The request an [`Event`] is about: what identifies it, without its
143/// headers or body.
144#[derive(Debug, Clone, PartialEq, Eq)]
145#[non_exhaustive]
146pub struct RequestInfo {
147    /// The method.
148    pub method: Method,
149    /// The URI.
150    pub uri: Uri,
151    /// The tag that routes it, if any.
152    pub tag: Option<Tag>,
153    /// How many links away from a seed it is.
154    pub depth: Depth,
155    /// Which attempt this is.
156    pub attempt: Attempt,
157}
158
159impl From<&Seed> for RequestInfo {
160    fn from(seed: &Seed) -> Self {
161        Self {
162            method: seed.method().clone(),
163            uri: seed.uri().clone(),
164            tag: seed.tag().cloned(),
165            depth: seed.depth(),
166            attempt: seed.attempt(),
167        }
168    }
169}
170
171/// What a request's handlers returned, see [`Outcome`].
172#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
173#[non_exhaustive]
174pub enum OutcomeKind {
175    /// [`Outcome::Continue`].
176    Continue,
177    /// [`Outcome::Retry`].
178    Retry,
179    /// [`Outcome::Backoff`].
180    Backoff,
181    /// [`Outcome::Abort`].
182    Abort,
183    /// [`Outcome::Stop`].
184    Stop,
185    /// [`Outcome::Fail`], including a
186    /// timeout or a panic.
187    Fail,
188    /// [`Outcome::Fatal`].
189    Fatal,
190}
191
192impl OutcomeKind {
193    /// Returns the outcome's name, e.g. `"continue"`.
194    #[must_use]
195    pub const fn as_str(self) -> &'static str {
196        match self {
197            Self::Continue => "continue",
198            Self::Retry => "retry",
199            Self::Backoff => "backoff",
200            Self::Abort => "abort",
201            Self::Stop => "stop",
202            Self::Fail => "fail",
203            Self::Fatal => "fatal",
204        }
205    }
206}
207
208impl fmt::Display for OutcomeKind {
209    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
210        f.write_str(self.as_str())
211    }
212}
213
214/// Why a request or robots.txt fetch failed: a copy of its [`Error`]'s
215/// kind, status and message.
216#[derive(Debug, Clone, PartialEq, Eq)]
217#[non_exhaustive]
218pub struct Failure {
219    /// The error's kind.
220    pub kind: ErrorKind,
221    /// The response's status, if the error came from one.
222    pub status: Option<StatusCode>,
223    /// The error, as it displays.
224    pub message: String,
225}
226
227impl From<&Error> for Failure {
228    fn from(error: &Error) -> Self {
229        Self {
230            kind: error.kind(),
231            status: error.status(),
232            message: error.to_string(),
233        }
234    }
235}
236
237impl fmt::Display for Failure {
238    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
239        f.write_str(&self.message)
240    }
241}
242
243/// Why a request was skipped.
244#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
245#[non_exhaustive]
246pub enum SkipReason {
247    /// robots.txt disallows it.
248    Disallowed,
249    /// Its tag was aborted, see [`Outcome::Abort`].
250    TagAborted,
251}
252
253impl SkipReason {
254    /// Returns the reason as it is logged, e.g. `"disallowed by robots.txt"`.
255    #[must_use]
256    pub const fn as_str(self) -> &'static str {
257        match self {
258            Self::Disallowed => "disallowed by robots.txt",
259            Self::TagAborted => "tag aborted",
260        }
261    }
262}
263
264impl fmt::Display for SkipReason {
265    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
266        f.write_str(self.as_str())
267    }
268}
269
270/// How a throttled host was slowed down.
271#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
272#[non_exhaustive]
273pub enum Slowdown {
274    /// Its adaptive concurrency was lowered, see
275    /// [`HostConcurrency::adaptive`](crate::crawler::HostConcurrency::adaptive).
276    Concurrency {
277        /// Requests in flight to it at most, from now on.
278        limit: usize,
279    },
280    /// The gap between starting two requests to it grew, see
281    /// [`HostDelay`](crate::crawler::HostDelay).
282    Gap {
283        /// The new gap.
284        gap: Duration,
285    },
286}
287
288/// Why a run stopped.
289#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
290#[non_exhaustive]
291pub enum StopReason {
292    /// The [shutdown token](crate::crawler::Crawler::shutdown_token) was
293    /// cancelled.
294    Shutdown,
295    /// The [time limit](crate::crawler::Crawler::with_time_limit) was
296    /// reached.
297    TimeLimit,
298    /// The [request budget](crate::crawler::Crawler::with_max_requests)
299    /// was spent.
300    RequestBudget,
301    /// A handler returned [`Outcome::Stop`].
302    Requested,
303    /// A handler returned [`Outcome::Fatal`],
304    /// or the frontier failed.
305    Failed,
306}
307
308impl StopReason {
309    /// Returns the reason as it is logged, e.g. `"time limit reached"`.
310    #[must_use]
311    pub const fn as_str(self) -> &'static str {
312        match self {
313            Self::Shutdown => "shutdown requested",
314            Self::TimeLimit => "time limit reached",
315            Self::RequestBudget => "request budget spent",
316            Self::Requested => "worker requested stop",
317            Self::Failed => "crawl failed",
318        }
319    }
320}
321
322impl fmt::Display for StopReason {
323    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
324        f.write_str(self.as_str())
325    }
326}
327
328/// Sends events to the subscribers of a crawler.
329///
330/// Subscribers are counted on the side, so that without one an event is
331/// never built: checking is a single atomic load, where the channel's own
332/// count takes a lock.
333#[derive(Clone)]
334pub(crate) struct Emitter {
335    sender: broadcast::Sender<Event>,
336    subscribers: Arc<AtomicUsize>,
337}
338
339impl Emitter {
340    /// Creates an emitter keeping `capacity` events, at least one, for each
341    /// subscriber.
342    pub(crate) fn new(capacity: usize) -> Self {
343        let (sender, _) = broadcast::channel(capacity.max(1));
344        Self {
345            sender,
346            subscribers: Arc::default(),
347        }
348    }
349
350    /// Whether anyone is subscribed.
351    pub(crate) fn is_active(&self) -> bool {
352        self.subscribers.load(Ordering::Relaxed) > 0
353    }
354
355    /// Sends the event `event` builds, only built if anyone is subscribed.
356    pub(crate) fn emit(&self, event: impl FnOnce() -> Event) {
357        if self.is_active() {
358            // A subscriber dropped since the check is not an error.
359            let _ = self.sender.send(event());
360        }
361    }
362
363    /// Sends the event `event` builds from `request`, if the events of a
364    /// request are wanted: `request` is only built while anyone subscribes.
365    pub(crate) fn emit_for(
366        &self,
367        request: Option<&RequestInfo>,
368        event: impl FnOnce(RequestInfo) -> Event,
369    ) {
370        if let Some(request) = request {
371            self.emit(|| event(request.clone()));
372        }
373    }
374
375    /// Subscribes to every event from now on.
376    pub(crate) fn subscribe(&self) -> Events {
377        self.subscribers.fetch_add(1, Ordering::Relaxed);
378        let subscription = Subscription {
379            receiver: self.sender.subscribe(),
380            subscribers: self.subscribers.clone(),
381        };
382        let stream = futures::stream::unfold(subscription, |mut subscription| async move {
383            let event = match subscription.receiver.recv().await {
384                Ok(event) => event,
385                Err(RecvError::Lagged(missed)) => Event::Lagged { missed },
386                Err(RecvError::Closed) => return None,
387            };
388            Some((event, subscription))
389        });
390        Events {
391            stream: Box::pin(stream),
392        }
393    }
394
395    /// Prepares reporting the fetches of the request for `seed`, if anyone
396    /// is subscribed.
397    pub(crate) fn notify(&self, seed: &Seed) -> Option<Notify> {
398        self.is_active().then(|| Notify {
399            emitter: self.clone(),
400            request: seed.into(),
401        })
402    }
403}
404
405impl fmt::Debug for Emitter {
406    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
407        f.debug_struct("Emitter")
408            .field("subscribers", &self.subscribers.load(Ordering::Relaxed))
409            .finish_non_exhaustive()
410    }
411}
412
413/// The [`Event`]s of a crawler, from [`Crawler::events`].
414///
415/// It ends once the crawler is dropped.
416///
417/// [`Crawler::events`]: crate::crawler::Crawler::events
418#[must_use = "streams do nothing unless polled"]
419pub struct Events {
420    stream: Pin<Box<dyn Stream<Item = Event> + Send>>,
421}
422
423impl Stream for Events {
424    type Item = Event;
425
426    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Event>> {
427        self.stream.as_mut().poll_next(cx)
428    }
429}
430
431impl fmt::Debug for Events {
432    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
433        f.debug_struct("Events").finish_non_exhaustive()
434    }
435}
436
437/// A subscriber's receiver, counted while it lives.
438struct Subscription {
439    receiver: broadcast::Receiver<Event>,
440    subscribers: Arc<AtomicUsize>,
441}
442
443impl Drop for Subscription {
444    fn drop(&mut self) {
445        self.subscribers.fetch_sub(1, Ordering::Relaxed);
446    }
447}
448
449/// Reports the fetches of a running request, from the task it runs on.
450pub(crate) struct Notify {
451    emitter: Emitter,
452    request: RequestInfo,
453}
454
455impl Notify {
456    /// Reports that the request was fetched.
457    pub(crate) fn fetched(&self, fetch: Fetch) {
458        self.emitter.emit(|| Event::Fetched {
459            request: self.request.clone(),
460            status: fetch.status,
461            latency: fetch.latency.unwrap_or_default(),
462        });
463    }
464}
465
466impl From<&Outcome> for OutcomeKind {
467    fn from(outcome: &Outcome) -> Self {
468        match outcome {
469            Outcome::Continue => Self::Continue,
470            Outcome::Retry { .. } => Self::Retry,
471            Outcome::Backoff { .. } => Self::Backoff,
472            Outcome::Abort(_) => Self::Abort,
473            Outcome::Stop => Self::Stop,
474            Outcome::Fail(_) => Self::Fail,
475            Outcome::Fatal(_) => Self::Fatal,
476        }
477    }
478}
spire_core/crawler/event/mod.rs

spire_core/crawler/event/
mod.rs
