From 94fc5ebd767169798358751d9f8cda448d57443d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Tue, 21 Jul 2026 19:01:18 -0300 Subject: [PATCH] feat(events): add chain_reorg and safe_target chain events Extend the /lean/v0/events stream with two cheap follow-up topics derived from the actor-layer snapshot diff. chain_reorg: fork choice switched to a head off the old head's chain. Reuses store::reorg_depth (widened to pub(crate) so the actor diff and block processing share one definition of "reorg"); recency-gated like head and emitted just before it. Mirrors the beacon chain_reorg payload minus epoch/execution_optimistic. safe_target: the interval-3 fork-choice safe attestation target advanced. Ungated since it coalesces to the latest value, the only one that matters; an ethlambda extension with no beacon analog. --- crates/blockchain/src/events.rs | 181 ++++++++++++++++++++++++++++++-- crates/blockchain/src/store.rs | 5 +- docs/rpc.md | 8 +- 3 files changed, 180 insertions(+), 14 deletions(-) diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 66386ffd..bcbc2d55 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -22,16 +22,20 @@ 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 -/// 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. +/// payload stays flat with the topic travelling out-of-band. Names match the +/// Ethereum beacon-API eventstream topics where an analog exists (`head`, +/// `block`, `finalized_checkpoint`, `chain_reorg`); `justified_checkpoint` and +/// `safe_target` are ethlambda extensions with no direct beacon topic. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Topic { Head, Block, JustifiedCheckpoint, FinalizedCheckpoint, + /// Fork choice switched to a head that is not a descendant of the old one. + ChainReorg, + /// The interval-3 fork-choice safe attestation target advanced. + SafeTarget, } impl Topic { @@ -41,6 +45,8 @@ impl Topic { Topic::Block => "block", Topic::JustifiedCheckpoint => "justified_checkpoint", Topic::FinalizedCheckpoint => "finalized_checkpoint", + Topic::ChainReorg => "chain_reorg", + Topic::SafeTarget => "safe_target", } } } @@ -60,6 +66,8 @@ impl FromStr for Topic { "block" => Ok(Topic::Block), "justified_checkpoint" => Ok(Topic::JustifiedCheckpoint), "finalized_checkpoint" => Ok(Topic::FinalizedCheckpoint), + "chain_reorg" => Ok(Topic::ChainReorg), + "safe_target" => Ok(Topic::SafeTarget), other => Err(UnknownTopic(other.to_string())), } } @@ -90,6 +98,19 @@ pub enum ChainEvent { JustifiedCheckpoint { slot: u64, block: H256, state: H256 }, /// The finalized checkpoint advanced. FinalizedCheckpoint { slot: u64, block: H256, state: H256 }, + /// Fork choice switched to a head off the old head's chain. `slot` is the + /// new head's slot; `depth` the number of blocks rolled back. Mirrors the + /// beacon `chain_reorg` payload minus `epoch`/`execution_optimistic`. + ChainReorg { + slot: u64, + depth: u64, + old_head_block: H256, + old_head_state: H256, + new_head_block: H256, + new_head_state: H256, + }, + /// The interval-3 safe attestation target advanced (ethlambda-specific). + SafeTarget { slot: u64, block: H256 }, } impl ChainEvent { @@ -99,6 +120,8 @@ impl ChainEvent { ChainEvent::Block { .. } => Topic::Block, ChainEvent::JustifiedCheckpoint { .. } => Topic::JustifiedCheckpoint, ChainEvent::FinalizedCheckpoint { .. } => Topic::FinalizedCheckpoint, + ChainEvent::ChainReorg { .. } => Topic::ChainReorg, + ChainEvent::SafeTarget { .. } => Topic::SafeTarget, } } } @@ -182,6 +205,7 @@ pub(crate) struct ChainEventSnapshot { head: H256, justified: Checkpoint, finalized: Checkpoint, + safe_target: H256, } impl ChainEventSnapshot { @@ -194,16 +218,20 @@ impl ChainEventSnapshot { finalized: store .latest_finalized() .expect("latest finalized checkpoint exists"), + safe_target: store.safe_target().expect("safe target 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.) + /// order: `chain_reorg` → `head` → `justified_checkpoint` → + /// `finalized_checkpoint` → `safe_target`. (`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. + /// `wall_clock_slot` is the caller's current slot, used to gate the `head` + /// and `chain_reorg` events against [`HEAD_EVENT_RECENCY_SLOTS`] (catch-up + /// head walks are noise); `justified_checkpoint`, `finalized_checkpoint`, + /// and `safe_target` are ungated (they coalesce to the latest value, which + /// is the only one that matters). 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 { @@ -215,6 +243,29 @@ impl ChainEventSnapshot { { // Skip stale heads (catch-up/backfill): see HEAD_EVENT_RECENCY_SLOTS. if header.slot + HEAD_EVENT_RECENCY_SLOTS >= wall_clock_slot { + // A head change that leaves the old head's chain is a reorg; + // surface it just before the `head` event (beacon ordering). + // `reorg_depth` returns None for a plain extension. Reading + // the old-head header can fail only on genuine store + // inconsistency; that just drops the reorg detail, never the + // head event. Note: during a multi-move catch-up tick the + // diff sees only net (pre, post) heads, so a transient reorg + // that reverts within one call is invisible here (store + // metrics still count it). + if let Some(depth) = crate::store::reorg_depth(self.head, head, store) + && let Some(old_header) = store + .get_block_header(&self.head) + .expect("block header read should succeed") + { + events.emit(ChainEvent::ChainReorg { + slot: header.slot, + depth, + old_head_block: self.head, + old_head_state: old_header.state_root, + new_head_block: head, + new_head_state: header.state_root, + }); + } events.emit(ChainEvent::Head { slot: header.slot, block: head, @@ -264,6 +315,27 @@ impl ChainEventSnapshot { ); } } + + // Safe target is a fork-choice block root (not a checkpoint), advanced + // at interval 3. Report its slot from the header; a missing header only + // drops this event. + let safe_target = store.safe_target().expect("safe target exists"); + if safe_target != self.safe_target { + if let Some(header) = store + .get_block_header(&safe_target) + .expect("block header read should succeed") + { + events.emit(ChainEvent::SafeTarget { + slot: header.slot, + block: safe_target, + }); + } else { + warn!( + safe_target_root = %ShortRoot(&safe_target.0), + "Safe target header missing while emitting event; skipping" + ); + } + } } } @@ -297,11 +369,13 @@ mod tests { } } - const ALL_TOPICS: [Topic; 4] = [ + const ALL_TOPICS: [Topic; 6] = [ Topic::Head, Topic::Block, Topic::JustifiedCheckpoint, Topic::FinalizedCheckpoint, + Topic::ChainReorg, + Topic::SafeTarget, ]; #[tokio::test] @@ -356,6 +430,18 @@ mod tests { }, Topic::FinalizedCheckpoint, ), + ( + ChainEvent::ChainReorg { + slot: 1, + depth: 1, + old_head_block: block, + old_head_state: state, + new_head_block: block, + new_head_state: state, + }, + Topic::ChainReorg, + ), + (ChainEvent::SafeTarget { slot: 1, block }, Topic::SafeTarget), ]; for (event, topic) in cases { assert_eq!(event.topic(), topic); @@ -516,4 +602,79 @@ mod tests { } assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); } + + /// Swinging the head onto a sibling chain fires `chain_reorg` ahead of the + /// `head` event, with the rolled-back depth. (A plain extension is covered + /// by `chain_event_diff_emits_recent_head`, which sees no reorg.) + #[test] + fn chain_event_diff_emits_chain_reorg_before_head() { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(16); + let mut rx = bus.subscribe(); + + // Two forks off genesis: A(1) ← A2(2), and B(1) on its own. + let a = H256([1u8; 32]); + let a2 = H256([2u8; 32]); + let b = H256([3u8; 32]); + insert_test_block(&mut store, a, 1, genesis, H256([11u8; 32])); + insert_test_block(&mut store, a2, 2, a, H256([22u8; 32])); + insert_test_block(&mut store, b, 1, genesis, H256([33u8; 32])); + + // Head starts on A2; snapshot; then fork choice picks B, off A2's chain. + store + .update_checkpoints(ForkCheckpoints::head_only(a2)) + .expect("head A2"); + let snapshot = ChainEventSnapshot::capture(&store); + store + .update_checkpoints(ForkCheckpoints::head_only(b)) + .expect("head B"); + + // Wall clock at B's slot so the recency gate lets both events through. + snapshot.diff_and_emit(&store, &bus, 1); + + match rx.try_recv().unwrap() { + ChainEvent::ChainReorg { + slot, + depth, + old_head_block, + new_head_block, + .. + } => { + assert_eq!(slot, 1, "reorg slot is the new head's slot"); + assert_eq!(depth, 1, "A2 rolls back one block to the A/B fork point"); + assert_eq!(old_head_block, a2); + assert_eq!(new_head_block, b); + } + other => panic!("expected chain_reorg first, got: {other:?}"), + } + match rx.try_recv().unwrap() { + ChainEvent::Head { block, .. } => assert_eq!(block, b), + other => panic!("expected head after reorg, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + /// A safe-target move (interval-3 output) emits `safe_target` and nothing + /// else when head/justified/finalized are unchanged. + #[test] + fn chain_event_diff_emits_safe_target() { + let mut store = test_store(); + let genesis = store.head().expect("store head exists"); + let bus = EventBus::new(16); + let mut rx = bus.subscribe(); + + let target = H256([6u8; 32]); + insert_test_block(&mut store, target, 2, genesis, H256([66u8; 32])); + let snapshot = ChainEventSnapshot::capture(&store); + store.set_safe_target(target).expect("set safe target"); + + snapshot.diff_and_emit(&store, &bus, 2); + + match rx.try_recv().unwrap() { + ChainEvent::SafeTarget { slot, block } => assert_eq!((slot, block), (2, target)), + other => panic!("expected safe_target, got: {other:?}"), + } + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } } diff --git a/crates/blockchain/src/store.rs b/crates/blockchain/src/store.rs index 2a3ff032..899cceff 100644 --- a/crates/blockchain/src/store.rs +++ b/crates/blockchain/src/store.rs @@ -1177,7 +1177,10 @@ pub fn verify_block_signatures( /// slot head to the lower slot head's slot, we don't arrive at the lower slot head. /// Returns `Some(depth)` where depth is the number of blocks walked back, or `None` /// if no reorg occurred. -fn reorg_depth(old_head: H256, new_head: H256, store: &Store) -> Option { +/// +/// `pub(crate)` so the actor-layer chain-event diff can reuse the same +/// detection for the `chain_reorg` event, keeping one definition of "reorg". +pub(crate) fn reorg_depth(old_head: H256, new_head: H256, store: &Store) -> Option { if new_head == old_head { return None; } diff --git a/docs/rpc.md b/docs/rpc.md index f560692d..cdf358d3 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -86,9 +86,9 @@ SSZ-encoded `SignedBlock` at the latest finalized checkpoint. The genesis/anchor ### `GET /lean/v0/events` -Server-Sent Events stream (`Content-Type: text/event-stream`) of live chain events published by the blockchain actor. Four event types: +Server-Sent Events stream (`Content-Type: text/event-stream`) of live chain events published by the blockchain actor. Six event types: -Payload fields mirror the Ethereum beacon-API eventstream: `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`. +Payload fields mirror the Ethereum beacon-API eventstream where an analog exists: `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`. `justified_checkpoint` and `safe_target` are ethlambda extensions with no beacon topic. | Event | Payload | Emitted when | |-------|---------|--------------| @@ -96,6 +96,8 @@ Payload fields mirror the Ethereum beacon-API eventstream: `block` is the block | `block` | `{ "slot": 128, "block": "0x…" }` | A block is imported into the store | | `justified_checkpoint` | `{ "slot": 120, "block": "0x…", "state": "0x…" }` | The justified checkpoint advances | | `finalized_checkpoint` | `{ "slot": 96, "block": "0x…", "state": "0x…" }` | The finalized checkpoint advances | +| `chain_reorg` | `{ "slot": 128, "depth": 2, "old_head_block": "0x…", "old_head_state": "0x…", "new_head_block": "0x…", "new_head_state": "0x…" }` | Fork choice switches to a head off the old head's chain (same recency gate as `head`); beacon shape minus `epoch`/`execution_optimistic` | +| `safe_target` | `{ "slot": 127, "block": "0x…" }` | The interval-3 safe attestation target advances | The topic name travels only on the SSE `event:` line; the `data:` line carries the flat JSON payload. Example frame: @@ -112,7 +114,7 @@ A **required** comma-separated list of event names selects which events to strea curl -N 'http://127.0.0.1:5052/lean/v0/events?topics=head,finalized_checkpoint' ``` -Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`. As in the Beacon API `eventstream` endpoint, `topics` is mandatory: there is no "subscribe to everything" default; list the topics you want. +Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`, `chain_reorg`, `safe_target`. As in the Beacon API `eventstream` endpoint, `topics` is mandatory: there is no "subscribe to everything" default; list the topics you want. | Status | Condition | |--------|-----------|