ipfrs_network/
network_event_bus.rs

1//! NetworkEventBus — synchronous publish-subscribe event bus for network events.
2//!
3//! Provides topic-based routing, subscriber filtering, and event replay for late
4//! subscribers. All operations are synchronous and suitable for use from both
5//! synchronous and asynchronous contexts without spawning tasks.
6//!
7//! # Design
8//!
9//! - Publishers call [`NetworkEventBus::publish`] to emit events.
10//! - Consumers call [`NetworkEventBus::subscribe`] to register interest in events.
11//! - Subscribers may specify an [`EventFilter`] to narrow the event stream.
12//! - A configurable replay buffer retains recent events so that late subscribers
13//!   can catch up via [`NetworkEventBus::replay_for_subscriber`].
14//! - No `unwrap()` is used anywhere in this module.
15
16use std::collections::{HashMap, VecDeque};
17
18// ---------------------------------------------------------------------------
19// EventTopic
20// ---------------------------------------------------------------------------
21
22/// Categorises the kind of network event being published.
23#[derive(Clone, Debug, PartialEq, Eq, Hash)]
24pub enum EventTopic {
25    /// A remote peer established a connection.
26    PeerConnected,
27    /// A remote peer disconnected or was dropped.
28    PeerDisconnected,
29    /// A content block was received from a peer.
30    BlockReceived,
31    /// A content block was requested by a peer.
32    BlockRequested,
33    /// A DHT lookup was initiated or completed.
34    DhtLookup,
35    /// A gossip-sub message arrived on some topic.
36    GossipMessage,
37    /// An application-defined event with a free-form tag.
38    Custom(String),
39}
40
41impl EventTopic {
42    /// Returns a string representation of the topic suitable for logging.
43    pub fn as_str(&self) -> &str {
44        match self {
45            Self::PeerConnected => "PeerConnected",
46            Self::PeerDisconnected => "PeerDisconnected",
47            Self::BlockReceived => "BlockReceived",
48            Self::BlockRequested => "BlockRequested",
49            Self::DhtLookup => "DhtLookup",
50            Self::GossipMessage => "GossipMessage",
51            Self::Custom(tag) => tag.as_str(),
52        }
53    }
54}
55
56impl std::fmt::Display for EventTopic {
57    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58        f.write_str(self.as_str())
59    }
60}
61
62// ---------------------------------------------------------------------------
63// NebNetworkEvent  (aliased at crate-root as NebNetworkEvent to avoid clash
64//                   with node::NetworkEvent)
65// ---------------------------------------------------------------------------
66
67/// A single network event, as published to the bus.
68#[derive(Clone, Debug, PartialEq, Eq)]
69pub struct NebNetworkEvent {
70    /// Monotonically increasing identifier assigned at publish time.
71    pub id: u64,
72    /// The topic this event belongs to.
73    pub topic: EventTopic,
74    /// Opaque payload bytes supplied by the publisher.
75    pub payload: Vec<u8>,
76    /// Optional peer-id string that identifies the originating peer.
77    pub source_peer: Option<String>,
78    /// Wall-clock timestamp (e.g. seconds or milliseconds since UNIX epoch)
79    /// supplied by the caller; the bus does not call the OS clock itself.
80    pub timestamp: u64,
81}
82
83// ---------------------------------------------------------------------------
84// SubscriberId
85// ---------------------------------------------------------------------------
86
87/// An opaque, monotonically increasing identifier for a subscription.
88///
89/// The sentinel value `SubscriberId(0)` is returned when subscription fails
90/// (e.g. because the bus is already at capacity).
91#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
92pub struct SubscriberId(pub u64);
93
94impl std::fmt::Display for SubscriberId {
95    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        write!(f, "Sub({})", self.0)
97    }
98}
99
100// ---------------------------------------------------------------------------
101// EventFilter
102// ---------------------------------------------------------------------------
103
104/// Determines which events are delivered to a particular subscriber.
105#[derive(Clone, Debug)]
106pub enum EventFilter {
107    /// Accept every event regardless of topic, source, or payload size.
108    All,
109    /// Accept only events whose topic is in the provided list.
110    TopicIn(Vec<EventTopic>),
111    /// Accept only events that originated from the specified peer.
112    FromPeer(String),
113    /// Accept only events whose payload is larger than `n` bytes.
114    PayloadSizeAbove(usize),
115}
116
117// ---------------------------------------------------------------------------
118// NebSubscription  (aliased at crate-root as NebSubscription)
119// ---------------------------------------------------------------------------
120
121/// Runtime state for a single subscriber.
122#[derive(Clone, Debug)]
123pub struct NebSubscription {
124    /// The unique identifier for this subscription.
125    pub id: SubscriberId,
126    /// The filter applied to inbound events.
127    pub filter: EventFilter,
128    /// Total number of events that passed the filter and were counted as
129    /// delivered to this subscriber.
130    pub events_received: u64,
131    /// Timestamp of the most recently delivered event (as supplied by the
132    /// publisher).  Zero if no event has been delivered yet.
133    pub last_event_at: u64,
134}
135
136// ---------------------------------------------------------------------------
137// EventBusConfig
138// ---------------------------------------------------------------------------
139
140/// Configuration knobs for [`NetworkEventBus`].
141#[derive(Clone, Debug)]
142pub struct EventBusConfig {
143    /// Maximum number of simultaneous subscribers.  Attempts to subscribe
144    /// beyond this limit return `SubscriberId(0)`.
145    pub max_subscribers: usize,
146    /// Number of events retained in the replay buffer.  Older events are
147    /// evicted (FIFO) when this limit is reached.
148    pub replay_buffer_size: usize,
149    /// Maximum number of bytes allowed in a single event payload.  Events
150    /// that exceed this limit are rejected with [`BusError::PayloadTooLarge`].
151    pub max_payload_bytes: usize,
152}
153
154impl Default for EventBusConfig {
155    fn default() -> Self {
156        Self {
157            max_subscribers: 1_000,
158            replay_buffer_size: 10_000,
159            max_payload_bytes: 1_048_576, // 1 MiB
160        }
161    }
162}
163
164// ---------------------------------------------------------------------------
165// BusError
166// ---------------------------------------------------------------------------
167
168/// Errors returned by [`NetworkEventBus`] operations.
169#[derive(Clone, Debug, PartialEq, Eq)]
170pub enum BusError {
171    /// No subscription with the given id exists.
172    SubscriberNotFound(u64),
173    /// The payload exceeds the configured maximum.
174    PayloadTooLarge {
175        /// Actual size of the rejected payload.
176        size: usize,
177        /// Configured maximum.
178        max: usize,
179    },
180    /// The bus already has `max_subscribers` active subscriptions.
181    MaxSubscribersReached,
182}
183
184impl std::fmt::Display for BusError {
185    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
186        match self {
187            Self::SubscriberNotFound(id) => {
188                write!(f, "subscriber {id} not found")
189            }
190            Self::PayloadTooLarge { size, max } => {
191                write!(f, "payload too large: {size} bytes (max {max})")
192            }
193            Self::MaxSubscribersReached => {
194                write!(f, "maximum number of subscribers reached")
195            }
196        }
197    }
198}
199
200impl std::error::Error for BusError {}
201
202// ---------------------------------------------------------------------------
203// EventBusStats
204// ---------------------------------------------------------------------------
205
206/// A snapshot of bus-wide statistics.
207#[derive(Clone, Debug)]
208pub struct EventBusStats {
209    /// Current number of active subscribers.
210    pub subscriber_count: usize,
211    /// Current number of events held in the replay buffer.
212    pub replay_buffer_size: usize,
213    /// Total number of events successfully published (not rejected) so far.
214    pub total_published: u64,
215    /// Total number of `(subscriber, event)` delivery pairs counted so far.
216    pub total_delivered: u64,
217    /// Average number of deliveries per published event (0.0 if no events
218    /// have been published yet).
219    pub avg_delivery_per_event: f64,
220}
221
222// ---------------------------------------------------------------------------
223// NetworkEventBus
224// ---------------------------------------------------------------------------
225
226/// Synchronous, in-process publish-subscribe bus for network events.
227///
228/// ## Thread safety
229///
230/// `NetworkEventBus` is intentionally **not** `Send` or `Sync`.  If you need
231/// to share it across threads, wrap it in `Arc<Mutex<NetworkEventBus>>`.
232///
233/// ## Example
234///
235/// ```rust
236/// use ipfrs_network::{
237///     NebNetworkEvent, NebSubscription,
238///     EventBusConfig, EventFilter, EventTopic, NetworkEventBus,
239/// };
240///
241/// let mut bus = NetworkEventBus::new(EventBusConfig::default());
242///
243/// // Subscribe to peer-connection events only.
244/// let sub_id = bus.subscribe(EventFilter::TopicIn(vec![EventTopic::PeerConnected]));
245///
246/// // Publish an event.
247/// let _event_id = bus
248///     .publish(
249///         EventTopic::PeerConnected,
250///         b"hello".to_vec(),
251///         Some("peer-abc".to_string()),
252///         1_000,
253///     )
254///     .expect("publish failed");
255///
256/// // Retrieve matching events for the subscriber.
257/// let events = bus.drain_events_for(sub_id).expect("drain failed");
258/// assert_eq!(events.len(), 1);
259/// ```
260pub struct NetworkEventBus {
261    /// Configuration that governs limits and buffer sizes.
262    pub config: EventBusConfig,
263    /// Live subscriptions, keyed by their numeric id.
264    pub subscriptions: HashMap<u64, NebSubscription>,
265    /// Ring buffer of recent events available for replay.
266    pub replay_buffer: VecDeque<NebNetworkEvent>,
267    /// Counter used to assign unique ids to new events.
268    pub next_event_id: u64,
269    /// Counter used to assign unique ids to new subscriptions.
270    pub next_sub_id: u64,
271    /// Total number of events that have been successfully published.
272    pub total_published: u64,
273    /// Total number of `(subscriber, event)` delivery pairs.
274    pub total_delivered: u64,
275}
276
277impl NetworkEventBus {
278    // -----------------------------------------------------------------------
279    // Construction
280    // -----------------------------------------------------------------------
281
282    /// Create a new bus with the supplied configuration.
283    pub fn new(config: EventBusConfig) -> Self {
284        // Start next_sub_id at 1 so that 0 is permanently reserved as the
285        // "failure sentinel" returned by [`subscribe`] when the bus is full.
286        Self {
287            replay_buffer: VecDeque::with_capacity(config.replay_buffer_size.min(4096)),
288            config,
289            subscriptions: HashMap::new(),
290            next_event_id: 1,
291            next_sub_id: 1,
292            total_published: 0,
293            total_delivered: 0,
294        }
295    }
296
297    // -----------------------------------------------------------------------
298    // Filter helpers
299    // -----------------------------------------------------------------------
300
301    /// Returns `true` if `event` satisfies `filter`.
302    pub fn matches_filter(filter: &EventFilter, event: &NebNetworkEvent) -> bool {
303        match filter {
304            EventFilter::All => true,
305            EventFilter::TopicIn(topics) => topics.contains(&event.topic),
306            EventFilter::FromPeer(peer) => event.source_peer.as_deref() == Some(peer.as_str()),
307            EventFilter::PayloadSizeAbove(n) => event.payload.len() > *n,
308        }
309    }
310
311    // -----------------------------------------------------------------------
312    // Subscription management
313    // -----------------------------------------------------------------------
314
315    /// Register a new subscriber with the given filter.
316    ///
317    /// Returns the assigned [`SubscriberId`].  If the bus is already at
318    /// capacity, `SubscriberId(0)` is returned as a sentinel (no subscription
319    /// is created).
320    pub fn subscribe(&mut self, filter: EventFilter) -> SubscriberId {
321        if self.subscriptions.len() >= self.config.max_subscribers {
322            return SubscriberId(0);
323        }
324        let id = self.next_sub_id;
325        self.next_sub_id = self.next_sub_id.saturating_add(1);
326        let sub = NebSubscription {
327            id: SubscriberId(id),
328            filter,
329            events_received: 0,
330            last_event_at: 0,
331        };
332        self.subscriptions.insert(id, sub);
333        SubscriberId(id)
334    }
335
336    /// Remove the subscription identified by `id`.
337    ///
338    /// Returns `true` if the subscription existed and was removed, `false` if
339    /// no subscription with that id was found.
340    pub fn unsubscribe(&mut self, id: SubscriberId) -> bool {
341        self.subscriptions.remove(&id.0).is_some()
342    }
343
344    /// Return an immutable reference to the [`NebSubscription`] for `id`, or
345    /// `None` if no such subscription exists.
346    pub fn subscriber_stats(&self, id: SubscriberId) -> Option<&NebSubscription> {
347        self.subscriptions.get(&id.0)
348    }
349
350    // -----------------------------------------------------------------------
351    // Publishing
352    // -----------------------------------------------------------------------
353
354    /// Publish an event to the bus.
355    ///
356    /// # Errors
357    ///
358    /// - [`BusError::PayloadTooLarge`] – `payload.len() > config.max_payload_bytes`.
359    ///
360    /// # Returns
361    ///
362    /// The assigned event id (monotonically increasing).
363    pub fn publish(
364        &mut self,
365        topic: EventTopic,
366        payload: Vec<u8>,
367        source_peer: Option<String>,
368        now: u64,
369    ) -> Result<u64, BusError> {
370        // Validate payload size before doing anything.
371        if payload.len() > self.config.max_payload_bytes {
372            return Err(BusError::PayloadTooLarge {
373                size: payload.len(),
374                max: self.config.max_payload_bytes,
375            });
376        }
377
378        // Assign event id and build the event.
379        let event_id = self.next_event_id;
380        self.next_event_id = self.next_event_id.saturating_add(1);
381
382        let event = NebNetworkEvent {
383            id: event_id,
384            topic,
385            payload,
386            source_peer,
387            timestamp: now,
388        };
389
390        // Deliver to matching subscribers and update their stats.
391        let mut delivered_count: u64 = 0;
392        for sub in self.subscriptions.values_mut() {
393            if Self::matches_filter(&sub.filter, &event) {
394                sub.events_received = sub.events_received.saturating_add(1);
395                sub.last_event_at = now;
396                delivered_count = delivered_count.saturating_add(1);
397            }
398        }
399        self.total_delivered = self.total_delivered.saturating_add(delivered_count);
400
401        // Push to replay buffer, evicting the oldest entry if needed.
402        // When replay_buffer_size == 0, no events are retained.
403        if self.config.replay_buffer_size > 0 {
404            if self.replay_buffer.len() >= self.config.replay_buffer_size {
405                self.replay_buffer.pop_front();
406            }
407            self.replay_buffer.push_back(event);
408        }
409
410        self.total_published = self.total_published.saturating_add(1);
411        Ok(event_id)
412    }
413
414    // -----------------------------------------------------------------------
415    // Replay / draining
416    // -----------------------------------------------------------------------
417
418    /// Return all buffered events with `id > since_event_id` that match the
419    /// subscriber's filter.
420    ///
421    /// # Errors
422    ///
423    /// - [`BusError::SubscriberNotFound`] if `id` does not correspond to an
424    ///   active subscription.
425    pub fn replay_for_subscriber(
426        &self,
427        id: SubscriberId,
428        since_event_id: u64,
429    ) -> Result<Vec<&NebNetworkEvent>, BusError> {
430        let sub = self
431            .subscriptions
432            .get(&id.0)
433            .ok_or(BusError::SubscriberNotFound(id.0))?;
434
435        let events = self
436            .replay_buffer
437            .iter()
438            .filter(|e| e.id > since_event_id && Self::matches_filter(&sub.filter, e))
439            .collect();
440
441        Ok(events)
442    }
443
444    /// Clone all buffered events that match the subscriber's filter and return
445    /// them.  The replay buffer is **not** modified.
446    ///
447    /// # Errors
448    ///
449    /// - [`BusError::SubscriberNotFound`] if `id` does not correspond to an
450    ///   active subscription.
451    pub fn drain_events_for(&mut self, id: SubscriberId) -> Result<Vec<NebNetworkEvent>, BusError> {
452        // Borrow subscriptions immutably first to retrieve the filter, then
453        // iterate over replay_buffer.  We clone matching events.
454        let filter = {
455            let sub = self
456                .subscriptions
457                .get(&id.0)
458                .ok_or(BusError::SubscriberNotFound(id.0))?;
459            sub.filter.clone()
460        };
461
462        let events: Vec<NebNetworkEvent> = self
463            .replay_buffer
464            .iter()
465            .filter(|e| Self::matches_filter(&filter, e))
466            .cloned()
467            .collect();
468
469        Ok(events)
470    }
471
472    // -----------------------------------------------------------------------
473    // Statistics
474    // -----------------------------------------------------------------------
475
476    /// Return a snapshot of bus-wide statistics.
477    pub fn event_bus_stats(&self) -> EventBusStats {
478        let avg_delivery_per_event = if self.total_published == 0 {
479            0.0_f64
480        } else {
481            self.total_delivered as f64 / self.total_published as f64
482        };
483
484        EventBusStats {
485            subscriber_count: self.subscriptions.len(),
486            replay_buffer_size: self.replay_buffer.len(),
487            total_published: self.total_published,
488            total_delivered: self.total_delivered,
489            avg_delivery_per_event,
490        }
491    }
492}
493
494impl Default for NetworkEventBus {
495    fn default() -> Self {
496        Self::new(EventBusConfig::default())
497    }
498}
499
500// ===========================================================================
501// Tests
502// ===========================================================================
503
504#[cfg(test)]
505mod tests {
506    use super::{
507        BusError, EventBusConfig, EventFilter, EventTopic, NebSubscription, NetworkEventBus,
508        SubscriberId,
509    };
510
511    // -----------------------------------------------------------------------
512    // Helper
513    // -----------------------------------------------------------------------
514
515    fn make_bus() -> NetworkEventBus {
516        NetworkEventBus::new(EventBusConfig::default())
517    }
518
519    fn make_small_bus(max_subs: usize, buf: usize, max_bytes: usize) -> NetworkEventBus {
520        NetworkEventBus::new(EventBusConfig {
521            max_subscribers: max_subs,
522            replay_buffer_size: buf,
523            max_payload_bytes: max_bytes,
524        })
525    }
526
527    fn publish_peer_connected(bus: &mut NetworkEventBus, peer: &str, now: u64) -> u64 {
528        bus.publish(
529            EventTopic::PeerConnected,
530            peer.as_bytes().to_vec(),
531            Some(peer.to_string()),
532            now,
533        )
534        .expect("publish failed")
535    }
536
537    // -----------------------------------------------------------------------
538    // EventTopic
539    // -----------------------------------------------------------------------
540
541    #[test]
542    fn test_event_topic_as_str_named_variants() {
543        assert_eq!(EventTopic::PeerConnected.as_str(), "PeerConnected");
544        assert_eq!(EventTopic::PeerDisconnected.as_str(), "PeerDisconnected");
545        assert_eq!(EventTopic::BlockReceived.as_str(), "BlockReceived");
546        assert_eq!(EventTopic::BlockRequested.as_str(), "BlockRequested");
547        assert_eq!(EventTopic::DhtLookup.as_str(), "DhtLookup");
548        assert_eq!(EventTopic::GossipMessage.as_str(), "GossipMessage");
549    }
550
551    #[test]
552    fn test_event_topic_custom_as_str() {
553        let tag = "my-custom-event";
554        let topic = EventTopic::Custom(tag.to_string());
555        assert_eq!(topic.as_str(), tag);
556    }
557
558    #[test]
559    fn test_event_topic_display() {
560        let s = format!("{}", EventTopic::BlockReceived);
561        assert_eq!(s, "BlockReceived");
562    }
563
564    #[test]
565    fn test_event_topic_equality() {
566        assert_eq!(EventTopic::PeerConnected, EventTopic::PeerConnected);
567        assert_ne!(EventTopic::PeerConnected, EventTopic::PeerDisconnected);
568        assert_eq!(
569            EventTopic::Custom("x".to_string()),
570            EventTopic::Custom("x".to_string())
571        );
572        assert_ne!(
573            EventTopic::Custom("x".to_string()),
574            EventTopic::Custom("y".to_string())
575        );
576    }
577
578    // -----------------------------------------------------------------------
579    // SubscriberId
580    // -----------------------------------------------------------------------
581
582    #[test]
583    fn test_subscriber_id_display() {
584        let s = format!("{}", SubscriberId(42));
585        assert_eq!(s, "Sub(42)");
586    }
587
588    #[test]
589    fn test_subscriber_id_equality() {
590        assert_eq!(SubscriberId(1), SubscriberId(1));
591        assert_ne!(SubscriberId(1), SubscriberId(2));
592    }
593
594    // -----------------------------------------------------------------------
595    // subscribe / unsubscribe
596    // -----------------------------------------------------------------------
597
598    #[test]
599    fn test_subscribe_returns_non_zero_id() {
600        let mut bus = make_bus();
601        let id = bus.subscribe(EventFilter::All);
602        assert_ne!(
603            id,
604            SubscriberId(0),
605            "sentinel value should not be returned on success"
606        );
607    }
608
609    #[test]
610    fn test_subscribe_ids_are_monotonically_increasing() {
611        let mut bus = make_bus();
612        let id1 = bus.subscribe(EventFilter::All);
613        let id2 = bus.subscribe(EventFilter::All);
614        let id3 = bus.subscribe(EventFilter::All);
615        assert!(id1.0 < id2.0);
616        assert!(id2.0 < id3.0);
617    }
618
619    #[test]
620    fn test_subscribe_at_capacity_returns_sentinel() {
621        let mut bus = make_small_bus(2, 100, 1024);
622        let id1 = bus.subscribe(EventFilter::All);
623        let id2 = bus.subscribe(EventFilter::All);
624        // Third subscribe should fail.
625        let id3 = bus.subscribe(EventFilter::All);
626        assert_ne!(id1, SubscriberId(0));
627        assert_ne!(id2, SubscriberId(0));
628        assert_eq!(
629            id3,
630            SubscriberId(0),
631            "should return sentinel when at capacity"
632        );
633    }
634
635    #[test]
636    fn test_subscribe_after_unsubscribe_frees_slot() {
637        let mut bus = make_small_bus(2, 100, 1024);
638        let id1 = bus.subscribe(EventFilter::All);
639        let _id2 = bus.subscribe(EventFilter::All);
640        // At capacity — unsubscribe one.
641        assert!(bus.unsubscribe(id1));
642        // Now there is room for one more.
643        let id3 = bus.subscribe(EventFilter::All);
644        assert_ne!(id3, SubscriberId(0));
645    }
646
647    #[test]
648    fn test_unsubscribe_unknown_id_returns_false() {
649        let mut bus = make_bus();
650        assert!(!bus.unsubscribe(SubscriberId(999)));
651    }
652
653    #[test]
654    fn test_unsubscribe_twice_returns_false_second_time() {
655        let mut bus = make_bus();
656        let id = bus.subscribe(EventFilter::All);
657        assert!(bus.unsubscribe(id));
658        assert!(!bus.unsubscribe(id));
659    }
660
661    // -----------------------------------------------------------------------
662    // matches_filter
663    // -----------------------------------------------------------------------
664
665    fn make_event(
666        id: u64,
667        topic: EventTopic,
668        payload: Vec<u8>,
669        source_peer: Option<&str>,
670    ) -> super::NebNetworkEvent {
671        super::NebNetworkEvent {
672            id,
673            topic,
674            payload,
675            source_peer: source_peer.map(|s| s.to_string()),
676            timestamp: 0,
677        }
678    }
679
680    #[test]
681    fn test_filter_all_matches_any_event() {
682        let event = make_event(1, EventTopic::BlockReceived, vec![1, 2, 3], None);
683        assert!(NetworkEventBus::matches_filter(&EventFilter::All, &event));
684    }
685
686    #[test]
687    fn test_filter_topic_in_matches_included_topic() {
688        let event = make_event(1, EventTopic::PeerConnected, vec![], None);
689        let filter = EventFilter::TopicIn(vec![EventTopic::PeerConnected, EventTopic::DhtLookup]);
690        assert!(NetworkEventBus::matches_filter(&filter, &event));
691    }
692
693    #[test]
694    fn test_filter_topic_in_rejects_excluded_topic() {
695        let event = make_event(1, EventTopic::GossipMessage, vec![], None);
696        let filter = EventFilter::TopicIn(vec![EventTopic::PeerConnected]);
697        assert!(!NetworkEventBus::matches_filter(&filter, &event));
698    }
699
700    #[test]
701    fn test_filter_from_peer_matches_correct_peer() {
702        let event = make_event(1, EventTopic::BlockReceived, vec![], Some("alice"));
703        let filter = EventFilter::FromPeer("alice".to_string());
704        assert!(NetworkEventBus::matches_filter(&filter, &event));
705    }
706
707    #[test]
708    fn test_filter_from_peer_rejects_wrong_peer() {
709        let event = make_event(1, EventTopic::BlockReceived, vec![], Some("bob"));
710        let filter = EventFilter::FromPeer("alice".to_string());
711        assert!(!NetworkEventBus::matches_filter(&filter, &event));
712    }
713
714    #[test]
715    fn test_filter_from_peer_rejects_no_peer() {
716        let event = make_event(1, EventTopic::BlockReceived, vec![], None);
717        let filter = EventFilter::FromPeer("alice".to_string());
718        assert!(!NetworkEventBus::matches_filter(&filter, &event));
719    }
720
721    #[test]
722    fn test_filter_payload_size_above_matches_larger_payload() {
723        let event = make_event(1, EventTopic::BlockReceived, vec![0u8; 100], None);
724        let filter = EventFilter::PayloadSizeAbove(50);
725        assert!(NetworkEventBus::matches_filter(&filter, &event));
726    }
727
728    #[test]
729    fn test_filter_payload_size_above_rejects_equal_size() {
730        // "above" means strictly greater than, not >=.
731        let event = make_event(1, EventTopic::BlockReceived, vec![0u8; 50], None);
732        let filter = EventFilter::PayloadSizeAbove(50);
733        assert!(!NetworkEventBus::matches_filter(&filter, &event));
734    }
735
736    #[test]
737    fn test_filter_payload_size_above_rejects_smaller_payload() {
738        let event = make_event(1, EventTopic::BlockReceived, vec![0u8; 10], None);
739        let filter = EventFilter::PayloadSizeAbove(50);
740        assert!(!NetworkEventBus::matches_filter(&filter, &event));
741    }
742
743    // -----------------------------------------------------------------------
744    // publish
745    // -----------------------------------------------------------------------
746
747    #[test]
748    fn test_publish_returns_monotonically_increasing_ids() {
749        let mut bus = make_bus();
750        let id1 = bus
751            .publish(EventTopic::PeerConnected, vec![], None, 1)
752            .expect("publish failed");
753        let id2 = bus
754            .publish(EventTopic::PeerDisconnected, vec![], None, 2)
755            .expect("publish failed");
756        assert!(id1 < id2);
757    }
758
759    #[test]
760    fn test_publish_rejects_oversized_payload() {
761        let mut bus = make_small_bus(100, 100, 10);
762        let result = bus.publish(EventTopic::BlockReceived, vec![0u8; 11], None, 0);
763        assert_eq!(result, Err(BusError::PayloadTooLarge { size: 11, max: 10 }));
764    }
765
766    #[test]
767    fn test_publish_accepts_max_size_payload() {
768        let mut bus = make_small_bus(100, 100, 10);
769        let result = bus.publish(EventTopic::BlockReceived, vec![0u8; 10], None, 0);
770        assert!(result.is_ok());
771    }
772
773    #[test]
774    fn test_publish_increments_total_published() {
775        let mut bus = make_bus();
776        bus.publish(EventTopic::PeerConnected, vec![], None, 0)
777            .expect("publish failed");
778        bus.publish(EventTopic::PeerConnected, vec![], None, 1)
779            .expect("publish failed");
780        assert_eq!(bus.total_published, 2);
781    }
782
783    #[test]
784    fn test_publish_increments_delivered_for_matching_subscribers() {
785        let mut bus = make_bus();
786        let _sub = bus.subscribe(EventFilter::All);
787        bus.publish(EventTopic::PeerConnected, vec![], None, 0)
788            .expect("publish failed");
789        bus.publish(EventTopic::PeerConnected, vec![], None, 1)
790            .expect("publish failed");
791        assert_eq!(bus.total_delivered, 2);
792    }
793
794    #[test]
795    fn test_publish_does_not_count_non_matching_subscribers() {
796        let mut bus = make_bus();
797        // Subscriber only wants DhtLookup events.
798        let _sub = bus.subscribe(EventFilter::TopicIn(vec![EventTopic::DhtLookup]));
799        // Publish a PeerConnected event — should not be counted.
800        bus.publish(EventTopic::PeerConnected, vec![], None, 0)
801            .expect("publish failed");
802        assert_eq!(bus.total_delivered, 0);
803    }
804
805    #[test]
806    fn test_publish_updates_subscriber_events_received() {
807        let mut bus = make_bus();
808        let id = bus.subscribe(EventFilter::All);
809        publish_peer_connected(&mut bus, "alice", 100);
810        publish_peer_connected(&mut bus, "bob", 200);
811        let stats = bus.subscriber_stats(id).expect("stats missing");
812        assert_eq!(stats.events_received, 2);
813    }
814
815    #[test]
816    fn test_publish_updates_subscriber_last_event_at() {
817        let mut bus = make_bus();
818        let id = bus.subscribe(EventFilter::All);
819        publish_peer_connected(&mut bus, "alice", 100);
820        publish_peer_connected(&mut bus, "bob", 200);
821        let stats = bus.subscriber_stats(id).expect("stats missing");
822        assert_eq!(stats.last_event_at, 200);
823    }
824
825    #[test]
826    fn test_publish_fills_replay_buffer() {
827        let mut bus = make_small_bus(100, 5, 1024);
828        for i in 0..5u64 {
829            bus.publish(EventTopic::PeerConnected, vec![], None, i)
830                .expect("publish failed");
831        }
832        assert_eq!(bus.replay_buffer.len(), 5);
833    }
834
835    #[test]
836    fn test_publish_evicts_oldest_when_buffer_full() {
837        let mut bus = make_small_bus(100, 3, 1024);
838        let id1 = bus
839            .publish(EventTopic::PeerConnected, vec![], None, 1)
840            .expect("publish failed");
841        bus.publish(EventTopic::PeerConnected, vec![], None, 2)
842            .expect("publish failed");
843        bus.publish(EventTopic::PeerConnected, vec![], None, 3)
844            .expect("publish failed");
845        // Buffer is full.  This should evict event with id == id1.
846        bus.publish(EventTopic::PeerConnected, vec![], None, 4)
847            .expect("publish failed");
848
849        assert_eq!(bus.replay_buffer.len(), 3);
850        // The oldest event should no longer be in the buffer.
851        let oldest_in_buffer = bus.replay_buffer.front().expect("buffer empty").id;
852        assert_ne!(oldest_in_buffer, id1);
853    }
854
855    // -----------------------------------------------------------------------
856    // replay_for_subscriber
857    // -----------------------------------------------------------------------
858
859    #[test]
860    fn test_replay_for_subscriber_unknown_id_returns_error() {
861        let bus = make_bus();
862        let result = bus.replay_for_subscriber(SubscriberId(999), 0);
863        assert_eq!(result, Err(BusError::SubscriberNotFound(999)));
864    }
865
866    #[test]
867    fn test_replay_for_subscriber_returns_events_after_given_id() {
868        let mut bus = make_bus();
869        let sub = bus.subscribe(EventFilter::All);
870        // Publish three events; their ids start at 1.
871        let id1 = publish_peer_connected(&mut bus, "a", 1);
872        let _id2 = publish_peer_connected(&mut bus, "b", 2);
873        let _id3 = publish_peer_connected(&mut bus, "c", 3);
874        // Replay events with id > id1.
875        let replayed = bus.replay_for_subscriber(sub, id1).expect("replay failed");
876        assert_eq!(replayed.len(), 2);
877    }
878
879    #[test]
880    fn test_replay_for_subscriber_returns_empty_when_up_to_date() {
881        let mut bus = make_bus();
882        let sub = bus.subscribe(EventFilter::All);
883        let id1 = publish_peer_connected(&mut bus, "a", 1);
884        let replayed = bus.replay_for_subscriber(sub, id1).expect("replay failed");
885        assert!(replayed.is_empty());
886    }
887
888    #[test]
889    fn test_replay_for_subscriber_respects_filter() {
890        let mut bus = make_bus();
891        let sub = bus.subscribe(EventFilter::TopicIn(vec![EventTopic::BlockReceived]));
892        // Publish mixed topics.
893        bus.publish(EventTopic::PeerConnected, vec![], None, 1)
894            .expect("publish failed");
895        bus.publish(EventTopic::BlockReceived, vec![], None, 2)
896            .expect("publish failed");
897        bus.publish(EventTopic::PeerConnected, vec![], None, 3)
898            .expect("publish failed");
899        // Replay from id 0 — only BlockReceived should be included.
900        let replayed = bus.replay_for_subscriber(sub, 0).expect("replay failed");
901        assert_eq!(replayed.len(), 1);
902        assert_eq!(replayed[0].topic, EventTopic::BlockReceived);
903    }
904
905    // -----------------------------------------------------------------------
906    // drain_events_for
907    // -----------------------------------------------------------------------
908
909    #[test]
910    fn test_drain_events_for_unknown_id_returns_error() {
911        let mut bus = make_bus();
912        let result = bus.drain_events_for(SubscriberId(999));
913        assert_eq!(result, Err(BusError::SubscriberNotFound(999)));
914    }
915
916    #[test]
917    fn test_drain_events_for_returns_matching_events() {
918        let mut bus = make_bus();
919        let sub = bus.subscribe(EventFilter::All);
920        publish_peer_connected(&mut bus, "x", 1);
921        publish_peer_connected(&mut bus, "y", 2);
922        let drained = bus.drain_events_for(sub).expect("drain failed");
923        assert_eq!(drained.len(), 2);
924    }
925
926    #[test]
927    fn test_drain_events_for_does_not_remove_from_buffer() {
928        let mut bus = make_bus();
929        let sub = bus.subscribe(EventFilter::All);
930        publish_peer_connected(&mut bus, "x", 1);
931        let _first = bus.drain_events_for(sub).expect("drain failed");
932        // The buffer should still contain the event.
933        let second = bus.drain_events_for(sub).expect("drain failed");
934        assert_eq!(second.len(), 1);
935    }
936
937    #[test]
938    fn test_drain_events_for_respects_filter() {
939        let mut bus = make_bus();
940        let sub = bus.subscribe(EventFilter::FromPeer("alice".to_string()));
941        bus.publish(
942            EventTopic::PeerConnected,
943            vec![],
944            Some("alice".to_string()),
945            1,
946        )
947        .expect("publish failed");
948        bus.publish(
949            EventTopic::PeerConnected,
950            vec![],
951            Some("bob".to_string()),
952            2,
953        )
954        .expect("publish failed");
955        let drained = bus.drain_events_for(sub).expect("drain failed");
956        assert_eq!(drained.len(), 1);
957        assert_eq!(drained[0].source_peer.as_deref(), Some("alice"));
958    }
959
960    // -----------------------------------------------------------------------
961    // subscriber_stats
962    // -----------------------------------------------------------------------
963
964    #[test]
965    fn test_subscriber_stats_none_for_unknown_id() {
966        let bus = make_bus();
967        assert!(bus.subscriber_stats(SubscriberId(42)).is_none());
968    }
969
970    #[test]
971    fn test_subscriber_stats_some_for_known_id() {
972        let mut bus = make_bus();
973        let id = bus.subscribe(EventFilter::All);
974        assert!(bus.subscriber_stats(id).is_some());
975    }
976
977    #[test]
978    fn test_subscriber_stats_initial_values() {
979        let mut bus = make_bus();
980        let id = bus.subscribe(EventFilter::All);
981        let stats: &NebSubscription = bus.subscriber_stats(id).expect("stats missing");
982        assert_eq!(stats.events_received, 0);
983        assert_eq!(stats.last_event_at, 0);
984        assert_eq!(stats.id, id);
985    }
986
987    // -----------------------------------------------------------------------
988    // event_bus_stats
989    // -----------------------------------------------------------------------
990
991    #[test]
992    fn test_event_bus_stats_initial_state() {
993        let bus = make_bus();
994        let stats = bus.event_bus_stats();
995        assert_eq!(stats.subscriber_count, 0);
996        assert_eq!(stats.replay_buffer_size, 0);
997        assert_eq!(stats.total_published, 0);
998        assert_eq!(stats.total_delivered, 0);
999        assert_eq!(stats.avg_delivery_per_event, 0.0);
1000    }
1001
1002    #[test]
1003    fn test_event_bus_stats_counts_subscribers() {
1004        let mut bus = make_bus();
1005        bus.subscribe(EventFilter::All);
1006        bus.subscribe(EventFilter::All);
1007        let stats = bus.event_bus_stats();
1008        assert_eq!(stats.subscriber_count, 2);
1009    }
1010
1011    #[test]
1012    fn test_event_bus_stats_counts_buffer_size() {
1013        let mut bus = make_bus();
1014        publish_peer_connected(&mut bus, "a", 1);
1015        publish_peer_connected(&mut bus, "b", 2);
1016        let stats = bus.event_bus_stats();
1017        assert_eq!(stats.replay_buffer_size, 2);
1018    }
1019
1020    #[test]
1021    fn test_event_bus_stats_avg_delivery_per_event() {
1022        let mut bus = make_bus();
1023        // Two subscribers, one matching only PeerConnected.
1024        bus.subscribe(EventFilter::All);
1025        bus.subscribe(EventFilter::TopicIn(vec![EventTopic::PeerConnected]));
1026        // Publish one PeerConnected → both subscribers receive it (2 deliveries).
1027        bus.publish(EventTopic::PeerConnected, vec![], None, 1)
1028            .expect("publish failed");
1029        let stats = bus.event_bus_stats();
1030        assert_eq!(stats.total_published, 1);
1031        assert_eq!(stats.total_delivered, 2);
1032        assert!((stats.avg_delivery_per_event - 2.0_f64).abs() < f64::EPSILON);
1033    }
1034
1035    #[test]
1036    fn test_event_bus_stats_avg_delivery_zero_when_no_events() {
1037        let bus = make_bus();
1038        let stats = bus.event_bus_stats();
1039        assert_eq!(stats.avg_delivery_per_event, 0.0);
1040    }
1041
1042    // -----------------------------------------------------------------------
1043    // BusError display / equality
1044    // -----------------------------------------------------------------------
1045
1046    #[test]
1047    fn test_bus_error_subscriber_not_found_display() {
1048        let e = BusError::SubscriberNotFound(7);
1049        assert!(e.to_string().contains("7"));
1050    }
1051
1052    #[test]
1053    fn test_bus_error_payload_too_large_display() {
1054        let e = BusError::PayloadTooLarge {
1055            size: 2000,
1056            max: 1000,
1057        };
1058        let s = e.to_string();
1059        assert!(s.contains("2000"));
1060        assert!(s.contains("1000"));
1061    }
1062
1063    #[test]
1064    fn test_bus_error_max_subscribers_reached_display() {
1065        let e = BusError::MaxSubscribersReached;
1066        assert!(!e.to_string().is_empty());
1067    }
1068
1069    #[test]
1070    fn test_bus_error_equality() {
1071        assert_eq!(
1072            BusError::SubscriberNotFound(1),
1073            BusError::SubscriberNotFound(1)
1074        );
1075        assert_ne!(
1076            BusError::SubscriberNotFound(1),
1077            BusError::SubscriberNotFound(2)
1078        );
1079        assert_eq!(
1080            BusError::PayloadTooLarge { size: 5, max: 4 },
1081            BusError::PayloadTooLarge { size: 5, max: 4 }
1082        );
1083        assert_eq!(
1084            BusError::MaxSubscribersReached,
1085            BusError::MaxSubscribersReached
1086        );
1087    }
1088
1089    // -----------------------------------------------------------------------
1090    // Default impl
1091    // -----------------------------------------------------------------------
1092
1093    #[test]
1094    fn test_default_bus_uses_default_config() {
1095        let bus = NetworkEventBus::default();
1096        assert_eq!(bus.config.max_subscribers, 1_000);
1097        assert_eq!(bus.config.replay_buffer_size, 10_000);
1098        assert_eq!(bus.config.max_payload_bytes, 1_048_576);
1099    }
1100
1101    // -----------------------------------------------------------------------
1102    // Edge-cases / integration
1103    // -----------------------------------------------------------------------
1104
1105    #[test]
1106    fn test_multiple_subscribers_with_different_filters() {
1107        let mut bus = make_bus();
1108        let sub_all = bus.subscribe(EventFilter::All);
1109        let sub_dht = bus.subscribe(EventFilter::TopicIn(vec![EventTopic::DhtLookup]));
1110
1111        bus.publish(EventTopic::PeerConnected, vec![], None, 1)
1112            .expect("publish failed");
1113        bus.publish(EventTopic::DhtLookup, vec![], None, 2)
1114            .expect("publish failed");
1115
1116        let all_stats = bus.subscriber_stats(sub_all).expect("stats");
1117        let dht_stats = bus.subscriber_stats(sub_dht).expect("stats");
1118        assert_eq!(all_stats.events_received, 2);
1119        assert_eq!(dht_stats.events_received, 1);
1120    }
1121
1122    #[test]
1123    fn test_custom_topic_filtering() {
1124        let mut bus = make_bus();
1125        let sub = bus.subscribe(EventFilter::TopicIn(vec![EventTopic::Custom(
1126            "app::hello".to_string(),
1127        )]));
1128        bus.publish(
1129            EventTopic::Custom("app::hello".to_string()),
1130            vec![],
1131            None,
1132            1,
1133        )
1134        .expect("publish failed");
1135        bus.publish(
1136            EventTopic::Custom("app::world".to_string()),
1137            vec![],
1138            None,
1139            2,
1140        )
1141        .expect("publish failed");
1142        let stats = bus.subscriber_stats(sub).expect("stats");
1143        assert_eq!(stats.events_received, 1);
1144    }
1145
1146    #[test]
1147    fn test_publish_rejected_payload_not_counted() {
1148        let mut bus = make_small_bus(100, 100, 5);
1149        let _ = bus.publish(EventTopic::BlockReceived, vec![0u8; 10], None, 0);
1150        assert_eq!(bus.total_published, 0);
1151    }
1152
1153    #[test]
1154    fn test_subscriber_receives_zero_events_when_filter_never_matches() {
1155        let mut bus = make_bus();
1156        let sub = bus.subscribe(EventFilter::FromPeer("ghost".to_string()));
1157        for i in 0..10u64 {
1158            bus.publish(
1159                EventTopic::PeerConnected,
1160                vec![],
1161                Some("real-peer".to_string()),
1162                i,
1163            )
1164            .expect("publish failed");
1165        }
1166        let stats = bus.subscriber_stats(sub).expect("stats");
1167        assert_eq!(stats.events_received, 0);
1168    }
1169
1170    #[test]
1171    fn test_replay_buffer_size_with_zero_capacity() {
1172        // A bus with replay_buffer_size == 0 never retains events.
1173        let mut bus = make_small_bus(100, 0, 1024);
1174        let sub = bus.subscribe(EventFilter::All);
1175        bus.publish(EventTopic::PeerConnected, vec![], None, 1)
1176            .expect("publish failed");
1177        let replayed = bus.replay_for_subscriber(sub, 0).expect("replay failed");
1178        assert!(replayed.is_empty(), "buffer size 0 should retain no events");
1179    }
1180}
ipfrs_network/network_event_bus.rs

ipfrs_network/
network_event_bus.rs
