From b4adbafc54bf4512c6457f6ec1a884dad29f3d7d 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, 4 Nov 2026 11:53:27 -0300 Subject: [PATCH 1/2] perf(beacon): cache fork-choice vote weights in a node array get_head tallied every validator's latest message on every call, about 1M on mainnet, and a live import pays it twice for the first timely block of a slot (on_block's boost-gate walk, then the actor's get_head). That tally was most of get_head's ~17 ms in the eth-5 replay. The store now keeps the vote weights in a node array (Lighthouse unstable's proto-array layout: one node per block, carrying its PENDING, EMPTY and FULL weights). One attestation is one vote batch: Store::record_latest_messages records the messages and moves the attesters' weight under one lock, and the next read refreshes the subtree totals in one reverse pass over the blocks. The head walk reads them through cached_node_weights. The proposer boost, gloas's previous-slot rule and the walk bound depend on the clock, so they stay read-time. There are no best-child pointers: under gloas the choice moves without any vote, which is why Lighthouse stopped maintaining them. What the cache cannot follow incrementally (finality pruning, a live-window removal, a restart, a new justified balances snapshot, the gloas rule switch, a block stored without its node) marks it stale, and the next read rebuilds it from the latest messages at the old tally's cost, about once per epoch. A rebuild rescans LiveChain under the lock, so an API read that scanned before a removal cannot rebuild the removed blocks back in. compute_node_weights stays, as the reference. Equivalence is checked by a proptest against a naive tally (with the incremental state also equal to a fresh rebuild after every operation), on the random trees at every step of every seed, by a new test that feeds votes, blocks and equivocations through update_latest_messages, on_block's insertion order and insert_equivocating_index, and at every head check of every fork-choice fixture. Local benchmark (1M validators, 73 blocks, laptop): the tally 27 ms; a warm cached read 7 us; a read after a 15.6k-vote batch 49 us; the batch itself 1.3 ms; a rebuild 17 ms. --- Cargo.lock | 1 + .../src/beacon/fork_choice.rs | 520 +++++++- .../tests/beacon_spec/fork_choice.rs | 27 +- crates/storage/Cargo.toml | 6 + crates/storage/src/lib.rs | 2 + crates/storage/src/metrics.rs | 32 + crates/storage/src/proto_array.rs | 1090 +++++++++++++++++ crates/storage/src/store.rs | 193 ++- docs/beacon_engine.md | 8 +- docs/metrics.md | 14 + docs/spec_deviations.md | 5 +- 11 files changed, 1838 insertions(+), 60 deletions(-) create mode 100644 crates/storage/src/proto_array.rs diff --git a/Cargo.lock b/Cargo.lock index 0bec6c9e0..7588eeed5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2216,6 +2216,7 @@ dependencies = [ "ethlambda-crypto", "ethlambda-metrics", "ethlambda-types", + "foldhash 0.2.0", "libssz 0.3.0 (git+https://github.com/lambdaclass/libssz?rev=5cb14378ffdadcb51b387c003567e2630d0c7073)", "libssz-derive 0.3.0 (git+https://github.com/lambdaclass/libssz?rev=5cb14378ffdadcb51b387c003567e2630d0c7073)", "libssz-types 0.3.0 (git+https://github.com/lambdaclass/libssz?rev=5cb14378ffdadcb51b387c003567e2630d0c7073)", diff --git a/crates/blockchain/state_transition/src/beacon/fork_choice.rs b/crates/blockchain/state_transition/src/beacon/fork_choice.rs index ac83d85fe..6f064265e 100644 --- a/crates/blockchain/state_transition/src/beacon/fork_choice.rs +++ b/crates/blockchain/state_transition/src/beacon/fork_choice.rs @@ -170,7 +170,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; -use ethlambda_storage::{CacheKey, ForkCheckpoints, StorageBackend}; +use ethlambda_storage::{CacheKey, ForkCheckpoints, ProtoArrayError, StorageBackend, WeightRules}; use ethlambda_types::ShortRoot; use tracing::{debug, error, warn}; @@ -789,8 +789,10 @@ pub fn resolve_invalid_block( /// branch the execution layer has since rejected keeps pointing at a root no /// longer in the index, until it attests again. /// -/// [`compute_node_weights`] already drops such a vote rather than raising, so -/// [`get_head`] is unaffected. [`get_weight`] deliberately does not: it is the +/// [`compute_node_weights`] already drops such a vote rather than raising, and +/// so do the cached weights [`get_head`] reads (the removal makes the store +/// rebuild them), so [`get_head`] is unaffected. [`get_weight`] deliberately +/// does not: it is the /// specification's own version, and /// `tests::a_vote_for_a_pruned_block_weighs_nothing_instead_of_failing` pins /// that divergence on purpose. Nothing on this node's paths calls it today @@ -1606,7 +1608,8 @@ fn build_justified_balances(checkpoint: Checkpoint, state: &BeaconState) -> Just /// [`is_parent_strong`] call it too, [`is_head_weak`] underneath the public /// [`should_apply_proposer_boost`] and [`is_parent_strong`] underneath /// [`get_proposer_head`]. The head walk does not: it reads the same scores -/// from [`compute_node_weights`], which drops such a vote too. +/// from [`cached_node_weights`], which drops such a vote too, as its reference +/// [`compute_node_weights`] does. pub fn get_attestation_score( store: &Store, index: &HashMap, @@ -2120,6 +2123,13 @@ impl PayloadTree<'_> { /// blocks; see the section documentation above for why this equals the /// specification's per-node `get_weight`. /// +/// Two producers fill the same table. The head walk reads +/// [`cached_node_weights`], whose vote weights the store keeps current as votes +/// arrive, so a head computation costs the blocks in the window and not the +/// voters. [`compute_node_weights`] tallies them from every latest message on +/// each call; it is kept as the reference the cached path is tested against. +/// The rest of this paragraph describes the tally's cost. +/// /// Work is proportional to the number of latest messages plus the number of /// indexed blocks at or above the finalized block (times two lookups each), /// which is the unfinalized window, not the history since the anchor (see @@ -2140,6 +2150,26 @@ pub struct NodeWeights { weights: HashMap, } +/// Equal when every node weighs the same under the same rules and slot: a +/// node absent from one table and present with `0` in the other is the same +/// weight, since [`NodeWeights::raw`] reads absence as `0`. What the tests +/// compare [`cached_node_weights`] with [`compute_node_weights`] by. +impl PartialEq for NodeWeights { + fn eq(&self, other: &Self) -> bool { + let nonzero = |weights: &NodeWeights| -> HashMap { + weights + .weights + .iter() + .filter(|&(_, &weight)| weight != 0) + .map(|(&node, &weight)| (node, weight)) + .collect() + }; + self.rules == other.rules + && self.current_slot == other.current_slot + && nonzero(self) == nonzero(other) + } +} + impl NodeWeights { /// `node`'s weight, `block_slot` being its block's slot. /// @@ -2234,6 +2264,13 @@ fn walk_bound(store: &Store, index: &HashMap) -> Slot { /// [`NodeWeights`] for `rules`, on the block tree `index`. /// +/// **A reference, not the node's head computation.** [`walk_head`] reads +/// [`cached_node_weights`], which gives the same table from the store's cached +/// vote weights instead of this pass over every latest message. This tally is +/// kept, public, as what that cache is tested against: on random trees, as +/// votes, blocks and equivocations arrive, and at every `head` check of every +/// fork-choice fixture. +/// /// `rules` selects the two fork-dependent rules: how a block's payload /// dimension is read (see [`PayloadTree`]) and whether the proposer boost is /// gated by `should_apply_proposer_boost` (gloas) or applies whenever a boost @@ -2380,11 +2417,45 @@ pub fn compute_node_weights( } } - // The boost. Gloas asks `should_apply_proposer_boost`, fed the parent's - // score from the table just built; before gloas it applies whenever a boost - // root is set and still indexed, the tolerance `compute_weights` documents. - // Neither runs for a boosted block outside the justified subtree, which no - // node the descent compares would gain from. + add_proposer_boost( + store, + index, + config, + committees, + rules, + &balances, + bound, + &mut weights, + )?; + Ok(weights) +} + +/// The proposer boost, added to `weights` that hold the vote weights alone: +/// the last step of [`compute_node_weights`], shared with +/// [`cached_node_weights`], which reads the vote weights from the store's +/// cache instead of tallying them. The boost depends on the clock (it moves +/// every slot) and, under gloas, on the boosted block's parent's score, so it +/// is never cached. +/// +/// Gloas asks `should_apply_proposer_boost`, fed the parent's score from +/// `weights`; before gloas the boost applies whenever a boost root is set and +/// still indexed, the tolerance `compute_weights` documents. Neither runs for a +/// boosted block outside the justified subtree, which no node the descent +/// compares would gain from. The chain stops at `bound`, for the reasons +/// [`compute_node_weights`]' `# The bound` section gives. +#[allow(clippy::too_many_arguments)] +fn add_proposer_boost( + store: &Store, + index: &HashMap, + config: &Config, + committees: &CommitteeCache, + rules: ForkRules, + balances: &JustifiedBalances, + bound: Slot, + weights: &mut NodeWeights, +) -> Result<()> { + let tree = PayloadTree { store, rules }; + let in_window = |root: &Root| index.get(root).is_some_and(|&(slot, _)| slot >= bound); let boost_root = store.proposer_boost_root(); let justified_root = store.beacon_justified_checkpoint().root; let boost_applies = boost_reaches_the_descent(index, justified_root, boost_root) @@ -2395,7 +2466,7 @@ pub fn compute_node_weights( config, committees, index, - &balances, + balances, |parent| { Ok(weights.raw(ForkChoiceNode { root: parent, @@ -2426,7 +2497,75 @@ pub fn compute_node_weights( root = parent_root; } } + Ok(()) +} +/// [`compute_node_weights`]' answer, with the vote weights read from the +/// store's cached fork-choice weights ([`Store::proto_array_weights`]) instead +/// of tallied from every latest message: the head walk's weights. +/// +/// The cache moves a validator's weight when its vote arrives +/// ([`update_latest_messages`] is one batch), so a head computation no longer +/// pays a pass over every voter. What depends on the clock stays here: the +/// boost ([`add_proposer_boost`], the same code the reference runs) and the +/// previous-slot rule ([`NodeWeights::weight`]). The tests hold this equal to +/// [`compute_node_weights`] on random trees, on every fork-choice fixture's +/// `head` check, and as votes, blocks and equivocations arrive. +/// +/// Under the gloas rules the cache places votes and weight by each block's +/// payload link, so the links of the blocks at or above the bound are +/// recorded first (decoding a block only after a restart lost its link, as +/// the tally did on first use). A block whose link cannot be derived fails the +/// computation only where the tally would: when something at or above the +/// bound needs that link, and then with the derivation's own error. +pub fn cached_node_weights( + store: &Store, + index: &HashMap, + config: &Config, + committees: &CommitteeCache, + rules: ForkRules, +) -> Result { + let balances = justified_balances(store, config)?; + let bound = walk_bound(store, index); + let mut link_errors: HashMap = HashMap::new(); + let weight_rules = match rules { + ForkRules::PreGloas => WeightRules::PreGloas, + ForkRules::Gloas => { + for (&root, &(slot, _)) in index { + if slot >= bound + && let Err(err) = ensure_payload_link(store, root) + { + link_errors.insert(root, err); + } + } + WeightRules::Gloas + } + }; + let weights = store + .proto_array_weights(index, &balances, bound, weight_rules) + .map_err(|err| match err { + ProtoArrayError::MissingLink(root) => link_errors + .remove(&root) + .unwrap_or(Error::SpecAssert("root in store.blocks")), + ProtoArrayError::MissingParentStatus(_) => { + Error::SpecAssert("block.parent_root in store.blocks") + } + })?; + let mut weights = NodeWeights { + rules, + current_slot: get_current_slot(store, config), + weights, + }; + add_proposer_boost( + store, + index, + config, + committees, + rules, + &balances, + bound, + &mut weights, + )?; Ok(weights) } @@ -2472,7 +2611,7 @@ fn walk_children( /// The LMD GHOST head as a [`ForkChoiceNode`], for `rules`: the descent from /// the justified checkpoint that both [`compute_head`] and [`gloas_get_head`] -/// perform, over the one weight table of [`compute_node_weights`]. +/// perform, over the one weight table of [`cached_node_weights`]. /// /// At every step the heaviest child wins, ties going to the higher root and /// then to [`get_payload_status_tiebreaker`], the key gloas's `get_head` sorts @@ -2491,7 +2630,7 @@ fn walk_head( ) -> Result { let tree = PayloadTree { store, rules }; let blocks = get_filtered_block_tree(store, index, config)?; - let weights = compute_node_weights(store, index, config, committees, rules)?; + let weights = cached_node_weights(store, index, config, committees, rules)?; let current_slot = get_current_slot(store, config); let mut children_of: HashMap> = HashMap::new(); @@ -2538,7 +2677,7 @@ fn walk_head( } /// The LMD GHOST head as a full [`ForkChoiceNode`] rather than just a root, -/// from the one bottom-up walk ([`compute_node_weights`] and the descent over +/// from the one bottom-up walk ([`cached_node_weights`] and the descent over /// it) that serves every fork. /// /// The current slot's own fork, via [`Config::fork_at_epoch`], not the head's @@ -4695,6 +4834,11 @@ fn validate_on_attestation_indexed( /// Takes `attesting_indices` and `data` rather than an [`Attestation`]: by /// the time [`on_attestation`] calls this, [`Attestation::verified_attesting_indices`] /// has already resolved the one fork-specific fact this needed out of it. +/// +/// One attestation is one vote batch for the store's cached fork-choice +/// weights: [`Store::record_latest_messages`] records every attester's message +/// and moves their weight under one lock, which is what keeps +/// [`get_head`] from tallying every latest message again. pub fn update_latest_messages( store: &mut Store, attesting_indices: &[ValidatorIndex], @@ -4702,30 +4846,16 @@ pub fn update_latest_messages( rules: ForkRules, ) { let target = data.target; - let beacon_block_root = data.beacon_block_root; - let payload_present = rules == ForkRules::Gloas && data.index == 1; - - for &index in attesting_indices { - if store.is_equivocating(index) { - continue; - } - let should_update = match (store.latest_message(index), rules) { - (None, _) => true, - (Some(existing), ForkRules::PreGloas) => target.epoch > existing.epoch, - (Some(existing), ForkRules::Gloas) => data.slot > existing.slot, - }; - if should_update { - store.set_latest_message( - index, - LatestMessage { - epoch: target.epoch, - slot: data.slot, - root: beacon_block_root, - payload_present, - }, - ); - } - } + let message = LatestMessage { + epoch: target.epoch, + slot: data.slot, + root: data.beacon_block_root, + payload_present: rules == ForkRules::Gloas && data.index == 1, + }; + store.record_latest_messages(attesting_indices, message, |existing| match rules { + ForkRules::PreGloas => target.epoch > existing.epoch, + ForkRules::Gloas => data.slot > existing.slot, + }); } // --------------------------------------------------------------------------- @@ -5059,8 +5189,7 @@ pub fn on_block( // head>`; the first two conditions are cheap and already decide most // blocks, especially every block a follower receives while syncing, which // is never timely. Computing that head (`walk_head`: a filtered-tree walk - // and a pass over every latest message and every block; see - // `compute_weights` for why the votes matter at scale) is not + // over every block, reading vote weights the store keeps cached) is not // worth paying for on a block the shuffling check could not change the // answer for anyway. // @@ -5095,8 +5224,11 @@ pub fn on_block( } // Add new block to the store, and the new state for this block to the - // store. + // store. Its node in the cached fork-choice weights goes in first, so a + // head computation on another thread never finds the block's index row + // without one (which would cost it a full rebuild). let signed_block_el_hash = signed_block.execution_block_hash_if_execution_block(); + store.insert_fork_choice_node(block_root, block_slot, parent_root, payload_link); store .insert_signed_block(block_root, signed_block) .expect("insert"); @@ -9729,6 +9861,7 @@ mod tests { previous_slot_decisions: usize, pre_fork_weights: usize, branches_filtered_out: usize, + cached_weight_tables: usize, } /// Whether `root` is `justified` or a block below it in `index`. A root @@ -9776,6 +9909,20 @@ mod tests { let index = store.block_index(); let walked = get_head_node(store, &config) .unwrap_or_else(|err| panic!("seed {seed} slot {current_slot}: walk failed: {err:?}")); + + // The cached weights the walk just read, against the tally they replace, + // node for node and under the rules the walk used. + let rules = ForkRules::of(config.fork_at_epoch(compute_epoch_at_slot(current_slot))); + let cached = cached_node_weights(store, &index, &config, &committees, rules) + .unwrap_or_else(|err| panic!("seed {seed} slot {current_slot}: cached: {err:?}")); + let tallied = compute_node_weights(store, &index, &config, &committees, rules) + .unwrap_or_else(|err| panic!("seed {seed} slot {current_slot}: tally: {err:?}")); + assert_eq!( + cached, tallied, + "seed {seed} slot {current_slot}: cached weights differ from the tally" + ); + stats.cached_weight_tables += 1; + let justified = store.beacon_justified_checkpoint().root; let bound = walk_bound(store, &index); let justified_subtree = tree @@ -9928,6 +10075,299 @@ mod tests { stats.branches_filtered_out > 0, "the block filter never dropped a branch: {stats:?}" ); + assert!( + stats.cached_weight_tables > 0, + "the cached weights were never compared: {stats:?}" + ); + } + + /// What [`the_cached_weights_follow_votes_blocks_and_equivocations`] did, + /// so it can insist it was not vacuous. + #[derive(Default, Debug)] + struct FollowStats { + vote_batches: usize, + blocks: usize, + equivocations: usize, + weighted_tables: usize, + gloas_steps: usize, + pre_gloas_steps: usize, + } + + /// The cached weights follow the store as the node changes it, without a + /// rebuild: after every vote batch (`update_latest_messages`), block import + /// (`on_block`'s own insertion sequence) and equivocation + /// (`insert_equivocating_index`), the cache equals a rebuild from the same + /// inputs node by node and vote by vote, its weights equal the tally they + /// replace, and the walk over them still names the spec-literal head. Under + /// the pre-gloas rules first, then the gloas ones, which the cache switches + /// to with one rebuild. + #[test] + fn the_cached_weights_follow_votes_blocks_and_equivocations() { + let mut stats = FollowStats::default(); + let gloas_state = test_state::with_validators_at(ForkName::Gloas, TREE_VALIDATORS); + for seed in 0..150 { + let mut tree = random_tree(seed, Arc::new(InMemoryBackend::new())); + let mut rng = TestRng(seed.wrapping_mul(0x5DEE_CE66D) ^ 0xA5A5); + let config = tree.config.clone(); + let latest = tree + .slots + .iter() + .copied() + .max() + .unwrap() + .max(tree.first_gloas_slot); + for current_slot in [tree.first_gloas_slot - 1, latest + 1] { + tree.store + .set_time_ms(current_slot * config.slot_duration_ms) + .unwrap(); + let rules = + ForkRules::of(config.fork_at_epoch(compute_epoch_at_slot(current_slot))); + // The first read builds the cache for these rules; every check + // below then sees only what the steps changed incrementally. + get_head_node(&tree.store, &config) + .unwrap_or_else(|err| panic!("seed {seed} slot {current_slot}: {err:?}")); + + for step in 0..10u8 { + let action = rng.below(4); + if action == 3 && rules == ForkRules::Gloas { + // A block, as `on_block` stores one: node, block, state, + // link, then the per-block records the walk reads. + let parents: Vec = (0..tree.roots.len()) + .filter(|&i| tree.slots[i] < current_slot) + .collect(); + let parent = parents[rng.below(parents.len() as u64) as usize]; + let parent_root = tree.roots[parent]; + let slot = (tree.slots[parent] + 1 + rng.below(2)) + .min(current_slot) + .max(tree.first_gloas_slot); + let parent_hash = match tree.store.get_signed_block(&parent_root).unwrap() { + Some(SignedBeaconBlock::Gloas(inner)) if rng.chance(1, 2) => { + inner + .message + .body + .signed_execution_payload_bid + .message + .block_hash + } + _ => ExecutionBlockHash::repeat_byte(0xe0 + rng.below(8) as u8), + }; + let root = rng.root(); + let block = gloas_block( + slot, + parent_root, + parent_hash, + ExecutionBlockHash::repeat_byte(0x40 + step), + ); + let link = derive_payload_link(&tree.store, &block).unwrap(); + tree.store + .insert_fork_choice_node(root, slot, parent_root, link); + tree.store.insert_signed_block(root, block).unwrap(); + tree.store.insert_state(root, gloas_state.clone()).unwrap(); + tree.store.set_payload_link(root, slot, link); + tree.store + .set_payload_timeliness_vote(root, vec![None; preset::PTC_SIZE]); + tree.store + .set_payload_data_availability_vote(root, vec![None; preset::PTC_SIZE]); + tree.store + .set_unrealized_justification(root, slot, Checkpoint::default()); + if rng.chance(1, 2) { + verify_payload(&mut tree.store, root); + } + tree.roots.push(root); + tree.slots.push(slot); + stats.blocks += 1; + } else if action == 2 { + let validator = rng.below(TREE_VALIDATORS as u64); + tree.store.insert_equivocating_index(validator); + stats.equivocations += 1; + } else { + let pick = rng.below(tree.roots.len() as u64) as usize; + let (root, block_slot) = (tree.roots[pick], tree.slots[pick]); + let vote_slot = block_slot + rng.below(3); + let validators: Vec = (0..TREE_VALIDATORS as u64) + .filter(|_| rng.chance(1, 2)) + .collect(); + let data = AttestationData { + slot: vote_slot, + index: rng.below(2), + beacon_block_root: root, + source: Checkpoint::default(), + target: Checkpoint { + epoch: compute_epoch_at_slot(vote_slot), + root, + }, + }; + update_latest_messages(&mut tree.store, &validators, data, rules); + stats.vote_batches += 1; + } + + let store = &tree.store; + let context = format!("seed {seed} slot {current_slot} step {step}"); + let index = store.block_index(); + store + .proto_array_diff_against_rebuild(&index) + .unwrap_or_else(|err| panic!("{context}: {err}")); + let committees = store.committee_cache(); + let cached = cached_node_weights(store, &index, &config, &committees, rules) + .unwrap_or_else(|err| panic!("{context}: cached: {err:?}")); + let tallied = compute_node_weights(store, &index, &config, &committees, rules) + .unwrap_or_else(|err| panic!("{context}: tally: {err:?}")); + assert_eq!(cached, tallied, "{context}: cached weights differ"); + if cached.weights.values().any(|&weight| weight != 0) { + stats.weighted_tables += 1; + } + + let walked = get_head_node(store, &config) + .unwrap_or_else(|err| panic!("{context}: walk: {err:?}")); + let reference = match rules { + ForkRules::Gloas => gloas_get_head(store, &config, &committees) + .unwrap_or_else(|err| panic!("{context}: reference: {err:?}")), + ForkRules::PreGloas => ForkChoiceNode { + root: compute_head(store, &index, &config) + .unwrap_or_else(|err| panic!("{context}: reference: {err:?}")), + payload_status: PayloadStatus::Full, + }, + }; + assert_eq!(walked, reference, "{context}: head differs"); + match rules { + ForkRules::Gloas => stats.gloas_steps += 1, + ForkRules::PreGloas => stats.pre_gloas_steps += 1, + } + } + } + } + assert!(stats.vote_batches > 0, "no vote batch: {stats:?}"); + assert!(stats.blocks > 0, "no block import: {stats:?}"); + assert!(stats.equivocations > 0, "no equivocation: {stats:?}"); + assert!( + stats.weighted_tables > 0, + "every table was empty: {stats:?}" + ); + assert!( + stats.gloas_steps > 0 && stats.pre_gloas_steps > 0, + "{stats:?}" + ); + } + + /// Mainnet-sized timing of the head computation's weights: the tally + /// (`compute_node_weights`) against the cache (`cached_node_weights`) cold + /// (a rebuild), warm, and after one committee-sized vote batch. Ignored: a + /// measurement, not a check. Run with + /// `cargo test -p ethlambda-state-transition --profile release-fast --lib + /// -- --ignored --nocapture cached_weights_timing`. + #[test] + #[ignore = "benchmark"] + fn cached_weights_timing() { + const VALIDATORS: u64 = 1_000_000; + const CHAIN: u64 = 64; + const RUNS: u32 = 20; + use std::time::Instant; + + let root_of = |n: u64| { + let mut bytes = [0u8; 32]; + bytes[..8].copy_from_slice(&n.to_be_bytes()); + Root::from_slice(&bytes) + }; + let anchor = Root::repeat_byte(0xaa); + let mut store = store_anchored_at(anchor); + store.insert_live_chain_entry(0, anchor, Root::ZERO); + let mut chain = vec![anchor]; + for slot in 1..=CHAIN { + let root = root_of(slot); + store.insert_live_chain_entry(slot, root, chain[slot as usize - 1]); + chain.push(root); + // A one-block fork off every eighth block. + if slot % 8 == 0 { + let fork = root_of(10_000 + slot); + store.insert_live_chain_entry(slot, fork, chain[slot as usize - 1]); + } + } + let checkpoint = store.beacon_justified_checkpoint(); + let increment = preset::EFFECTIVE_BALANCE_INCREMENT; + store.insert_justified_balances(Arc::new(JustifiedBalances::new( + checkpoint, + vec![32 * increment; VALIDATORS as usize].into(), + VALIDATORS * 32 * increment, + vec![32; VALIDATORS as usize].into(), + ))); + let config = Config::active(); + store + .set_time_ms((CHAIN + 1) * config.slot_duration_ms) + .unwrap(); + + // Every validator attests once over the last two epochs, committee by + // committee: one batch per slot, each slot's validators voting for + // that slot's block. + let batch = |slot: u64| -> (Vec, LatestMessage) { + let validators = (0..VALIDATORS).filter(|v| v % CHAIN == slot - 1).collect(); + let message = LatestMessage { + epoch: compute_epoch_at_slot(slot), + slot, + root: chain[slot as usize], + payload_present: false, + }; + (validators, message) + }; + for slot in 1..=CHAIN { + let (validators, message) = batch(slot); + store.record_latest_messages(&validators, message, |existing| { + message.epoch > existing.epoch + }); + } + + let index = store.block_index(); + let committees = store.committee_cache(); + let rules = ForkRules::PreGloas; + let mean = |f: &mut dyn FnMut()| { + let start = Instant::now(); + for _ in 0..RUNS { + f(); + } + start.elapsed() / RUNS + }; + + let tally = mean(&mut || { + compute_node_weights(&store, &index, &config, &committees, rules).unwrap(); + }); + let start = Instant::now(); + let cold = cached_node_weights(&store, &index, &config, &committees, rules).unwrap(); + let rebuild = start.elapsed(); + assert_eq!( + cold, + compute_node_weights(&store, &index, &config, &committees, rules).unwrap() + ); + let warm = mean(&mut || { + cached_node_weights(&store, &index, &config, &committees, rules).unwrap(); + }); + + // The next slot's committee moves its votes from the oldest block to + // the newest: one batch, then the read that refreshes the totals. + let (validators, mut message) = batch(1); + message.epoch += 10; + message.root = chain[CHAIN as usize]; + let start = Instant::now(); + store.record_latest_messages(&validators, message, |existing| { + message.epoch > existing.epoch + }); + let write = start.elapsed(); + let start = Instant::now(); + let after = cached_node_weights(&store, &index, &config, &committees, rules).unwrap(); + let refreshed = start.elapsed(); + assert_eq!( + after, + compute_node_weights(&store, &index, &config, &committees, rules).unwrap() + ); + + eprintln!( + "validators {VALIDATORS}, blocks {}, batch {} votes\n\ + tally (compute_node_weights) mean {tally:?}\n\ + cached, rebuild (first read, one run) {rebuild:?}\n\ + cached, warm read mean {warm:?}\n\ + one vote batch (record_latest_messages) {write:?}\n\ + cached read after the batch {refreshed:?}", + index.len(), + validators.len(), + ); } /// The specification's `filter_block_tree` finds a block's children by diff --git a/crates/blockchain/state_transition/tests/beacon_spec/fork_choice.rs b/crates/blockchain/state_transition/tests/beacon_spec/fork_choice.rs index 4c2f4ea74..1d942825a 100644 --- a/crates/blockchain/state_transition/tests/beacon_spec/fork_choice.rs +++ b/crates/blockchain/state_transition/tests/beacon_spec/fork_choice.rs @@ -715,15 +715,34 @@ fn check_head_against_reference(store: &Store, config: &Config) -> Result<(), St } } }; - if walked == reference { - Ok(()) - } else { - Err(format!( + if walked != reference { + return Err(format!( "the bottom-up walk chose 0x{} ({:?}) but the reference chose 0x{} ({:?})", hex::encode(walked.root.0), walked.payload_status, hex::encode(reference.root.0), reference.payload_status, + )); + } + check_cached_weights(store, config, ForkRules::of(fork)) +} + +/// Checks that the weights the walk reads from the store's cache equal the +/// tally they replaced ([`fork_choice::compute_node_weights`]), node for node: +/// at every `head` check of every fixture, the cache has followed each vote, +/// block and equivocation the fixture fed the store. +fn check_cached_weights(store: &Store, config: &Config, rules: ForkRules) -> Result<(), String> { + let index = store.block_index(); + let committees = store.committee_cache(); + let cached = fork_choice::cached_node_weights(store, &index, config, &committees, rules) + .map_err(|err| format!("cached_node_weights: {err:?}"))?; + let tallied = fork_choice::compute_node_weights(store, &index, config, &committees, rules) + .map_err(|err| format!("compute_node_weights: {err:?}"))?; + if cached == tallied { + Ok(()) + } else { + Err(format!( + "the cached weights {cached:?} differ from the tally {tallied:?}" )) } } diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index 9c25dd644..4f49096f4 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -25,6 +25,12 @@ libssz-types.workspace = true lru.workspace = true xdelta3.workspace = true +# The cached fork-choice weights look a block root up once per vote batch, and +# once per latest message on a rebuild; std's SipHash was one of the largest +# costs of the head computation's tally. Already in the lockfile through other +# crates, so this adds no new version. +foldhash = "0.2.0" + [dev-dependencies] tempfile = "3" proptest = "1" diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 85ad30c5a..4bbed63f3 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -7,6 +7,7 @@ mod error; mod liveness; mod metrics; pub mod pools; +mod proto_array; mod state_codec; mod state_diff; mod state_writer; @@ -21,6 +22,7 @@ pub use committee_cache::{CommitteeCache, Lookup, ShufflingKey}; /// callers can match on it (e.g. to distinguish [`Error::DbVersionMismatch`]). pub use error::Error; pub use liveness::ObservedLiveness; +pub use proto_array::{ProtoArrayError, WeightRules}; // `CacheKey` lives in `state_writer` (beside the `StateCache` it keys), not // `store`; re-exported here so the public path (`ethlambda_storage::CacheKey`) // is unaffected by which module owns it. diff --git a/crates/storage/src/metrics.rs b/crates/storage/src/metrics.rs index 99093a8fb..bb60d687f 100644 --- a/crates/storage/src/metrics.rs +++ b/crates/storage/src/metrics.rs @@ -101,3 +101,35 @@ pub(crate) fn inc_state_cache_lookups(method: StateCacheMethod, key: &CacheKey, .with_label_values(&[method.label(), kind, result]) .inc(); } + +static LEAN_BEACON_PROTO_ARRAY_REBUILDS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "lean_beacon_proto_array_rebuilds_total", + "Full rebuilds of the cached beacon fork-choice weights, by what made the cache stale", + &["reason"] + ) + .unwrap() +}); + +static LEAN_BEACON_PROTO_ARRAY_REBUILD_SECONDS: LazyLock = LazyLock::new(|| { + register_histogram!( + "lean_beacon_proto_array_rebuild_seconds", + "Time to rebuild the cached beacon fork-choice weights from every latest message", + vec![ + 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5 + ] + ) + .unwrap() +}); + +/// Count one rebuild of the cached fork-choice weights, by `reason`. +pub(crate) fn inc_proto_array_rebuilds(reason: &str) { + LEAN_BEACON_PROTO_ARRAY_REBUILDS_TOTAL + .with_label_values(&[reason]) + .inc(); +} + +/// Time one rebuild of the cached fork-choice weights; the guard records on drop. +pub(crate) fn time_proto_array_rebuild() -> TimingGuard { + TimingGuard::new(&LEAN_BEACON_PROTO_ARRAY_REBUILD_SECONDS) +} diff --git a/crates/storage/src/proto_array.rs b/crates/storage/src/proto_array.rs new file mode 100644 index 000000000..7386dc528 --- /dev/null +++ b/crates/storage/src/proto_array.rs @@ -0,0 +1,1090 @@ +//! Cached LMD GHOST weights for the beacon fork choice. +//! +//! A node array over the block tree fork choice weighs (Lighthouse's +//! proto-array layout, one entry per block), whose weights are kept current as +//! votes arrive instead of being re-tallied from every validator's latest +//! message on each head computation. That tally was the one part of the head +//! walk proportional to the validator set; everything else it does is +//! proportional to the unfinalized blocks. +//! +//! # What is cached, and what is not +//! +//! Each block carries the three gloas fork-choice nodes, `PENDING`, `EMPTY` and +//! `FULL`, as two arrays indexed by [`PayloadStatus`]: +//! +//! - `own`: the summed balance of the votes that land on that node itself; +//! - `total`: that node's subtree weight, the value +//! `fork_choice::compute_node_weights` gives it (before the proposer boost). +//! `total[EMPTY]` and `total[FULL]` add the `PENDING` totals of the children +//! that build on that branch; `total[PENDING]` adds both branches. +//! +//! A vote batch updates `own` and marks the array dirty; the next read +//! refreshes every `total` in one reverse pass over the nodes, which is the +//! least work when many batches arrive between two reads. The proposer boost, +//! gloas's previous-slot rule and the walk bound depend on the clock and are +//! applied by the reader, never cached. +//! +//! # Staleness +//! +//! Anything the array cannot follow incrementally marks it stale (finality +//! pruning, a live-window removal, a restart), and so does a read whose +//! balances snapshot, rule set or block index differs from the one the array +//! follows. A stale array ignores votes; the next read rebuilds it from the +//! store's latest messages, which is the tally the head walk used to pay on +//! every call. So incremental updates are an optimization over a rebuild that +//! is always available, and the tests hold the two equal. +//! +//! Each validator's counted vote is remembered by node index (`applied`), not by +//! root: a root removed from the block index can come back (the restart head +//! repair drops rows whose state was never written, and those blocks are +//! imported again), and a root-keyed record would then subtract a vote from a +//! node it was never added to. A removal marks the array stale, so every index +//! it holds stays valid until the rebuild replaces them. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +use ethlambda_types::{ + beacon::{ + fork_choice::{ + BlockPayloadLink, ForkChoiceNode, JustifiedBalances, LatestMessage, PayloadStatus, + }, + primitives::{Gwei, Slot, ValidatorIndex}, + }, + primitives::H256, +}; +use tracing::warn; + +/// The rule set votes are mapped and weight flows under. Mirrors +/// `ethlambda-state-transition`'s `ForkRules`, which this crate cannot name: +/// under pre-gloas rules every block is a single full node whatever its own +/// payload link says, which is how the head walk reads a tree before the gloas +/// fork epoch. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WeightRules { + PreGloas, + Gloas, +} + +/// Why a read rebuilt the array, for the rebuild counter's `reason` label. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RebuildReason { + /// Never built in this process: a fresh store or a restart. + Unbuilt, + /// Finality pruning dropped rows from the block index. + Prune, + /// Rows were removed from the live window (payload invalidation, or the + /// restart head repair). + Removal, + /// The justified checkpoint's balances snapshot changed. + Balances, + /// The current slot crossed into (or, in tests, out of) the gloas rules. + Rules, + /// The block index holds a root the array has no node for: a block + /// written without `Store::insert_fork_choice_node`. + Coverage, + /// A payload link the array lacked at build time is now recorded, and + /// votes already counted on that block depend on it. + Link, + /// A vote's removal would have taken a weight below zero. Only a bug + /// reaches it; the rebuild repairs the array, and the warning names it. + Underflow, +} + +impl RebuildReason { + pub(crate) fn label(self) -> &'static str { + match self { + RebuildReason::Unbuilt => "unbuilt", + RebuildReason::Prune => "prune", + RebuildReason::Removal => "removal", + RebuildReason::Balances => "balances", + RebuildReason::Rules => "rules", + RebuildReason::Coverage => "coverage", + RebuildReason::Link => "link", + RebuildReason::Underflow => "underflow", + } + } +} + +/// Why a read could not produce the weights: the same two conditions on which +/// `fork_choice::compute_node_weights` fails its head computation. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ProtoArrayError { + /// A block at or above the bound needed its payload link (a vote made after + /// its slot, or weight to hand to its parent) and the store has none. + MissingLink(H256), + /// A gloas block at or above the bound, whose parent is too, has weight to + /// hand to a parent branch its link does not name. + MissingParentStatus(H256), +} + +/// Where one validator's vote is counted. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Applied { + node: u32, + status: PayloadStatus, + /// Counted on a block whose payload link was unknown while the vote needed + /// it (made after the block's slot, under the gloas rules). It sits on the + /// `PENDING` node as a placeholder, and the node's `unresolved_votes` + /// counts it, so a read that weighs that block fails the way the tally + /// does instead of answering with a guessed branch. + unresolved: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Node { + root: H256, + slot: Slot, + parent: Option, + /// `None` when neither a recorded link nor the block was available when + /// the node was built; see [`Applied::unresolved`]. + link: Option, + own: [Gwei; 3], + total: [Gwei; 3], + unresolved_votes: u32, +} + +/// One node as [`ProtoArray::diff_against_rebuild`] compares it: the parent +/// named by root, since two arrays can number the same nodes differently. +#[derive(Debug, PartialEq, Eq)] +struct ComparedNode { + slot: Slot, + parent: Option, + own: [Gwei; 3], + total: [Gwei; 3], + unresolved_votes: u32, +} + +/// The cache itself; see the module documentation. +#[derive(Debug, Clone)] +pub(crate) struct ProtoArray { + /// Every parent before its children. + nodes: Vec, + indices: HashMap, + /// By validator index; `None` for a validator whose vote counts nowhere + /// (none yet, an equivocator, a zero balance, or a root with no node). + applied: Vec>, + /// The snapshot every counted vote is weighed by. + balances: Option>, + rules: WeightRules, + /// `Some` while the array must be rebuilt before it is read. + stale: Option, + /// `own` changed since `total` was last refreshed. + dirty: bool, +} + +impl Default for ProtoArray { + fn default() -> Self { + Self { + nodes: Vec::new(), + indices: HashMap::default(), + applied: Vec::new(), + balances: None, + rules: WeightRules::PreGloas, + stale: Some(RebuildReason::Unbuilt), + dirty: false, + } + } +} + +const fn slot_of(status: PayloadStatus) -> usize { + status as usize +} + +const PENDING: usize = slot_of(PayloadStatus::Pending); +const EMPTY: usize = slot_of(PayloadStatus::Empty); +const FULL: usize = slot_of(PayloadStatus::Full); + +impl ProtoArray { + /// Marks the array for a rebuild on its next read, keeping the first + /// reason if it is already stale. + pub(crate) fn mark_stale(&mut self, reason: RebuildReason) { + if self.stale.is_none() { + self.stale = Some(reason); + } + } + + /// Why a read with these inputs must rebuild first, if it must. + pub(crate) fn rebuild_reason( + &self, + index: &HashMap, + balances: &Arc, + rules: WeightRules, + ) -> Option { + if let Some(reason) = self.stale { + return Some(reason); + } + if !self + .balances + .as_ref() + .is_some_and(|current| Arc::ptr_eq(current, balances)) + { + return Some(RebuildReason::Balances); + } + if self.rules != rules { + return Some(RebuildReason::Rules); + } + if index.keys().any(|root| !self.indices.contains_key(root)) { + return Some(RebuildReason::Coverage); + } + // Votes counted while their block's link was unknown sit on a + // placeholder; once [`Self::fill_missing_links`] has found the link, + // only a rebuild puts them on their branch. + if self + .nodes + .iter() + .any(|node| node.unresolved_votes > 0 && node.link.is_some()) + { + return Some(RebuildReason::Link); + } + None + } + + /// Replaces the whole array with one built from scratch: a node per + /// `index` entry and every non-equivocating latest message counted at the + /// node it lands on. `link` answers a block's recorded payload link. + pub(crate) fn rebuild( + &mut self, + index: &HashMap, + link: impl Fn(&H256) -> Option, + latest_messages: &[Option], + equivocating: &HashSet, + balances: Arc, + rules: WeightRules, + ) { + // Ancestors first: a parent's slot is below its child's. + let mut entries: Vec<(Slot, H256, H256)> = index + .iter() + .map(|(&root, &(slot, parent_root))| (slot, root, parent_root)) + .collect(); + entries.sort_unstable(); + + self.nodes = Vec::with_capacity(entries.len()); + self.indices = HashMap::with_capacity_and_hasher(entries.len(), Default::default()); + for (slot, root, parent_root) in entries { + let parent = self.indices.get(&parent_root).copied(); + let node_index = u32::try_from(self.nodes.len()).expect("node count fits u32"); + self.nodes.push(Node { + root, + slot, + parent, + link: link(&root), + own: [0; 3], + total: [0; 3], + unresolved_votes: 0, + }); + self.indices.insert(root, node_index); + } + + self.applied = vec![None; latest_messages.len()]; + self.balances = Some(balances); + self.rules = rules; + self.stale = None; + self.dirty = true; + + let any_equivocators = !equivocating.is_empty(); + for (validator, message) in latest_messages.iter().enumerate() { + let Some(message) = message else { + continue; + }; + let validator_index = validator as ValidatorIndex; + if any_equivocators && equivocating.contains(&validator_index) { + continue; + } + let Some(&node_index) = self.indices.get(&message.root) else { + continue; + }; + let landing = self.landing(node_index, message); + self.move_vote(validator, Some(landing)); + } + } + + /// Adds a block imported after the last rebuild. Ignored while stale, since + /// the rebuild reads every block from the index, and for a root the array + /// already holds. + pub(crate) fn insert_node( + &mut self, + root: H256, + slot: Slot, + parent_root: H256, + link: BlockPayloadLink, + ) { + if self.stale.is_some() || self.indices.contains_key(&root) { + return; + } + let parent = self.indices.get(&parent_root).copied(); + let node_index = u32::try_from(self.nodes.len()).expect("node count fits u32"); + self.nodes.push(Node { + root, + slot, + parent, + link: Some(link), + own: [0; 3], + total: [0; 3], + unresolved_votes: 0, + }); + self.indices.insert(root, node_index); + } + + /// One vote batch: `validators` all now have `message` as their latest + /// message. The caller has already left out equivocators and validators + /// whose existing message is fresher. + pub(crate) fn apply_votes(&mut self, validators: &[ValidatorIndex], message: &LatestMessage) { + if self.stale.is_some() || validators.is_empty() { + return; + } + let landing = self + .indices + .get(&message.root) + .map(|&node_index| self.landing(node_index, message)); + for &validator in validators { + let validator = usize::try_from(validator).expect("validator index fits usize"); + self.move_vote(validator, landing); + } + } + + /// `validator`'s vote no longer counts anywhere (it equivocated). + pub(crate) fn remove_vote(&mut self, validator: ValidatorIndex) { + if self.stale.is_some() { + return; + } + let validator = usize::try_from(validator).expect("validator index fits usize"); + self.move_vote(validator, None); + } + + /// Where `message` lands on the block at `node_index`: the specification's + /// `get_supported_node`. A vote from the block's own slot supports the + /// block before its payload is decided; a later one names a branch, which + /// before gloas (or for a pre-gloas block) is always the full one. + fn landing(&self, node_index: u32, message: &LatestMessage) -> Applied { + let node = &self.nodes[node_index as usize]; + if node.slot >= message.slot { + return Applied { + node: node_index, + status: PayloadStatus::Pending, + unresolved: false, + }; + } + match self.effective_link(node) { + Some(link) => Applied { + node: node_index, + status: if !link.is_gloas() || message.payload_present { + PayloadStatus::Full + } else { + PayloadStatus::Empty + }, + unresolved: false, + }, + None => Applied { + node: node_index, + status: PayloadStatus::Pending, + unresolved: true, + }, + } + } + + fn effective_link(&self, node: &Node) -> Option { + match self.rules { + WeightRules::PreGloas => Some(BlockPayloadLink::PreGloas), + WeightRules::Gloas => node.link, + } + } + + fn balance(&self, validator: usize) -> Gwei { + self.balances + .as_ref() + .map_or(0, |balances| balances.get(validator as ValidatorIndex)) + } + + /// Moves `validator`'s counted vote to `target` (`None`: counted nowhere). + /// A zero balance counts nowhere either, as in the tally. + fn move_vote(&mut self, validator: usize, target: Option) { + let balance = self.balance(validator); + let target = target.filter(|_| balance > 0); + let current = self.applied.get(validator).copied().flatten(); + if current == target { + return; + } + if let Some(current) = current { + let node = &mut self.nodes[current.node as usize]; + let weight = &mut node.own[slot_of(current.status)]; + match weight.checked_sub(balance) { + Some(rest) => *weight = rest, + None => { + warn!( + validator, + balance, + weight = *weight, + "Fork-choice weight would go below zero; rebuilding the cached weights" + ); + *weight = 0; + self.stale = Some(RebuildReason::Underflow); + } + } + if current.unresolved { + node.unresolved_votes = node.unresolved_votes.saturating_sub(1); + } + } + if let Some(target) = target { + let node = &mut self.nodes[target.node as usize]; + let weight = &mut node.own[slot_of(target.status)]; + *weight = weight.saturating_add(balance); + if target.unresolved { + node.unresolved_votes += 1; + } + } + if self.applied.len() <= validator { + self.applied.resize(validator + 1, None); + } + self.applied[validator] = target; + self.dirty = true; + } + + /// Picks up payload links recorded since the nodes were built, for nodes + /// that had none. Call before [`Self::rebuild_reason`]: votes already + /// counted on such a node are left to the rebuild it then asks for + /// (`Link`). + pub(crate) fn fill_missing_links( + &mut self, + payload_links: &HashMap, + ) { + for node in &mut self.nodes { + if node.link.is_none() + && let Some(&(_, link)) = payload_links.get(&node.root) + { + node.link = Some(link); + self.dirty = true; + } + } + } + + /// Recomputes every `total` from `own`, if a vote moved since the last + /// refresh: one pass from the newest node back to the oldest, each node + /// handing its `PENDING` total to the parent branch it builds on. + pub(crate) fn refresh(&mut self) { + if !self.dirty { + return; + } + for node in &mut self.nodes { + node.total = node.own; + } + for node_index in (0..self.nodes.len()).rev() { + let node = &mut self.nodes[node_index]; + node.total[PENDING] = node.total[PENDING] + .saturating_add(node.total[EMPTY]) + .saturating_add(node.total[FULL]); + let (pending, parent, link) = (node.total[PENDING], node.parent, node.link); + let Some(parent) = parent else { + continue; + }; + let link = match self.rules { + WeightRules::PreGloas => Some(BlockPayloadLink::PreGloas), + WeightRules::Gloas => link, + }; + let Some(parent_status) = link.and_then(BlockPayloadLink::parent_status) else { + continue; + }; + let bucket = &mut self.nodes[parent as usize].total[slot_of(parent_status)]; + *bucket = bucket.saturating_add(pending); + } + self.dirty = false; + } + + /// The refreshed totals of every node at or above `bound` that `index` + /// holds, zero totals left out, keyed as the head walk reads them. + /// + /// Fails exactly where the tally fails for the same inputs: a block in the + /// window holds votes that need its unknown payload link, or has weight for + /// a parent in the window that its link cannot place. + pub(crate) fn weights( + &self, + index: &HashMap, + bound: Slot, + ) -> Result, ProtoArrayError> { + debug_assert!( + self.stale.is_none() && !self.dirty, + "read a refreshed array" + ); + let in_window = |node: &Node| node.slot >= bound && index.contains_key(&node.root); + let mut weights = HashMap::with_capacity(self.nodes.len() * 2); + for node in &self.nodes { + if !in_window(node) { + continue; + } + if node.unresolved_votes > 0 { + return Err(ProtoArrayError::MissingLink(node.root)); + } + if node.total[PENDING] != 0 + && let Some(parent) = node.parent + && in_window(&self.nodes[parent as usize]) + { + match self.effective_link(node) { + None => return Err(ProtoArrayError::MissingLink(node.root)), + Some(link) if link.parent_status().is_none() => { + return Err(ProtoArrayError::MissingParentStatus(node.root)); + } + Some(_) => {} + } + } + for payload_status in [ + PayloadStatus::Pending, + PayloadStatus::Empty, + PayloadStatus::Full, + ] { + let weight = node.total[slot_of(payload_status)]; + if weight != 0 { + weights.insert( + ForkChoiceNode { + root: node.root, + payload_status, + }, + weight, + ); + } + } + } + Ok(weights) + } + + /// Compares this array, refreshed, with one rebuilt from the same inputs, + /// node by node and vote by vote, keyed by root since the two can order + /// their nodes differently. For tests: the incremental path must leave the + /// array exactly where a rebuild would. + pub(crate) fn diff_against_rebuild( + &self, + index: &HashMap, + link: impl Fn(&H256) -> Option, + latest_messages: &[Option], + equivocating: &HashSet, + ) -> Result<(), String> { + if self.stale.is_some() { + return Err("the array is stale, so there is nothing incremental to compare".into()); + } + let balances = self.balances.clone().expect("a built array has balances"); + let mut current = self.clone(); + current.refresh(); + let mut rebuilt = ProtoArray::default(); + rebuilt.rebuild( + index, + link, + latest_messages, + equivocating, + balances, + self.rules, + ); + rebuilt.refresh(); + + let by_root = |array: &ProtoArray| -> HashMap { + array + .nodes + .iter() + .map(|node| { + let compared = ComparedNode { + slot: node.slot, + parent: node.parent.map(|parent| array.nodes[parent as usize].root), + own: node.own, + total: node.total, + unresolved_votes: node.unresolved_votes, + }; + (node.root, compared) + }) + .collect() + }; + let (current_nodes, rebuilt_nodes) = (by_root(¤t), by_root(&rebuilt)); + if current_nodes != rebuilt_nodes { + return Err(format!( + "nodes differ:\nincremental {current_nodes:?}\nrebuilt {rebuilt_nodes:?}" + )); + } + + let votes = |array: &ProtoArray| -> Vec> { + let mut votes: Vec<_> = array + .applied + .iter() + .map(|applied| { + applied.map(|applied| { + ( + array.nodes[applied.node as usize].root, + applied.status, + applied.unresolved, + ) + }) + }) + .collect(); + while votes.last() == Some(&None) { + votes.pop(); + } + votes + }; + let (current_votes, rebuilt_votes) = (votes(¤t), votes(&rebuilt)); + if current_votes != rebuilt_votes { + return Err(format!( + "counted votes differ:\nincremental {current_votes:?}\nrebuilt {rebuilt_votes:?}" + )); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use ethlambda_types::beacon::containers::Checkpoint; + use proptest::prelude::*; + + use super::*; + + const VALIDATORS: usize = 10; + + fn root(byte: u8) -> H256 { + H256([byte; 32]) + } + + fn balances(values: &[Gwei]) -> Arc { + let total = values.iter().sum::().max(1); + let increments = vec![0u16; values.len()].into(); + Arc::new(JustifiedBalances::new( + Checkpoint::default(), + values.to_vec().into(), + total, + increments, + )) + } + + /// The store-side state the array follows, kept by the test the way + /// `Store` keeps it, plus a naive reference computation of the weights. + struct World { + /// root -> (slot, parent_root, link), the block index plus links. + blocks: HashMap, + latest_messages: Vec>, + equivocating: HashSet, + balances: Arc, + rules: WeightRules, + array: ProtoArray, + } + + impl World { + fn new() -> Self { + let mut blocks = HashMap::new(); + blocks.insert(root(1), (0, H256::ZERO, BlockPayloadLink::PreGloas)); + World { + blocks, + latest_messages: vec![None; VALIDATORS], + equivocating: HashSet::new(), + balances: balances(&[32; VALIDATORS]), + rules: WeightRules::Gloas, + array: ProtoArray::default(), + } + } + + fn index(&self) -> HashMap { + self.blocks + .iter() + .map(|(&root, &(slot, parent, _))| (root, (slot, parent))) + .collect() + } + + fn links(&self) -> HashMap { + self.blocks + .iter() + .map(|(&root, &(slot, _, link))| (root, (slot, link))) + .collect() + } + + /// What the store's read does: rebuild if needed, refresh, copy out. + fn read(&mut self) -> HashMap { + let index = self.index(); + let links = self.links(); + self.array.fill_missing_links(&links); + if self + .array + .rebuild_reason(&index, &self.balances, self.rules) + .is_some() + { + self.array.rebuild( + &index, + |root| links.get(root).map(|&(_, link)| link), + &self.latest_messages, + &self.equivocating, + Arc::clone(&self.balances), + self.rules, + ); + } + self.array.refresh(); + self.array.weights(&index, 0).expect("every link is known") + } + + /// The tally and fold `compute_node_weights` performs, without the + /// bound or the boost. + fn reference(&self) -> HashMap { + let link = |root: &H256| match self.rules { + WeightRules::PreGloas => BlockPayloadLink::PreGloas, + WeightRules::Gloas => self.blocks[root].2, + }; + let mut weights: HashMap = HashMap::new(); + for (validator, message) in self.latest_messages.iter().enumerate() { + let Some(message) = message else { continue }; + if self.equivocating.contains(&(validator as u64)) { + continue; + } + let balance = self.balances.get(validator as u64); + let Some(&(block_slot, _, _)) = self.blocks.get(&message.root) else { + continue; + }; + if balance == 0 { + continue; + } + let payload_status = if block_slot < message.slot { + if !link(&message.root).is_gloas() || message.payload_present { + PayloadStatus::Full + } else { + PayloadStatus::Empty + } + } else { + PayloadStatus::Pending + }; + *weights + .entry(ForkChoiceNode { + root: message.root, + payload_status, + }) + .or_default() += balance; + } + let mut order: Vec<(Slot, H256, H256)> = self + .blocks + .iter() + .map(|(&root, &(slot, parent, _))| (slot, root, parent)) + .collect(); + order.sort_unstable_by_key(|block| std::cmp::Reverse(block.0)); + for (_, root, parent) in order { + let total: Gwei = [ + PayloadStatus::Pending, + PayloadStatus::Empty, + PayloadStatus::Full, + ] + .into_iter() + .map(|payload_status| { + weights + .get(&ForkChoiceNode { + root, + payload_status, + }) + .copied() + .unwrap_or_default() + }) + .sum(); + if total == 0 { + continue; + } + weights.insert( + ForkChoiceNode { + root, + payload_status: PayloadStatus::Pending, + }, + total, + ); + if self.blocks.contains_key(&parent) { + let parent_status = link(&root).parent_status().expect("known parent status"); + *weights + .entry(ForkChoiceNode { + root: parent, + payload_status: parent_status, + }) + .or_default() += total; + } + } + weights.retain(|_, weight| *weight != 0); + weights + } + + fn apply(&mut self, op: &Op) { + match *op { + Op::Block { + parent, + gap, + gloas, + on_full, + } => { + let mut roots: Vec = self.blocks.keys().copied().collect(); + roots.sort_unstable(); + let parent_root = roots[parent % roots.len()]; + let slot = self.blocks[&parent_root].0 + gap; + let new_root = root(self.blocks.len() as u8 + 1); + let link = if gloas { + BlockPayloadLink::Gloas { + parent_status: Some(if on_full { + PayloadStatus::Full + } else { + PayloadStatus::Empty + }), + } + } else { + BlockPayloadLink::PreGloas + }; + // `on_block`'s order: the node, then the row. + self.array.insert_node(new_root, slot, parent_root, link); + self.blocks.insert(new_root, (slot, parent_root, link)); + } + Op::Votes { + mask, + block, + later, + payload_present, + } => { + let mut roots: Vec = self.blocks.keys().copied().collect(); + roots.sort_unstable(); + let target = roots[block % roots.len()]; + let message = LatestMessage { + epoch: 0, + slot: self.blocks[&target].0 + later, + root: target, + payload_present, + }; + // `Store::record_latest_messages` with the gloas ordering. + let mut changed = Vec::new(); + for validator in 0..VALIDATORS { + if mask & (1 << validator) == 0 + || self.equivocating.contains(&(validator as u64)) + { + continue; + } + let replace = self.latest_messages[validator] + .is_none_or(|existing| message.slot > existing.slot); + if replace { + self.latest_messages[validator] = Some(message); + changed.push(validator as u64); + } + } + self.array.apply_votes(&changed, &message); + } + Op::Equivocate { validator } => { + let validator = (validator % VALIDATORS) as u64; + self.equivocating.insert(validator); + self.array.remove_vote(validator); + } + Op::Balances { seed } => { + let values: Vec = (0..VALIDATORS as u64) + .map(|validator| (seed.wrapping_mul(validator + 7) >> 3) % 5 * 8) + .collect(); + self.balances = balances(&values); + } + Op::Rules { gloas } => { + self.rules = if gloas { + WeightRules::Gloas + } else { + WeightRules::PreGloas + }; + } + Op::Stale => self.array.mark_stale(RebuildReason::Removal), + Op::Read => {} + } + } + } + + #[derive(Debug, Clone)] + enum Op { + Block { + parent: usize, + gap: u64, + gloas: bool, + on_full: bool, + }, + Votes { + mask: u16, + block: usize, + later: u64, + payload_present: bool, + }, + Equivocate { + validator: usize, + }, + Balances { + seed: u64, + }, + Rules { + gloas: bool, + }, + Stale, + Read, + } + + fn op() -> impl Strategy { + prop_oneof![ + 4 => (any::(), 1..4u64, any::(), any::()).prop_map( + |(parent, gap, gloas, on_full)| Op::Block { parent, gap, gloas, on_full } + ), + 8 => (any::(), any::(), 0..3u64, any::()).prop_map( + |(mask, block, later, payload_present)| Op::Votes { mask, block, later, payload_present } + ), + 1 => any::().prop_map(|validator| Op::Equivocate { validator }), + 1 => any::().prop_map(|seed| Op::Balances { seed }), + 1 => any::().prop_map(|gloas| Op::Rules { gloas }), + 1 => Just(Op::Stale), + 3 => Just(Op::Read), + ] + } + + proptest! { + /// After every operation the cached weights equal the reference tally, + /// and the incrementally kept array equals a fresh rebuild. + #[test] + fn cached_weights_match_a_fresh_tally(ops in proptest::collection::vec(op(), 1..60)) { + let mut world = World::new(); + world.read(); + for op in &ops { + world.apply(op); + // The incremental state, before the read can rebuild it. The + // rebuild it is compared with uses the array's own balances + // and rules, so a pending balance or rule change does not + // excuse a difference. + if world.array.stale.is_none() { + let index = world.index(); + let links = world.links(); + world + .array + .diff_against_rebuild( + &index, + |root| links.get(root).map(|&(_, link)| link), + &world.latest_messages, + &world.equivocating, + ) + .map_err(|err| TestCaseError::fail(format!("after {op:?}: {err}")))?; + } + let cached = world.read(); + prop_assert_eq!(cached, world.reference(), "after {:?}", op); + } + } + } + + #[test] + fn a_fresh_array_is_stale_and_ignores_votes() { + let mut array = ProtoArray::default(); + let message = LatestMessage { + epoch: 0, + slot: 1, + root: root(1), + payload_present: false, + }; + array.apply_votes(&[0, 1], &message); + array.insert_node(root(2), 1, root(1), BlockPayloadLink::PreGloas); + assert!(array.nodes.is_empty()); + assert!(array.applied.is_empty()); + assert_eq!(array.stale, Some(RebuildReason::Unbuilt)); + } + + #[test] + fn a_vote_moving_between_branches_moves_its_weight() { + let mut world = World::new(); + world.apply(&Op::Block { + parent: 0, + gap: 1, + gloas: true, + on_full: true, + }); + world.read(); + let child = root(2); + let vote = |later, payload_present| Op::Votes { + mask: 0b1, + block: 1, + later, + payload_present, + }; + world.apply(&vote(1, false)); + let weights = world.read(); + assert_eq!( + weights.get(&ForkChoiceNode { + root: child, + payload_status: PayloadStatus::Empty + }), + Some(&32) + ); + // A later message for the full branch: same root, one slot on. + world.apply(&vote(2, true)); + let weights = world.read(); + assert_eq!(weights, world.reference()); + assert_eq!( + weights.get(&ForkChoiceNode { + root: child, + payload_status: PayloadStatus::Full + }), + Some(&32) + ); + assert!(!weights.contains_key(&ForkChoiceNode { + root: child, + payload_status: PayloadStatus::Empty + })); + // The child builds on the anchor's full branch. + assert_eq!( + weights.get(&ForkChoiceNode { + root: root(1), + payload_status: PayloadStatus::Full + }), + Some(&32) + ); + } + + #[test] + fn an_underflow_marks_the_array_for_a_rebuild() { + let mut world = World::new(); + world.read(); + world.apply(&Op::Votes { + mask: 0b1, + block: 0, + later: 0, + payload_present: false, + }); + // Corrupt the count the way only a bug could. + world.array.nodes[0].own[PENDING] = 0; + world.array.remove_vote(0); + assert_eq!(world.array.stale, Some(RebuildReason::Underflow)); + world.equivocating.insert(0); + assert_eq!(world.read(), world.reference()); + } + + #[test] + fn a_vote_needing_an_unknown_link_fails_the_read_only_inside_the_window() { + let mut array = ProtoArray::default(); + let mut index = HashMap::new(); + index.insert(root(1), (0, H256::ZERO)); + index.insert(root(2), (5, root(1))); + let message = LatestMessage { + epoch: 0, + slot: 6, + root: root(2), + payload_present: true, + }; + array.rebuild( + &index, + |_| None, + &[Some(message)], + &HashSet::new(), + balances(&[32]), + WeightRules::Gloas, + ); + array.refresh(); + assert_eq!( + array.weights(&index, 0), + Err(ProtoArrayError::MissingLink(root(2))) + ); + // Above the block: nothing in the window needs the link. + assert!(array.weights(&index, 6).unwrap().is_empty()); + + // Once the link is recorded, the read asks for a rebuild and then + // places the vote on its branch. + let mut links = HashMap::new(); + links.insert( + root(2), + ( + 5, + BlockPayloadLink::Gloas { + parent_status: Some(PayloadStatus::Empty), + }, + ), + ); + array.fill_missing_links(&links); + let snapshot = Arc::clone(array.balances.as_ref().unwrap()); + assert_eq!( + array.rebuild_reason(&index, &snapshot, WeightRules::Gloas), + Some(RebuildReason::Link) + ); + } +} diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 14ac9116f..34cd8b65e 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -12,6 +12,7 @@ use crate::committee_cache::CommitteeCache; use crate::error::Error; use crate::liveness::ObservedLiveness; use crate::pools::{AttestationPool, OperationPool}; +use crate::proto_array::{ProtoArray, ProtoArrayError, RebuildReason, WeightRules}; use ethlambda_crypto::signature::ValidatorSignature; use ethlambda_types::{ @@ -29,8 +30,8 @@ use ethlambda_types::{ }, fork::ForkName, fork_choice::{ - BlockPayloadLink, JustifiedBalances, LatestMessage, PayloadStatus, PayloadStatusEnum, - PayloadStatusV1, PowBlock, + BlockPayloadLink, ForkChoiceNode, JustifiedBalances, LatestMessage, PayloadStatus, + PayloadStatusEnum, PayloadStatusV1, PowBlock, }, preset::{PTC_SIZE, Preset, SLOTS_PER_EPOCH}, primitives::ExecutionBlockHash, @@ -809,6 +810,12 @@ pub(crate) struct BeaconScratch { /// snapshot balances in index order, not hash order. `None` is a validator /// that has not voted (or an index below the highest one seen). pub(crate) latest_messages: Vec>, + /// The head walk's weights, kept current as `latest_messages` and + /// `equivocating_indices` change (see [`crate::proto_array`]). Beside them + /// under the one lock, so a vote and its weight change are one step. + /// Scratch like its inputs: a restart starts it stale, and the first head + /// computation rebuilds it from the latest messages it has by then. + pub(crate) proto_array: ProtoArray, /// Epoch-boundary balances, flattened for the vote loop /// and keyed by their own checkpoint. A derived cache: a miss is rebuilt /// from `checkpoint_states`, so nothing needs to persist it. @@ -2408,6 +2415,12 @@ impl Store { .delete_batch(Table::LiveChain, keys_to_delete) .expect("delete non-finalized chain entries"); batch.commit().expect("commit"); + // The cached fork-choice weights drop the same blocks by rebuilding. + self.beacon + .lock() + .unwrap() + .proto_array + .mark_stale(RebuildReason::Prune); Ok(count) } @@ -2434,10 +2447,11 @@ impl Store { /// the blocks either side of them at the same slots must survive. /// /// Dropping the row is the whole of "remove this block from fork choice": - /// [`block_index`](Self::block_index) is the only source + /// [`block_index`](Self::block_index) is the only block source /// `filter_block_tree`, `compute_node_weights` and `walk_head` read, so a /// root with no row contributes no weight to any ancestor and can never be - /// walked to. + /// walked to. The cached fork-choice weights are marked stale here, so the + /// next head computation rebuilds them from the index without the rows. /// /// The block and its state stay in their own tables. Nothing reads them /// once the index row is gone, and keeping them means an operator can still @@ -2455,6 +2469,14 @@ impl Store { .delete_batch(Table::LiveChain, keys) .expect("delete live chain entries"); batch.commit().expect("commit"); + // A removed block's votes must stop counting for its ancestors. The + // cached fork-choice weights drop them by rebuilding, which also + // forgets the node indices the removal invalidates. + self.beacon + .lock() + .unwrap() + .proto_array + .mark_stale(RebuildReason::Removal); } /// Prune gossip signatures for slots <= finalized_slot. @@ -3886,13 +3908,12 @@ impl Store { .contains(&index) } - /// Marks `index` as equivocating. + /// Marks `index` as equivocating, and takes its vote out of the cached + /// fork-choice weights: an equivocator's vote counts for neither side. pub fn insert_equivocating_index(&mut self, index: u64) { - self.beacon - .lock() - .unwrap() - .equivocating_indices - .insert(index); + let mut beacon = self.beacon.lock().unwrap(); + beacon.equivocating_indices.insert(index); + beacon.proto_array.remove_vote(index); } /// Gloas: whether `root`'s execution payload envelope has been locally @@ -4268,17 +4289,167 @@ impl Store { /// attestation (a member of a committee of the registry), never from raw /// input: a wild index would allocate a table to match. The assertion is /// the tripwire for that in debug builds and fixtures. + /// + /// A batch of one for the cached fork-choice weights; an equivocator's + /// message is recorded but counts nowhere. The production path records a + /// whole attestation at once through [`Store::record_latest_messages`]. pub fn set_latest_message(&mut self, index: u64, message: LatestMessage) { debug_assert!( index < MAX_PLAUSIBLE_VALIDATOR_INDEX, "validator index {index} would grow the dense vote table far past any registry" ); let slot = usize::try_from(index).expect("validator index fits in usize"); - let mut beacon = self.beacon.lock().unwrap(); + let mut guard = self.beacon.lock().unwrap(); + let beacon = &mut *guard; if beacon.latest_messages.len() <= slot { beacon.latest_messages.resize(slot + 1, None); } beacon.latest_messages[slot] = Some(message); + let equivocating = !beacon.equivocating_indices.is_empty() + && beacon.equivocating_indices.contains(&index); + if equivocating { + beacon.proto_array.remove_vote(index); + } else { + beacon.proto_array.apply_votes(&[index], &message); + } + } + + /// Records `message` as the latest message of every validator in + /// `indices` that is not a known equivocator and whose current latest + /// message `should_replace` answers `true` for (a validator with none is + /// always recorded), and moves their weight in the cached fork-choice + /// weights: one vote batch, under one lock. + /// + /// `should_replace` is the fork's ordering of messages, which this crate + /// cannot name: target epoch before gloas, slot from gloas on. + /// + /// The indices must come from a validated attestation, for the reason + /// [`Store::set_latest_message`] gives. + pub fn record_latest_messages( + &mut self, + indices: &[u64], + message: LatestMessage, + should_replace: impl Fn(&LatestMessage) -> bool, + ) { + let mut guard = self.beacon.lock().unwrap(); + let beacon = &mut *guard; + let mut recorded = Vec::with_capacity(indices.len()); + // Most of the time nobody has equivocated, so skip the set probe per + // attester, as `for_each_non_equivocating_latest_message` does. + let any_equivocators = !beacon.equivocating_indices.is_empty(); + for &index in indices { + debug_assert!( + index < MAX_PLAUSIBLE_VALIDATOR_INDEX, + "validator index {index} would grow the dense vote table far past any registry" + ); + if any_equivocators && beacon.equivocating_indices.contains(&index) { + continue; + } + let slot = usize::try_from(index).expect("validator index fits in usize"); + if beacon.latest_messages.len() <= slot { + beacon.latest_messages.resize(slot + 1, None); + } + let entry = &mut beacon.latest_messages[slot]; + if entry.as_ref().is_none_or(&should_replace) { + *entry = Some(message); + recorded.push(index); + } + } + beacon.proto_array.apply_votes(&recorded, &message); + } + + /// Adds a block to the cached fork-choice weights, for `on_block` to call + /// just before it stores the block itself. In that order, a head + /// computation on another thread never reads a block index row whose node + /// is missing, which would cost it a full rebuild. + /// + /// A block stored any other way still counts: the next head computation + /// finds its row without a node and rebuilds. + pub fn insert_fork_choice_node( + &self, + root: H256, + slot: u64, + parent_root: H256, + link: BlockPayloadLink, + ) { + self.beacon + .lock() + .unwrap() + .proto_array + .insert_node(root, slot, parent_root, link); + } + + /// The head walk's node weights from the cached fork-choice weights: + /// every node at or above `bound` that `index` holds, before the proposer + /// boost, with zero weights left out. The same map + /// `fork_choice::compute_node_weights` builds from its tally, which is what + /// this replaces on the head computation's path. + /// + /// Rebuilds the cache first when it is stale or does not follow these + /// inputs (another balances snapshot, the other rule set, a block in + /// `index` it has no node for), and refreshes its subtree totals when a + /// vote moved since the last read. Payload links are read from this + /// store's records: under the gloas rules the caller records those of the + /// blocks at or above `bound` first. + /// + /// Holds the scratch lock throughout, calling nothing that takes it again. + pub fn proto_array_weights( + &self, + index: &HashMap, + balances: &Arc, + bound: u64, + rules: WeightRules, + ) -> Result, ProtoArrayError> { + let mut guard = self.beacon.lock().unwrap(); + let BeaconScratch { + proto_array, + payload_links, + latest_messages, + equivocating_indices, + .. + } = &mut *guard; + proto_array.fill_missing_links(payload_links); + if let Some(reason) = proto_array.rebuild_reason(index, balances, rules) { + crate::metrics::inc_proto_array_rebuilds(reason.label()); + let _timing = crate::metrics::time_proto_array_rebuild(); + // Rebuilt from a scan made now, under the lock, not from the + // caller's `index`: a caller on another thread scanned before + // locking, and a removal committed in between (whose stale mark + // comes after its commit) would otherwise be rebuilt back in and + // its mark cleared. The chain actor's own `index` is this scan, + // since nothing else writes the table. The caller's `index` still + // filters what is read out below. + let live_chain = self.get_live_chain().expect("live chain scan"); + proto_array.rebuild( + &live_chain, + |root| payload_links.get(root).map(|&(_, link)| link), + latest_messages, + equivocating_indices, + Arc::clone(balances), + rules, + ); + } + proto_array.refresh(); + proto_array.weights(index, bound) + } + + /// Test support: whether the cached fork-choice weights, as votes, blocks + /// and equivocations have moved them since their last rebuild, equal a + /// rebuild from the same inputs, node by node and vote by vote. `Err` + /// describes the first difference, or says the cache is stale (nothing + /// incremental to compare). `index` is the block index the cache follows. + #[doc(hidden)] + pub fn proto_array_diff_against_rebuild( + &self, + index: &HashMap, + ) -> Result<(), String> { + let beacon = self.beacon.lock().unwrap(); + beacon.proto_array.diff_against_rebuild( + index, + |root| beacon.payload_links.get(root).map(|&(_, link)| link), + &beacon.latest_messages, + &beacon.equivocating_indices, + ) } /// Gloas: the payload timeliness committee's per-member votes on whether diff --git a/docs/beacon_engine.md b/docs/beacon_engine.md index 1aa13b184..333e70a04 100644 --- a/docs/beacon_engine.md +++ b/docs/beacon_engine.md @@ -108,9 +108,11 @@ take the head from a `VALID` one on attestation weight, and then requires the head to fall back when the `SYNCING` branch turns out to be invalid. An invalidated block leaves fork choice by having its `LiveChain` index row -deleted. That is sufficient because `Store::block_index()` is the only source -`filter_block_tree`, `compute_node_weights` and `walk_head` read: a root with no -row contributes no weight to any ancestor and can never be walked to. The block +deleted. That is sufficient because `Store::block_index()` is the only block +source `filter_block_tree`, `compute_node_weights` and `walk_head` read: a root +with no row contributes no weight to any ancestor and can never be walked to. +The deletion also marks the store's cached fork-choice weights stale, so the next +head computation rebuilds them without the removed blocks. The block and its state stay in their own tables, so an operator can still inspect what was rejected. diff --git a/docs/metrics.md b/docs/metrics.md index 7107778d1..76458dc9e 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -335,6 +335,20 @@ In practice the distribution is bimodal and dominated by production rather than | `lean_state_write_queue_depth` | Gauge | States handed to the background writer but not yet committed | On every hand-off and on every commit | | | `lean_state_write_seconds` | Histogram | Time the background writer spends encoding, diffing and committing one state | Per state written | | | `lean_state_cache_lookups_total` | Counter | State cache lookups, by the `Store` method that made them, the kind of state asked for, and whether the cache held it | On every `get_state` (and the writer's own parent read), `has_state` and `cached_state` call | method=get_state,has_state,cached_state
kind=block,checkpoint
result=hit,miss | +| `lean_beacon_proto_array_rebuilds_total` | Counter | Full rebuilds of the cached beacon fork-choice weights, by what made the cache stale | On a head computation that found the cache stale or not following its inputs | reason=unbuilt,prune,removal,balances,rules,coverage,link,underflow | +| `lean_beacon_proto_array_rebuild_seconds` | Histogram | Time to rebuild the cached beacon fork-choice weights from every latest message | Per rebuild | | + +**`lean_beacon_proto_array_rebuilds_total` should move about once per epoch.** +The beacon head computation reads vote weights the store keeps current as votes +arrive, and rebuilds them from every latest message only when its inputs move +under it: `balances` when the justified checkpoint moves, `prune` when finality +prunes the block index (the two usually land together, and one rebuild covers +both), `unbuilt` once per process. `removal` follows a payload invalidation or the +restart head repair, `rules` the gloas fork boundary. `coverage` means a block +reached the index without `Store::insert_fork_choice_node`: `on_block` always +inserts one, so outside tests and bootstrap a steady rate of it is worth a look +(an API read racing a block import can cost one now and then). `underflow` is a +bug: the cache repaired itself, and a warning in the log names the validator. **On a beacon follower, watch `lean_table_bytes{table="data_columns"}`.** Every other table's series either stays flat or is bounded by pruning; `data_columns` diff --git a/docs/spec_deviations.md b/docs/spec_deviations.md index 191be46cd..bdcbdf779 100644 --- a/docs/spec_deviations.md +++ b/docs/spec_deviations.md @@ -402,8 +402,9 @@ Three places in the gloas head read something the specification would abort on, because a live node holds gaps a fixture tree never has. Each is marked "Implementation choice, not spec text" at its definition. The head the node runs is `walk_head` (through `get_head_node`) over the weights of -`compute_node_weights`; the spec-literal references carry the same tolerances so -that the tests compare like with like. +`cached_node_weights`, the store's cached vote weights, which the tests hold +equal to `compute_node_weights`; the spec-literal references carry the same +tolerances so that the tests compare like with like. - **Pruned votes** (`compute_node_weights`; in the references, `get_attestation_score`, and through it `is_head_weak` and From b951d472d774fbcdac3469a1bfcf877bf0f86a96 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:48:14 -0300 Subject: [PATCH 2/2] style(storage): rustfmt the equivocating check in set_latest_message Claude-Session: https://claude.ai/code/session_011vghNmndKPkZinHRkVTwVB --- crates/storage/src/store.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 34cd8b65e..924dcaba6 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -4305,8 +4305,8 @@ impl Store { beacon.latest_messages.resize(slot + 1, None); } beacon.latest_messages[slot] = Some(message); - let equivocating = !beacon.equivocating_indices.is_empty() - && beacon.equivocating_indices.contains(&index); + let equivocating = + !beacon.equivocating_indices.is_empty() && beacon.equivocating_indices.contains(&index); if equivocating { beacon.proto_array.remove_vote(index); } else {