From 60d9c2e8bbc07b473643166194866bf1202853ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 15 Jul 2026 10:38:04 -0300 Subject: [PATCH 1/9] 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. Emission is actor-layer only, not threaded through the store. Before each state-changing store call (store::on_tick, store::on_block) the actor 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 -> justified_checkpoint -> head -> finalized_checkpoint. This keeps store.rs, both spec-test files, and test_driver.rs 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 bus itself is a facade over a single bounded tokio broadcast channel: emit() never blocks or fails, since a slow subscriber lags and drops instead of back-pressuring consensus, and a receiver-count guard makes emission free when nobody is listening. 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 `head` event is additionally gated by HEAD_EVENT_RECENCY_SLOTS, adopted from a review of Lighthouse's canonical-head recompute path: a new head is only emitted when its slot is within that many slots of the wall clock, so startup catch-up and backfill don't spam subscribers with historical head moves on the way to the tip. `block`, justified_checkpoint, and finalized_checkpoint stay ungated; consumers tracking sync progress should watch `block` instead. --- bin/ethlambda/src/main.rs | 10 +- crates/blockchain/Cargo.toml | 2 +- crates/blockchain/src/events.rs | 204 ++++++++++++++++ crates/blockchain/src/lib.rs | 407 +++++++++++++++++++++++++++++++- 4 files changed, 618 insertions(+), 5 deletions(-) create mode 100644 crates/blockchain/src/events.rs diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 5234a84e..643240df 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -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, diff --git a/crates/blockchain/Cargo.toml b/crates/blockchain/Cargo.toml index f25234cd..a08262eb 100644 --- a/crates/blockchain/Cargo.toml +++ b/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 diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs new file mode 100644 index 00000000..7a7c0dff --- /dev/null +++ b/crates/blockchain/src/events.rs @@ -0,0 +1,204 @@ +//! 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. +const CHAIN_EVENT_CHANNEL_CAPACITY: usize = 256; + +/// Cloneable handle to the chain-event broadcast channel. +/// +/// Owned solely by [`crate::BlockChainServer`] (never `Option`, never +/// threaded into `store.rs`): the actor snapshots store state before a call +/// to `store::on_tick`/`store::on_block`, runs it unchanged, then diffs and +/// calls [`EventBus::emit`] itself. Call sites that must stay eventless (spec +/// tests, `test_driver.rs`) simply never construct a live bus. +#[derive(Clone)] +pub struct EventBus { + tx: broadcast::Sender, +} + +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 { + self.tx.subscribe() + } +} + +impl Default for EventBus { + /// A live bus with the default channel capacity + /// ([`CHAIN_EVENT_CHANNEL_CAPACITY`]). + fn default() -> Self { + Self::new(CHAIN_EVENT_CHANNEL_CAPACITY) + } +} + +#[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::default(); + 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::default(); + // 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()); + } +} diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 05fa8b03..24db4bc6 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -9,6 +9,7 @@ use ethlambda_types::{ aggregator::AggregatorController, attestation::{SignedAggregatedAttestation, SignedAttestation}, block::{ByteList512KiB, MultiMessageAggregate, SignedBlock}, + checkpoint::Checkpoint, primitives::{H256, HashTreeRoot as _}, signature::{ValidatorPublicKey, ValidatorSignature}, }; @@ -30,9 +31,12 @@ use tracing::{debug, error, info, trace, warn}; use crate::block_builder::ProposerConfig; use crate::store::StoreError; +pub use events::{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; @@ -124,10 +128,15 @@ fn unix_now_ms() -> u64 { } impl BlockChain { + /// Spawn the blockchain actor. + /// + /// `events` is the chain-event publication bus: the spawned actor is its + /// sole publisher; consumers subscribe read-only receivers. pub fn spawn( store: Store, validator_keys: HashMap, config: BlockChainConfig, + events: EventBus, ) -> BlockChain { let BlockChainConfig { aggregator, @@ -169,6 +178,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 +265,105 @@ 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, +} + +/// Suppresses `head` events whose slot has fallen too far behind the wall +/// clock: during startup catch-up or backfill, fork choice can walk through +/// many historical heads on its way to the tip, and none of those are +/// interesting to a live subscriber. Consumers that need to track sync +/// progress should watch `block` events instead, which are never gated on +/// recency. Mirrors Lighthouse's recency filter on its head SSE event +/// (`EARLY_ATTESTER_CACHE_HISTORIC_SLOTS`), but with a wider window since a +/// lagging head is far more common here during multi-slot catch-up ticks. +const HEAD_EVENT_RECENCY_SLOTS: u64 = 32; + +/// Pre-call snapshot of the store values the chain-event bus reports on. +/// +/// The actor — not the store — publishes chain events: it captures this +/// snapshot before a store call (`store::on_tick`, `store::on_block`) and +/// diffs the store against it afterwards, so `store.rs` needs no event +/// plumbing. Consequences of diffing at this level, both intentional: +/// +/// - Multiple head moves within one store call coalesce into a single `head` +/// event; subscribers only care about the latest. +/// - The proposer's pre-build catch-up (`get_proposal_head`) runs outside any +/// snapshot window, so a head move never surfaces before the block it +/// belongs to exists; the block's own import then emits the final head. +struct ChainEventSnapshot { + head: H256, + justified: Checkpoint, + finalized: Checkpoint, +} + +impl ChainEventSnapshot { + fn capture(store: &Store) -> Self { + Self { + head: store.head().expect("head block exists"), + justified: store + .latest_justified() + .expect("latest justified checkpoint exists"), + finalized: store + .latest_finalized() + .expect("latest finalized checkpoint exists"), + } + } + + /// Emit one event per value that changed since the snapshot, in a fixed + /// order: `justified_checkpoint` → `head` → `finalized_checkpoint`. + /// (`block` is emitted separately by the import path, ahead of this diff.) + /// + /// `wall_clock_slot` is the caller's current slot, used only to gate the + /// `head` event against [`HEAD_EVENT_RECENCY_SLOTS`]; the other events are + /// ungated. + fn diff_and_emit(&self, store: &Store, events: &EventBus, wall_clock_slot: u64) { + let justified = store + .latest_justified() + .expect("latest justified checkpoint exists"); + if justified != self.justified { + events.emit(ChainEvent::JustifiedCheckpoint { + slot: justified.slot, + root: justified.root, + }); + } + + let head = store.head().expect("head block exists"); + if head != self.head { + // Read the header once and reuse it for slot and parent_root so + // they stay consistent. + if let Some(header) = store + .get_block_header(&head) + .expect("block header read should succeed") + { + // Skip stale heads (catch-up/backfill): see HEAD_EVENT_RECENCY_SLOTS. + if header.slot + HEAD_EVENT_RECENCY_SLOTS >= wall_clock_slot { + events.emit(ChainEvent::Head { + slot: header.slot, + root: head, + parent_root: header.parent_root, + }); + } + } else { + warn!( + head_root = %ShortRoot(&head.0), + "Head header missing while emitting head event; skipping" + ); + } + } + + let finalized = store + .latest_finalized() + .expect("latest finalized checkpoint exists"); + if finalized != self.finalized { + events.emit(ChainEvent::FinalizedCheckpoint { + slot: finalized.slot, + root: finalized.root, + }); + } + } } impl BlockChainServer { @@ -330,8 +439,14 @@ impl BlockChainServer { .flatten() .is_some(); - // Tick the store first - this accepts attestations at interval 0 if we have a proposal + // Tick the store first - this accepts attestations at interval 0 if we have a proposal. + // Snapshot/diff around the call so attestation-driven head or + // finalization moves surface as chain events. + let pre_tick = ChainEventSnapshot::capture(&self.store); store::on_tick(&mut self.store, timestamp_ms, is_proposer); + // `slot` above is already derived from `timestamp_ms` (the wall clock + // at tick time), so it doubles as the wall-clock slot for the gate. + pre_tick.diff_and_emit(&self.store, &self.events, slot); // Per-interval duties for this tick. Intervals 0 (block publish) and 3 // (safe-target update) are driven inside `store::on_tick` above, so they @@ -855,9 +970,34 @@ impl BlockChainServer { info!(%slot, %validator_id, "Published block"); } - /// Run block import and refresh metrics. + /// Run block import, emit the resulting chain events, and refresh metrics. fn process_block(&mut self, signed_block: SignedBlock) -> Result<(), StoreError> { + // `on_block` returns Ok early for an already-imported block, so gate + // the `block` event on whether this root is actually new. + let slot = signed_block.message.slot; + let block_root = signed_block.message.hash_tree_root(); + let is_new = !self + .store + .has_state(&block_root) + .expect("DB read should succeed"); + let pre_import = ChainEventSnapshot::capture(&self.store); + store::on_block(&mut self.store, signed_block)?; + + // `block` goes out first so subscribers see it ahead of the + // justified/head/finalized moves its import triggers. + if is_new { + self.events.emit(ChainEvent::Block { + slot, + root: block_root, + }); + } + // Block import has no ready-made "now" slot like `on_tick`'s, so + // compute the wall-clock slot fresh for the head-recency gate. + let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000; + let wall_clock_slot = unix_now_ms().saturating_sub(genesis_time_ms) / MILLISECONDS_PER_SLOT; + pre_import.diff_and_emit(&self.store, &self.events, wall_clock_slot); + metrics::update_head_slot(self.store.head_slot()); let latest_justified_slot = self .store @@ -1360,3 +1500,266 @@ impl Handler for BlockChainServer { } } } + +#[cfg(test)] +mod tests { + use super::*; + use ethlambda_storage::{ForkCheckpoints, backend::InMemoryBackend}; + use ethlambda_types::{ + block::{Block, BlockBody}, + state::State, + }; + use std::sync::Arc; + use tokio::sync::broadcast::error::TryRecvError; + + fn test_store() -> Store { + let genesis_state = State::from_genesis(1000, vec![]); + Store::from_anchor_state(Arc::new(InMemoryBackend::new()), genesis_state) + } + + /// Insert a header-only block at `root` so head-header reads resolve. + fn insert_test_block(store: &mut Store, root: H256, slot: u64, parent_root: H256) { + let signed_block = SignedBlock { + message: Block { + slot, + proposer_index: 0, + parent_root, + state_root: H256::ZERO, + body: BlockBody::default(), + }, + proof: MultiMessageAggregate::default(), + }; + store + .insert_signed_block(root, signed_block) + .expect("insert test block should succeed"); + } + + #[test] + fn chain_event_diff_emits_nothing_when_unchanged() { + let store = test_store(); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + snapshot.diff_and_emit(&store, &bus, 0); + + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A store call that moved everything yields exactly one event per value, + /// in the fixed justified → head → finalized order. + #[test] + fn chain_event_diff_emits_moves_in_order() { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, 1, genesis); + let checkpoint = Checkpoint { + root: new_root, + slot: 1, + }; + store + .update_checkpoints(ForkCheckpoints::new( + new_root, + Some(checkpoint), + Some(checkpoint), + )) + .expect("update_checkpoints should succeed"); + + // Wall-clock slot equal to the new head's slot: well within the + // recency window, so the head event fires normally. + snapshot.diff_and_emit(&store, &bus, 1); + + match rx.try_recv().unwrap() { + ChainEvent::JustifiedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected justified_checkpoint first, got: {other:?}"), + } + match rx.try_recv().unwrap() { + ChainEvent::Head { + slot, + root, + parent_root, + } => { + assert_eq!((slot, root, parent_root), (1, new_root, genesis)); + } + other => panic!("expected head second, got: {other:?}"), + } + match rx.try_recv().unwrap() { + ChainEvent::FinalizedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected finalized_checkpoint last, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A head whose header cannot be read is skipped (warn), while checkpoint + /// moves still emit. + #[test] + fn chain_event_diff_skips_head_with_missing_header() { + let mut store = test_store(); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + + // Point the head at a root with no stored header; advance finalized to + // a real block so its event still fires. + let orphan_head = H256([7u8; 32]); + let genesis = store.head().expect("store head exists"); + let finalized_root = H256([8u8; 32]); + insert_test_block(&mut store, finalized_root, 1, genesis); + let finalized = Checkpoint { + root: finalized_root, + slot: 1, + }; + store + .update_checkpoints(ForkCheckpoints::new(orphan_head, None, Some(finalized))) + .expect("update_checkpoints should succeed"); + + snapshot.diff_and_emit(&store, &bus, 1); + + match rx.try_recv().unwrap() { + ChainEvent::FinalizedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, finalized_root)); + } + other => panic!("expected finalized_checkpoint only, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A head far below the wall-clock slot (catch-up/backfill) emits no + /// `head` event, but `justified_checkpoint`/`finalized_checkpoint` still + /// fire since only `head` is gated. + #[test] + fn chain_event_diff_gates_stale_head() { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, 1, genesis); + let checkpoint = Checkpoint { + root: new_root, + slot: 1, + }; + store + .update_checkpoints(ForkCheckpoints::new( + new_root, + Some(checkpoint), + Some(checkpoint), + )) + .expect("update_checkpoints should succeed"); + + // Wall clock far ahead of the new head's slot (1): well past + // HEAD_EVENT_RECENCY_SLOTS, so the head event must be suppressed. + let wall_clock_slot = 1 + HEAD_EVENT_RECENCY_SLOTS + 100; + snapshot.diff_and_emit(&store, &bus, wall_clock_slot); + + match rx.try_recv().unwrap() { + ChainEvent::JustifiedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected justified_checkpoint first, got: {other:?}"), + } + match rx.try_recv().unwrap() { + ChainEvent::FinalizedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected finalized_checkpoint second (head gated), got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A head within the recency window emits normally, mirroring + /// `chain_event_diff_emits_moves_in_order` but stated explicitly against + /// the gate for clarity. + #[test] + fn chain_event_diff_emits_recent_head() { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, 1, genesis); + store + .update_checkpoints(ForkCheckpoints::head_only(new_root)) + .expect("update_checkpoints should succeed"); + + // Wall clock equal to the head's own slot: as recent as it gets. + snapshot.diff_and_emit(&store, &bus, 1); + + match rx.try_recv().unwrap() { + ChainEvent::Head { slot, root, .. } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected head event, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// Boundary case: `new_head_slot + HEAD_EVENT_RECENCY_SLOTS >= + /// wall_clock_slot` is the emit condition, so equality still emits, while + /// one slot further pushes it over into "gated". + #[test] + fn chain_event_diff_head_recency_boundary() { + let head_slot = 1; + + // Exactly at the boundary: emits (`>=` includes equality). + { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, head_slot, genesis); + store + .update_checkpoints(ForkCheckpoints::head_only(new_root)) + .expect("update_checkpoints should succeed"); + + let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS; + snapshot.diff_and_emit(&store, &bus, wall_clock_slot); + + match rx.try_recv().unwrap() { + ChainEvent::Head { slot, .. } => assert_eq!(slot, head_slot), + other => panic!("expected head event at the boundary, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + // One slot past the boundary: gated. + { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, head_slot, genesis); + store + .update_checkpoints(ForkCheckpoints::head_only(new_root)) + .expect("update_checkpoints should succeed"); + + let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS + 1; + snapshot.diff_and_emit(&store, &bus, wall_clock_slot); + + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + } +} From 03118fcc9bf96561693edbe8e559f461909b487d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 15 Jul 2026 10:47:04 -0300 Subject: [PATCH 2/9] 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. --- Cargo.lock | 2 + bin/ethlambda/src/main.rs | 16 +++-- crates/net/rpc/Cargo.toml | 2 + crates/net/rpc/src/events.rs | 114 +++++++++++++++++++++++++++++++++++ crates/net/rpc/src/lib.rs | 8 ++- docs/rpc.md | 22 +++++++ 6 files changed, 157 insertions(+), 7 deletions(-) create mode 100644 crates/net/rpc/src/events.rs diff --git a/Cargo.lock b/Cargo.lock index c57423fa..dcd004b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2135,6 +2135,7 @@ dependencies = [ "ethlambda-storage", "ethlambda-test-fixtures", "ethlambda-types", + "futures-core", "hex", "http-body-util", "jemalloc_pprof", @@ -2142,6 +2143,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-stream", "tokio-util", "tower", "tracing", diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 643240df..b4d2828d 100644 --- a/bin/ethlambda/src/main.rs +++ b/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 diff --git a/crates/net/rpc/Cargo.toml b/crates/net/rpc/Cargo.toml index bebd3218..cee21b9d 100644 --- a/crates/net/rpc/Cargo.toml +++ b/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 diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs new file mode 100644 index 00000000..99995439 --- /dev/null +++ b/crates/net/rpc/src/events.rs @@ -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, +) -> Sse>> { + 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 { + 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}" + ); + } +} diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs index 8ea2ade7..217543dd 100644 --- a/crates/net/rpc/src/lib.rs +++ b/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)) diff --git a/docs/rpc.md b/docs/rpc.md index 2ec8c44c..209a28f9 100644 --- a/docs/rpc.md +++ b/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,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 within `HEAD_EVENT_RECENCY_SLOTS` (32 slots) of the wall clock; no head events fire during catch-up | +| `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. 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` | From d9b4c37c27750dd72d2470268ed95c0c3cfbb616 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:18:05 -0300 Subject: [PATCH 3/9] refactor(blockchain): emit head before justified_checkpoint in diff_and_emit Reorder chain-event emission so the head event fires before justified_checkpoint, giving subscribers the head move ahead of the checkpoint updates that follow it. Drop the ordering assertion test, which only pinned the old sequence. --- crates/blockchain/src/lib.rs | 76 ++++++------------------------------ 1 file changed, 11 insertions(+), 65 deletions(-) diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 24db4bc6..b746a9de 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -313,23 +313,13 @@ impl ChainEventSnapshot { } /// Emit one event per value that changed since the snapshot, in a fixed - /// order: `justified_checkpoint` → `head` → `finalized_checkpoint`. + /// order: `head` → `justified_checkpoint` → `finalized_checkpoint`. /// (`block` is emitted separately by the import path, ahead of this diff.) /// /// `wall_clock_slot` is the caller's current slot, used only to gate the /// `head` event against [`HEAD_EVENT_RECENCY_SLOTS`]; the other events are /// ungated. fn diff_and_emit(&self, store: &Store, events: &EventBus, wall_clock_slot: u64) { - let justified = store - .latest_justified() - .expect("latest justified checkpoint exists"); - if justified != self.justified { - events.emit(ChainEvent::JustifiedCheckpoint { - slot: justified.slot, - root: justified.root, - }); - } - let head = store.head().expect("head block exists"); if head != self.head { // Read the header once and reuse it for slot and parent_root so @@ -354,6 +344,16 @@ impl ChainEventSnapshot { } } + let justified = store + .latest_justified() + .expect("latest justified checkpoint exists"); + if justified != self.justified { + events.emit(ChainEvent::JustifiedCheckpoint { + slot: justified.slot, + root: justified.root, + }); + } + let finalized = store .latest_finalized() .expect("latest finalized checkpoint exists"); @@ -1546,60 +1546,6 @@ mod tests { assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); } - /// A store call that moved everything yields exactly one event per value, - /// in the fixed justified → head → finalized order. - #[test] - fn chain_event_diff_emits_moves_in_order() { - let mut store = test_store(); - let genesis = store.head().expect("store head exists"); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - - let snapshot = ChainEventSnapshot::capture(&store); - - let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, 1, genesis); - let checkpoint = Checkpoint { - root: new_root, - slot: 1, - }; - store - .update_checkpoints(ForkCheckpoints::new( - new_root, - Some(checkpoint), - Some(checkpoint), - )) - .expect("update_checkpoints should succeed"); - - // Wall-clock slot equal to the new head's slot: well within the - // recency window, so the head event fires normally. - snapshot.diff_and_emit(&store, &bus, 1); - - match rx.try_recv().unwrap() { - ChainEvent::JustifiedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, new_root)); - } - other => panic!("expected justified_checkpoint first, got: {other:?}"), - } - match rx.try_recv().unwrap() { - ChainEvent::Head { - slot, - root, - parent_root, - } => { - assert_eq!((slot, root, parent_root), (1, new_root, genesis)); - } - other => panic!("expected head second, got: {other:?}"), - } - match rx.try_recv().unwrap() { - ChainEvent::FinalizedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, new_root)); - } - other => panic!("expected finalized_checkpoint last, got: {other:?}"), - } - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - /// A head whose header cannot be read is skipped (warn), while checkpoint /// moves still emit. #[test] From 4b7a2aed8b75c7fe664d0acc022e3086998bc978 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:25:21 -0300 Subject: [PATCH 4/9] refactor(blockchain): move ChainEventSnapshot into events module Colocate the pre-call snapshot type (and HEAD_EVENT_RECENCY_SLOTS) with the ChainEvent/EventBus definitions it depends on, keeping all chain-event machinery in events.rs. The actor references it via crate::events; the snapshot's diff/gate tests move alongside it. No behavior change. --- crates/blockchain/src/events.rs | 302 +++++++++++++++++++++++++++++++ crates/blockchain/src/lib.rs | 306 +------------------------------- 2 files changed, 303 insertions(+), 305 deletions(-) diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 7a7c0dff..e74221bd 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -9,9 +9,13 @@ //! a slow subscriber loses events (the bounded broadcast channel overwrites //! its backlog) rather than back-pressuring consensus. +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; /// Wire-visible topic names for chain events. /// @@ -125,9 +129,111 @@ impl Default for EventBus { } } +/// Suppresses `head` events whose slot has fallen too far behind the wall +/// clock: during startup catch-up or backfill, fork choice can walk through +/// many historical heads on its way to the tip, and none of those are +/// interesting to a live subscriber. Consumers that need to track sync +/// progress should watch `block` events instead, which are never gated on +/// recency. Mirrors Lighthouse's recency filter on its head SSE event +/// (`EARLY_ATTESTER_CACHE_HISTORIC_SLOTS`), but with a wider window since a +/// lagging head is far more common here during multi-slot catch-up ticks. +const HEAD_EVENT_RECENCY_SLOTS: u64 = 32; + +/// Pre-call snapshot of the store values the chain-event bus reports on. +/// +/// The actor — not the store — publishes chain events: it captures this +/// snapshot before a store call (`store::on_tick`, `store::on_block`) and +/// diffs the store against it afterwards, so `store.rs` needs no event +/// plumbing. Consequences of diffing at this level, both intentional: +/// +/// - Multiple head moves within one store call coalesce into a single `head` +/// event; subscribers only care about the latest. +/// - The proposer's pre-build catch-up (`get_proposal_head`) runs outside any +/// snapshot window, so a head move never surfaces before the block it +/// belongs to exists; the block's own import then emits the final head. +pub(crate) struct ChainEventSnapshot { + head: H256, + justified: Checkpoint, + finalized: Checkpoint, +} + +impl ChainEventSnapshot { + pub(crate) fn capture(store: &Store) -> Self { + Self { + head: store.head().expect("head block exists"), + justified: store + .latest_justified() + .expect("latest justified checkpoint exists"), + finalized: store + .latest_finalized() + .expect("latest finalized checkpoint exists"), + } + } + + /// Emit one event per value that changed since the snapshot, in a fixed + /// order: `head` → `justified_checkpoint` → `finalized_checkpoint`. + /// (`block` is emitted separately by the import path, ahead of this diff.) + /// + /// `wall_clock_slot` is the caller's current slot, used only to gate the + /// `head` event against [`HEAD_EVENT_RECENCY_SLOTS`]; the other events are + /// ungated. + pub(crate) fn diff_and_emit(&self, store: &Store, events: &EventBus, wall_clock_slot: u64) { + let head = store.head().expect("head block exists"); + if head != self.head { + // Read the header once and reuse it for slot and parent_root so + // they stay consistent. + if let Some(header) = store + .get_block_header(&head) + .expect("block header read should succeed") + { + // Skip stale heads (catch-up/backfill): see HEAD_EVENT_RECENCY_SLOTS. + if header.slot + HEAD_EVENT_RECENCY_SLOTS >= wall_clock_slot { + events.emit(ChainEvent::Head { + slot: header.slot, + root: head, + parent_root: header.parent_root, + }); + } + } else { + warn!( + head_root = %ShortRoot(&head.0), + "Head header missing while emitting head event; skipping" + ); + } + } + + let justified = store + .latest_justified() + .expect("latest justified checkpoint exists"); + if justified != self.justified { + events.emit(ChainEvent::JustifiedCheckpoint { + slot: justified.slot, + root: justified.root, + }); + } + + let finalized = store + .latest_finalized() + .expect("latest finalized checkpoint exists"); + if finalized != self.finalized { + events.emit(ChainEvent::FinalizedCheckpoint { + slot: finalized.slot, + root: finalized.root, + }); + } + } +} + #[cfg(test)] mod tests { use super::*; + use ethlambda_storage::{ForkCheckpoints, backend::InMemoryBackend}; + use ethlambda_types::{ + block::{Block, BlockBody, MultiMessageAggregate, SignedBlock}, + state::State, + }; + use std::sync::Arc; + use tokio::sync::broadcast::error::TryRecvError; fn head_event(slot: u64) -> ChainEvent { ChainEvent::Head { @@ -201,4 +307,200 @@ mod tests { assert!(json.get("event").is_none()); assert!(json.get("data").is_none()); } + + fn test_store() -> Store { + let genesis_state = State::from_genesis(1000, vec![]); + Store::from_anchor_state(Arc::new(InMemoryBackend::new()), genesis_state) + } + + /// Insert a header-only block at `root` so head-header reads resolve. + fn insert_test_block(store: &mut Store, root: H256, slot: u64, parent_root: H256) { + let signed_block = SignedBlock { + message: Block { + slot, + proposer_index: 0, + parent_root, + state_root: H256::ZERO, + body: BlockBody::default(), + }, + proof: MultiMessageAggregate::default(), + }; + store + .insert_signed_block(root, signed_block) + .expect("insert test block should succeed"); + } + + #[test] + fn chain_event_diff_emits_nothing_when_unchanged() { + let store = test_store(); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + snapshot.diff_and_emit(&store, &bus, 0); + + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A head whose header cannot be read is skipped (warn), while checkpoint + /// moves still emit. + #[test] + fn chain_event_diff_skips_head_with_missing_header() { + let mut store = test_store(); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + + // Point the head at a root with no stored header; advance finalized to + // a real block so its event still fires. + let orphan_head = H256([7u8; 32]); + let genesis = store.head().expect("store head exists"); + let finalized_root = H256([8u8; 32]); + insert_test_block(&mut store, finalized_root, 1, genesis); + let finalized = Checkpoint { + root: finalized_root, + slot: 1, + }; + store + .update_checkpoints(ForkCheckpoints::new(orphan_head, None, Some(finalized))) + .expect("update_checkpoints should succeed"); + + snapshot.diff_and_emit(&store, &bus, 1); + + match rx.try_recv().unwrap() { + ChainEvent::FinalizedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, finalized_root)); + } + other => panic!("expected finalized_checkpoint only, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A head far below the wall-clock slot (catch-up/backfill) emits no + /// `head` event, but `justified_checkpoint`/`finalized_checkpoint` still + /// fire since only `head` is gated. + #[test] + fn chain_event_diff_gates_stale_head() { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, 1, genesis); + let checkpoint = Checkpoint { + root: new_root, + slot: 1, + }; + store + .update_checkpoints(ForkCheckpoints::new( + new_root, + Some(checkpoint), + Some(checkpoint), + )) + .expect("update_checkpoints should succeed"); + + // Wall clock far ahead of the new head's slot (1): well past + // HEAD_EVENT_RECENCY_SLOTS, so the head event must be suppressed. + let wall_clock_slot = 1 + HEAD_EVENT_RECENCY_SLOTS + 100; + snapshot.diff_and_emit(&store, &bus, wall_clock_slot); + + match rx.try_recv().unwrap() { + ChainEvent::JustifiedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected justified_checkpoint first, got: {other:?}"), + } + match rx.try_recv().unwrap() { + ChainEvent::FinalizedCheckpoint { slot, root } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected finalized_checkpoint second (head gated), got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A head within the recency window emits normally, stated explicitly + /// against the gate for clarity. + #[test] + fn chain_event_diff_emits_recent_head() { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, 1, genesis); + store + .update_checkpoints(ForkCheckpoints::head_only(new_root)) + .expect("update_checkpoints should succeed"); + + // Wall clock equal to the head's own slot: as recent as it gets. + snapshot.diff_and_emit(&store, &bus, 1); + + match rx.try_recv().unwrap() { + ChainEvent::Head { slot, root, .. } => { + assert_eq!((slot, root), (1, new_root)); + } + other => panic!("expected head event, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// Boundary case: `new_head_slot + HEAD_EVENT_RECENCY_SLOTS >= + /// wall_clock_slot` is the emit condition, so equality still emits, while + /// one slot further pushes it over into "gated". + #[test] + fn chain_event_diff_head_recency_boundary() { + let head_slot = 1; + + // Exactly at the boundary: emits (`>=` includes equality). + { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, head_slot, genesis); + store + .update_checkpoints(ForkCheckpoints::head_only(new_root)) + .expect("update_checkpoints should succeed"); + + let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS; + snapshot.diff_and_emit(&store, &bus, wall_clock_slot); + + match rx.try_recv().unwrap() { + ChainEvent::Head { slot, .. } => assert_eq!(slot, head_slot), + other => panic!("expected head event at the boundary, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + // One slot past the boundary: gated. + { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + let snapshot = ChainEventSnapshot::capture(&store); + + let new_root = H256([9u8; 32]); + insert_test_block(&mut store, new_root, head_slot, genesis); + store + .update_checkpoints(ForkCheckpoints::head_only(new_root)) + .expect("update_checkpoints should succeed"); + + let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS + 1; + snapshot.diff_and_emit(&store, &bus, wall_clock_slot); + + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + } } diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index b746a9de..8fb170aa 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -9,7 +9,6 @@ use ethlambda_types::{ aggregator::AggregatorController, attestation::{SignedAggregatedAttestation, SignedAttestation}, block::{ByteList512KiB, MultiMessageAggregate, SignedBlock}, - checkpoint::Checkpoint, primitives::{H256, HashTreeRoot as _}, signature::{ValidatorPublicKey, ValidatorSignature}, }; @@ -29,6 +28,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; use crate::block_builder::ProposerConfig; +use crate::events::ChainEventSnapshot; use crate::store::StoreError; pub use events::{ChainEvent, EventBus, Topic}; @@ -271,101 +271,6 @@ pub struct BlockChainServer { events: EventBus, } -/// Suppresses `head` events whose slot has fallen too far behind the wall -/// clock: during startup catch-up or backfill, fork choice can walk through -/// many historical heads on its way to the tip, and none of those are -/// interesting to a live subscriber. Consumers that need to track sync -/// progress should watch `block` events instead, which are never gated on -/// recency. Mirrors Lighthouse's recency filter on its head SSE event -/// (`EARLY_ATTESTER_CACHE_HISTORIC_SLOTS`), but with a wider window since a -/// lagging head is far more common here during multi-slot catch-up ticks. -const HEAD_EVENT_RECENCY_SLOTS: u64 = 32; - -/// Pre-call snapshot of the store values the chain-event bus reports on. -/// -/// The actor — not the store — publishes chain events: it captures this -/// snapshot before a store call (`store::on_tick`, `store::on_block`) and -/// diffs the store against it afterwards, so `store.rs` needs no event -/// plumbing. Consequences of diffing at this level, both intentional: -/// -/// - Multiple head moves within one store call coalesce into a single `head` -/// event; subscribers only care about the latest. -/// - The proposer's pre-build catch-up (`get_proposal_head`) runs outside any -/// snapshot window, so a head move never surfaces before the block it -/// belongs to exists; the block's own import then emits the final head. -struct ChainEventSnapshot { - head: H256, - justified: Checkpoint, - finalized: Checkpoint, -} - -impl ChainEventSnapshot { - fn capture(store: &Store) -> Self { - Self { - head: store.head().expect("head block exists"), - justified: store - .latest_justified() - .expect("latest justified checkpoint exists"), - finalized: store - .latest_finalized() - .expect("latest finalized checkpoint exists"), - } - } - - /// Emit one event per value that changed since the snapshot, in a fixed - /// order: `head` → `justified_checkpoint` → `finalized_checkpoint`. - /// (`block` is emitted separately by the import path, ahead of this diff.) - /// - /// `wall_clock_slot` is the caller's current slot, used only to gate the - /// `head` event against [`HEAD_EVENT_RECENCY_SLOTS`]; the other events are - /// ungated. - fn diff_and_emit(&self, store: &Store, events: &EventBus, wall_clock_slot: u64) { - let head = store.head().expect("head block exists"); - if head != self.head { - // Read the header once and reuse it for slot and parent_root so - // they stay consistent. - if let Some(header) = store - .get_block_header(&head) - .expect("block header read should succeed") - { - // Skip stale heads (catch-up/backfill): see HEAD_EVENT_RECENCY_SLOTS. - if header.slot + HEAD_EVENT_RECENCY_SLOTS >= wall_clock_slot { - events.emit(ChainEvent::Head { - slot: header.slot, - root: head, - parent_root: header.parent_root, - }); - } - } else { - warn!( - head_root = %ShortRoot(&head.0), - "Head header missing while emitting head event; skipping" - ); - } - } - - let justified = store - .latest_justified() - .expect("latest justified checkpoint exists"); - if justified != self.justified { - events.emit(ChainEvent::JustifiedCheckpoint { - slot: justified.slot, - root: justified.root, - }); - } - - let finalized = store - .latest_finalized() - .expect("latest finalized checkpoint exists"); - if finalized != self.finalized { - events.emit(ChainEvent::FinalizedCheckpoint { - slot: finalized.slot, - root: finalized.root, - }); - } - } -} - impl BlockChainServer { async fn on_tick(&mut self, timestamp_ms: u64, ctx: &Context) { let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000; @@ -1500,212 +1405,3 @@ impl Handler for BlockChainServer { } } } - -#[cfg(test)] -mod tests { - use super::*; - use ethlambda_storage::{ForkCheckpoints, backend::InMemoryBackend}; - use ethlambda_types::{ - block::{Block, BlockBody}, - state::State, - }; - use std::sync::Arc; - use tokio::sync::broadcast::error::TryRecvError; - - fn test_store() -> Store { - let genesis_state = State::from_genesis(1000, vec![]); - Store::from_anchor_state(Arc::new(InMemoryBackend::new()), genesis_state) - } - - /// Insert a header-only block at `root` so head-header reads resolve. - fn insert_test_block(store: &mut Store, root: H256, slot: u64, parent_root: H256) { - let signed_block = SignedBlock { - message: Block { - slot, - proposer_index: 0, - parent_root, - state_root: H256::ZERO, - body: BlockBody::default(), - }, - proof: MultiMessageAggregate::default(), - }; - store - .insert_signed_block(root, signed_block) - .expect("insert test block should succeed"); - } - - #[test] - fn chain_event_diff_emits_nothing_when_unchanged() { - let store = test_store(); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - - let snapshot = ChainEventSnapshot::capture(&store); - snapshot.diff_and_emit(&store, &bus, 0); - - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - - /// A head whose header cannot be read is skipped (warn), while checkpoint - /// moves still emit. - #[test] - fn chain_event_diff_skips_head_with_missing_header() { - let mut store = test_store(); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - - let snapshot = ChainEventSnapshot::capture(&store); - - // Point the head at a root with no stored header; advance finalized to - // a real block so its event still fires. - let orphan_head = H256([7u8; 32]); - let genesis = store.head().expect("store head exists"); - let finalized_root = H256([8u8; 32]); - insert_test_block(&mut store, finalized_root, 1, genesis); - let finalized = Checkpoint { - root: finalized_root, - slot: 1, - }; - store - .update_checkpoints(ForkCheckpoints::new(orphan_head, None, Some(finalized))) - .expect("update_checkpoints should succeed"); - - snapshot.diff_and_emit(&store, &bus, 1); - - match rx.try_recv().unwrap() { - ChainEvent::FinalizedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, finalized_root)); - } - other => panic!("expected finalized_checkpoint only, got: {other:?}"), - } - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - - /// A head far below the wall-clock slot (catch-up/backfill) emits no - /// `head` event, but `justified_checkpoint`/`finalized_checkpoint` still - /// fire since only `head` is gated. - #[test] - fn chain_event_diff_gates_stale_head() { - let mut store = test_store(); - let genesis = store.head().expect("store head exists"); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - - let snapshot = ChainEventSnapshot::capture(&store); - - let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, 1, genesis); - let checkpoint = Checkpoint { - root: new_root, - slot: 1, - }; - store - .update_checkpoints(ForkCheckpoints::new( - new_root, - Some(checkpoint), - Some(checkpoint), - )) - .expect("update_checkpoints should succeed"); - - // Wall clock far ahead of the new head's slot (1): well past - // HEAD_EVENT_RECENCY_SLOTS, so the head event must be suppressed. - let wall_clock_slot = 1 + HEAD_EVENT_RECENCY_SLOTS + 100; - snapshot.diff_and_emit(&store, &bus, wall_clock_slot); - - match rx.try_recv().unwrap() { - ChainEvent::JustifiedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, new_root)); - } - other => panic!("expected justified_checkpoint first, got: {other:?}"), - } - match rx.try_recv().unwrap() { - ChainEvent::FinalizedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, new_root)); - } - other => panic!("expected finalized_checkpoint second (head gated), got: {other:?}"), - } - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - - /// A head within the recency window emits normally, mirroring - /// `chain_event_diff_emits_moves_in_order` but stated explicitly against - /// the gate for clarity. - #[test] - fn chain_event_diff_emits_recent_head() { - let mut store = test_store(); - let genesis = store.head().expect("store head exists"); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - - let snapshot = ChainEventSnapshot::capture(&store); - - let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, 1, genesis); - store - .update_checkpoints(ForkCheckpoints::head_only(new_root)) - .expect("update_checkpoints should succeed"); - - // Wall clock equal to the head's own slot: as recent as it gets. - snapshot.diff_and_emit(&store, &bus, 1); - - match rx.try_recv().unwrap() { - ChainEvent::Head { slot, root, .. } => { - assert_eq!((slot, root), (1, new_root)); - } - other => panic!("expected head event, got: {other:?}"), - } - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - - /// Boundary case: `new_head_slot + HEAD_EVENT_RECENCY_SLOTS >= - /// wall_clock_slot` is the emit condition, so equality still emits, while - /// one slot further pushes it over into "gated". - #[test] - fn chain_event_diff_head_recency_boundary() { - let head_slot = 1; - - // Exactly at the boundary: emits (`>=` includes equality). - { - let mut store = test_store(); - let genesis = store.head().expect("store head exists"); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - let snapshot = ChainEventSnapshot::capture(&store); - - let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, head_slot, genesis); - store - .update_checkpoints(ForkCheckpoints::head_only(new_root)) - .expect("update_checkpoints should succeed"); - - let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS; - snapshot.diff_and_emit(&store, &bus, wall_clock_slot); - - match rx.try_recv().unwrap() { - ChainEvent::Head { slot, .. } => assert_eq!(slot, head_slot), - other => panic!("expected head event at the boundary, got: {other:?}"), - } - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - - // One slot past the boundary: gated. - { - let mut store = test_store(); - let genesis = store.head().expect("store head exists"); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - let snapshot = ChainEventSnapshot::capture(&store); - - let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, head_slot, genesis); - store - .update_checkpoints(ForkCheckpoints::head_only(new_root)) - .expect("update_checkpoints should succeed"); - - let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS + 1; - snapshot.diff_and_emit(&store, &bus, wall_clock_slot); - - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - } -} From 9cbe5b7273fc6d6c1521bdbe47faff619d3c8c7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:42:38 -0300 Subject: [PATCH 5/9] test(blockchain): drop chain_event_diff_head_recency_boundary The recency gate is already covered by chain_event_diff_gates_stale_head (gated) and chain_event_diff_emits_recent_head (within window); the explicit boundary case was redundant. --- crates/blockchain/src/events.rs | 52 --------------------------------- 1 file changed, 52 deletions(-) diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index e74221bd..4f07fb1b 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -451,56 +451,4 @@ mod tests { } assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); } - - /// Boundary case: `new_head_slot + HEAD_EVENT_RECENCY_SLOTS >= - /// wall_clock_slot` is the emit condition, so equality still emits, while - /// one slot further pushes it over into "gated". - #[test] - fn chain_event_diff_head_recency_boundary() { - let head_slot = 1; - - // Exactly at the boundary: emits (`>=` includes equality). - { - let mut store = test_store(); - let genesis = store.head().expect("store head exists"); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - let snapshot = ChainEventSnapshot::capture(&store); - - let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, head_slot, genesis); - store - .update_checkpoints(ForkCheckpoints::head_only(new_root)) - .expect("update_checkpoints should succeed"); - - let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS; - snapshot.diff_and_emit(&store, &bus, wall_clock_slot); - - match rx.try_recv().unwrap() { - ChainEvent::Head { slot, .. } => assert_eq!(slot, head_slot), - other => panic!("expected head event at the boundary, got: {other:?}"), - } - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - - // One slot past the boundary: gated. - { - let mut store = test_store(); - let genesis = store.head().expect("store head exists"); - let bus = EventBus::new(8); - let mut rx = bus.subscribe(); - let snapshot = ChainEventSnapshot::capture(&store); - - let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, head_slot, genesis); - store - .update_checkpoints(ForkCheckpoints::head_only(new_root)) - .expect("update_checkpoints should succeed"); - - let wall_clock_slot = head_slot + HEAD_EVENT_RECENCY_SLOTS + 1; - snapshot.diff_and_emit(&store, &bus, wall_clock_slot); - - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - } - } } From 8464e6a9353fa4592d319faa12f4d0b857fe5483 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:45:28 -0300 Subject: [PATCH 6/9] refactor(blockchain): drop unused EventBus::disabled Eventless call paths (spec tests, test_driver.rs) never construct an EventBus at all, so the dormant-bus constructor had no production callers; its only user was its own test. --- crates/blockchain/src/events.rs | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 4f07fb1b..33260288 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -99,11 +99,6 @@ impl EventBus { 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 @@ -263,16 +258,6 @@ mod tests { 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; From 55d1310cb213f003c6cc689c9bde1e6e355ddaff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 17:16:18 -0300 Subject: [PATCH 7/9] fix(blockchain): beacon-align chain events; emit proposer checkpoints Shape ChainEvent payloads after the Ethereum beacon-API eventstream (v4.0.0): `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`. Drop the `#[serde(untagged)]` Serialize derive; the wire encoding (decimal-string ints, 0x-hex roots, out-of-band topic) returns with the SSE endpoint in a follow-up PR. This also removes the ambiguous untagged shape flagged in review. Since serde is no longer used in the crate, drop the serde/serde_json deps. `justified_checkpoint` has no beacon analog and is kept as an ethlambda extension mirroring `finalized_checkpoint`. Also fix a checkpoint event dropped on the proposer's own slots: the interval-0 catch-up in `get_proposal_head` advances head/finalized outside every snapshot window, so a finalization moved there was folded into the later block-import diff's baseline and never emitted. `propose_block` now snapshots around the build and emits those moves on both build success and failure, matching what a non-proposing node emits at its interval-0 tick. --- Cargo.lock | 2 - crates/blockchain/Cargo.toml | 2 - crates/blockchain/src/events.rs | 174 ++++++++++++++++++++------------ crates/blockchain/src/lib.rs | 37 +++++-- 4 files changed, 134 insertions(+), 81 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c57423fa..f131b8b3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2051,8 +2051,6 @@ dependencies = [ "libssz", "libssz-types", "rayon", - "serde", - "serde_json", "spawned-concurrency 0.5.0", "thiserror 2.0.18", "tokio", diff --git a/crates/blockchain/Cargo.toml b/crates/blockchain/Cargo.toml index a08262eb..5f47b334 100644 --- a/crates/blockchain/Cargo.toml +++ b/crates/blockchain/Cargo.toml @@ -29,13 +29,11 @@ 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_json = { workspace = true } hex = { workspace = true } libssz.workspace = true libssz-types.workspace = true diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 33260288..bc1716c2 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -13,7 +13,6 @@ 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; @@ -21,7 +20,10 @@ use tracing::warn; /// /// 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. +/// payload stays flat with the topic travelling out-of-band. The names match +/// the Ethereum beacon-API eventstream topics (`head`, `block`, +/// `finalized_checkpoint`); `justified_checkpoint` is an ethlambda extension +/// with no beacon analog. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Topic { Head, @@ -43,24 +45,27 @@ impl Topic { /// 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)] +/// Fields mirror the Ethereum beacon-API eventstream payloads so a future SSE +/// endpoint can render 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)] pub enum ChainEvent { /// Fork choice selected a new head. - Head { - slot: u64, - root: H256, - parent_root: H256, - }, + Head { slot: u64, block: H256, state: H256 }, /// A block was imported into the store. - Block { slot: u64, root: H256 }, + Block { slot: u64, block: H256 }, /// The justified checkpoint advanced. - JustifiedCheckpoint { slot: u64, root: H256 }, + JustifiedCheckpoint { slot: u64, block: H256, state: H256 }, /// The finalized checkpoint advanced. - FinalizedCheckpoint { slot: u64, root: H256 }, + FinalizedCheckpoint { slot: u64, block: H256, state: H256 }, } impl ChainEvent { @@ -139,13 +144,16 @@ const HEAD_EVENT_RECENCY_SLOTS: u64 = 32; /// The actor — not the store — publishes chain events: it captures this /// snapshot before a store call (`store::on_tick`, `store::on_block`) and /// diffs the store against it afterwards, so `store.rs` needs no event -/// plumbing. Consequences of diffing at this level, both intentional: +/// plumbing. +/// +/// Multiple head moves within one store call coalesce into a single `head` +/// event; subscribers only care about the latest. /// -/// - Multiple head moves within one store call coalesce into a single `head` -/// event; subscribers only care about the latest. -/// - The proposer's pre-build catch-up (`get_proposal_head`) runs outside any -/// snapshot window, so a head move never surfaces before the block it -/// belongs to exists; the block's own import then emits the final head. +/// The proposer's pre-build catch-up (`get_proposal_head`) advances the store +/// too, so `propose_block` wraps that call in its own snapshot: the +/// head/justified/finalized moves it triggers surface exactly as they would on +/// a non-proposing node's interval-0 tick, rather than being silently folded +/// into the later block-import diff's baseline. pub(crate) struct ChainEventSnapshot { head: H256, justified: Checkpoint, @@ -175,8 +183,8 @@ impl ChainEventSnapshot { pub(crate) fn diff_and_emit(&self, store: &Store, events: &EventBus, wall_clock_slot: u64) { let head = store.head().expect("head block exists"); if head != self.head { - // Read the header once and reuse it for slot and parent_root so - // they stay consistent. + // Read the header once and reuse it for slot and state root so they + // stay consistent. if let Some(header) = store .get_block_header(&head) .expect("block header read should succeed") @@ -185,8 +193,8 @@ impl ChainEventSnapshot { if header.slot + HEAD_EVENT_RECENCY_SLOTS >= wall_clock_slot { events.emit(ChainEvent::Head { slot: header.slot, - root: head, - parent_root: header.parent_root, + block: head, + state: header.state_root, }); } } else { @@ -201,24 +209,51 @@ impl ChainEventSnapshot { .latest_justified() .expect("latest justified checkpoint exists"); if justified != self.justified { - events.emit(ChainEvent::JustifiedCheckpoint { - slot: justified.slot, - root: justified.root, - }); + if let Some(state) = checkpoint_state_root(store, justified.root) { + events.emit(ChainEvent::JustifiedCheckpoint { + slot: justified.slot, + block: justified.root, + state, + }); + } else { + warn!( + justified_root = %ShortRoot(&justified.root.0), + "Justified block header missing while emitting event; skipping" + ); + } } let finalized = store .latest_finalized() .expect("latest finalized checkpoint exists"); if finalized != self.finalized { - events.emit(ChainEvent::FinalizedCheckpoint { - slot: finalized.slot, - root: finalized.root, - }); + if let Some(state) = checkpoint_state_root(store, finalized.root) { + events.emit(ChainEvent::FinalizedCheckpoint { + slot: finalized.slot, + block: finalized.root, + state, + }); + } else { + warn!( + finalized_root = %ShortRoot(&finalized.root.0), + "Finalized block header missing while emitting event; skipping" + ); + } } } } +/// Look up the state root of a checkpoint's block for the `{block, state}` +/// event shape. Returns `None` if the header is absent so the caller can skip +/// emission; finalized/justified block headers are never pruned, so this only +/// fails on genuine store inconsistency. +fn checkpoint_state_root(store: &Store, root: H256) -> Option { + store + .get_block_header(&root) + .expect("block header read should succeed") + .map(|header| header.state_root) +} + #[cfg(test)] mod tests { use super::*; @@ -233,8 +268,8 @@ mod tests { fn head_event(slot: u64) -> ChainEvent { ChainEvent::Head { slot, - root: H256([1u8; 32]), - parent_root: H256([2u8; 32]), + block: H256([1u8; 32]), + state: H256([2u8; 32]), } } @@ -260,16 +295,25 @@ mod tests { #[test] fn topic_maps_every_variant() { - let root = H256::ZERO; + let block = H256::ZERO; + let state = H256::ZERO; let cases = [ (head_event(1), Topic::Head), - (ChainEvent::Block { slot: 1, root }, Topic::Block), + (ChainEvent::Block { slot: 1, block }, Topic::Block), ( - ChainEvent::JustifiedCheckpoint { slot: 1, root }, + ChainEvent::JustifiedCheckpoint { + slot: 1, + block, + state, + }, Topic::JustifiedCheckpoint, ), ( - ChainEvent::FinalizedCheckpoint { slot: 1, root }, + ChainEvent::FinalizedCheckpoint { + slot: 1, + block, + state, + }, Topic::FinalizedCheckpoint, ), ]; @@ -279,33 +323,26 @@ mod tests { } } - /// 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()); - } - fn test_store() -> Store { let genesis_state = State::from_genesis(1000, vec![]); Store::from_anchor_state(Arc::new(InMemoryBackend::new()), genesis_state) } - /// Insert a header-only block at `root` so head-header reads resolve. - fn insert_test_block(store: &mut Store, root: H256, slot: u64, parent_root: H256) { + /// Insert a header-only block at `root` so header reads (block root, slot, + /// state root) resolve for the event payloads. + fn insert_test_block( + store: &mut Store, + root: H256, + slot: u64, + parent_root: H256, + state_root: H256, + ) { let signed_block = SignedBlock { message: Block { slot, proposer_index: 0, parent_root, - state_root: H256::ZERO, + state_root, body: BlockBody::default(), }, proof: MultiMessageAggregate::default(), @@ -342,7 +379,8 @@ mod tests { let orphan_head = H256([7u8; 32]); let genesis = store.head().expect("store head exists"); let finalized_root = H256([8u8; 32]); - insert_test_block(&mut store, finalized_root, 1, genesis); + let finalized_state = H256([88u8; 32]); + insert_test_block(&mut store, finalized_root, 1, genesis, finalized_state); let finalized = Checkpoint { root: finalized_root, slot: 1, @@ -354,8 +392,8 @@ mod tests { snapshot.diff_and_emit(&store, &bus, 1); match rx.try_recv().unwrap() { - ChainEvent::FinalizedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, finalized_root)); + ChainEvent::FinalizedCheckpoint { slot, block, state } => { + assert_eq!((slot, block, state), (1, finalized_root, finalized_state)); } other => panic!("expected finalized_checkpoint only, got: {other:?}"), } @@ -375,7 +413,8 @@ mod tests { let snapshot = ChainEventSnapshot::capture(&store); let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, 1, genesis); + let new_state = H256([99u8; 32]); + insert_test_block(&mut store, new_root, 1, genesis, new_state); let checkpoint = Checkpoint { root: new_root, slot: 1, @@ -394,14 +433,14 @@ mod tests { snapshot.diff_and_emit(&store, &bus, wall_clock_slot); match rx.try_recv().unwrap() { - ChainEvent::JustifiedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, new_root)); + ChainEvent::JustifiedCheckpoint { slot, block, state } => { + assert_eq!((slot, block, state), (1, new_root, new_state)); } other => panic!("expected justified_checkpoint first, got: {other:?}"), } match rx.try_recv().unwrap() { - ChainEvent::FinalizedCheckpoint { slot, root } => { - assert_eq!((slot, root), (1, new_root)); + ChainEvent::FinalizedCheckpoint { slot, block, state } => { + assert_eq!((slot, block, state), (1, new_root, new_state)); } other => panic!("expected finalized_checkpoint second (head gated), got: {other:?}"), } @@ -420,7 +459,8 @@ mod tests { let snapshot = ChainEventSnapshot::capture(&store); let new_root = H256([9u8; 32]); - insert_test_block(&mut store, new_root, 1, genesis); + let new_state = H256([99u8; 32]); + insert_test_block(&mut store, new_root, 1, genesis, new_state); store .update_checkpoints(ForkCheckpoints::head_only(new_root)) .expect("update_checkpoints should succeed"); @@ -429,8 +469,8 @@ mod tests { snapshot.diff_and_emit(&store, &bus, 1); match rx.try_recv().unwrap() { - ChainEvent::Head { slot, root, .. } => { - assert_eq!((slot, root), (1, new_root)); + ChainEvent::Head { slot, block, state } => { + assert_eq!((slot, block, state), (1, new_root, new_state)); } other => panic!("expected head event, got: {other:?}"), } diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 8fb170aa..e7b82edb 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -702,16 +702,33 @@ impl BlockChainServer { // stashed for a later tick), and the real interval-0 tick is then skipped // by the idempotency guard in `on_tick`, since the store clock is already // here. + // + // That interval-0 catch-up can move head/justified/finalized (it is the + // same attestation-acceptance step a non-proposing node runs at its + // interval-0 tick). Snapshot around the build so those moves surface as + // chain events here, matching an observer node; otherwise they would + // land outside every snapshot window and be silently absorbed into the + // later block-import diff's baseline. + let pre_build = ChainEventSnapshot::capture(&self.store); let timing = metrics::time_block_building(); - let Ok((block, single_message_aggregates, _post_checkpoints)) = - store::produce_block_with_signatures( - &mut self.store, - slot, - validator_id, - self.proposer_config, - ) - .inspect_err(|err| error!(%slot, %validator_id, %err, "Failed to build block")) - else { + let build_result = store::produce_block_with_signatures( + &mut self.store, + slot, + validator_id, + self.proposer_config, + ) + .inspect_err(|err| error!(%slot, %validator_id, %err, "Failed to build block")); + + // `get_proposal_head` advances the store (interval-0 catch-up) inside + // `produce_block_with_signatures` *before* the build can fail, so emit + // the resulting head/checkpoint moves on both paths — a build failure + // must not strand a real finalization move outside every snapshot + // window. Ordered before the freshly built block's own import (which + // emits its `block` + head/checkpoint events). The catch-up advanced + // the store to `slot`'s interval 0, so the head-recency gate uses `slot`. + pre_build.diff_and_emit(&self.store, &self.events, slot); + + let Ok((block, single_message_aggregates, _post_checkpoints)) = build_result else { metrics::inc_block_building_failures(); return; }; @@ -894,7 +911,7 @@ impl BlockChainServer { if is_new { self.events.emit(ChainEvent::Block { slot, - root: block_root, + block: block_root, }); } // Block import has no ready-made "now" slot like `on_tick`'s, so From d83ea930abb0dbff9220d5469cb288d16515f500 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 20 Jul 2026 16:11:12 -0300 Subject: [PATCH 8/9] fix(rpc): adapt SSE events to beacon-aligned ChainEvent shape #516 reshaped ChainEvent (root/parent_root -> block/state) and dropped its Serialize derive; move the wire encoding into an RPC-crate DTO. --- crates/net/rpc/src/events.rs | 55 ++++++++++++++++++++++++++++++++---- docs/rpc.md | 12 ++++---- 2 files changed, 56 insertions(+), 11 deletions(-) diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs index 99995439..3cd6c6b5 100644 --- a/crates/net/rpc/src/events.rs +++ b/crates/net/rpc/src/events.rs @@ -17,14 +17,51 @@ use axum::{ response::{Sse, sse::Event}, routing::get, }; -use ethlambda_blockchain::EventBus; +use ethlambda_blockchain::{ChainEvent, EventBus}; use ethlambda_storage::Store; +use ethlambda_types::primitives::H256; use futures_core::Stream; +use serde::Serialize; use tokio_stream::{ StreamExt, wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}, }; +/// Wire form of a [`ChainEvent`] for the SSE `data:` line. +/// +/// [`ChainEvent`] is intentionally serialization-agnostic (see #516): the wire +/// encoding lives here in the RPC crate. The body is flat, with the topic name +/// carried out-of-band on the SSE `event:` line ([`ChainEvent::topic`]), so no +/// `event`/`data` wrapper keys ever appear inside the JSON. `slot` is a plain +/// JSON number (matching every other ethlambda RPC payload); roots are +/// `0x`-hex via [`H256`]'s serializer. +/// +/// `untagged` yields only the variant's fields; the two variants are +/// serialize-only, so their structural overlap is never a deserialization +/// hazard. +#[derive(Serialize)] +#[serde(untagged)] +enum ChainEventPayload { + /// `head` / `justified_checkpoint` / `finalized_checkpoint`: block root + /// plus state root. + WithState { slot: u64, block: H256, state: H256 }, + /// `block`: block root only. + BlockOnly { slot: u64, block: H256 }, +} + +impl From<&ChainEvent> for ChainEventPayload { + fn from(event: &ChainEvent) -> Self { + match *event { + ChainEvent::Head { slot, block, state } + | ChainEvent::JustifiedCheckpoint { slot, block, state } + | ChainEvent::FinalizedCheckpoint { slot, block, state } => { + Self::WithState { slot, block, state } + } + ChainEvent::Block { slot, block } => Self::BlockOnly { slot, block }, + } + } +} + async fn get_events( Extension(events): Extension, ) -> Sse>> { @@ -42,7 +79,7 @@ async fn get_events( }; Some(Ok(Event::default() .event(ev.topic().as_str()) - .json_data(&ev) + .json_data(ChainEventPayload::from(&ev)) .inspect_err(|err| tracing::warn!(%err, "Failed to serialize SSE chain event")) .ok()?)) }); @@ -84,8 +121,8 @@ mod tests { events.emit(ChainEvent::Head { slot: 3, - root: Default::default(), - parent_root: Default::default(), + block: Default::default(), + state: Default::default(), }); let mut body = resp.into_body().into_data_stream(); @@ -100,12 +137,18 @@ mod tests { 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). + // ...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}" diff --git a/docs/rpc.md b/docs/rpc.md index 209a28f9..baf50e2c 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -88,18 +88,20 @@ SSZ-encoded `SignedBlock` at the latest finalized checkpoint. The genesis/anchor 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, "root": "0x…", "parent_root": "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, "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 | +| `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,"root":"0x1a2b…","parent_root":"0x3c4d…"} +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. From acf7d5af56161e59d8be1877a0bfb7bf791fe0de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 20 Jul 2026 16:47:28 -0300 Subject: [PATCH 9/9] refactor(rpc): derive Serialize on ChainEvent instead of a DTO #516 deferred the chain-event wire encoding to this SSE PR. Derive it on ChainEvent directly (untagged, flat body) rather than mirroring the enum in an RPC-crate DTO, so the handler stays a direct json_data(&ev) with no parallel type to keep in sync. Wire shape is unchanged. --- Cargo.lock | 1 + crates/blockchain/Cargo.toml | 1 + crates/blockchain/src/events.rs | 19 ++++++++------- crates/net/rpc/src/events.rs | 41 ++------------------------------- 4 files changed, 15 insertions(+), 47 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 099d6227..413f049d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2051,6 +2051,7 @@ dependencies = [ "libssz", "libssz-types", "rayon", + "serde", "spawned-concurrency 0.5.0", "thiserror 2.0.18", "tokio", diff --git a/crates/blockchain/Cargo.toml b/crates/blockchain/Cargo.toml index 042a0b4d..342ae0b0 100644 --- a/crates/blockchain/Cargo.toml +++ b/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 diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index bc1716c2..3ff1e1b1 100644 --- a/crates/blockchain/src/events.rs +++ b/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 }, diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs index 3cd6c6b5..d398834d 100644 --- a/crates/net/rpc/src/events.rs +++ b/crates/net/rpc/src/events.rs @@ -17,51 +17,14 @@ use axum::{ response::{Sse, sse::Event}, routing::get, }; -use ethlambda_blockchain::{ChainEvent, EventBus}; +use ethlambda_blockchain::EventBus; use ethlambda_storage::Store; -use ethlambda_types::primitives::H256; use futures_core::Stream; -use serde::Serialize; use tokio_stream::{ StreamExt, wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}, }; -/// Wire form of a [`ChainEvent`] for the SSE `data:` line. -/// -/// [`ChainEvent`] is intentionally serialization-agnostic (see #516): the wire -/// encoding lives here in the RPC crate. The body is flat, with the topic name -/// carried out-of-band on the SSE `event:` line ([`ChainEvent::topic`]), so no -/// `event`/`data` wrapper keys ever appear inside the JSON. `slot` is a plain -/// JSON number (matching every other ethlambda RPC payload); roots are -/// `0x`-hex via [`H256`]'s serializer. -/// -/// `untagged` yields only the variant's fields; the two variants are -/// serialize-only, so their structural overlap is never a deserialization -/// hazard. -#[derive(Serialize)] -#[serde(untagged)] -enum ChainEventPayload { - /// `head` / `justified_checkpoint` / `finalized_checkpoint`: block root - /// plus state root. - WithState { slot: u64, block: H256, state: H256 }, - /// `block`: block root only. - BlockOnly { slot: u64, block: H256 }, -} - -impl From<&ChainEvent> for ChainEventPayload { - fn from(event: &ChainEvent) -> Self { - match *event { - ChainEvent::Head { slot, block, state } - | ChainEvent::JustifiedCheckpoint { slot, block, state } - | ChainEvent::FinalizedCheckpoint { slot, block, state } => { - Self::WithState { slot, block, state } - } - ChainEvent::Block { slot, block } => Self::BlockOnly { slot, block }, - } - } -} - async fn get_events( Extension(events): Extension, ) -> Sse>> { @@ -79,7 +42,7 @@ async fn get_events( }; Some(Ok(Event::default() .event(ev.topic().as_str()) - .json_data(ChainEventPayload::from(&ev)) + .json_data(&ev) .inspect_err(|err| tracing::warn!(%err, "Failed to serialize SSE chain event")) .ok()?)) });