diff --git a/crates/blockchain/src/epoch_precompute.rs b/crates/blockchain/src/epoch_precompute.rs new file mode 100644 index 000000000..9c77e137a --- /dev/null +++ b/crates/blockchain/src/epoch_precompute.rs @@ -0,0 +1,369 @@ +//! Beacon epoch-transition precompute. +//! +//! The first block of an epoch runs `process_epoch` inline on the import path, +//! then pays the post-epoch rehash for its state-root check. Both can be done +//! before the block arrives, since the parent is known by the end of the last +//! slot of the previous epoch: clone the head state, advance it to the epoch's +//! first slot on a blocking worker, hash it, and park it in the store's state +//! cache under [`CacheKey::CheckpointState`]. `fork_choice::on_block` resumes +//! from that entry when it finds one, and falls back to the inline path when it +//! does not, so the precompute is a pure speed trade. +//! +//! Two triggers feed the same worker: +//! +//! | Trigger | When | Covers | +//! |---|---|---| +//! | [`Trigger::Head`] | an import leaves a last-slot block as head | a late last-slot block | +//! | [`Trigger::Timer`] | three quarters into a last slot | a skipped last slot | +//! +//! Either trigger advances the head across one epoch boundary at most. At most +//! one worker runs at a time, and a key already in the cache is never +//! recomputed. A block that arrives while the worker is still running imports +//! inline; the late result is stored and simply goes unused by that block. + +use std::sync::Arc; +use std::time::Duration; + +use ethlambda_state_transition::beacon::fork_choice; +use ethlambda_state_transition::beacon::helpers::misc::compute_start_slot_at_epoch; +use ethlambda_state_transition::metrics as stf_metrics; +use ethlambda_storage::{CacheKey, Chain}; +use ethlambda_types::ShortRoot; +use ethlambda_types::beacon::containers::BeaconState; +use ethlambda_types::beacon::preset; +use ethlambda_types::primitives::H256; +use spawned_concurrency::message::Message; +use spawned_concurrency::tasks::{Context, Handler, send_after}; +use tracing::{debug, info, warn}; + +use crate::BlockChainServer; + +/// What caused a precompute, and the label it is counted under. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Trigger { + /// An import made a last-slot block the head. + Head, + /// The three-quarter point of a last slot, carrying that slot. + Timer { slot: u64 }, +} + +impl Trigger { + fn label(self) -> &'static str { + match self { + Trigger::Head => "head", + Trigger::Timer { .. } => "timer", + } + } +} + +/// The state a precompute produces: the head's post-state advanced to the first +/// slot of `epoch`. The same identity as `CacheKey::CheckpointState`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct PrecomputeKey { + pub(crate) epoch: u64, + pub(crate) root: H256, +} + +/// Everything [`decide`] reads, gathered so the decision is a pure function. +pub(crate) struct DecisionInputs { + pub(crate) trigger: Trigger, + pub(crate) syncing: bool, + pub(crate) head_slot: u64, + pub(crate) head_root: H256, + /// The wall-clock slot now. + pub(crate) current_slot: u64, + /// The worker running right now, if any. + pub(crate) in_flight: Option, + /// Whether the store already holds a state for the candidate key. + pub(crate) already_cached: bool, +} + +fn is_last_slot_of_epoch(slot: u64) -> bool { + (slot + 1).is_multiple_of(preset::SLOTS_PER_EPOCH) +} + +/// Whether to start a precompute, and for which key. +/// +/// `None` when the node is syncing, when the trigger does not sit on the last +/// slot of an epoch, when a worker is already running (at most one), or when +/// the store already holds the state. A head trigger also requires the head to +/// be fresh (at most one slot behind the wall clock): the sync tracker moves +/// once per slot, so a follower importing history would otherwise precompute +/// every epoch it replays. A timer trigger requires the head to be in the +/// timer slot's own epoch, so a worker crosses one epoch boundary at most: in +/// a network-wide stall the sync tracker reports the node synced, and a timer +/// epochs past a stalled head would otherwise replay every epoch in between. +pub(crate) fn decide(inputs: &DecisionInputs) -> Option { + if inputs.syncing || inputs.in_flight.is_some() { + return None; + } + let epoch = match inputs.trigger { + Trigger::Head => { + if !is_last_slot_of_epoch(inputs.head_slot) + || inputs.current_slot > inputs.head_slot.saturating_add(1) + { + return None; + } + inputs.head_slot / preset::SLOTS_PER_EPOCH + 1 + } + Trigger::Timer { slot } => { + // A head past the slot would be a stale message racing a newer + // import; one already at the boundary has nothing left to advance. + // A head before the slot's own epoch is a stalled chain: advancing + // it would replay every epoch it missed, and each later timer would + // replay them again under a new key from the same head. + let slot_epoch = slot / preset::SLOTS_PER_EPOCH; + if !is_last_slot_of_epoch(slot) + || inputs.head_slot > slot + || inputs.head_slot < compute_start_slot_at_epoch(slot_epoch) + { + return None; + } + slot_epoch + 1 + } + }; + if inputs.head_slot >= compute_start_slot_at_epoch(epoch) { + return None; + } + let key = PrecomputeKey { + epoch, + root: inputs.head_root, + }; + (!inputs.already_cached).then_some(key) +} + +/// Delay from the start of a slot to its three-quarter point. +pub(crate) fn three_quarter_slot(slot_duration_ms: u64) -> Duration { + Duration::from_millis(slot_duration_ms / 4 * 3) +} + +/// Self-message armed at a last slot's start, delivered at its three-quarter point. +pub(crate) struct EpochPrecomputeCheck { + pub(crate) slot: u64, +} +impl Message for EpochPrecomputeCheck { + type Result = (); +} + +/// A worker's result, sent back to the actor. +pub(crate) struct EpochPrecomputed { + pub(crate) key: PrecomputeKey, + pub(crate) result: Result, String>, +} +impl Message for EpochPrecomputed { + type Result = (); +} + +impl BlockChainServer { + /// Arm the three-quarter check if `slot` is the last of its epoch. + /// + /// Called from the once-per-slot tick, which fires at the slot boundary, so + /// the delay is the three-quarter point less whatever of the slot the tick + /// has already consumed (a tick delayed by a long import). + pub(crate) fn arm_epoch_precompute_check(&self, slot: u64, ctx: &Context) { + if self.store.chain() != Chain::Beacon || !is_last_slot_of_epoch(slot) { + return; + } + let slot_duration_ms = self.store.config().time_grid().milliseconds_per_slot; + let into_slot = Duration::from_millis(self.ms_into_slot(slot).max(0) as u64); + let delay = three_quarter_slot(slot_duration_ms).saturating_sub(into_slot); + send_after(delay, ctx.clone(), EpochPrecomputeCheck { slot }); + } + + /// Start a precompute for the current head if [`decide`] allows one. + pub(crate) fn maybe_start_epoch_precompute(&mut self, trigger: Trigger, ctx: &Context) { + if self.store.chain() != Chain::Beacon { + return; + } + let Some((head_slot, head_root)) = self.store.beacon_head() else { + return; + }; + let candidate_epoch = match trigger { + Trigger::Head => head_slot / preset::SLOTS_PER_EPOCH + 1, + Trigger::Timer { slot } => slot / preset::SLOTS_PER_EPOCH + 1, + }; + let already_cached = self + .store + .cached_state(CacheKey::CheckpointState { + epoch: candidate_epoch, + root: head_root, + }) + .is_some(); + let inputs = DecisionInputs { + trigger, + syncing: self.sync_status.is_syncing(), + head_slot, + head_root, + current_slot: self.wall_clock_slot(), + in_flight: self.epoch_precompute_in_flight, + already_cached, + }; + let Some(key) = decide(&inputs) else { + return; + }; + // The head's post-state is resident (it was just imported); a read that + // cannot find it means the head moved to a block this node has no state + // for, which a later trigger can retry. + let Ok(Some(head_state)) = self.store.get_state(&head_root) else { + return; + }; + + self.epoch_precompute_in_flight = Some(key); + stf_metrics::inc_epoch_precompute_started(trigger.label()); + debug!( + epoch = key.epoch, + head_slot, + head_root = %ShortRoot(&head_root.0), + trigger = trigger.label(), + "Starting epoch precompute" + ); + + let config = self.store.config(); + let actor = ctx.actor_ref(); + tokio::task::spawn_blocking(move || { + let result = { + let _timing = stf_metrics::time_epoch_precompute(); + fork_choice::advance_to_epoch_start(&head_state, key.epoch, &config) + }; + let result = result.map(Arc::new).map_err(|err| err.to_string()); + let _ = actor.send(EpochPrecomputed { key, result }); + }); + } +} + +impl Handler for BlockChainServer { + async fn handle(&mut self, msg: EpochPrecomputeCheck, ctx: &Context) { + self.maybe_start_epoch_precompute(Trigger::Timer { slot: msg.slot }, ctx); + } +} + +impl Handler for BlockChainServer { + async fn handle(&mut self, msg: EpochPrecomputed, _ctx: &Context) { + self.epoch_precompute_in_flight = None; + match msg.result { + Ok(state) => { + // Stored even if the block it was meant for already imported + // inline: attestation targets for the epoch read the same key. + self.store.cache_state( + CacheKey::CheckpointState { + epoch: msg.key.epoch, + root: msg.key.root, + }, + state, + ); + info!( + epoch = msg.key.epoch, + head_root = %ShortRoot(&msg.key.root.0), + "Epoch precompute stored" + ); + } + Err(err) => warn!( + epoch = msg.key.epoch, + head_root = %ShortRoot(&msg.key.root.0), + %err, + "Epoch precompute failed" + ), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const SPE: u64 = preset::SLOTS_PER_EPOCH; + + fn inputs(trigger: Trigger, head_slot: u64) -> DecisionInputs { + DecisionInputs { + trigger, + syncing: false, + head_slot, + head_root: H256([7; 32]), + current_slot: head_slot, + in_flight: None, + already_cached: false, + } + } + + #[test] + fn a_last_slot_head_precomputes_the_next_epoch() { + let key = decide(&inputs(Trigger::Head, 2 * SPE - 1)).expect("starts"); + assert_eq!(key.epoch, 2); + assert_eq!(key.root, H256([7; 32])); + } + + #[test] + fn a_head_elsewhere_in_the_epoch_starts_nothing() { + assert!(decide(&inputs(Trigger::Head, SPE)).is_none()); + assert!(decide(&inputs(Trigger::Head, 2 * SPE - 2)).is_none()); + } + + #[test] + fn a_syncing_node_starts_nothing() { + let mut i = inputs(Trigger::Head, 2 * SPE - 1); + i.syncing = true; + assert!(decide(&i).is_none()); + } + + #[test] + fn a_stale_head_trigger_starts_nothing() { + let mut i = inputs(Trigger::Head, 2 * SPE - 1); + i.current_slot = 2 * SPE + 5; + assert!(decide(&i).is_none()); + } + + #[test] + fn a_running_worker_blocks_a_second_one() { + let mut i = inputs(Trigger::Head, 2 * SPE - 1); + i.in_flight = Some(PrecomputeKey { + epoch: 2, + root: H256([7; 32]), + }); + assert!(decide(&i).is_none()); + } + + #[test] + fn a_state_already_cached_is_not_recomputed() { + let mut i = inputs(Trigger::Timer { slot: 2 * SPE - 1 }, 2 * SPE - 3); + i.already_cached = true; + assert!(decide(&i).is_none()); + } + + #[test] + fn the_timer_covers_a_skipped_last_slot() { + let key = + decide(&inputs(Trigger::Timer { slot: 2 * SPE - 1 }, 2 * SPE - 3)).expect("starts"); + assert_eq!(key.epoch, 2); + } + + #[test] + fn the_timer_covers_an_epoch_whose_only_block_is_its_first() { + let key = decide(&inputs(Trigger::Timer { slot: 3 * SPE - 1 }, 2 * SPE)).expect("starts"); + assert_eq!(key.epoch, 3); + } + + #[test] + fn a_stale_timer_trigger_starts_nothing() { + // The timer's epoch holds no block: the head is the previous epoch's + // last slot. + assert!(decide(&inputs(Trigger::Timer { slot: 3 * SPE - 1 }, 2 * SPE - 1)).is_none()); + // A stall several epochs long. + assert!(decide(&inputs(Trigger::Timer { slot: 6 * SPE - 1 }, 2 * SPE - 5)).is_none()); + } + + #[test] + fn the_timer_ignores_slots_that_are_not_the_last() { + assert!(decide(&inputs(Trigger::Timer { slot: SPE }, SPE - 1)).is_none()); + } + + #[test] + fn the_timer_skips_a_head_already_at_the_boundary() { + assert!(decide(&inputs(Trigger::Timer { slot: 2 * SPE - 1 }, 2 * SPE)).is_none()); + } + + #[test] + fn three_quarters_follows_the_configured_slot_duration() { + assert_eq!(three_quarter_slot(12_000), Duration::from_secs(9)); + assert_eq!(three_quarter_slot(6_000), Duration::from_millis(4_500)); + } +} diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index c9dc378df..a9cd8fc1c 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -62,6 +62,7 @@ mod beacon_payloads; pub use beacon_payloads::checkpoint_hash; pub mod block_builder; pub(crate) mod coverage; +mod epoch_precompute; pub mod events; pub(crate) mod fork_choice_tree; pub mod import_timing; @@ -421,6 +422,7 @@ impl BlockChain { engine, safe_slots_to_import_optimistically, last_tick_instant: None, + epoch_precompute_in_flight: None, sync_status, sync_status_controller, events, @@ -547,6 +549,10 @@ pub struct BlockChainServer { /// Last tick instant for measuring interval duration. last_tick_instant: Option, + /// The epoch precompute worker running right now, if any (beacon only). At + /// most one runs at a time; see [`epoch_precompute`]. + epoch_precompute_in_flight: Option, + /// Stateful sync heuristic used by `lean_node_sync_status`. Also gates /// validator duties while syncing, unless that gating was disabled at /// startup via `--disable-duty-sync-gate` (then it is metric-only). On a @@ -1066,6 +1072,12 @@ impl BlockChainServer { self.redrive_missing_envelopes(); self.settle_envelopes().await; + // Beacon only (both calls are no-ops on lean): a last slot arms its + // three-quarter precompute check, and a redrive above may have made a + // last-slot block the head without passing through the block handler. + self.arm_epoch_precompute_check(slot, ctx); + self.maybe_start_epoch_precompute(epoch_precompute::Trigger::Head, ctx); + // Per-interval duties for this tick. Lean-only, so this is where a // beacon follower's tick ends: it has no validator duties (see // [`ChainDuties::Beacon`]), and everything a tick owes it happened @@ -1996,6 +2008,7 @@ impl BlockChainServer { engine, safe_slots_to_import_optimistically, last_tick_instant: None, + epoch_precompute_in_flight: None, sync_status: SyncStatusTracker::new(false), sync_status_controller: SyncStatusController::default(), events: ChainEvents::new(EventBus::default(), &store), @@ -3895,6 +3908,10 @@ impl Handler for BlockChainServer { self.on_block(msg.block, timings).await; // A gloas envelope that arrived ahead of this block is judgeable now. self.settle_envelopes().await; + + // The import above may have left a last-slot block as head: the + // moment to start the next epoch's transition (no-op on lean). + self.maybe_start_epoch_precompute(epoch_precompute::Trigger::Head, ctx); } } @@ -4439,6 +4456,7 @@ mod tests { engine: None, safe_slots_to_import_optimistically: constants::SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY, last_tick_instant: None, + epoch_precompute_in_flight: None, sync_status: SyncStatusTracker::new(false), sync_status_controller: SyncStatusController::default(), duties: ChainDuties::Beacon, diff --git a/crates/blockchain/src/sync_status.rs b/crates/blockchain/src/sync_status.rs index 5e968a5c4..27dff999b 100644 --- a/crates/blockchain/src/sync_status.rs +++ b/crates/blockchain/src/sync_status.rs @@ -122,6 +122,12 @@ impl SyncStatusTracker { } } + /// Whether the tracker currently considers the node to be syncing, + /// regardless of whether that gates duties. + pub(crate) fn is_syncing(&self) -> bool { + self.syncing + } + pub(crate) fn duties_allowed(&self) -> bool { // Gate disabled: the syncing state is observe-only, never suppresses duties. !self.gate_duties || !self.syncing diff --git a/crates/blockchain/state_transition/src/beacon/fork_choice.rs b/crates/blockchain/state_transition/src/beacon/fork_choice.rs index 73d285afd..f99165f0a 100644 --- a/crates/blockchain/state_transition/src/beacon/fork_choice.rs +++ b/crates/blockchain/state_transition/src/beacon/fork_choice.rs @@ -179,7 +179,8 @@ use crate::beacon::containers::{bellatrix, deneb, electra, fulu, gloas, phase0}; use crate::beacon::error::{Error, Result, verify}; use crate::beacon::fork::ForkName; use crate::beacon::helpers::accessors::{ - CommitteeCache, CommitteeCacheExt, get_active_validator_indices, get_current_epoch, + ActiveBalanceCache, CommitteeCache, CommitteeCacheExt, get_active_validator_indices, + get_current_epoch, }; use crate::beacon::helpers::attestation as phase0_attestation; use crate::beacon::helpers::electra as electra_helpers; @@ -1004,6 +1005,117 @@ pub fn checkpoint_state( Ok(state) } +/// The state of `parent_root` advanced to the first slot of `epoch`, computed +/// without the store. +/// +/// What [`checkpoint_state`] derives on a miss, taking the checkpoint block's +/// post-state as an argument so a worker thread can run it off the actor. +/// Hashes the result after flushing buffered writes, so the import that later +/// resumes from a clone of it finds the tree nodes' hashes already memoized +/// and pays only for the flat fields. A state already at or past the epoch's +/// first slot comes back unchanged, as `checkpoint_state` returns it. +pub fn advance_to_epoch_start( + parent_state: &BeaconState, + epoch: Epoch, + config: &Config, +) -> Result { + let target_slot = compute_start_slot_at_epoch(epoch); + let mut state = parent_state.clone(); + if state.slot() < target_slot { + stf::process_slots(&mut state, target_slot, config)?; + } + state.apply_pending_mutations(); + let _ = state.hash_tree_root(); + Ok(state) +} + +/// Applies `signed_block` to a copy of its parent's post-state, resuming from +/// the precomputed epoch-boundary state when the block crosses into an epoch +/// and one is cached. +/// +/// Returns the copy with the transition's outcome, since a failed transition +/// still leaves the caller a state to inspect. Both starting points reach the +/// same post-state: the precomputed one is what `process_slots` would produce +/// from the parent. Either way the copy is independent, so a failing block +/// cannot corrupt a cached entry. +#[allow(clippy::too_many_arguments)] +fn transition_block( + store: &Store, + parent_state: &BeaconState, + parent_root: Root, + signed_block: &SignedBeaconBlock, + validate_result: bool, + config: &Config, + engine: &stf::ExecutionEngine, + committees: &CommitteeCache, + active_balances: &ActiveBalanceCache, +) -> (BeaconState, Result<()>) { + let block_slot = signed_block.slot(); + let Some(precomputed) = + precomputed_epoch_state(store, parent_state.slot(), parent_root, block_slot) + else { + let mut state = parent_state.clone(); + let outcome = stf::state_transition( + &mut state, + signed_block, + validate_result, + config, + engine, + committees, + active_balances, + ); + return (state, outcome); + }; + + let mut state = (*precomputed).clone(); + // Only the slots skipped past the boundary are left to advance; none when + // the block sits on the boundary itself. + let outcome = if state.slot() < block_slot { + stf::process_slots(&mut state, block_slot, config) + } else { + Ok(()) + } + .and_then(|()| { + stf::apply_block( + &mut state, + signed_block, + validate_result, + config, + engine, + committees, + active_balances, + ) + }); + (state, outcome) +} + +/// The precomputed epoch-boundary state an import can resume from, if any. +/// +/// Applies to a block that crosses into a new epoch: its parent's post-state +/// is before the first slot of the block's epoch, and the block is at or after +/// that slot. The precompute worker stores it under the key +/// [`checkpoint_state`] uses for `(block_epoch, parent_root)`, since the two +/// name the same state. Counts a hit or a miss for every such import and for +/// no other. +fn precomputed_epoch_state( + store: &Store, + parent_slot: Slot, + parent_root: Root, + block_slot: Slot, +) -> Option> { + let epoch = compute_epoch_at_slot(block_slot); + let epoch_start = compute_start_slot_at_epoch(epoch); + if !(parent_slot < epoch_start && epoch_start <= block_slot) { + return None; + } + let cached = store.cached_state(CacheKey::CheckpointState { + epoch, + root: parent_root, + }); + crate::metrics::inc_epoch_precompute_lookups(if cached.is_some() { "hit" } else { "miss" }); + cached +} + // --------------------------------------------------------------------------- // get_forkchoice_store // --------------------------------------------------------------------------- @@ -4578,7 +4690,6 @@ pub fn on_block( .get_state(&parent_root) .expect("get") .ok_or(Error::SpecAssert("block.parent_root in store.block_states"))?; - let mut state = (*parent_state).clone(); // [New in Gloas:EIP7732] If this block builds on its parent's full payload, // that payload must have been verified by `on_execution_payload_envelope`. @@ -4698,11 +4809,22 @@ pub fn on_block( stf::ExecutionEngine::valid() } }; + // + // An epoch-crossing block starts from the precomputed boundary state when + // one is cached, and from the parent's own post-state otherwise. Both reach + // the same post-state: the precomputed one is exactly what `process_slots` + // would have produced from the parent (a gloas parent included: a gloas + // block's post-state never holds its own payload, which only the child + // applies, after `process_slots`, so no second parent state exists). + // Either way `state` is an independent clone, so a failing block cannot + // corrupt a cached entry. // The store's own cache, so every block of this chain shares one entry per // epoch; see `ActiveBalanceCacheExt`. let active_balances = store.active_balance_cache(); - let transition = stf::state_transition( - &mut state, + let (mut state, transition) = transition_block( + store, + &parent_state, + parent_root, &signed_block, true, config, @@ -9824,4 +9946,244 @@ mod tests { ); assert_eq!(store.payload_link(&root), Some(link)); } + + // ---- epoch precompute ---- + + use crate::beacon::block_production::{ + BlockInputs, advance_to_slot, assemble_block, empty_sync_aggregate, payload_inputs, + }; + use crate::beacon::containers::electra; + use crate::beacon::helpers::accessors::{get_beacon_proposer_index, get_domain}; + use crate::beacon::helpers::test_state::{sign_for, with_signing_validators_at}; + use crate::beacon::primitives::Bytes32; + use ethlambda_types::beacon::containers::deneb::ExecutionPayload; + use ethlambda_types::beacon::containers::electra::{BeaconBlockBody, ExecutionRequests}; + + /// A fulu state at the last slot of epoch 1, the parent of every + /// epoch-crossing block below. + fn last_slot_parent() -> BeaconState { + let mut state = with_signing_validators_at(crate::beacon::ForkName::Fulu, 64); + let lookahead = + crate::beacon::helpers::fulu::initialize_proposer_lookahead(&state).expect("lookahead"); + let sync_committee = + crate::beacon::helpers::altair::get_next_sync_committee(&state).expect("committee"); + let BeaconState::Fulu(inner) = &mut state else { + unreachable!("built as fulu") + }; + inner.proposer_lookahead = lookahead.try_into().expect("lookahead length"); + inner.current_sync_committee = sync_committee.clone(); + inner.next_sync_committee = sync_committee; + let last_slot = 2 * preset::SLOTS_PER_EPOCH - 1; + advance_to_slot(&state, last_slot, &Config::mainnet()).expect("advance") + } + + /// A block at `slot` built on `parent` the way a proposer would, and the + /// post-state root it commits to. + fn block_on(parent: &BeaconState, slot: Slot) -> (SignedBeaconBlock, Root) { + let config = Config::mainnet(); + let advanced = advance_to_slot(parent, slot, &config).expect("advance"); + let proposer = get_beacon_proposer_index(&advanced).expect("proposer"); + let epoch = get_current_epoch(&advanced); + let domain = get_domain(&advanced, constants::DOMAIN_RANDAO, Some(epoch)); + let randao_reveal: BlsSignature = sign_for( + proposer as usize, + compute_signing_root(epoch.hash_tree_root(), domain), + ); + let payload = payload_inputs(&advanced, &config).expect("payload inputs"); + let inputs = BlockInputs { + randao_reveal, + graffiti: Bytes32::ZERO, + attestations: Vec::new(), + operations: Default::default(), + execution_payload: ExecutionPayload { + parent_hash: payload.parent_hash, + prev_randao: payload.prev_randao, + timestamp: payload.timestamp, + withdrawals: payload.withdrawals.try_into().expect("withdrawals"), + ..BeaconBlockBody::empty().execution_payload + }, + blob_kzg_commitments: Vec::new(), + execution_requests: ExecutionRequests::default(), + sync_aggregate: empty_sync_aggregate(), + }; + let message = assemble_block( + &advanced, + inputs, + &config, + crate::beacon::stf::block::BlockProcessingOptions::default(), + ) + .expect("assemble"); + let state_root = message.state_root; + let signed = SignedBeaconBlock::Fulu(electra::SignedBeaconBlock { + message, + signature: BlsSignature::default(), + }); + (signed, state_root) + } + + fn post_state_root( + store: &Store, + parent: &BeaconState, + parent_root: Root, + block: &SignedBeaconBlock, + ) -> Root { + let (mut state, outcome) = transition_block( + store, + parent, + parent_root, + block, + false, + &Config::mainnet(), + &stf::ExecutionEngine::valid(), + &CommitteeCache::default(), + &ActiveBalanceCache::default(), + ); + outcome.expect("the block applies"); + state.apply_pending_mutations(); + state.hash_tree_root() + } + + fn store_with_precompute(parent: &BeaconState, parent_root: Root, epoch: Epoch) -> Store { + let store = empty_store(); + let precomputed = + advance_to_epoch_start(parent, epoch, &Config::mainnet()).expect("precompute"); + store.cache_state( + CacheKey::CheckpointState { + epoch, + root: parent_root, + }, + Arc::new(precomputed), + ); + store + } + + /// Resuming from the precomputed state is not a different transition: the + /// post-state is the one the inline path commits to, for a block on the + /// boundary and for one past a skipped first slot. + #[test] + fn resuming_from_a_precomputed_state_gives_the_inline_post_state() { + let parent = last_slot_parent(); + let parent_root = Root::repeat_byte(0x42); + let epoch = 2; + let first_slot = compute_start_slot_at_epoch(epoch); + + for slot in [first_slot, first_slot + 1, first_slot + 3] { + let (block, committed) = block_on(&parent, slot); + let hit_store = store_with_precompute(&parent, parent_root, epoch); + assert!( + precomputed_epoch_state(&hit_store, parent.slot(), parent_root, slot).is_some() + ); + + let inline = post_state_root(&empty_store(), &parent, parent_root, &block); + let resumed = post_state_root(&hit_store, &parent, parent_root, &block); + assert_eq!(inline, committed, "slot {slot}: inline matches the block"); + assert_eq!(resumed, inline, "slot {slot}: resumed matches inline"); + } + } + + #[test] + fn only_an_epoch_crossing_import_looks_for_a_precomputed_state() { + let parent = last_slot_parent(); + let parent_root = Root::repeat_byte(0x42); + let store = store_with_precompute(&parent, parent_root, 2); + let first_slot = compute_start_slot_at_epoch(2); + + // Crosses: parent before the boundary, block at or after it. + assert!(precomputed_epoch_state(&store, parent.slot(), parent_root, first_slot).is_some()); + // Same epoch as its parent: nothing to resume. + assert!(precomputed_epoch_state(&store, first_slot, parent_root, first_slot + 1).is_none()); + // A different parent has no entry. + assert!( + precomputed_epoch_state(&store, parent.slot(), Root::repeat_byte(1), first_slot) + .is_none() + ); + } + + /// The worker's result is what `checkpoint_state` derives for the same + /// `(epoch, root)`, which is why one cache key can serve both. + #[test] + fn the_precompute_equals_checkpoint_state_for_the_same_key() { + let config = Config::mainnet(); + let parent = last_slot_parent(); + let root = Root::repeat_byte(0x42); + let mut store = empty_store(); + store + .insert_signed_block(root, block(parent.slot(), Root::ZERO)) + .expect("insert block"); + store + .insert_state(root, parent.clone()) + .expect("insert state"); + let epoch = 2; + + let worker = advance_to_epoch_start(&parent, epoch, &config).expect("precompute"); + let derived = + checkpoint_state(&store, &Checkpoint { epoch, root }, &config).expect("checkpoint"); + + assert_eq!(worker.slot(), compute_start_slot_at_epoch(epoch)); + assert_eq!(worker.slot(), derived.slot()); + assert_eq!(worker.hash_tree_root(), derived.hash_tree_root()); + } + + /// A gloas state at the last slot of epoch 1, the parent of the gloas + /// epoch-crossing cases below. + fn gloas_last_slot_parent() -> BeaconState { + let state = test_state::with_validators_at(crate::beacon::ForkName::Gloas, 64); + let last_slot = 2 * preset::SLOTS_PER_EPOCH - 1; + advance_to_slot(&state, last_slot, &Config::mainnet()).expect("advance") + } + + /// A gloas block's post-state never holds its own payload (the child applies + /// it inside `process_block`, after `process_slots`), and the store keeps + /// one state per root, so the precompute starts from exactly the state an + /// import of a child would. Checked three ways for a gloas parent: the + /// worker's result is `checkpoint_state`'s for the same key, an import + /// finds it under that key, and advancing past a skipped first slot from it + /// gives the state a plain `process_slots` from the parent gives. + #[test] + fn a_gloas_precompute_starts_from_the_state_an_import_would() { + let config = Config::mainnet(); + let parent = gloas_last_slot_parent(); + let root = Root::repeat_byte(0x42); + let mut store = empty_store(); + store + .insert_signed_block(root, block(parent.slot(), Root::ZERO)) + .expect("insert block"); + store + .insert_state(root, parent.clone()) + .expect("insert state"); + let epoch = 2; + let first_slot = compute_start_slot_at_epoch(epoch); + + let worker = advance_to_epoch_start(&parent, epoch, &config).expect("precompute"); + let derived = + checkpoint_state(&store, &Checkpoint { epoch, root }, &config).expect("checkpoint"); + assert_eq!(worker.fork_name(), crate::beacon::ForkName::Gloas); + assert_eq!(worker.hash_tree_root(), derived.hash_tree_root()); + + store.cache_state(CacheKey::CheckpointState { epoch, root }, Arc::new(worker)); + let stored = store + .get_state(&root) + .expect("get") + .expect("the parent's own post-state"); + assert!( + precomputed_epoch_state(&store, stored.slot(), root, first_slot).is_some(), + "an import of a child of `root` finds the entry" + ); + + for slot in [first_slot + 1, first_slot + 3] { + let mut resumed = (*precomputed_epoch_state(&store, stored.slot(), root, slot) + .expect("cached")) + .clone(); + stf::process_slots(&mut resumed, slot, &config).expect("advance the rest"); + let mut inline = (*stored).clone(); + stf::process_slots(&mut inline, slot, &config).expect("advance inline"); + resumed.apply_pending_mutations(); + inline.apply_pending_mutations(); + assert_eq!( + resumed.hash_tree_root(), + inline.hash_tree_root(), + "slot {slot}" + ); + } + } } diff --git a/crates/blockchain/state_transition/src/beacon/stf/mod.rs b/crates/blockchain/state_transition/src/beacon/stf/mod.rs index 8a57ca638..3dba6eef8 100644 --- a/crates/blockchain/state_transition/src/beacon/stf/mod.rs +++ b/crates/blockchain/state_transition/src/beacon/stf/mod.rs @@ -137,7 +137,35 @@ pub fn state_transition( active_balances: &ActiveBalanceCache, ) -> Result<()> { process_slots(state, signed_block.slot(), config)?; + apply_block( + state, + signed_block, + validate_result, + config, + engine, + committees, + active_balances, + ) +} +/// [`state_transition`] from the point where `state` is already at the block's +/// own slot: everything after `process_slots`. +/// +/// For a caller that advanced the state itself (fork choice resuming from a +/// precomputed epoch-boundary state). `process_slots` rejects a target equal to +/// the current slot, so such a state cannot go through [`state_transition`] +/// when the block sits exactly on the boundary. A state at the wrong slot is +/// still refused: the block header check in `process_block` compares the +/// block's slot with the state's. +pub fn apply_block( + state: &mut BeaconState, + signed_block: &containers::SignedBeaconBlock, + validate_result: bool, + config: &Config, + engine: &ExecutionEngine, + committees: &CommitteeCache, + active_balances: &ActiveBalanceCache, +) -> Result<()> { // After `process_slots`, never before it. A block proposed at the first slot // of a fork's activation epoch is the *post*-fork shape while the state // arriving here is still the pre-fork one, which is exactly the case a fork diff --git a/crates/blockchain/state_transition/src/metrics.rs b/crates/blockchain/state_transition/src/metrics.rs index dc2b4c9f1..63c029166 100644 --- a/crates/blockchain/state_transition/src/metrics.rs +++ b/crates/blockchain/state_transition/src/metrics.rs @@ -80,7 +80,6 @@ pub fn inc_total_active_balance_lookups(result: &str) { .inc(); } - static LEAN_BEACON_JUSTIFIED_BALANCES_LOOKUPS_TOTAL: LazyLock = LazyLock::new( || { register_int_counter_vec!( @@ -115,6 +114,54 @@ pub fn time_justified_balances_build() -> TimingGuard { TimingGuard::new(&LEAN_BEACON_JUSTIFIED_BALANCES_BUILD_SECONDS) } +static LEAN_BEACON_EPOCH_PRECOMPUTE_LOOKUPS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "lean_beacon_epoch_precompute_lookups_total", + "Epoch-crossing block imports, by whether a precomputed epoch-boundary state was cached", + &["result"] + ) + .unwrap() +}); + +/// Count one epoch-crossing import's lookup of a precomputed state: `hit` or +/// `miss`. +pub fn inc_epoch_precompute_lookups(result: &str) { + LEAN_BEACON_EPOCH_PRECOMPUTE_LOOKUPS_TOTAL + .with_label_values(&[result]) + .inc(); +} + +static LEAN_BEACON_EPOCH_PRECOMPUTE_SECONDS: LazyLock = LazyLock::new(|| { + register_histogram!( + "lean_beacon_epoch_precompute_seconds", + "Duration of one epoch-transition precompute: advance, flush and hash", + vec![0.1, 0.25, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0, 8.0] + ) + .unwrap() +}); + +/// Start timing one epoch precompute. Records duration when the guard is dropped. +pub fn time_epoch_precompute() -> TimingGuard { + TimingGuard::new(&LEAN_BEACON_EPOCH_PRECOMPUTE_SECONDS) +} + +static LEAN_BEACON_EPOCH_PRECOMPUTE_STARTED_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "lean_beacon_epoch_precompute_started_total", + "Epoch precompute workers started, by what triggered them", + &["trigger"] + ) + .unwrap() +}); + +/// Count one precompute worker started: `head` (a last-slot block became head) +/// or `timer` (three quarters into a last slot with nothing computed). +pub fn inc_epoch_precompute_started(trigger: &str) { + LEAN_BEACON_EPOCH_PRECOMPUTE_STARTED_TOTAL + .with_label_values(&[trigger]) + .inc(); +} + static LEAN_STATE_TRANSITION_TIME_SECONDS: LazyLock = LazyLock::new(|| { register_histogram!( "lean_state_transition_time_seconds", diff --git a/docs/beacon_stf.md b/docs/beacon_stf.md index 0752eebff..b2df5a0f2 100644 --- a/docs/beacon_stf.md +++ b/docs/beacon_stf.md @@ -617,6 +617,35 @@ eleven cores busy to about eight. The 3.3x understates the hashing change, because the later run does strictly more work: `transition` went from failing immediately to running every case. +## Epoch-transition precompute + +The first block of an epoch used to run `process_epoch` inline on the import +path (`fork_choice::on_block`, then `process_slots`), followed by the rehash +the state-root check needs. The beacon chain actor now does that work ahead of +time, on a blocking worker, and the import resumes from the result. + +| Step | What happens | +| --- | --- | +| Trigger: head | An import leaves a block at the last slot of epoch `E` as head: precompute `(E+1, head_root)`. Covers a late last-slot block. | +| Trigger: timer | Three quarters into the last slot of `E` (derived from the configured slot duration), if nothing for the current head is cached or running: precompute from the head, which is an earlier block when the last slot was skipped. | +| Gate | Skipped while the sync tracker says syncing. The head trigger skips a head more than a slot behind the wall clock; the timer trigger skips a head before `E`'s first slot, since the tracker reports a stalled network as synced and advancing such a head would replay every epoch it missed. Either way a worker crosses one epoch boundary at most. At most one worker runs; a key already cached is not recomputed. | +| Work | Clone the head state, `process_slots` to the first slot of `E+1`, flush pending writes, `hash_tree_root` (so the tree nodes' hashes are memoized), send the state back to the actor. | +| Store | The store's state cache under `CacheKey::CheckpointState { epoch: E+1, root: head_root }`. `fork_choice::checkpoint_state` derives the same value for that checkpoint (the checkpoint block's post-state advanced to the epoch's first slot), so attestation targets for `E+1` hit it too. | +| Consume | `on_block` looks up `CheckpointState { block_epoch, parent_root }` when the parent's slot is before the epoch's first slot and the block is at or after it. A hit clones the entry, advances any remaining skipped slots, and applies the block with `stf::apply_block`; the post-state is identical to the inline path's. | +| Miss | Today's path: clone the parent state and run `stf::state_transition`. A block that arrives while the worker is still running takes this path too; the late result is stored and goes unused by it. | + +The code lives in `crates/blockchain/src/epoch_precompute.rs` (triggers, worker, +actor handlers) and `fork_choice::advance_to_epoch_start` / +`fork_choice::transition_block` in the state-transition crate. The upgrade to a +new fork at the boundary happens inside `process_slots`, so the precompute needs +no fork-specific code. Gloas is covered by the same key: a gloas block's +post-state never contains its own payload (the next block applies it inside +`process_block`, after `process_slots`), and the store keeps exactly one state +per root, so the entry for `(E+1, root)` is advanced from the same state an +import of a child of `root` would clone. Hits and misses, worker time and +trigger counts are exported as `lean_beacon_epoch_precompute_*`; see +[metrics](metrics.md). + ## One test per fixture case The suites were once one test apiece, each looping over its own cases and diff --git a/docs/metrics.md b/docs/metrics.md index f96969b4e..a68219212 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -502,6 +502,26 @@ per justified checkpoint change, so a handful per hour on a healthy chain. Misses that track `get_head` calls mean the justified checkpoint is flapping between branches, or the snapshot is being replaced between two readers. +### Beacon Epoch Precompute + +The beacon chain actor advances the head state across the next epoch boundary +ahead of time, so the first block of an epoch does not run `process_epoch` on +the import path. See "Epoch-transition precompute" in +[`beacon_stf.md`](beacon_stf.md). This is ethlambda-specific, not part of the +leanMetrics spec. + +| Name | Type | Usage | Sample collection event | Labels | Buckets | +|------|------|-------|-------------------------|--------|---------| +| `lean_beacon_epoch_precompute_lookups_total` | Counter | Epoch-crossing imports, by whether a precomputed boundary state was cached | On each import whose parent is before the block's epoch start and whose block is at or after it | result=hit,miss | | +| `lean_beacon_epoch_precompute_seconds` | Histogram | Worker time to advance, flush and hash one state | When a precompute worker finishes | | 0.1, 0.25, 0.5, 0.75, 1, 1.5, 2, 3, 4, 8 | +| `lean_beacon_epoch_precompute_started_total` | Counter | Workers started, by trigger | When a worker is spawned | trigger=head,timer | | + +**Read the hit ratio against epochs.** On a synced follower nearly every +epoch-crossing import should hit. Misses mean the worker had not finished (the +block raced it, see `lean_beacon_epoch_precompute_seconds` against the slot +duration), the node was syncing, or the entry was evicted from the state cache. +A `timer` start with no matching `head` start means the last slot was skipped. + ### Beacon Pubkey Cache Every BLS signature check `ethlambda beacon` runs (block import, fork choice,