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

Second PR of the chain-events series: the transport for the event bus
introduced in the previous PR. Kept separate from the mechanism so the
axum plumbing reviews on its own; topic filtering (?topics=) follows in
the next PR, so this stream is unfiltered.

Each connection subscribes its own receiver and forwards events as SSE
frames: the topic name goes on the event: line via ChainEvent::topic()
(no hardcoded name match to drift from the enum) and the data: line
carries the flat JSON payload, so the topic is never duplicated inside
the body. A lagged client skips past the events the bounded channel
overwrote (debug log, stream continues) rather than ending the stream:
the stream is best-effort by contract and clients re-sync via the
blocks endpoints. Keep-alive comments hold idle connections open
through proxies.

Author: MegaRedHand

Cargo.lock

@@ -237,10 +237,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::new(CHAIN_EVENT_CHANNEL_CAPACITY);
 
     let blockchain_config = BlockChainConfig {
@@ -253,7 +253,7 @@ async fn main() -> eyre::Result<()> {
             enable_proposer_aggregation: options.enable_proposer_aggregation,
             max_attestations_per_block: options.max_attestations_per_block,
         },
-        events,
+        events: events.clone(),
     };
 
     let blockchain = BlockChain::spawn(store.clone(), validator_keys, blockchain_config);
@@ -299,6 +299,7 @@ async fn main() -> eyre::Result<()> {
             aggregator,
             sync_status,
             local_peer_id,
+            events,
             rpc_shutdown,
         )
         .await

bin/ethlambda/src/main.rs

@@ -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/Cargo.toml

@@ -0,0 +1,114 @@
+//! `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,
+            root: Default::default(),
+            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);
+
+        // 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: fields at the top level, 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("\"data\":") && !text.contains("\"event\":"),
+            "payload is not flat: {text}"
+        );
+    }
+}

crates/net/rpc/src/events.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))

crates/net/rpc/src/lib.rs

@@ -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,26 @@ 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:
+
+| Event | Payload | Emitted when |
+|-------|---------|--------------|
+| `head` | `{ "slot": 128, "root": "0x…", "parent_root": "0x…" }` | Fork choice selects a new head |
+| `block` | `{ "slot": 128, "root": "0x…" }` | A block is imported into the store |
+| `justified_checkpoint` | `{ "slot": 120, "root": "0x…" }` | The justified checkpoint advances |
+| `finalized_checkpoint` | `{ "slot": 96, "root": "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,"root":"0x1a2b…","parent_root":"0x3c4d…"}
+```
+
+Events are fanned out over a bounded broadcast channel (`CHAIN_EVENT_CHANNEL_CAPACITY`). 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 +216,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` |