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

Adds a chain-event pub-sub mechanism to the blockchain actor: a
`ChainEvent` enum (`head`, `block`, `justified_checkpoint`,
`finalized_checkpoint`) published through an `EventBus` facade over a
bounded tokio broadcast channel. The actor is the sole publisher;
subscribers attach their own receivers, and a slow consumer skips missed
events instead of back-pressuring the actor. Near-zero cost with no
subscribers (`receiver_count` guard). No HTTP surface in this PR: the
RPC consumer arrives in the next PR of the series.

Emission is actor-layer only, not threaded through the store. Before
each state-changing store call (`store::on_tick`, `store::on_block`)
`BlockChainServer` snapshots `(head, latest_justified,
latest_finalized)`, runs the store function completely unchanged, then
diffs the snapshot against the post-call state and emits in order
`block` → `head` → `justified_checkpoint` → `finalized_checkpoint`.
`store.rs`, both spec-test files, and `test_driver.rs` stay
byte-identical to `main`: no `Option<&EventBus>` parameter spreads
across `on_tick`/`on_block`/`update_head` or any of their test call
sites, and the sole-publisher property is structural (only
`BlockChainServer` ever holds a live `EventBus`) rather than something
every call site has to preserve by convention.

Two consequences of diffing at this level, both accepted:

- `block` needs a was-it-new guard: `on_block` returns `Ok` early for an
already-imported root, so the actor checks block-known-ness (the same
`has_state` check the store itself uses) before importing, and only
emits `block` for a genuinely new root.
- Several head moves inside a single tick or block import coalesce into
one `head` event carrying the net move; fine for subscribers, since the
beacon-API analog is also last-value-wins per slot.

The `head` event is additionally recency-gated: it only fires when the
new head is within `HEAD_EVENT_RECENCY_SLOTS` (32) of the wall-clock
slot, so catch-up/backfill never spams head events (adopted from
Lighthouse's head-SSE recency filter); the other events stay ungated.

Payload serialization is untagged, so the JSON body stays flat and topic
names live only on the transport's `event:` line.

Supersedes #460, re-cut as a stacked series for easier review:

1. this PR: the event mechanism
2. `GET /lean/v0/events` SSE endpoint (all events)
3. `?topics=` server-side filtering
4. `chain_reorg` + `safe_target` events
5. high-rate `attestation`/`aggregate` events

Has unit tests and passes clippy with `-D warnings`.

Author: MegaRedHand

Cargo.lock

@@ -52,7 +52,7 @@ 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, EventBus, SyncStatusController};
 use ethlambda_rpc::RpcConfig;
 use ethlambda_storage::{
     MAX_RESUMABLE_DB_STATE_AGE, StorageBackend, Store, backend::RocksDBBackend,
@@ -235,6 +235,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::default();
+
     let blockchain_config = BlockChainConfig {
         aggregator: aggregator.clone(),
         sync_status_controller: sync_status.clone(),
@@ -247,7 +253,7 @@ async fn main() -> eyre::Result<()> {
         },
     };
 
-    let blockchain = BlockChain::spawn(store.clone(), validator_keys, blockchain_config);
+    let blockchain = BlockChain::spawn(store.clone(), validator_keys, blockchain_config, events);
 
     let built = build_swarm(SwarmConfig {
         node_key: node_p2p_key,

bin/ethlambda/src/main.rs

@@ -34,8 +34,6 @@ tracing.workspace = true
 hex.workspace = true
 
 [dev-dependencies]
-serde = { workspace = true }
-serde_json = { workspace = true }
 hex = { workspace = true }
 libssz.workspace = true
 libssz-types.workspace = true