From 252b760c1d3a348575775af601af4cba04b4f2c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 15 Jul 2026 10:50:38 -0300 Subject: [PATCH] feat(rpc): add ?topics= filtering to the events stream Third PR of the chain-events series: server-side topic filtering for GET /lean/v0/events, split out so the previous PR's transport stayed a review of pure axum plumbing. The filter lives in the bus (EventBus::subscribe(TopicSet) returning an EventSubscription with a receive-side skip loop) rather than in the SSE handler, so future non-SSE consumers get filtering for free and the handler stays transport-only. TopicSet is a Copy bitmask: no allocation and one AND per event; lag is surfaced to the caller rather than swallowed by the skip loop. The signature also survives a later per-topic-channel split behind the facade, since callers only ever see subscribe(TopicSet). The handler parses ?topics=head,block via Topic::from_str (the exact inverse of as_str, so names cannot drift) and returns 400 naming the offending value on an unknown topic. A missing or empty parameter defaults to all topics: the Beacon API makes topics required, lean is friendlier here (documented as a divergence in docs/rpc.md). BroadcastStream no longer fits (it wraps a raw receiver, not the filtered subscription), so the bridge is futures-util's stream::unfold around EventSubscription::recv, preserving the Lagged skip + debug-log behavior; tokio-stream and futures-core are dropped for futures-util, already present in the dependency tree. EventSubscription also gains a non-blocking try_recv (same skip loop as recv, over the receiver's try_recv), so the previous PR's lib.rs unit tests (which assert on already-buffered events without spinning up a runtime) can move onto the filtered type after subscribe()'s signature changed here. On Lagged, the bridge now also yields an `: error - dropped N messages` SSE comment frame before resuming the stream, wire-compatible with Lighthouse's lagged-client marker so raw-stream consumers get an explicit re-sync trigger instead of a silent gap. --- Cargo.lock | 3 +- crates/blockchain/src/events.rs | 159 +++++++++++++++++++++- crates/blockchain/src/lib.rs | 16 +-- crates/net/rpc/Cargo.toml | 3 +- crates/net/rpc/src/events.rs | 229 +++++++++++++++++++++++++------- docs/rpc.md | 19 ++- 6 files changed, 365 insertions(+), 64 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index dcd004b4..dd3337cc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2135,7 +2135,7 @@ dependencies = [ "ethlambda-storage", "ethlambda-test-fixtures", "ethlambda-types", - "futures-core", + "futures-util", "hex", "http-body-util", "jemalloc_pprof", @@ -2143,7 +2143,6 @@ dependencies = [ "serde", "serde_json", "tokio", - "tokio-stream", "tokio-util", "tower", "tracing", diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 7a7c0dff..fe5f5ecb 100644 --- a/crates/blockchain/src/events.rs +++ b/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 { + 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 for TopicSet { + fn from_iter>(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 { - 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, + 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 { + 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 { + 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::().unwrap(), topic); + } + let err = "bogus".parse::().unwrap_err(); + assert_eq!(err.to_string(), "unknown topic: 'bogus'"); + } + #[test] fn emit_without_subscribers_is_a_noop() { let bus = EventBus::default(); diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 24db4bc6..7ac1f521 100644 --- a/crates/blockchain/src/lib.rs +++ b/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]); diff --git a/crates/net/rpc/Cargo.toml b/crates/net/rpc/Cargo.toml index cee21b9d..e6e5f0ea 100644 --- a/crates/net/rpc/Cargo.toml +++ b/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 diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs index 99995439..34cb5f29 100644 --- a/crates/net/rpc/src/events.rs +++ b/crates/net/rpc/src/events.rs @@ -9,44 +9,92 @@ //! Framing: the topic name goes on the SSE `event:` line //! ([`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. -use std::convert::Infallible; +use std::{convert::Infallible, str::FromStr}; use axum::{ Extension, Router, - response::{Sse, sse::Event}, + extract::Query, + http::StatusCode, + response::{IntoResponse, Response, Sse, sse::Event}, routing::get, }; -use ethlambda_blockchain::EventBus; +use ethlambda_blockchain::{EventBus, EventSubscription, Topic, TopicSet}; use ethlambda_storage::Store; -use futures_core::Stream; -use tokio_stream::{ - StreamExt, - wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}, -}; +use futures_util::{Stream, stream::unfold}; +use serde::Deserialize; +use tokio::sync::broadcast::error::RecvError; + +#[derive(Deserialize)] +struct EventsParams { + topics: Option, +} async fn get_events( Extension(events): Extension, -) -> Sse>> { - let stream = BroadcastStream::new(events.subscribe()).filter_map(|res| { - // A slow client falls behind and the bounded broadcast channel - // overwrites events it never read; skip past the gap rather than - // ending the stream. The stream is best-effort by contract: clients - // re-sync via the blocks endpoints after a gap. - let ev = match res { - Ok(ev) => ev, - Err(BroadcastStreamRecvError::Lagged(skipped)) => { - tracing::debug!(skipped, "SSE client lagged; dropped chain events"); - return None; + Query(params): Query, +) -> Response { + // A missing or empty `topics` selects every topic. This diverges from the + // Beacon API, which makes the parameter mandatory; see docs/rpc.md. + let topics = match params.topics.as_deref() { + None | Some("") => TopicSet::ALL, + Some(list) => { + let parsed: Result = list.split(',').map(Topic::from_str).collect(); + match parsed { + Ok(topics) => topics, + Err(err) => return (StatusCode::BAD_REQUEST, err.to_string()).into_response(), + } + } + }; + + Sse::new(event_stream(events.subscribe(topics))) + .keep_alive(axum::response::sse::KeepAlive::default()) + .into_response() +} + +/// Bridge the subscription's `recv` loop into a stream of SSE frames. +fn event_stream(subscription: EventSubscription) -> impl Stream> { + unfold(subscription, |mut subscription| async move { + loop { + match subscription.recv().await { + Ok(ev) => { + let frame = Event::default() + .event(ev.topic().as_str()) + .json_data(&ev) + .inspect_err( + |err| tracing::warn!(%err, "Failed to serialize SSE chain event"), + ); + // A frame that fails to serialize is skipped, not fatal. + if let Ok(frame) = frame { + return Some((Ok(frame), subscription)); + } + } + // A slow client falls behind and the bounded broadcast channel + // overwrites events it never read; skip past the gap rather + // than ending the stream. The stream is best-effort by + // contract: clients re-sync via the blocks endpoints. + // + // The comment frame below is wire-compatible with + // Lighthouse's lagged-client marker + // (`beacon_node/http_api/src/lib.rs:3298-3302`): same text, + // same `Event::comment` mechanism. Comment lines are invisible + // to browser `EventSource` consumers (only raw-stream readers + // see them), so this is a best-effort signal, not a guarantee. + Err(RecvError::Lagged(skipped)) => { + tracing::debug!(skipped, "SSE client lagged; dropped chain events"); + let frame = + Event::default().comment(format!("error - dropped {skipped} messages")); + return Some((Ok(frame), subscription)); + } + // Publisher dropped: the node is shutting down. + Err(RecvError::Closed) => return None, } - }; - Some(Ok(Event::default() - .event(ev.topic().as_str()) - .json_data(&ev) - .inspect_err(|err| tracing::warn!(%err, "Failed to serialize SSE chain event")) - .ok()?)) - }); - Sse::new(stream).keep_alive(axum::response::sse::KeepAlive::default()) + } + }) } pub(crate) fn routes() -> Router { @@ -55,32 +103,42 @@ pub(crate) fn routes() -> Router { #[cfg(test)] mod tests { - use axum::{Extension, body::Body, http::Request}; + use axum::{ + Extension, + body::Body, + http::{Request, StatusCode}, + }; 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 crate::test_utils::create_test_state; + 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())); + app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap()) + .await + .unwrap() + } + + async fn first_frame(resp: axum::response::Response) -> String { + let mut body = resp.into_body().into_data_stream(); + let chunk = body.next().await.unwrap().unwrap(); + String::from_utf8_lossy(&chunk).into_owned() + } + #[tokio::test] async fn events_streams_head_with_flat_payload() { let events = EventBus::new(16); - 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())); // Issue the request first so the handler subscribes its receiver // before we publish — `emit` drops events with no live receivers. - let resp = app - .oneshot( - Request::builder() - .uri("/lean/v0/events") - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - assert_eq!(resp.status(), axum::http::StatusCode::OK); + let resp = events_response(&events, "/lean/v0/events").await; + assert_eq!(resp.status(), StatusCode::OK); events.emit(ChainEvent::Head { slot: 3, @@ -88,12 +146,7 @@ mod tests { parent_root: Default::default(), }); - let mut body = resp.into_body().into_data_stream(); - let chunk = tokio_stream::StreamExt::next(&mut body) - .await - .unwrap() - .unwrap(); - let text = String::from_utf8_lossy(&chunk); + let text = first_frame(resp).await; // Topic on the `event:` line... assert!( @@ -111,4 +164,88 @@ mod tests { "payload is not flat: {text}" ); } + + #[tokio::test] + async fn events_topics_filter_skips_unmatched() { + let events = EventBus::new(16); + + let resp = events_response(&events, "/lean/v0/events?topics=head").await; + assert_eq!(resp.status(), StatusCode::OK); + + // The block event lands first but is filtered out server-side, so the + // first frame on the wire must be the head event. + events.emit(ChainEvent::Block { + slot: 1, + root: Default::default(), + }); + events.emit(ChainEvent::Head { + slot: 2, + root: Default::default(), + parent_root: Default::default(), + }); + + let text = first_frame(resp).await; + assert!( + text.contains("event:head") || text.contains("event: head"), + "missing head event name in frame: {text}" + ); + assert!( + !text.contains("block"), + "filtered block event leaked into the stream: {text}" + ); + } + + #[tokio::test] + async fn events_lagged_subscriber_gets_dropped_comment() { + // Capacity 2: emitting 3 events before anything is read forces the + // broadcast channel to overwrite the oldest one, so the subscriber's + // next recv() reports RecvError::Lagged(1). + let events = EventBus::new(2); + + let resp = events_response(&events, "/lean/v0/events").await; + assert_eq!(resp.status(), StatusCode::OK); + + // Emit before polling the body: the handler already subscribed while + // handling the request above, so these land in the channel unread. + for slot in 1..=3 { + events.emit(ChainEvent::Head { + slot, + root: Default::default(), + parent_root: Default::default(), + }); + } + + // The lag comment should appear among the first frames, ahead of (or + // interleaved with) the surviving head events. + let mut body = resp.into_body().into_data_stream(); + let mut frames = String::new(); + for _ in 0..4 { + let Some(chunk) = body.next().await else { + break; + }; + frames.push_str(&String::from_utf8_lossy(&chunk.unwrap())); + if frames.contains("error - dropped") { + break; + } + } + assert!( + frames.contains(": error - dropped 1 messages"), + "missing lagged-client comment frame: {frames}" + ); + } + + #[tokio::test] + async fn events_unknown_topic_returns_400() { + let events = EventBus::new(16); + + let resp = events_response(&events, "/lean/v0/events?topics=head,bogus").await; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + + let body = resp.into_body().collect().await.unwrap().to_bytes(); + let text = String::from_utf8_lossy(&body); + assert!( + text.contains("unknown topic: 'bogus'"), + "unhelpful 400 body: {text}" + ); + } } diff --git a/docs/rpc.md b/docs/rpc.md index 209a28f9..6ac316e6 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -102,7 +102,24 @@ event: head data: {"slot":128,"root":"0x1a2b…","parent_root":"0x3c4d…"} ``` -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. Keep-alive comments are sent periodically to hold idle connections open. +#### Filtering with `?topics=` + +An optional comma-separated list of event names narrows the stream server-side: + +```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`. + +| 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. + +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. ### `GET /lean/v0/blocks/{block_id}` and `/header`