From 585b55342361e40016357ccf62d91bf57996912e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Tue, 21 Jul 2026 17:18:12 -0300 Subject: [PATCH] feat(rpc): add required ?topics= filtering to the events stream Adds server-side `?topics=` filtering to `GET /lean/v0/events`. The comma-separated list of event names is required (Beacon-API-aligned): a missing, empty, or unknown topic is a 400 that names the offending value. Filtering lives in the SSE consumer, not the bus: the handler parses the selection into a `Vec` and skips events outside it. The stream is driven by `futures_util::unfold` over the broadcast receiver directly (dropping the tokio-stream wrapper), and a lagged subscriber now receives a Lighthouse-compatible `: error - dropped N messages` comment frame before the stream continues. `Topic` gains a `FromStr` inverse of `as_str` with a `thiserror` `UnknownTopic` error. Documented in docs/rpc.md. --- Cargo.lock | 3 +- crates/blockchain/src/events.rs | 37 +++++ crates/blockchain/src/lib.rs | 2 +- crates/net/rpc/Cargo.toml | 3 +- crates/net/rpc/src/events.rs | 256 ++++++++++++++++++++++++++------ docs/rpc.md | 17 ++- 6 files changed, 266 insertions(+), 52 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7e31a87a..ddd13c0c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2136,7 +2136,7 @@ dependencies = [ "ethlambda-storage", "ethlambda-test-fixtures", "ethlambda-types", - "futures-core", + "futures-util", "hex", "http-body-util", "jemalloc_pprof", @@ -2144,7 +2144,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 3ff1e1b1..66386ffd 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -14,6 +14,7 @@ use ethlambda_types::ShortRoot; use ethlambda_types::checkpoint::Checkpoint; use ethlambda_types::primitives::H256; use serde::Serialize; +use std::str::FromStr; use tokio::sync::broadcast; use tracing::warn; @@ -44,6 +45,26 @@ impl Topic { } } +/// Error returned by [`Topic::from_str`] for a name matching no topic. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("unknown topic: '{0}'")] +pub struct UnknownTopic(String); + +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())), + } + } +} + /// A consensus event published by the blockchain actor. /// /// Fields mirror the Ethereum beacon-API eventstream payloads so the SSE @@ -276,6 +297,13 @@ 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(); @@ -289,6 +317,15 @@ mod tests { } } + #[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 3427d3ff..73d3a9cd 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -31,7 +31,7 @@ use crate::block_builder::ProposerConfig; use crate::events::ChainEventSnapshot; use crate::store::StoreError; -pub use events::{ChainEvent, EventBus, Topic}; +pub use events::{ChainEvent, EventBus, Topic, UnknownTopic}; pub mod aggregation; pub mod block_builder; 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 d398834d..69482239 100644 --- a/crates/net/rpc/src/events.rs +++ b/crates/net/rpc/src/events.rs @@ -9,44 +9,104 @@ //! 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) 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; +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::{ChainEvent, EventBus, Topic}; 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::{self, 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 { + // `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. + // + // 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 = match params.topics.as_deref() { + None | Some("") => { + return ( + StatusCode::BAD_REQUEST, + "missing required query parameter: topics", + ) + .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)) + .keep_alive(axum::response::sse::KeepAlive::default()) + .into_response() +} + +/// Bridge the receiver's `recv` loop into a stream of SSE frames, dropping +/// events the client's `topics` filter excludes. +fn event_stream( + rx: broadcast::Receiver, + topics: Vec, +) -> impl Stream> { + 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) => { + 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), (rx, topics))); + } + } + // 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), (rx, topics))); + } + // 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 +115,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?topics=head").await; + assert_eq!(resp.status(), StatusCode::OK); events.emit(ChainEvent::Head { slot: 3, @@ -88,12 +158,7 @@ mod tests { state: 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!( @@ -117,4 +182,103 @@ 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, + block: Default::default(), + }); + events.emit(ChainEvent::Head { + slot: 2, + block: Default::default(), + state: Default::default(), + }); + + let text = first_frame(resp).await; + assert!( + text.contains("event:head") || text.contains("event: head"), + "missing head event name in frame: {text}" + ); + // Check the SSE `event:` topic line, not the body: every event now + // carries a `block` *field*, so a substring check for "block" would + // false-positive on the head payload itself. + assert!( + !text.contains("event:block") && !text.contains("event: 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?topics=head").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, + block: Default::default(), + state: 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}" + ); + } + + #[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}"); + } + } } diff --git a/docs/rpc.md b/docs/rpc.md index baf50e2c..f560692d 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -104,7 +104,22 @@ event: head data: {"slot":128,"block":"0x1a2b…","state":"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=` + +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`. 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 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. ### `GET /lean/v0/blocks/{block_id}` and `/header`