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/6] 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 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 2/6] 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 3/6] 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 4/6] 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 5/6] 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 6/6] 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