diff --git a/Cargo.lock b/Cargo.lock index 145334e0..7e31a87a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2053,6 +2053,7 @@ dependencies = [ "libssz-types", "rand 0.10.1", "rayon", + "serde", "spawned-concurrency 0.5.0", "thiserror 2.0.18", "tokio", @@ -2135,6 +2136,7 @@ dependencies = [ "ethlambda-storage", "ethlambda-test-fixtures", "ethlambda-types", + "futures-core", "hex", "http-body-util", "jemalloc_pprof", @@ -2142,6 +2144,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-stream", "tokio-util", "tower", "tracing", diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 643240df..b4d2828d 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -235,10 +235,10 @@ async fn main() -> eyre::Result<()> { // metric's startup value. let sync_status = SyncStatusController::default(); - // Chain-event bus: the blockchain actor is the sole publisher. No consumer - // subscribes yet — the RPC SSE endpoint (follow-up PR) will attach here; - // until then the receiver-count guard in `emit` makes every emission a - // no-op. + // Chain-event bus: the blockchain actor is the sole publisher; each SSE + // client (`GET /lean/v0/events`) subscribes its own receiver through the + // clone handed to the RPC server below. With no subscribers attached, the + // receiver-count guard in `emit` makes every emission a no-op. let events = EventBus::default(); let blockchain_config = BlockChainConfig { @@ -253,7 +253,12 @@ async fn main() -> eyre::Result<()> { }, }; - let blockchain = BlockChain::spawn(store.clone(), validator_keys, blockchain_config, events); + let blockchain = BlockChain::spawn( + store.clone(), + validator_keys, + blockchain_config, + events.clone(), + ); let built = build_swarm(SwarmConfig { node_key: node_p2p_key, @@ -296,6 +301,7 @@ async fn main() -> eyre::Result<()> { aggregator, sync_status, local_peer_id, + events, rpc_shutdown, ) .await diff --git a/crates/blockchain/Cargo.toml b/crates/blockchain/Cargo.toml index d6c3bcea..cacd50ad 100644 --- a/crates/blockchain/Cargo.toml +++ b/crates/blockchain/Cargo.toml @@ -28,6 +28,7 @@ tokio.workspace = true tokio-util = { version = "0.7", default-features = false } rayon.workspace = true +serde.workspace = true thiserror.workspace = true tracing.workspace = true diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index bc1716c2..3ff1e1b1 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -13,6 +13,7 @@ use ethlambda_storage::Store; use ethlambda_types::ShortRoot; use ethlambda_types::checkpoint::Checkpoint; use ethlambda_types::primitives::H256; +use serde::Serialize; use tokio::sync::broadcast; use tracing::warn; @@ -45,18 +46,20 @@ impl Topic { /// A consensus event published by the blockchain actor. /// -/// Fields mirror the Ethereum beacon-API eventstream payloads so a future SSE -/// endpoint can render a beacon-compatible stream: `block` is the block root, +/// Fields mirror the Ethereum beacon-API eventstream payloads so the SSE +/// endpoint renders a beacon-compatible stream: `block` is the block root, /// `state` the state root, and `slot` stands in for the beacon `epoch`. /// [`ChainEvent::JustifiedCheckpoint`] has no beacon analog; it mirrors /// [`ChainEvent::FinalizedCheckpoint`]'s shape as an ethlambda extension. /// -/// No `Serialize` is derived: the wire encoding (decimal-string integers, -/// `0x`-hex roots, and the out-of-band topic on the SSE `event:` line) lands -/// with the SSE endpoint in a follow-up PR. Until then this type is -/// internal-only, so there is no risk of an untagged shape being deserialized -/// ambiguously (the `{slot, block, state}` variants are structurally identical). -#[derive(Clone, Debug)] +/// `#[serde(untagged)]` serializes only the active variant's fields, so the SSE +/// `data:` body stays flat (`0x`-hex roots, numeric slots) while the topic name +/// travels out-of-band on the `event:` line via [`ChainEvent::topic`]. Only +/// `Serialize` is derived, never `Deserialize`: an untagged shape would +/// deserialize ambiguously since the `{slot, block, state}` variants are +/// structurally identical, but serialization always knows its variant. +#[derive(Clone, Debug, Serialize)] +#[serde(untagged)] pub enum ChainEvent { /// Fork choice selected a new head. Head { slot: u64, block: H256, state: H256 }, diff --git a/crates/net/rpc/Cargo.toml b/crates/net/rpc/Cargo.toml index bebd3218..cee21b9d 100644 --- a/crates/net/rpc/Cargo.toml +++ b/crates/net/rpc/Cargo.toml @@ -26,6 +26,8 @@ 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" [dev-dependencies] ethlambda-types.workspace = true diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs new file mode 100644 index 00000000..d398834d --- /dev/null +++ b/crates/net/rpc/src/events.rs @@ -0,0 +1,120 @@ +//! `GET /lean/v0/events` — Server-Sent Events stream of chain events. +//! +//! The [`ethlambda_blockchain::BlockChainServer`] actor publishes +//! [`ethlambda_blockchain::ChainEvent`]s on the [`EventBus`]; this read-only +//! handler subscribes a new receiver per connection and forwards each event as +//! one SSE frame. The flow is strictly one-directional (actor → bus → SSE), so +//! RPC never writes into the actor. +//! +//! 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. + +use std::convert::Infallible; + +use axum::{ + Extension, Router, + response::{Sse, sse::Event}, + routing::get, +}; +use ethlambda_blockchain::EventBus; +use ethlambda_storage::Store; +use futures_core::Stream; +use tokio_stream::{ + StreamExt, + wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}, +}; + +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; + } + }; + 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 { + Router::new().route("/lean/v0/events", get(get_events)) +} + +#[cfg(test)] +mod tests { + use axum::{Extension, body::Body, http::Request}; + use ethlambda_blockchain::{ChainEvent, EventBus}; + use ethlambda_storage::{Store, backend::InMemoryBackend}; + use std::sync::Arc; + use tower::ServiceExt; + + use crate::test_utils::create_test_state; + + #[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); + + events.emit(ChainEvent::Head { + slot: 3, + block: Default::default(), + 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); + + // Topic on the `event:` line... + assert!( + text.contains("event:head") || text.contains("event: head"), + "missing head event name in frame: {text}" + ); + // ...and a flat payload: the variant's own fields (`slot`, `block`, + // `state`) at the top level, `slot` as a plain number, with no + // `event`/`data` wrapper keys inside the JSON body (the #460 + // double-tag bug). + assert!( + text.contains("\"slot\":3"), + "missing top-level slot in frame: {text}" + ); + assert!( + text.contains("\"block\":") && text.contains("\"state\":"), + "missing beacon-aligned block/state fields in frame: {text}" + ); + assert!( + !text.contains("\"data\":") && !text.contains("\"event\":"), + "payload is not flat: {text}" + ); + } +} diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs index 8ea2ade7..217543dd 100644 --- a/crates/net/rpc/src/lib.rs +++ b/crates/net/rpc/src/lib.rs @@ -1,7 +1,7 @@ use std::net::{IpAddr, SocketAddr}; use axum::{Extension, Router}; -use ethlambda_blockchain::SyncStatusController; +use ethlambda_blockchain::{EventBus, SyncStatusController}; use ethlambda_storage::Store; use ethlambda_types::aggregator::AggregatorController; use tokio_util::sync::CancellationToken; @@ -12,6 +12,7 @@ pub(crate) const SSZ_CONTENT_TYPE: &str = "application/octet-stream"; mod admin; mod base; mod blocks; +mod events; mod fork_choice; mod genesis; mod heap_profiling; @@ -64,11 +65,13 @@ pub async fn start_rpc_server( aggregator: AggregatorController, sync_status: SyncStatusController, peer_id: String, + events: EventBus, shutdown: CancellationToken, ) -> Result<(), std::io::Error> { let api_router = build_api_router(store, config.version, peer_id) .layer(Extension(aggregator)) - .layer(Extension(sync_status)); + .layer(Extension(sync_status)) + .layer(Extension(events)); let metrics_router = metrics::start_prometheus_metrics_api(); let debug_router = build_debug_router(); @@ -115,6 +118,7 @@ fn build_api_router(store: Store, version: &'static str, peer_id: String) -> Rou Router::new() .merge(base::routes()) .merge(blocks::routes()) + .merge(events::routes()) .merge(fork_choice::routes()) .merge(admin::routes()) .merge(node::routes(version, peer_id)) diff --git a/docs/rpc.md b/docs/rpc.md index 2ec8c44c..baf50e2c 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -28,6 +28,7 @@ If `--api-port` and `--metrics-port` are equal, all routers are merged onto a si | `GET` | `/lean/v0/states/finalized` | SSZ | Latest finalized `State` | | `GET` | `/lean/v0/blocks/finalized` | SSZ | Latest finalized `SignedBlock` | | `GET` | `/lean/v0/checkpoints/justified` | JSON | Latest justified `Checkpoint` | +| `GET` | `/lean/v0/events` | SSE | Live stream of chain events | | `GET` | `/lean/v0/blocks/{block_id}` | JSON | Block by root or slot | | `GET` | `/lean/v0/blocks/{block_id}/header` | JSON | Block header by root or slot | | `GET` | `/lean/v0/fork_choice` | JSON | Fork-choice tree with per-block weights | @@ -83,6 +84,28 @@ SSZ-encoded `SignedBlock` at the latest finalized checkpoint. The genesis/anchor { "slot": 128, "root": "0x1a2b…" } ``` +### `GET /lean/v0/events` + +Server-Sent Events stream (`Content-Type: text/event-stream`) of live chain events published by the blockchain actor. Four event types: + +Payload fields mirror the Ethereum beacon-API eventstream: `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`. + +| Event | Payload | Emitted when | +|-------|---------|--------------| +| `head` | `{ "slot": 128, "block": "0x…", "state": "0x…" }` | Fork choice selects a new head within `HEAD_EVENT_RECENCY_SLOTS` (32 slots) of the wall clock; no head events fire during catch-up | +| `block` | `{ "slot": 128, "block": "0x…" }` | A block is imported into the store | +| `justified_checkpoint` | `{ "slot": 120, "block": "0x…", "state": "0x…" }` | The justified checkpoint advances | +| `finalized_checkpoint` | `{ "slot": 96, "block": "0x…", "state": "0x…" }` | The finalized checkpoint advances | + +The topic name travels only on the SSE `event:` line; the `data:` line carries the flat JSON payload. Example frame: + +``` +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. + ### `GET /lean/v0/blocks/{block_id}` and `/header` `block_id` is either: @@ -195,6 +218,7 @@ When the binary boots with `HIVE_LEAN_TEST_DRIVER=1` (any of `1`/`true`/`yes`), | Kind | `Content-Type` | |------|----------------| | JSON | `application/json; charset=utf-8` | +| SSE | `text/event-stream` | | SSZ | `application/octet-stream` | | Prometheus metrics | `text/plain; version=0.0.4; charset=utf-8` | | HTML | `text/html; charset=utf-8` |