feat(blockchain): add chain-event bus (head/block/justified/finalized)

First PR of the chain-events series (re-cut of #460), shipping only the
pub-sub mechanism so the real design decisions (event shape, publisher
model, back-pressure policy) are reviewable on their own. The SSE
transport and topic filtering follow in separate PRs; the bus stays
unconsumed until the endpoint lands.

The bus is a facade over a single bounded tokio broadcast channel: the
blockchain actor is the sole publisher, and emit never blocks or fails,
since a slow subscriber lags and drops instead of back-pressuring
consensus; a receiver-count guard makes emission free when nobody
listens. Threading &EventBus (never Option) through the store functions
keeps the sole-publisher property visible in signatures, and
EventBus::disabled() removes the Option noise from test call sites.

ChainEvent serializes untagged so payloads stay flat: the topic name
travels out-of-band via ChainEvent::topic(), fixing #460's
double-tagged SSE payload before any transport exists.

The justified_checkpoint emission lives at its advance site in
on_block_core instead of an old/new diff in update_head: update_head
never modifies latest_justified (its update_checkpoints call passes
None for it), so a diff there could never fire.

Author: MegaRedHand

bin/ethlambda/src/main.rs

@@ -52,7 +52,9 @@ use serde::Deserialize;
 use tracing::{error, info, warn};
 use tracing_subscriber::{EnvFilter, Layer, Registry, layer::SubscriberExt};
 
-use ethlambda_blockchain::{BlockChain, BlockChainConfig, SyncStatusController};
+use ethlambda_blockchain::{
+    BlockChain, BlockChainConfig, CHAIN_EVENT_CHANNEL_CAPACITY, EventBus, SyncStatusController,
+};
 use ethlambda_rpc::RpcConfig;
 use ethlambda_storage::{
     MAX_RESUMABLE_DB_STATE_AGE, StorageBackend, Store, backend::RocksDBBackend,
@@ -235,6 +237,12 @@ 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.
+    let events = EventBus::new(CHAIN_EVENT_CHANNEL_CAPACITY);
+
     let blockchain_config = BlockChainConfig {
         aggregator: aggregator.clone(),
         sync_status_controller: sync_status.clone(),
@@ -245,6 +253,7 @@ async fn main() -> eyre::Result<()> {
             enable_proposer_aggregation: options.enable_proposer_aggregation,
             max_attestations_per_block: options.max_attestations_per_block,
         },
+        events,
     };
 
     let blockchain = BlockChain::spawn(store.clone(), validator_keys, blockchain_config);

crates/blockchain/Cargo.toml

@@ -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/src/events.rs

@@ -0,0 +1,194 @@
+//! Chain-event pub-sub bus.
+//!
+//! The [`crate::BlockChainServer`] actor is the **sole publisher**: it emits a
+//! [`ChainEvent`] whenever consensus state changes (block import, head move,
+//! justification, finalization). Consumers subscribe read-only receivers and
+//! never write back into the actor, keeping the write flow one-directional.
+//!
+//! The bus is intentionally best-effort: emission never blocks the actor, and
+//! a slow subscriber loses events (the bounded broadcast channel overwrites
+//! its backlog) rather than back-pressuring consensus.
+
+use ethlambda_types::primitives::H256;
+use serde::Serialize;
+use tokio::sync::broadcast;
+
+/// Wire-visible topic names for chain events.
+///
+/// These are the names consumers address events by (the SSE `event:` line and,
+/// later, `?topics=` filtering), kept separate from [`ChainEvent`] so the
+/// payloads stay flat JSON with the topic travelling out-of-band.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Topic {
+    Head,
+    Block,
+    JustifiedCheckpoint,
+    FinalizedCheckpoint,
+}
+
+impl Topic {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Topic::Head => "head",
+            Topic::Block => "block",
+            Topic::JustifiedCheckpoint => "justified_checkpoint",
+            Topic::FinalizedCheckpoint => "finalized_checkpoint",
+        }
+    }
+}
+
+/// A consensus event published by the blockchain actor.
+///
+/// `untagged`: serializing yields only the variant's fields. The topic name
+/// travels out-of-band (via [`ChainEvent::topic`], e.g. on the SSE `event:`
+/// line), so the JSON body stays flat with no `event`/`data` wrapper.
+#[derive(Clone, Debug, Serialize)]
+#[serde(untagged)]
+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 justified checkpoint advanced.
+    JustifiedCheckpoint { slot: u64, root: H256 },
+    /// The finalized checkpoint advanced.
+    FinalizedCheckpoint { slot: u64, root: H256 },
+}
+
+impl ChainEvent {
+    pub fn topic(&self) -> Topic {
+        match self {
+            ChainEvent::Head { .. } => Topic::Head,
+            ChainEvent::Block { .. } => Topic::Block,
+            ChainEvent::JustifiedCheckpoint { .. } => Topic::JustifiedCheckpoint,
+            ChainEvent::FinalizedCheckpoint { .. } => Topic::FinalizedCheckpoint,
+        }
+    }
+}
+
+/// Capacity of the chain-event broadcast channel.
+///
+/// Chosen so a briefly-stalled subscriber is skipped past (lagged) rather than
+/// back-pressuring the actor. Lagged subscribers re-sync via the blocks
+/// endpoints.
+pub const CHAIN_EVENT_CHANNEL_CAPACITY: usize = 256;
+
+/// Cloneable handle to the chain-event broadcast channel.
+///
+/// Threaded as `&EventBus` (never `Option`) through the store's processing
+/// functions: emission sites call [`EventBus::emit`], and paths that must not
+/// surface events pass [`EventBus::disabled`].
+#[derive(Clone)]
+pub struct EventBus {
+    tx: broadcast::Sender<ChainEvent>,
+}
+
+impl EventBus {
+    pub fn new(capacity: usize) -> Self {
+        let (tx, _) = broadcast::channel(capacity);
+        Self { tx }
+    }
+
+    /// Dormant bus: emits go nowhere. For tests and eventless call paths.
+    pub fn disabled() -> Self {
+        Self::new(1)
+    }
+
+    /// Publish an event to all current subscribers.
+    ///
+    /// Never blocks, never fails: without subscribers this is a no-op, and a
+    /// send error (every subscriber dropped since the guard) is ignored.
+    pub fn emit(&self, event: ChainEvent) {
+        if self.tx.receiver_count() == 0 {
+            return;
+        }
+        let _ = self.tx.send(event);
+    }
+
+    /// Subscribe a new receiver observing every event emitted from now on.
+    pub fn subscribe(&self) -> broadcast::Receiver<ChainEvent> {
+        self.tx.subscribe()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn head_event(slot: u64) -> ChainEvent {
+        ChainEvent::Head {
+            slot,
+            root: H256([1u8; 32]),
+            parent_root: H256([2u8; 32]),
+        }
+    }
+
+    #[tokio::test]
+    async fn subscriber_receives_emitted_event() {
+        let bus = EventBus::new(CHAIN_EVENT_CHANNEL_CAPACITY);
+        let mut rx = bus.subscribe();
+
+        bus.emit(head_event(7));
+
+        match rx.recv().await.unwrap() {
+            ChainEvent::Head { slot, .. } => assert_eq!(slot, 7),
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn emit_without_subscribers_is_a_noop() {
+        let bus = EventBus::new(CHAIN_EVENT_CHANNEL_CAPACITY);
+        // No subscriber attached: must neither error nor panic.
+        bus.emit(head_event(1));
+    }
+
+    #[test]
+    fn disabled_bus_accepts_emits() {
+        let bus = EventBus::disabled();
+        bus.emit(head_event(1));
+        bus.emit(ChainEvent::Block {
+            slot: 2,
+            root: H256::ZERO,
+        });
+    }
+
+    #[test]
+    fn topic_maps_every_variant() {
+        let root = H256::ZERO;
+        let cases = [
+            (head_event(1), Topic::Head),
+            (ChainEvent::Block { slot: 1, root }, Topic::Block),
+            (
+                ChainEvent::JustifiedCheckpoint { slot: 1, root },
+                Topic::JustifiedCheckpoint,
+            ),
+            (
+                ChainEvent::FinalizedCheckpoint { slot: 1, root },
+                Topic::FinalizedCheckpoint,
+            ),
+        ];
+        for (event, topic) in cases {
+            assert_eq!(event.topic(), topic);
+            assert_eq!(event.topic().as_str(), topic.as_str());
+        }
+    }
+
+    /// The JSON body must be the variant's fields only: the topic name travels
+    /// out-of-band, so no `event`/`data` wrapper keys may appear (the #460
+    /// double-tag bug).
+    #[test]
+    fn serialization_is_flat_untagged_json() {
+        let json = serde_json::to_value(head_event(3)).unwrap();
+
+        assert_eq!(json["slot"], 3);
+        assert!(json["root"].is_string());
+        assert!(json["parent_root"].is_string());
+        assert!(json.get("event").is_none());
+        assert!(json.get("data").is_none());
+    }
+}

crates/blockchain/src/lib.rs

@@ -30,9 +30,12 @@ use tracing::{debug, error, info, trace, warn};
 use crate::block_builder::ProposerConfig;
 use crate::store::StoreError;
 
+pub use events::{CHAIN_EVENT_CHANNEL_CAPACITY, ChainEvent, EventBus, Topic};
+
 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;
@@ -61,6 +64,9 @@ pub struct BlockChainConfig {
     pub subscribed_subnets: HashSet<u64>,
     /// Proposer-side block-building policy.
     pub proposer_config: ProposerConfig,
+    /// Chain-event publication bus. The actor is the sole publisher;
+    /// consumers subscribe read-only receivers.
+    pub events: EventBus,
 }
 
 /// Milliseconds per interval (800ms ticks).
@@ -136,6 +142,7 @@ impl BlockChain {
             gate_duties,
             subscribed_subnets,
             proposer_config,
+            events,
         } = config;
 
         metrics::set_is_aggregator(aggregator.is_enabled());
@@ -169,6 +176,7 @@ impl BlockChain {
             pre_merge_coverage: None,
             sync_status: SyncStatusTracker::new(gate_duties),
             sync_status_controller,
+            events,
         }
         .start();
         let time_until_genesis = (SystemTime::UNIX_EPOCH + Duration::from_secs(genesis_time))
@@ -255,6 +263,10 @@ pub struct BlockChainServer {
     /// (the RPC `/lean/v0/node/syncing` endpoint). Written from
     /// `update_sync_status` with the same `SyncStatus` fed to the metric.
     sync_status_controller: SyncStatusController,
+
+    /// Chain-event publication bus. The actor is the sole publisher; consumers
+    /// only subscribe, preserving the one-directional write flow.
+    events: EventBus,
 }
 
 impl BlockChainServer {
@@ -331,7 +343,7 @@ impl BlockChainServer {
             .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, &self.events);
 
         // Per-interval duties for this tick. Intervals 0 (block publish) and 3
         // (safe-target update) are driven inside `store::on_tick` above, so they
@@ -857,7 +869,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, &self.events)?;
         metrics::update_head_slot(self.store.head_slot());
         let latest_justified_slot = self
             .store