millipede_core/
events.rs

1//! Crawler lifecycle events and broadcast delivery.
2
3use crate::request::Request;
4use std::{sync::Arc, time::Duration};
5
6/// The terminal result of processing one request.
7#[derive(Debug, Clone)]
8pub struct HandledRequest {
9    /// The request that was processed.
10    pub request: Arc<Request>,
11    /// The final URL after redirects, when navigation occurred.
12    pub loaded_url: Option<url::Url>,
13    /// The request's terminal state.
14    pub outcome: RequestFinalState,
15    /// The response status, when a response was received.
16    pub response_status: Option<http::StatusCode>,
17    /// The number of retry attempts made.
18    pub retry_count: u32,
19    /// The total processing duration accumulated across every attempt of this request, including retried attempts.
20    pub duration: Duration,
21}
22
23/// A request's terminal processing state.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25#[non_exhaustive]
26pub enum RequestFinalState {
27    /// The request completed successfully.
28    Succeeded,
29    /// The request permanently failed.
30    Failed,
31    /// The request was deliberately skipped.
32    Skipped,
33}
34
35/// A snapshot of host resource usage.
36///
37/// This is a placeholder data carrier; system-information collection is wired in Phase 4.
38#[derive(Debug, Clone, Default)]
39#[non_exhaustive]
40pub struct SystemSnapshot {
41    /// When the snapshot was collected, if known.
42    pub created_at: Option<time::OffsetDateTime>,
43    /// The fraction of CPU capacity in use, if known.
44    pub cpu_used_ratio: Option<f32>,
45    /// The number of memory bytes in use, if known.
46    pub memory_used_bytes: Option<u64>,
47}
48
49/// A control-plane event emitted during a crawler run.
50#[derive(Debug, Clone)]
51#[non_exhaustive]
52pub enum CrawlerEvent {
53    /// Requests that subscribers persist crawler state.
54    PersistState {
55        /// Whether persistence is occurring before migration.
56        is_migrating: bool,
57    },
58    /// Reports a terminal request snapshot.
59    RequestFinished(HandledRequest),
60    /// Reports a request processing failure.
61    RequestFailed {
62        /// The request that failed.
63        request: Arc<Request>,
64        /// A displayable error description.
65        error: String,
66    },
67    /// Reports current host resource usage.
68    SystemInfo(SystemSnapshot),
69    /// Indicates that immediate cancellation has begun.
70    Aborting,
71    /// Indicates that the crawler is exiting.
72    Exiting,
73}
74
75/// A receiver for crawler events.
76pub type EventStream = tokio::sync::broadcast::Receiver<CrawlerEvent>;
77
78/// A receiver for terminal request snapshots (the data-plane feed).
79pub type ResultStream = tokio::sync::broadcast::Receiver<HandledRequest>;
80
81/// A broadcast channel for crawler control-plane events.
82#[derive(Debug, Clone)]
83pub struct EventBus {
84    tx: tokio::sync::broadcast::Sender<CrawlerEvent>,
85}
86
87impl EventBus {
88    /// Creates an event bus with the specified retained-message capacity.
89    ///
90    /// # Panics
91    ///
92    /// Panics when `capacity` is zero.
93    pub fn new(capacity: usize) -> Self {
94        assert!(capacity > 0, "event bus capacity must be greater than zero");
95        let (tx, _) = tokio::sync::broadcast::channel(capacity);
96        Self { tx }
97    }
98
99    /// Creates a subscriber that receives events emitted after subscription.
100    pub fn subscribe(&self) -> EventStream {
101        self.tx.subscribe()
102    }
103
104    /// Broadcasts an event to current subscribers.
105    ///
106    /// The send error produced when no subscribers exist is intentionally ignored.
107    pub fn emit(&self, event: CrawlerEvent) {
108        let _ = self.tx.send(event);
109    }
110
111    /// Returns the number of active subscribers.
112    pub fn subscriber_count(&self) -> usize {
113        self.tx.receiver_count()
114    }
115}
116
117impl Default for EventBus {
118    fn default() -> Self {
119        Self::new(1024)
120    }
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126
127    #[tokio::test]
128    async fn emit_without_subscribers_does_not_panic() {
129        EventBus::default().emit(CrawlerEvent::Exiting);
130    }
131
132    #[tokio::test]
133    async fn subscriber_receives_emitted_event() {
134        let bus = EventBus::default();
135        let mut subscriber = bus.subscribe();
136        bus.emit(CrawlerEvent::Aborting);
137        assert!(matches!(
138            subscriber.recv().await,
139            Ok(CrawlerEvent::Aborting)
140        ));
141    }
142
143    #[tokio::test]
144    async fn all_subscribers_receive_the_same_event() {
145        let bus = EventBus::default();
146        let mut first = bus.subscribe();
147        let mut second = bus.subscribe();
148        bus.emit(CrawlerEvent::PersistState { is_migrating: true });
149        assert!(matches!(
150            first.recv().await,
151            Ok(CrawlerEvent::PersistState { is_migrating: true })
152        ));
153        assert!(matches!(
154            second.recv().await,
155            Ok(CrawlerEvent::PersistState { is_migrating: true })
156        ));
157    }
158
159    #[tokio::test]
160    async fn late_subscriber_does_not_receive_earlier_event() {
161        let bus = EventBus::default();
162        let mut existing = bus.subscribe();
163        bus.emit(CrawlerEvent::Exiting);
164        let mut late = bus.subscribe();
165        assert!(matches!(existing.recv().await, Ok(CrawlerEvent::Exiting)));
166        assert!(matches!(
167            late.try_recv(),
168            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
169        ));
170    }
171}
millipede_core/events.rs

millipede_core/
events.rs
