refactor(rpc): use Vec<Topic> for the events filter instead of a bitmask

The TopicSet newtype (a Copy u16 bitmask plus a FromIterator impl) was more
machinery than a handful of topics warrant. Parse `?topics=` straight into a
Vec<Topic> and membership-test with Vec::contains in the stream loop. Drops
the type and its unit test; filtering stays covered end-to-end by
events_topics_filter_skips_unmatched.

Author: MegaRedHand

crates/net/rpc/src/events.rs

@@ -34,50 +34,27 @@ struct EventsParams {
     topics: Option<String>,
 }
 
-/// Topics an SSE client asked for: a `Copy` bitmask over [`Topic`]
-/// discriminants, tested once per received event.
-///
-/// The filter lives here, in the sole consumer that needs it, rather than in
-/// the [`EventBus`]: the bus stays a plain fan-out, and any other subscriber
-/// pays nothing for a feature only this endpoint uses.
-#[derive(Clone, Copy)]
-struct TopicSet(u16);
-
-impl TopicSet {
-    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(0), |set, topic| {
-            TopicSet(set.0 | (1 << topic as u16))
-        })
-    }
-}
-
 async fn get_events(
     Extension(events): Extension<EventBus>,
     Query(params): Query<EventsParams>,
 ) -> Response {
     // `topics` is required, matching the Beacon API: a missing or empty value
     // is a 400, as is any unknown topic name. See docs/rpc.md.
-    let topics = match params.topics.as_deref() {
+    //
+    // The parsed selection lives here, in the sole consumer that filters, not
+    // in the bus: a plain `Vec` since there are only a handful of topics.
+    let topics: Vec<Topic> = match params.topics.as_deref() {
         None | Some("") => {
             return (
                 StatusCode::BAD_REQUEST,
                 "missing required query parameter: topics",
             )
                 .into_response();
         }
-        Some(list) => {
-            let parsed: Result<TopicSet, _> = list.split(',').map(Topic::from_str).collect();
-            match parsed {
-                Ok(topics) => topics,
-                Err(err) => return (StatusCode::BAD_REQUEST, err.to_string()).into_response(),
-            }
-        }
+        Some(list) => match list.split(',').map(Topic::from_str).collect() {
+            Ok(topics) => topics,
+            Err(err) => return (StatusCode::BAD_REQUEST, err.to_string()).into_response(),
+        },
     };
 
     Sse::new(event_stream(events.subscribe(), topics))
@@ -89,13 +66,13 @@ async fn get_events(
 /// events the client's `topics` filter excludes.
 fn event_stream(
     rx: broadcast::Receiver<ChainEvent>,
-    topics: TopicSet,
+    topics: Vec<Topic>,
 ) -> impl Stream<Item = Result<Event, Infallible>> {
     unfold((rx, topics), |(mut rx, topics)| async move {
         loop {
             match rx.recv().await {
                 // Not in the client's `?topics=` set: skip without a frame.
-                Ok(ev) if !topics.contains(ev.topic()) => continue,
+                Ok(ev) if !topics.contains(&ev.topic()) => continue,
                 Ok(ev) => {
                     let frame = Event::default()
                         .event(ev.topic().as_str())
@@ -143,40 +120,15 @@ mod tests {
         body::Body,
         http::{Request, StatusCode},
     };
-    use ethlambda_blockchain::{ChainEvent, EventBus, Topic};
+    use ethlambda_blockchain::{ChainEvent, EventBus};
     use ethlambda_storage::{Store, backend::InMemoryBackend};
     use futures_util::StreamExt;
     use http_body_util::BodyExt;
     use std::sync::Arc;
     use tower::ServiceExt;
 
-    use super::TopicSet;
     use crate::test_utils::create_test_state;
 
-    #[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));
-
-        // The empty set matches nothing; a fully-populated set matches all.
-        let all = [
-            Topic::Head,
-            Topic::Block,
-            Topic::JustifiedCheckpoint,
-            Topic::FinalizedCheckpoint,
-        ];
-        let full: TopicSet = all.into_iter().collect();
-        for topic in all {
-            assert!(!TopicSet(0).contains(topic));
-            assert!(full.contains(topic));
-        }
-    }
-
     async fn events_response(events: &EventBus, uri: &str) -> axum::response::Response {
         let store = Store::from_anchor_state(Arc::new(InMemoryBackend::new()), create_test_state());
         let app = crate::test_utils::test_api_router(store).layer(Extension(events.clone()));