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] 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))); + } + } +}