feat(rpc): add GET /lean/v0/events SSE stream

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

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
 

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 },

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

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<EventBus>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    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<Store> {
+    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}"
+        );
+    }
+}

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))

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` |