feat(blockchain): emit chain events from actor on head/block/finalized

Add a ChainEvent enum and a broadcast channel owned by the BlockChainServer
actor. The store's update_head path emits Head and FinalizedCheckpoint when
fork choice moves the head or finalization advances; on_block_core emits Block
on import. The sender is threaded as Option<&ChainEventTx> so spec-test and
test-driver entry points pass None. Keeps the actor as the sole writer: the
flow is strictly one-directional (actor -> broadcast).

Author: MegaRedHand

Cargo.lock

@@ -29,12 +29,12 @@ tokio-util = { version = "0.7", default-features = false }
 rayon.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
+serde.workspace = true
 
 hex.workspace = true
 
 [dev-dependencies]
 ethlambda-test-fixtures.workspace = true
-serde = { workspace = true }
 serde_json = { workspace = true }
 hex = { workspace = true }
 libssz.workspace = true

crates/blockchain/Cargo.toml

@@ -0,0 +1,56 @@
+//! Chain events emitted by the [`crate::BlockChainServer`] actor and streamed
+//! to RPC clients over Server-Sent Events (`GET /lean/v0/events`).
+//!
+//! The flow is strictly one-directional: the actor (the sole writer) publishes
+//! events on a [`broadcast`] channel, and the read-only RPC handler subscribes.
+//! RPC never writes back into the actor.
+
+use ethlambda_types::primitives::H256;
+use serde::Serialize;
+use tokio::sync::broadcast;
+
+/// A consensus event broadcast to SSE subscribers.
+///
+/// Serialized with an external `event`/`data` tag so the JSON payload mirrors
+/// the SSE framing (`event: head\ndata: {...}`).
+#[derive(Clone, Debug, Serialize)]
+#[serde(tag = "event", content = "data", rename_all = "snake_case")]
+pub enum ChainEvent {
+    /// Fork choice selected a new head.
+    Head {
+        slot: u64,
+        root: H256,
+        parent_root: H256,
+    },
+    /// A block was imported into the store.
+    Block { slot: u64, root: H256 },
+    /// The finalized checkpoint advanced.
+    FinalizedCheckpoint { slot: u64, root: H256 },
+}
+
+/// Sender half of the chain-event broadcast channel, owned by the actor.
+pub type ChainEventTx = broadcast::Sender<ChainEvent>;
+
+/// Capacity chosen so a briefly-stalled SSE client is dropped (lagged) rather
+/// than back-pressuring the actor. Lagged clients re-sync via backfill.
+pub const CHAIN_EVENT_CHANNEL_CAPACITY: usize = 256;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn channel_delivers_head_event() {
+        let (tx, mut rx) = broadcast::channel::<ChainEvent>(CHAIN_EVENT_CHANNEL_CAPACITY);
+        tx.send(ChainEvent::Head {
+            slot: 7,
+            root: H256::ZERO,
+            parent_root: H256::ZERO,
+        })
+        .unwrap();
+        match rx.recv().await.unwrap() {
+            ChainEvent::Head { slot, .. } => assert_eq!(slot, 7),
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+}

crates/blockchain/src/events.rs

@@ -28,9 +28,12 @@ use tracing::{debug, error, info, trace, warn};
 
 use crate::store::StoreError;
 
+pub use events::{CHAIN_EVENT_CHANNEL_CAPACITY, ChainEvent, ChainEventTx};
+
 pub mod aggregation;
 pub mod block_builder;
 pub(crate) mod coverage;
+pub mod events;
 pub(crate) mod fork_choice_tree;
 pub mod key_manager;
 pub mod metrics;
@@ -82,6 +85,7 @@ impl BlockChain {
         aggregator: AggregatorController,
         attestation_committee_count: u64,
         gate_duties: bool,
+        chain_events: ChainEventTx,
     ) -> BlockChain {
         metrics::set_is_aggregator(aggregator.is_enabled());
         metrics::set_node_sync_status(metrics::SyncStatus::Idle);
@@ -108,6 +112,7 @@ impl BlockChain {
             attestation_committee_count,
             pre_merge_coverage: None,
             sync_status: SyncStatusTracker::new(gate_duties),
+            chain_events,
         }
         .start();
         let time_until_genesis = (SystemTime::UNIX_EPOCH + Duration::from_secs(genesis_time))
@@ -177,6 +182,11 @@ pub struct BlockChainServer {
     /// validator duties while syncing, unless that gating was disabled at
     /// startup via `--disable-duty-sync-gate` (then it is metric-only).
     sync_status: SyncStatusTracker,
+
+    /// Broadcast sender for chain events streamed to SSE subscribers
+    /// (`GET /lean/v0/events`). The actor is the sole publisher; the RPC
+    /// handler only subscribes, preserving the one-directional write flow.
+    chain_events: ChainEventTx,
 }
 
 impl BlockChainServer {
@@ -258,7 +268,12 @@ impl BlockChainServer {
         let is_proposer = scheduled_proposer.is_some();
 
         // Tick the store first - this accepts attestations at interval 0 if we have a proposal
-        store::on_tick(&mut self.store, timestamp_ms, is_proposer);
+        store::on_tick(
+            &mut self.store,
+            timestamp_ms,
+            is_proposer,
+            Some(&self.chain_events),
+        );
 
         // ==== interval 0 ====
 
@@ -593,7 +608,7 @@ impl BlockChainServer {
 
     /// Run block import and refresh metrics.
     fn process_block(&mut self, signed_block: SignedBlock) -> Result<(), StoreError> {
-        store::on_block(&mut self.store, signed_block)?;
+        store::on_block(&mut self.store, signed_block, Some(&self.chain_events))?;
         metrics::update_head_slot(self.store.head_slot());
         metrics::update_latest_justified_slot(self.store.latest_justified().slot);
         metrics::update_latest_finalized_slot(self.store.latest_finalized().slot);

crates/blockchain/src/lib.rs

@@ -20,27 +20,33 @@ use crate::{
     GOSSIP_DISPARITY_INTERVALS, INTERVALS_PER_SLOT, MAX_ATTESTATIONS_DATA,
     MILLISECONDS_PER_INTERVAL, MILLISECONDS_PER_SLOT,
     block_builder::{PostBlockCheckpoints, build_block},
+    events::{ChainEvent, ChainEventTx},
     metrics,
 };
 
 const JUSTIFICATION_LOOKBACK_SLOTS: u64 = 3;
 
 /// Accept new aggregated payloads, promoting them to known for fork choice.
-fn accept_new_attestations(store: &mut Store, log_tree: bool) {
+fn accept_new_attestations(store: &mut Store, log_tree: bool, events: Option<&ChainEventTx>) {
     store.promote_new_aggregated_payloads();
     metrics::update_latest_new_aggregated_payloads(store.new_aggregated_payloads_count());
     metrics::update_latest_known_aggregated_payloads(store.known_aggregated_payloads_count());
-    update_head(store, log_tree);
+    update_head(store, log_tree, events);
 }
 
 /// Update the head based on the fork choice rule.
 ///
 /// When `log_tree` is true, also computes block weights and logs an ASCII
 /// fork choice tree to the terminal.
-pub fn update_head(store: &mut Store, log_tree: bool) {
+///
+/// When `events` is `Some`, emits a [`ChainEvent::Head`] whenever the head
+/// changes and a [`ChainEvent::FinalizedCheckpoint`] whenever finalization
+/// advances. Send errors (no subscribers) are ignored.
+pub fn update_head(store: &mut Store, log_tree: bool, events: Option<&ChainEventTx>) {
     let blocks = store.get_live_chain();
     let attestations = store.extract_latest_known_attestations();
     let old_head = store.head();
+    let old_finalized = store.latest_finalized();
     let (new_head, weights) = ethlambda_fork_choice::compute_lmd_ghost_head(
         store.latest_justified().root,
         &blocks,
@@ -60,6 +66,30 @@ pub fn update_head(store: &mut Store, log_tree: bool) {
         .filter(|finalized| store.get_block_header(&finalized.root).is_some());
     store.update_checkpoints(ForkCheckpoints::new(new_head, None, finalized));
 
+    if let Some(events) = events {
+        // Emit the new head whenever fork choice moved it.
+        if old_head != new_head {
+            let parent_root = store
+                .get_block_header(&new_head)
+                .map(|h| h.parent_root)
+                .unwrap_or(H256::ZERO);
+            let _ = events.send(ChainEvent::Head {
+                slot: store.head_slot(),
+                root: new_head,
+                parent_root,
+            });
+        }
+
+        // Emit a finalized-checkpoint event only when finalization advanced.
+        let new_finalized = store.latest_finalized();
+        if new_finalized.slot > old_finalized.slot || new_finalized.root != old_finalized.root {
+            let _ = events.send(ChainEvent::FinalizedCheckpoint {
+                slot: new_finalized.slot,
+                root: new_finalized.root,
+            });
+        }
+    }
+
     if old_head != new_head {
         let old_slot = store
             .get_block_header(&old_head)
@@ -254,7 +284,12 @@ fn validate_attestation_data(store: &Store, data: &AttestationData) -> Result<()
 /// 800ms interval. Slot and interval-within-slot are derived as:
 ///   slot     = store.time() / INTERVALS_PER_SLOT
 ///   interval = store.time() % INTERVALS_PER_SLOT
-pub fn on_tick(store: &mut Store, timestamp_ms: u64, has_proposal: bool) {
+pub fn on_tick(
+    store: &mut Store,
+    timestamp_ms: u64,
+    has_proposal: bool,
+    events: Option<&ChainEventTx>,
+) {
     // Convert UNIX timestamp (ms) to interval count since genesis
     let genesis_time_ms = store.config().genesis_time * 1000;
     let time_delta_ms = timestamp_ms.saturating_sub(genesis_time_ms);
@@ -287,7 +322,7 @@ pub fn on_tick(store: &mut Store, timestamp_ms: u64, has_proposal: bool) {
             0 => {
                 // Start of slot - process attestations if proposal exists
                 if should_signal_proposal {
-                    accept_new_attestations(store, false);
+                    accept_new_attestations(store, false, events);
                 }
             }
             1 => {
@@ -302,7 +337,7 @@ pub fn on_tick(store: &mut Store, timestamp_ms: u64, has_proposal: bool) {
             }
             4 => {
                 // End of slot - accept accumulated attestations and log tree
-                accept_new_attestations(store, true);
+                accept_new_attestations(store, true, events);
             }
             _ => unreachable!("slots only have 5 intervals"),
         }
@@ -481,8 +516,12 @@ fn on_gossip_aggregated_attestation_core(
 ///
 /// This is the safe default: it always verifies cryptographic signatures
 /// and stores them for future block building. Use this for all production paths.
-pub fn on_block(store: &mut Store, signed_block: SignedBlock) -> Result<(), StoreError> {
-    on_block_core(store, signed_block, true)
+pub fn on_block(
+    store: &mut Store,
+    signed_block: SignedBlock,
+    events: Option<&ChainEventTx>,
+) -> Result<(), StoreError> {
+    on_block_core(store, signed_block, true, events)
 }
 
 /// Process a new block without signature verification.
@@ -493,7 +532,7 @@ pub fn on_block_without_verification(
     store: &mut Store,
     signed_block: SignedBlock,
 ) -> Result<(), StoreError> {
-    on_block_core(store, signed_block, false)
+    on_block_core(store, signed_block, false, None)
 }
 
 /// Core block processing logic.
@@ -504,6 +543,7 @@ fn on_block_core(
     store: &mut Store,
     signed_block: SignedBlock,
     verify: bool,
+    events: Option<&ChainEventTx>,
 ) -> Result<(), StoreError> {
     let _timing = metrics::time_fork_choice_block_processing();
     let block_start = std::time::Instant::now();
@@ -586,8 +626,17 @@ fn on_block_core(
         metrics::inc_attestations_valid(count);
     }
 
+    // Emit the imported block before fork choice runs, so subscribers see the
+    // `block` event ahead of any `head` move it triggers.
+    if let Some(events) = events {
+        let _ = events.send(ChainEvent::Block {
+            slot,
+            root: block_root,
+        });
+    }
+
     // Update forkchoice head based on new block and attestations
-    update_head(store, false);
+    update_head(store, false, events);
 
     let block_total = block_start.elapsed();
     info!(
@@ -758,11 +807,16 @@ fn get_proposal_head(store: &mut Store, slot: u64) -> H256 {
     // Calculate time corresponding to this slot
     let slot_time_ms = store.config().genesis_time * 1000 + slot * MILLISECONDS_PER_SLOT;
 
-    // Advance time to current slot (ticking intervals)
-    on_tick(store, slot_time_ms, true);
+    // Advance time to current slot (ticking intervals).
+    //
+    // No event sender here: this is the proposer's pre-build catch-up, and the
+    // block it produces is imported via `on_block` (which emits the resulting
+    // `Block`/`Head`). Emitting from here would surface a head move before the
+    // block exists.
+    on_tick(store, slot_time_ms, true, None);
 
     // Process any pending attestations before proposal
-    accept_new_attestations(store, false);
+    accept_new_attestations(store, false, None);
 
     store.head()
 }

crates/blockchain/src/store.rs

@@ -103,7 +103,7 @@ fn run(path: &Path) -> datatest_stable::Result<()> {
                     if step.tick_to_slot {
                         let block_time_ms =
                             genesis_time * 1000 + signed_block.message.slot * MILLISECONDS_PER_SLOT;
-                        store::on_tick(&mut store, block_time_ms, true);
+                        store::on_tick(&mut store, block_time_ms, true, None);
                     }
                     let result = store::on_block_without_verification(&mut store, signed_block);
                     let import_ok = result.is_ok();
@@ -137,7 +137,7 @@ fn run(path: &Path) -> datatest_stable::Result<()> {
                         // on_block already ran the head update before these votes
                         // existed; recompute so the head reflects the block's own
                         // attestations, matching the proposer-view store.
-                        store::update_head(&mut store, false);
+                        store::update_head(&mut store, false, None);
                     }
                 }
                 "tick" => {
@@ -152,7 +152,7 @@ fn run(path: &Path) -> datatest_stable::Result<()> {
                         (None, None) => panic!("tick step missing both time and interval"),
                     };
                     let has_proposal = step.has_proposal.unwrap_or(false);
-                    store::on_tick(&mut store, timestamp_ms, has_proposal);
+                    store::on_tick(&mut store, timestamp_ms, has_proposal, None);
                 }
                 "attestation" => {
                     let att_data = step

crates/blockchain/tests/forkchoice_spectests.rs

@@ -61,10 +61,10 @@ fn run(path: &Path) -> datatest_stable::Result<()> {
 
         // Advance time to the block's slot
         let block_time_ms = genesis_time * 1000 + signed_block.message.slot * MILLISECONDS_PER_SLOT;
-        store::on_tick(&mut st, block_time_ms, true);
+        store::on_tick(&mut st, block_time_ms, true, None);
 
         // Process the block (this includes signature verification)
-        let result = store::on_block(&mut st, signed_block);
+        let result = store::on_block(&mut st, signed_block, None);
 
         // Step 3: Check that it succeeded or failed as expected
         match (result.is_ok(), test.expect_exception.as_ref()) {