feat(rpc): add ?topics= filtering to the events stream

crates/blockchain/src/events.rs

@@ -9,6 +9,8 @@
 //! a slow subscriber loses events (the bounded broadcast channel overwrites
 //! its backlog) rather than back-pressuring consensus.
 
+use std::{fmt, str::FromStr};
+
 use ethlambda_types::primitives::H256;
 use serde::Serialize;
 use tokio::sync::broadcast;
@@ -37,6 +39,58 @@ impl Topic {
     }
 }
 
+/// Error returned by [`Topic::from_str`] for a name matching no topic.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct UnknownTopic(String);
+
+impl fmt::Display for UnknownTopic {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "unknown topic: '{}'", self.0)
+    }
+}
+
+impl std::error::Error for UnknownTopic {}
+
+impl FromStr for Topic {
+    type Err = UnknownTopic;
+
+    /// Exact inverse of [`Topic::as_str`].
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "head" => Ok(Topic::Head),
+            "block" => Ok(Topic::Block),
+            "justified_checkpoint" => Ok(Topic::JustifiedCheckpoint),
+            "finalized_checkpoint" => Ok(Topic::FinalizedCheckpoint),
+            other => Err(UnknownTopic(other.to_string())),
+        }
+    }
+}
+
+/// Set of topics a subscriber wants.
+///
+/// A bitmask over [`Topic`] discriminants: `Copy`, no allocation, one AND per
+/// received event. `Default` is the empty set; matching everything must be
+/// explicit via [`TopicSet::ALL`].
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct TopicSet(u16);
+
+impl TopicSet {
+    /// Every topic, including ones added later.
+    pub const ALL: TopicSet = TopicSet(u16::MAX);
+
+    pub fn contains(self, topic: Topic) -> bool {
+        self.0 & (1 << topic as u16) != 0
+    }
+}
+
+impl FromIterator<Topic> for TopicSet {
+    fn from_iter<I: IntoIterator<Item = Topic>>(iter: I) -> Self {
+        iter.into_iter().fold(TopicSet::default(), |set, topic| {
+            TopicSet(set.0 | (1 << topic as u16))
+        })
+    }
+}
+
 /// A consensus event published by the blockchain actor.
 ///
 /// `untagged`: serializing yields only the variant's fields. The topic name
@@ -111,9 +165,52 @@ impl EventBus {
         let _ = self.tx.send(event);
     }
 
-    /// Subscribe a new receiver observing every event emitted from now on.
-    pub fn subscribe(&self) -> broadcast::Receiver<ChainEvent> {
-        self.tx.subscribe()
+    /// Subscribe a filtered view observing matching events emitted from now on.
+    ///
+    /// The filter lives in the bus rather than in each consumer so every
+    /// subscriber (SSE or otherwise) gets the same skip semantics.
+    pub fn subscribe(&self, topics: TopicSet) -> EventSubscription {
+        EventSubscription {
+            rx: self.tx.subscribe(),
+            topics,
+        }
+    }
+}
+
+/// A topic-filtered view over the chain-event broadcast channel.
+///
+/// All subscribers share one ring buffer: filtering happens at receive time,
+/// so skipped events still occupy channel slots and still count toward a
+/// subscriber's lag window.
+pub struct EventSubscription {
+    rx: broadcast::Receiver<ChainEvent>,
+    topics: TopicSet,
+}
+
+impl EventSubscription {
+    /// Next event matching the filter. Non-matching events are skipped; lag
+    /// is surfaced to the caller, not swallowed.
+    pub async fn recv(&mut self) -> Result<ChainEvent, broadcast::error::RecvError> {
+        loop {
+            match self.rx.recv().await {
+                Ok(event) if self.topics.contains(event.topic()) => return Ok(event),
+                Ok(_) => continue,
+                Err(err) => return Err(err), // Lagged(n) | Closed
+            }
+        }
+    }
+
+    /// Non-blocking variant of [`EventSubscription::recv`], for callers (e.g.
+    /// tests) that want to assert on already-buffered events without an
+    /// executor.
+    pub fn try_recv(&mut self) -> Result<ChainEvent, broadcast::error::TryRecvError> {
+        loop {
+            match self.rx.try_recv() {
+                Ok(event) if self.topics.contains(event.topic()) => return Ok(event),
+                Ok(_) => continue,
+                Err(err) => return Err(err), // Empty | Lagged(n) | Closed
+            }
+        }
     }
 }
 
@@ -137,19 +234,71 @@ mod tests {
         }
     }
 
+    const ALL_TOPICS: [Topic; 4] = [
+        Topic::Head,
+        Topic::Block,
+        Topic::JustifiedCheckpoint,
+        Topic::FinalizedCheckpoint,
+    ];
+
     #[tokio::test]
     async fn subscriber_receives_emitted_event() {
         let bus = EventBus::default();
-        let mut rx = bus.subscribe();
+        let mut sub = bus.subscribe(TopicSet::ALL);
 
         bus.emit(head_event(7));
 
-        match rx.recv().await.unwrap() {
+        match sub.recv().await.unwrap() {
             ChainEvent::Head { slot, .. } => assert_eq!(slot, 7),
             other => panic!("unexpected event: {other:?}"),
         }
     }
 
+    #[tokio::test]
+    async fn filtered_subscription_skips_unmatched_events() {
+        let bus = EventBus::default();
+        let mut sub = bus.subscribe([Topic::Head].into_iter().collect());
+
+        // The block event lands in the channel first but must never surface
+        // through the head-only subscription.
+        bus.emit(ChainEvent::Block {
+            slot: 1,
+            root: H256::ZERO,
+        });
+        bus.emit(head_event(2));
+
+        match sub.recv().await.unwrap() {
+            ChainEvent::Head { slot, .. } => assert_eq!(slot, 2),
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn topic_set_collects_and_contains() {
+        let set: TopicSet = [Topic::Head, Topic::FinalizedCheckpoint]
+            .into_iter()
+            .collect();
+        assert!(set.contains(Topic::Head));
+        assert!(set.contains(Topic::FinalizedCheckpoint));
+        assert!(!set.contains(Topic::Block));
+        assert!(!set.contains(Topic::JustifiedCheckpoint));
+
+        let empty = TopicSet::default();
+        for topic in ALL_TOPICS {
+            assert!(!empty.contains(topic));
+            assert!(TopicSet::ALL.contains(topic));
+        }
+    }
+
+    #[test]
+    fn topic_from_str_inverts_as_str() {
+        for topic in ALL_TOPICS {
+            assert_eq!(topic.as_str().parse::<Topic>().unwrap(), topic);
+        }
+        let err = "bogus".parse::<Topic>().unwrap_err();
+        assert_eq!(err.to_string(), "unknown topic: 'bogus'");
+    }
+
     #[test]
     fn emit_without_subscribers_is_a_noop() {
         let bus = EventBus::default();

crates/blockchain/src/lib.rs

@@ -31,7 +31,7 @@ use tracing::{debug, error, info, trace, warn};
 use crate::block_builder::ProposerConfig;
 use crate::store::StoreError;
 
-pub use events::{ChainEvent, EventBus, Topic};
+pub use events::{ChainEvent, EventBus, EventSubscription, Topic, TopicSet};
 
 pub mod aggregation;
 pub mod block_builder;
@@ -1538,7 +1538,7 @@ mod tests {
     fn chain_event_diff_emits_nothing_when_unchanged() {
         let store = test_store();
         let bus = EventBus::new(8);
-        let mut rx = bus.subscribe();
+        let mut rx = bus.subscribe(TopicSet::ALL);
 
         let snapshot = ChainEventSnapshot::capture(&store);
         snapshot.diff_and_emit(&store, &bus, 0);
@@ -1553,7 +1553,7 @@ mod tests {
         let mut store = test_store();
         let genesis = store.head().expect("store head exists");
         let bus = EventBus::new(8);
-        let mut rx = bus.subscribe();
+        let mut rx = bus.subscribe(TopicSet::ALL);
 
         let snapshot = ChainEventSnapshot::capture(&store);
 
@@ -1606,7 +1606,7 @@ mod tests {
     fn chain_event_diff_skips_head_with_missing_header() {
         let mut store = test_store();
         let bus = EventBus::new(8);
-        let mut rx = bus.subscribe();
+        let mut rx = bus.subscribe(TopicSet::ALL);
 
         let snapshot = ChainEventSnapshot::capture(&store);
 
@@ -1643,7 +1643,7 @@ mod tests {
         let mut store = test_store();
         let genesis = store.head().expect("store head exists");
         let bus = EventBus::new(8);
-        let mut rx = bus.subscribe();
+        let mut rx = bus.subscribe(TopicSet::ALL);
 
         let snapshot = ChainEventSnapshot::capture(&store);
 
@@ -1689,7 +1689,7 @@ mod tests {
         let mut store = test_store();
         let genesis = store.head().expect("store head exists");
         let bus = EventBus::new(8);
-        let mut rx = bus.subscribe();
+        let mut rx = bus.subscribe(TopicSet::ALL);
 
         let snapshot = ChainEventSnapshot::capture(&store);
 
@@ -1723,7 +1723,7 @@ mod tests {
             let mut store = test_store();
             let genesis = store.head().expect("store head exists");
             let bus = EventBus::new(8);
-            let mut rx = bus.subscribe();
+            let mut rx = bus.subscribe(TopicSet::ALL);
             let snapshot = ChainEventSnapshot::capture(&store);
 
             let new_root = H256([9u8; 32]);
@@ -1747,7 +1747,7 @@ mod tests {
             let mut store = test_store();
             let genesis = store.head().expect("store head exists");
             let bus = EventBus::new(8);
-            let mut rx = bus.subscribe();
+            let mut rx = bus.subscribe(TopicSet::ALL);
             let snapshot = ChainEventSnapshot::capture(&store);
 
             let new_root = H256([9u8; 32]);

crates/net/rpc/Cargo.toml

@@ -26,8 +26,7 @@ serde_json.workspace = true
 hex.workspace = true
 tracing.workspace = true
 jemalloc_pprof.workspace = true
-tokio-stream = { version = "0.1", features = ["sync"] }
-futures-core = "0.3"
+futures-util = "0.3"
 
 [dev-dependencies]
 ethlambda-types.workspace = true