feat(rpc): require the events ?topics= query parameter

Match the Beacon API eventstream, which makes `topics` mandatory: a missing
or empty parameter now returns 400 instead of defaulting to all topics. This
removes the divergence the endpoint previously documented.

Also drop the now-unused TopicSet::ALL sentinel.

Author: MegaRedHand

crates/blockchain/src/events.rs

@@ -140,10 +140,6 @@ impl EventBus {
     }
 
     /// Subscribe a new receiver observing every event emitted from now on.
-    ///
-    /// The bus is a plain fan-out: it does no filtering. A consumer that wants
-    /// only some topics filters on its own side (see the SSE `?topics=`
-    /// handler in `ethlambda-rpc`).
     pub fn subscribe(&self) -> broadcast::Receiver<ChainEvent> {
         self.tx.subscribe()
     }

crates/net/rpc/src/events.rs

@@ -10,9 +10,9 @@
 //! ([`ethlambda_blockchain::ChainEvent::topic`]) and the `data:` line carries
 //! the event's flat JSON payload; the topic is never repeated inside the body.
 //!
-//! Filtering: `?topics=head,block` (comma-separated [`Topic`] names) narrows
-//! the stream server-side. An unknown name is a 400; a missing or empty
-//! parameter selects every topic.
+//! Filtering: `?topics=head,block` (comma-separated [`Topic`] names) selects
+//! which events to stream. `topics` is required (matching the Beacon API): a
+//! missing, empty, or unknown value is a 400.
 
 use std::{convert::Infallible, str::FromStr};
 
@@ -44,9 +44,6 @@ struct EventsParams {
 struct TopicSet(u16);
 
 impl TopicSet {
-    /// Every topic, including ones added later.
-    const ALL: TopicSet = TopicSet(u16::MAX);
-
     fn contains(self, topic: Topic) -> bool {
         self.0 & (1 << topic as u16) != 0
     }
@@ -64,10 +61,16 @@ async fn get_events(
     Extension(events): Extension<EventBus>,
     Query(params): Query<EventsParams>,
 ) -> Response {
-    // A missing or empty `topics` selects every topic. This diverges from the
-    // Beacon API, which makes the parameter mandatory; see docs/rpc.md.
+    // `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() {
-        None | Some("") => TopicSet::ALL,
+        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 {
@@ -160,16 +163,17 @@ mod tests {
         assert!(!set.contains(Topic::Block));
         assert!(!set.contains(Topic::JustifiedCheckpoint));
 
-        // The empty set matches nothing; ALL matches every topic.
+        // 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!(TopicSet::ALL.contains(topic));
+            assert!(full.contains(topic));
         }
     }
 
@@ -193,7 +197,7 @@ mod tests {
 
         // Issue the request first so the handler subscribes its receiver
         // before we publish — `emit` drops events with no live receivers.
-        let resp = events_response(&events, "/lean/v0/events").await;
+        let resp = events_response(&events, "/lean/v0/events?topics=head").await;
         assert_eq!(resp.status(), StatusCode::OK);
 
         events.emit(ChainEvent::Head {
@@ -267,7 +271,7 @@ mod tests {
         // next recv() reports RecvError::Lagged(1).
         let events = EventBus::new(2);
 
-        let resp = events_response(&events, "/lean/v0/events").await;
+        let resp = events_response(&events, "/lean/v0/events?topics=head").await;
         assert_eq!(resp.status(), StatusCode::OK);
 
         // Emit before polling the body: the handler already subscribed while
@@ -313,4 +317,16 @@ mod tests {
             "unhelpful 400 body: {text}"
         );
     }
+
+    #[tokio::test]
+    async fn events_missing_or_empty_topics_returns_400() {
+        let events = EventBus::new(16);
+
+        // `topics` is required (Beacon-API-aligned): a fully absent parameter
+        // and a present-but-empty one are both rejected, never defaulted.
+        for uri in ["/lean/v0/events", "/lean/v0/events?topics="] {
+            let resp = events_response(&events, uri).await;
+            assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "uri: {uri}");
+        }
+    }
 }

docs/rpc.md

@@ -106,20 +106,18 @@ data: {"slot":128,"block":"0x1a2b…","state":"0x3c4d…"}
 
 #### Filtering with `?topics=`
 
-An optional comma-separated list of event names narrows the stream server-side:
+A **required** comma-separated list of event names selects which events to stream:
 
 ```bash
 curl -N 'http://127.0.0.1:5052/lean/v0/events?topics=head,finalized_checkpoint'
 ```
 
-Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`.
+Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`. As in the Beacon API `eventstream` endpoint, `topics` is mandatory: there is no "subscribe to everything" default; list the topics you want.
 
 | Status | Condition |
 |--------|-----------|
-| `200` | Stream opened (filtered, or unfiltered when the parameter is missing/empty) |
-| `400` | Any listed name is not a known topic (body names the offending value) |
-
-> **Divergence from the Beacon API:** the Beacon `eventstream` endpoint makes `topics` required; lean defaults a missing or empty parameter to all topics.
+| `200` | Stream opened for the listed topics |
+| `400` | `topics` is missing or empty, or any listed name is not a known topic (body names the offending value) |
 
 Events are fanned out over a bounded broadcast channel. A client that reads too slowly skips past the events it missed: they are dropped for that subscriber rather than back-pressured onto the actor, so treat the stream as best-effort and re-sync via the blocks endpoints after a gap. A client that falls behind receives an SSE comment line `: error - dropped N messages` marking the gap (wire-compatible with Lighthouse) before the stream continues; re-sync via the blocks endpoints rather than trusting the skipped range. Keep-alive comments are sent periodically to hold idle connections open.