From 63fa159e6608617edf826e1908e5eda1b0b1923e Mon Sep 17 00:00:00 2001 From: bomanaps Date: Fri, 10 Jul 2026 10:28:51 +0100 Subject: [PATCH 1/4] add opt-1, opt-2, single-flight, leanSpec fixes --- lean_client/Cargo.lock | 30 +-- lean_client/Cargo.toml | 12 +- lean_client/containers/src/slot.rs | 15 +- lean_client/containers/src/state.rs | 145 +++++++++++- lean_client/fork_choice/src/handlers.rs | 33 ++- lean_client/fork_choice/src/store.rs | 55 +++-- .../fork_choice/tests/unit_tests/validator.rs | 47 ++-- lean_client/src/main.rs | 221 +++++++++++++++++- 8 files changed, 470 insertions(+), 88 deletions(-) diff --git a/lean_client/Cargo.lock b/lean_client/Cargo.lock index 276680bf..fb127322 100644 --- a/lean_client/Cargo.lock +++ b/lean_client/Cargo.lock @@ -160,7 +160,7 @@ checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" [[package]] name = "arithmetic" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "easy-ext", "typenum", @@ -705,7 +705,7 @@ dependencies = [ [[package]] name = "bls" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "bls-blst", "bls-core", @@ -714,7 +714,7 @@ dependencies = [ [[package]] name = "bls-blst" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "bls-core", "blst", @@ -737,7 +737,7 @@ dependencies = [ [[package]] name = "bls-core" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "derivative", "hex", @@ -1805,7 +1805,7 @@ dependencies = [ [[package]] name = "features" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "enum-iterator", "log", @@ -2199,7 +2199,7 @@ checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" [[package]] name = "hashing" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "ethereum-types", "generic-array", @@ -2391,7 +2391,7 @@ dependencies = [ [[package]] name = "http_api_utils" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "anyhow", "axum", @@ -3462,7 +3462,7 @@ checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" [[package]] name = "logging" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "derive_more", "tracing", @@ -4558,7 +4558,7 @@ dependencies = [ [[package]] name = "prometheus_metrics" version = "0.1.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "anyhow", "features", @@ -5403,7 +5403,7 @@ dependencies = [ [[package]] name = "serde_utils" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "const-hex", "generic-array", @@ -5658,7 +5658,7 @@ dependencies = [ [[package]] name = "ssz" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "arithmetic", "bit_field", @@ -5690,7 +5690,7 @@ dependencies = [ [[package]] name = "ssz_derive" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "darling 0.21.3", "easy-ext", @@ -5716,7 +5716,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "std_ext" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "easy-ext", "triomphe", @@ -6302,7 +6302,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "try_from_iterator" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" [[package]] name = "twox-hash" @@ -6319,7 +6319,7 @@ checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" [[package]] name = "types" version = "0.0.0" -source = "git+https://github.com/grandinetech/grandine?rev=64afdee3c6be79fceffb66933dcb69a943f3f1ae#64afdee3c6be79fceffb66933dcb69a943f3f1ae" +source = "git+https://github.com/grandinetech/grandine?rev=c4b676e3daa0ddcb86d0bcb321b7166d59f920f9#c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" dependencies = [ "anyhow", "arithmetic", diff --git a/lean_client/Cargo.toml b/lean_client/Cargo.toml index 83eb13e9..53e70523 100644 --- a/lean_client/Cargo.toml +++ b/lean_client/Cargo.toml @@ -238,7 +238,7 @@ anyhow = "1.0.100" async-trait = "0.1" axum = "0.8.8" bitvec = "1.0.1" -bls = { git = "https://github.com/grandinetech/grandine", package = "bls", features = ["blst"], rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } +bls = { git = "https://github.com/grandinetech/grandine", package = "bls", features = ["blst"], rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" } clap = { version = "4", features = ["derive"] } derive_more = "2.1.1" discv5 = "0.10.2" @@ -246,12 +246,12 @@ enr = { version = "0.13", features = ["k256"] } eth_ssz = { package = "ethereum_ssz", version = "0.10.0" } ethereum-types = "0.14" futures = "0.3" -features = { git = "https://github.com/grandinetech/grandine", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } +features = { git = "https://github.com/grandinetech/grandine", rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" } git-version = "0.3" hex = "0.4.3" indexmap = "2" http-body-util = "0.1" -http_api_utils = { git = "https://github.com/grandinetech/grandine", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } +http_api_utils = { git = "https://github.com/grandinetech/grandine", rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" } k256 = "0.13" rec_aggregation = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e2592df4e30fdddbbf8ae26a333116c68cec7026" } backend = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e2592df4e30fdddbbf8ae26a333116c68cec7026" } @@ -279,7 +279,7 @@ parking_lot = "0.12" paste = "1.0.15" pretty_assertions = "1.4" prometheus = { version = "0.14", features = ["process"] } -prometheus_metrics = { git = "https://github.com/grandinetech/grandine", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } +prometheus_metrics = { git = "https://github.com/grandinetech/grandine", rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" } rand = "0.10" rand_chacha = "0.10" rayon = "1" @@ -290,7 +290,7 @@ serde_json = "1.0" serde_yaml = "0.9" sha2 = "0.10" snap = "1.1" -ssz = { git = "https://github.com/grandinetech/grandine", package = "ssz", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } +ssz = { git = "https://github.com/grandinetech/grandine", package = "ssz", rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" } ssz-types = "0.3" itertools = "0.14" test-generator = "0.3.1" @@ -303,7 +303,7 @@ tower-http = { version = '0.6', features = ['cors', 'trace'] } tracing = "0.1.41" tracing-subscriber = { version = "0.3.20", features = ["env-filter"] } tree-hash = "0.4.0" -try_from_iterator = { git = "https://github.com/grandinetech/grandine", package = "try_from_iterator", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } +try_from_iterator = { git = "https://github.com/grandinetech/grandine", package = "try_from_iterator", rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" } typenum = "1.19" yamux = "0.12" zeroize = "1.8" diff --git a/lean_client/containers/src/slot.rs b/lean_client/containers/src/slot.rs index cda0d335..a2dfe0f9 100644 --- a/lean_client/containers/src/slot.rs +++ b/lean_client/containers/src/slot.rs @@ -35,17 +35,12 @@ impl Slot { /// /// # Returns /// - /// True if the slot is justifiable, False otherwise. - /// - /// # Panics - /// - /// Panics if this slot is earlier than the finalized slot. + /// True if the slot is justifiable, False otherwise (including any slot + /// before the finalized slot). pub fn is_justifiable_after(self, finalized: Slot) -> bool { - assert!( - self >= finalized, - "Candidate slot must not be before finalized slot" - ); - let delta = self.0 - finalized.0; + let Some(delta) = self.0.checked_sub(finalized.0) else { + return false; + }; // Rule 1: The first 5 slots after finalization are always justifiable. // Examples: delta = 0, 1, 2, 3, 4, 5 diff --git a/lean_client/containers/src/state.rs b/lean_client/containers/src/state.rs index 318d08fb..5c206ed3 100644 --- a/lean_client/containers/src/state.rs +++ b/lean_client/containers/src/state.rs @@ -451,6 +451,20 @@ impl State { "zero hash is not allowed in justification roots" ); + let validator_count = self.validators.len_usize(); + ensure!( + validator_count > 0, + "state holds no validators to segment justification votes against" + ); + + let expected_vote_count = self.justifications_roots.len_usize() * validator_count; + ensure!( + self.justifications_validators.len() == expected_vote_count, + "justification vote list length {} does not equal tracked-root count times validator count {}", + self.justifications_validators.len(), + expected_vote_count, + ); + let mut justifications = self .justifications_roots .into_iter() @@ -458,8 +472,7 @@ impl State { .map(|(i, root)| { ( root.clone(), - self.justifications_validators - [i * self.validators.len_usize()..(i + 1) * self.validators.len_usize()] + self.justifications_validators[i * validator_count..(i + 1) * validator_count] .to_bitvec(), ) }) @@ -585,13 +598,17 @@ impl State { { justified_slots = justified_slots.shift_window(delta); - ensure!( - justifications - .keys() - .all(|root| root_to_slot.contains_key(root)), - "Justification root missing from root_to_slot" - ); - justifications.retain(|root, _| root_to_slot[root].0 > finalized_slot.0); + justifications.retain(|root, _| match root_to_slot.get(root) { + Some(slot) => slot.0 > finalized_slot.0, + None => { + warn!( + root = %root, + finalized_slot = finalized_slot.0, + "Justification root missing from root_to_slot, pruning" + ); + false + } + }); } } // justified_slots = justified_slots @@ -652,6 +669,7 @@ impl State { known_block_roots: &HashSet, aggregated_payloads: &HashMap)>, log_inv_rate: usize, + enable_proposer_aggregation: bool, ) -> Result<( Block, Self, @@ -805,7 +823,12 @@ impl State { .with_label_values(&["compact"]) .start_timer() }); - let compacted = self.compact_proofs_by_data(selected, log_inv_rate)?; + let compacted = if enable_proposer_aggregation { + self.compact_proofs_by_data(selected, log_inv_rate)? + } else { + let running_votes = build_running_votes(self); + keep_best_proof_per_data(selected, &running_votes, slot.0) + }; drop(compact_timer); METRICS.get().map(|metrics| { @@ -966,6 +989,108 @@ fn is_proposer_for(validator_index: u64, slot: Slot, num_validators: u64) -> boo slot.0 % num_validators == validator_index } +/// Read `state.justifications_validators` into a per-target-root voter set. +/// Layout: `bit[i * N + j]` = validator `j` has voted for +/// `justifications_roots[i]`, where `N = validators.len()`. +pub(crate) fn build_running_votes(state: &State) -> HashMap> { + let validator_count = state.validators.len_usize(); + let mut votes: HashMap> = HashMap::new(); + for (i, root) in (&state.justifications_roots).into_iter().enumerate() { + let mut voters: HashSet = HashSet::new(); + for j in 0..validator_count { + if state + .justifications_validators + .get(i * validator_count + j) + .map(|b| *b) + .unwrap_or(false) + { + voters.insert(j as u64); + } + } + votes.insert(*root, voters); + } + votes +} + +/// Keep one proof per AttestationData without merging. +/// +/// For each group of entries sharing an AttestationData, pick the proof with +/// the largest count of voters that are neither already in `running_votes` +/// for the target root nor already claimed by an earlier same-target-root +/// group in this block. Ties broken by the densest proof (most set bits), +/// then by earliest occurrence in the input. +pub(crate) fn keep_best_proof_per_data( + entries: Vec<(AggregatedAttestation, AggregatedSignatureProof)>, + running_votes: &HashMap>, + block_slot: u64, +) -> Vec<(AggregatedAttestation, AggregatedSignatureProof)> { + let mut order: Vec = Vec::new(); + let mut groups: HashMap> = HashMap::new(); + for (i, (att, _)) in entries.iter().enumerate() { + let dr = att.data.hash_tree_root(); + match groups.entry(dr) { + std::collections::hash_map::Entry::Vacant(e) => { + order.push(*e.key()); + e.insert(vec![i]); + } + std::collections::hash_map::Entry::Occupied(mut e) => { + e.get_mut().push(i); + } + } + } + + if order.len() == entries.len() { + return entries; + } + + info!( + slot = block_slot, + entries = entries.len(), + unique = order.len(), + "Skipping attestation compaction" + ); + + let mut claimed: HashMap> = HashMap::new(); + let mut best_per_data: Vec = Vec::with_capacity(order.len()); + for dr in &order { + let group = &groups[dr]; + let target_root = entries[group[0]].0.data.target.root; + let prior = running_votes.get(&target_root); + let block_claimed = claimed.get(&target_root); + let best = *group + .iter() + .max_by_key(|&&idx| { + let (att, proof) = &entries[idx]; + let marginal = proof + .get_participant_indices() + .into_iter() + .filter(|vid| { + prior.map_or(true, |voted| !voted.contains(vid)) + && block_claimed.map_or(true, |voted| !voted.contains(vid)) + }) + .count(); + ( + marginal, + att.aggregation_bits.to_validator_indices().len(), + std::cmp::Reverse(idx), + ) + }) + .expect("group is non-empty"); + claimed + .entry(target_root) + .or_default() + .extend(entries[best].1.get_participant_indices()); + best_per_data.push(best); + } + + let mut items: Vec> = + entries.into_iter().map(Some).collect(); + best_per_data + .into_iter() + .map(|idx| items[idx].take().expect("best index taken once")) + .collect() +} + #[cfg(test)] mod tests { use super::*; diff --git a/lean_client/fork_choice/src/handlers.rs b/lean_client/fork_choice/src/handlers.rs index e6fcf7fd..01e7ef25 100644 --- a/lean_client/fork_choice/src/handlers.rs +++ b/lean_client/fork_choice/src/handlers.rs @@ -14,8 +14,9 @@ use xmss::PublicKey; use crate::block_cache::BlockCache; use crate::store::{ - BLOCKS_TO_KEEP, GOSSIP_DISPARITY_INTERVALS, HEAD_RETENTION_SLOTS, INTERVALS_PER_SLOT, - MILLIS_PER_INTERVAL, STATE_PRUNE_BUFFER, STATES_TO_KEEP, Store, tick_interval, update_head, + BLOCKS_TO_KEEP, GOSSIP_DISPARITY_INTERVALS, HEAD_RETENTION_SLOTS, HISTORICAL_ROOTS_LIMIT, + INTERVALS_PER_SLOT, MILLIS_PER_INTERVAL, STATE_PRUNE_BUFFER, STATES_TO_KEEP, Store, + tick_interval, update_head, }; #[inline] @@ -98,6 +99,14 @@ fn validate_attestation_data(store: &Store, data: &AttestationData) -> Result<() data.head.root, data.head.slot.0, ); + ensure!( + checkpoint_is_ancestor(store, &store.latest_finalized, &data.head), + "Head checkpoint {} (slot {}) does not descend from latest finalized {} (slot {})", + data.head.root, + data.head.slot.0, + store.latest_finalized.root, + store.latest_finalized.slot.0, + ); // Honest validators emit votes only after their slot has begun. Allow exactly // one interval (~800 ms) of clock skew between peers; a whole-slot margin would @@ -673,6 +682,26 @@ pub fn on_block( ); } + let block_slot = signed_block.block.slot.0; + if let Some(parent_state) = store.states.get(&parent_root) { + let slot_gap = block_slot.saturating_sub(parent_state.slot.0); + ensure!( + slot_gap <= HISTORICAL_ROOTS_LIMIT, + "block slot gap {} exceeds HISTORICAL_ROOTS_LIMIT {} (block.slot={}, parent.slot={})", + slot_gap, + HISTORICAL_ROOTS_LIMIT, + block_slot, + parent_state.slot.0, + ); + } + let current_slot = store.time / INTERVALS_PER_SLOT; + ensure!( + block_slot <= current_slot + 1, + "block slot {} is more than one slot beyond current slot {}", + block_slot, + current_slot, + ); + process_block_internal(store, signed_block, block_root, verify_signatures)?; process_pending_blocks(store, cache, vec![block_root], verify_signatures); diff --git a/lean_client/fork_choice/src/store.rs b/lean_client/fork_choice/src/store.rs index 6951b9c2..18ffe095 100644 --- a/lean_client/fork_choice/src/store.rs +++ b/lean_client/fork_choice/src/store.rs @@ -24,6 +24,10 @@ pub const MILLIS_PER_INTERVAL: u64 = (SECONDS_PER_SLOT * 1000) / INTERVALS_PER_S /// analogue of mainnet's MAXIMUM_GOSSIP_CLOCK_DISPARITY. pub const GOSSIP_DISPARITY_INTERVALS: u64 = 1; +/// Maximum permitted slot gap between a block and its parent. +/// Must equal the `HistoricalRootsLimit` typenum in `containers/src/state.rs`. +pub const HISTORICAL_ROOTS_LIMIT: u64 = 262_144; + /// Forkchoice store tracking chain state and validator attestations #[derive(Debug, Clone, Default)] @@ -78,16 +82,13 @@ pub struct Store { /// Aggregated signature proofs from block bodies (on-chain). /// These are attestations that have been included in blocks and are part of /// the "known" pool for safe target computation. - /// Keyed by attestation data root (H256). `IndexMap` preserves insertion - /// order so same-slot equivocation tie-breaks are deterministic and match - /// leanSpec's first-vote-wins semantics (Python dict insertion order). + /// Keyed by attestation data root (H256). pub latest_known_aggregated_payloads: IndexMap>, /// Aggregated signature proofs from gossip aggregation topic. /// These are newly received aggregations that haven't been migrated to "known" yet. /// At interval 3, we merge this with latest_known_aggregated_payloads for safe target. - /// Keyed by attestation data root (H256). See note on the `known` pool above - /// for why this is `IndexMap`. + /// Keyed by attestation data root (H256). pub latest_new_aggregated_payloads: IndexMap>, /// Attestation data indexed by hash (data_root). @@ -498,28 +499,26 @@ fn extract_attestations_from_aggregated_payloads( ) -> HashMap { let mut attestations: HashMap = HashMap::new(); - for (data_root, proofs) in payloads { - // Look up the attestation data for this data root - let Some(attestation_data) = attestation_data_by_root.get(data_root) else { - continue; - }; + let mut ordered: Vec<(&H256, &AttestationData, &Vec)> = payloads + .iter() + .filter_map(|(root, proofs)| { + attestation_data_by_root + .get(root) + .map(|data| (root, data, proofs)) + }) + .filter(|(_, data, _)| data.head.slot > latest_finalized_slot) + .collect(); - if attestation_data.head.slot <= latest_finalized_slot { - continue; - } + ordered.sort_unstable_by(|a, b| (b.1.slot, b.0).cmp(&(a.1.slot, a.0))); - // For each proof, extract participating validators + for (_data_root, attestation_data, proofs) in ordered { for proof in proofs { for (bit_idx, bit) in proof.participants.0.iter().enumerate() { if *bit { let validator_id = bit_idx as u64; - // Only update if this is a newer attestation for this validator - if attestations - .get(&validator_id) - .map_or(true, |existing| existing.slot < attestation_data.slot) - { - attestations.insert(validator_id, attestation_data.clone()); - } + attestations + .entry(validator_id) + .or_insert_with(|| attestation_data.clone()); } } } @@ -639,6 +638,7 @@ pub struct BlockProductionInputs { /// counted in `lean_build_block_pool_missing_att_data`. pub aggregated_payloads: HashMap)>, pub log_inv_rate: usize, + pub enable_proposer_aggregation: bool, pub store_latest_justified: Checkpoint, } @@ -647,6 +647,7 @@ pub fn prepare_block_production( slot: Slot, validator_index: u64, log_inv_rate: usize, + enable_proposer_aggregation: bool, ) -> Result { let head_root = get_proposal_head(store, slot); let head_state = store @@ -697,6 +698,7 @@ pub fn prepare_block_production( known_block_roots, aggregated_payloads, log_inv_rate, + enable_proposer_aggregation, store_latest_justified: store.latest_justified.clone(), }) } @@ -712,6 +714,7 @@ pub fn execute_block_production( known_block_roots, aggregated_payloads, log_inv_rate, + enable_proposer_aggregation, store_latest_justified, } = inputs; @@ -740,6 +743,7 @@ pub fn execute_block_production( &known_block_roots, &aggregated_payloads, log_inv_rate, + enable_proposer_aggregation, )?; info!( @@ -772,7 +776,14 @@ pub fn produce_block_with_signatures( slot: Slot, validator_index: u64, log_inv_rate: usize, + enable_proposer_aggregation: bool, ) -> Result<(H256, Block, Vec)> { - let inputs = prepare_block_production(store, slot, validator_index, log_inv_rate)?; + let inputs = prepare_block_production( + store, + slot, + validator_index, + log_inv_rate, + enable_proposer_aggregation, + )?; execute_block_production(inputs).map(|(root, block, _post_state, sigs)| (root, block, sigs)) } diff --git a/lean_client/fork_choice/tests/unit_tests/validator.rs b/lean_client/fork_choice/tests/unit_tests/validator.rs index 97e5f37d..70da5388 100644 --- a/lean_client/fork_choice/tests/unit_tests/validator.rs +++ b/lean_client/fork_choice/tests/unit_tests/validator.rs @@ -139,7 +139,7 @@ fn test_produce_block_basic() { let validator_idx = 1; let (block_root, block, _signatures) = - produce_block_with_signatures(&mut store, slot, validator_idx, 1) + produce_block_with_signatures(&mut store, slot, validator_idx, 1, true) .expect("block production should succeed"); // Verify block structure @@ -161,7 +161,7 @@ fn test_produce_block_unauthorized_proposer() { let slot = Slot(1); let wrong_validator = 2; // Not proposer for slot 1 - let result = produce_block_with_signatures(&mut store, slot, wrong_validator, 1); + let result = produce_block_with_signatures(&mut store, slot, wrong_validator, 1, true); assert!(result.is_err()); let err = format!("{:?}", result.unwrap_err()); assert!( @@ -195,7 +195,7 @@ fn test_produce_block_with_attestations() { let validator_idx = 2; let (_root, block, signatures) = - produce_block_with_signatures(&mut store, slot, validator_idx, 1) + produce_block_with_signatures(&mut store, slot, validator_idx, 1, true) .expect("block production should succeed"); // Block should include the 2 attestations we added (validators 5 and 6). @@ -243,7 +243,8 @@ fn test_produce_block_sequential_slots() { // Produce block for slot 1 let (block1_root, block1, _sig1) = - produce_block_with_signatures(&mut store, Slot(1), 1, 1).expect("block1 should succeed"); + produce_block_with_signatures(&mut store, Slot(1), 1, 1, true) + .expect("block1 should succeed"); // Verify first block is properly created assert_eq!(block1.slot, Slot(1)); @@ -258,7 +259,8 @@ fn test_produce_block_sequential_slots() { // Produce block for slot 2 (will build on genesis due to forkchoice) let (block2_root, block2, _sig2) = - produce_block_with_signatures(&mut store, Slot(2), 2, 1).expect("block2 should succeed"); + produce_block_with_signatures(&mut store, Slot(2), 2, 1, true) + .expect("block2 should succeed"); // Verify block properties assert_eq!(block2.slot, Slot(2)); @@ -286,8 +288,9 @@ fn test_produce_block_empty_attestations() { let slot = Slot(3); let validator_idx = 3; - let (_root, block, _sig) = produce_block_with_signatures(&mut store, slot, validator_idx, 1) - .expect("block production should succeed"); + let (_root, block, _sig) = + produce_block_with_signatures(&mut store, slot, validator_idx, 1, true) + .expect("block production should succeed"); // Should produce valid block with empty attestations assert_eq!(block.body.attestations.len_usize(), 0); @@ -320,7 +323,7 @@ fn test_produce_block_state_consistency() { let validator_idx = 4; let (block_root, block, signatures) = - produce_block_with_signatures(&mut store, slot, validator_idx, 1) + produce_block_with_signatures(&mut store, slot, validator_idx, 1, true) .expect("block production should succeed"); apply_block(&mut store, &block); @@ -372,8 +375,8 @@ fn test_block_production_then_attestation() { let mut store = create_test_store(); // Proposer produces block for slot 1 - let (_root, _block, _sig) = - produce_block_with_signatures(&mut store, Slot(1), 1, 1).expect("block should succeed"); + let (_root, _block, _sig) = produce_block_with_signatures(&mut store, Slot(1), 1, 1, true) + .expect("block should succeed"); // Update store state after block production update_head(&mut store); @@ -413,7 +416,8 @@ fn test_multiple_validators_coordination() { // Validator 1 produces block for slot 1 let (block1_root, block1, _sig1) = - produce_block_with_signatures(&mut store, Slot(1), 1, 1).expect("block1 should succeed"); + produce_block_with_signatures(&mut store, Slot(1), 1, 1, true) + .expect("block1 should succeed"); let block1_hash = block1_root; apply_block(&mut store, &block1); @@ -443,7 +447,8 @@ fn test_multiple_validators_coordination() { // After processing block1, head should be block1 (fork choice walks the tree) // So block2 will build on block1 let (block2_root, block2, _sig2) = - produce_block_with_signatures(&mut store, Slot(2), 2, 1).expect("block2 should succeed"); + produce_block_with_signatures(&mut store, Slot(2), 2, 1, true) + .expect("block2 should succeed"); // Verify block properties assert_eq!(block2.slot, Slot(2)); @@ -477,8 +482,9 @@ fn test_validator_edge_cases() { let slot = Slot(9); // This validator's slot // Should be able to produce block - let (_root, block, _sig) = produce_block_with_signatures(&mut store, slot, max_validator, 1) - .expect("max validator block should succeed"); + let (_root, block, _sig) = + produce_block_with_signatures(&mut store, slot, max_validator, 1, true) + .expect("max validator block should succeed"); assert_eq!(block.proposer_index, max_validator); // Should be able to produce attestation @@ -517,8 +523,8 @@ fn test_validator_operations_empty_store() { let mut store = get_forkchoice_store(state, signed_block, config, true, 1); // Should be able to produce block and attestation - let (_root, block, _sig) = - produce_block_with_signatures(&mut store, Slot(1), 1, 1).expect("block should succeed"); + let (_root, block, _sig) = produce_block_with_signatures(&mut store, Slot(1), 1, 1, true) + .expect("block should succeed"); let attestation_data = store .produce_attestation_data(Slot(1)) .expect("failed to produce attestation data"); @@ -541,7 +547,7 @@ fn test_produce_block_wrong_proposer() { let slot = Slot(5); let wrong_proposer = 3; // Should be validator 5 for slot 5 - let result = produce_block_with_signatures(&mut store, slot, wrong_proposer, 1); + let result = produce_block_with_signatures(&mut store, slot, wrong_proposer, 1, true); assert!(result.is_err()); assert!(format!("{:?}", result.unwrap_err()).contains("is not the proposer for slot")); } @@ -567,7 +573,7 @@ fn test_produce_block_missing_parent_state() { }; let mut s = store; - let result = produce_block_with_signatures(&mut s, Slot(1), 1, 1); + let result = produce_block_with_signatures(&mut s, Slot(1), 1, 1, true); assert!(result.is_err()); } @@ -636,7 +642,8 @@ fn produce_and_apply( let proposer = slot.0 % num_validators; let _ = keys; let (_block_root, block, _sigs) = - produce_block_with_signatures(store, slot, proposer, 1).expect("block production failed"); + produce_block_with_signatures(&mut store, slot, proposer, 1, true) + .expect("block production failed"); let signed = SignedBlock { block, proof: MultiMessageAggregate::default(), @@ -769,7 +776,7 @@ fn test_produce_block_closes_justification_gap() { let proposer_7 = Slot(7).0 % num_validators; let (_block_7_root, block_7, _sigs_7) = - produce_block_with_signatures(&mut store, Slot(7), proposer_7, 1) + produce_block_with_signatures(&mut store, Slot(7), proposer_7, 1, true) .expect("block production for block_7 failed"); assert_eq!(block_7.parent_root, block_5_root); diff --git a/lean_client/src/main.rs b/lean_client/src/main.rs index de675c45..657be947 100644 --- a/lean_client/src/main.rs +++ b/lean_client/src/main.rs @@ -420,6 +420,14 @@ struct Args { #[arg(long)] checkpoint_sync_url: Option, + /// Merge same-`AttestationData` proofs into a single recursive proof at + /// proposal time. Off by default: the proposer keeps only the single + /// best-coverage proof per attestation data and skips the leanVM merge, + /// which trades a small per-block coverage loss for a lower proposal + /// deadline miss rate. + #[arg(long, default_value_t = false)] + enable_proposer_aggregation: bool, + #[cfg(shadow_mode)] #[command(flatten)] shadow: ShadowOptions, @@ -1153,6 +1161,7 @@ async fn main() -> Result<()> { }; let chain_log_inv_rate = genesis_log_inv_rate as usize; + let chain_enable_proposer_aggregation = args.enable_proposer_aggregation; // ── Aggregation background task ──────────────────────────────────────────── // XMSS aggregation takes 1-3 s. Running it inside the tick loop via .await @@ -1195,6 +1204,11 @@ async fn main() -> Result<()> { // Tracks the previous tick instant so the duration between consecutive // chain-task ticks can be observed at the top of each tick arm. let mut last_tick_instant: Option = None; + // Single-flight guard for ProduceBlock: caps concurrent block-build + // spawns to 1. Messages arriving while a build is in flight are + // rejected via the oneshot sender, which surfaces to the caller as an + // rx error; the caller can retry on the next tick. + let mut current_produce_block: Option<(Slot, tokio::task::JoinHandle<()>)> = None; let mut block_cache = BlockCache::new(); let mut sync_state = if has_aggregator { SyncState::Syncing @@ -1881,10 +1895,32 @@ async fn main() -> Result<()> { METRICS.get().map(|m| m.grandine_validator_chain_message_channel_depth.dec()); match v_message { ValidatorChainMessage::ProduceBlock { slot, proposer_index, sender } => { + if let Some((_, ref h)) = current_produce_block { + if h.is_finished() { + current_produce_block = None; + } + } + if let Some((in_flight_slot, _)) = ¤t_produce_block { + let in_flight = in_flight_slot.0; + info!( + new_slot = slot.0, + in_flight_slot = in_flight, + "ProduceBlock skipped: previous build still in flight" + ); + let _ = sender.send(Err(anyhow::anyhow!( + "produce_block dropped: previous build still in flight for slot {in_flight}" + ))); + } else { let block_build_start = Instant::now(); let prepare_result = { let mut w = store.write(); - prepare_block_production(&mut *w, slot, proposer_index, chain_log_inv_rate) + prepare_block_production( + &mut *w, + slot, + proposer_index, + chain_log_inv_rate, + chain_enable_proposer_aggregation, + ) }; match prepare_result { @@ -1897,7 +1933,7 @@ async fn main() -> Result<()> { let validators = inputs.head_state.validators.clone(); let exec = cpu_normal_executor.clone(); let exec_start = Instant::now(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { let job = exec.spawn(async move { execute_block_production(inputs) .map(|(_, block, post_state, sigs)| (block, post_state, sigs)) @@ -1942,8 +1978,10 @@ async fn main() -> Result<()> { } let _ = sender.send(result); }); + current_produce_block = Some((slot, handle)); } } + } } ValidatorChainMessage::BuildAttestationData { slot, sender } => { let store_read = store.read(); @@ -1981,6 +2019,8 @@ async fn main() -> Result<()> { let mut last_proposal_slot: Option = None; let mut last_attestation_slot: Option = None; let mut propose_handle: Option> = None; + let mut prebuild_handle: Option> = None; + let last_prebuild_success: Arc>> = Arc::new(Mutex::new(None)); loop { v_tick_interval.tick().await; @@ -1996,7 +2036,8 @@ async fn main() -> Result<()> { match current_interval { 0 => { - if last_proposal_slot != Some(current_slot) { + let prebuilt = *last_prebuild_success.lock() == Some(current_slot); + if last_proposal_slot != Some(current_slot) && !prebuilt { if current_slot > 0 { if let Some(proposer_idx) = vs.get_proposer_for_slot(Slot(current_slot)) { @@ -2009,6 +2050,15 @@ async fn main() -> Result<()> { ); } } + if let Some(prev) = prebuild_handle.take() { + if !prev.is_finished() { + prev.abort(); + info!( + slot = current_slot, + "Validator task: aborting stale pre-build before interval-0 fallback" + ); + } + } info!( slot = current_slot, @@ -2087,6 +2137,24 @@ async fn main() -> Result<()> { { Ok(signed_block) => { let block_root = signed_block.block.hash_tree_root(); + let target_slot_start_ms = genesis_millis_for_spawn + + propose_slot + * (MILLIS_PER_INTERVAL * INTERVALS_PER_SLOT); + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() + as u64; + if now_ms < target_slot_start_ms { + let sleep_ms = target_slot_start_ms - now_ms; + info!( + slot = propose_slot, + sleep_ms, + "Validator task: block built early, holding until slot boundary" + ); + tokio::time::sleep(Duration::from_millis(sleep_ms)) + .await; + } info!( slot = propose_slot, block_root = %format!("0x{:x}", block_root), @@ -2220,6 +2288,153 @@ async fn main() -> Result<()> { last_attestation_slot = Some(current_slot); } } + 4 => { + // Pre-build the next slot's block one interval early if we + // propose it. The marker is set synchronously at dispatch, + // before the spawn is created, so interval-0 of the next + // slot always sees "pre-build in progress" and no-ops. + // If any step in the spawn errors, the marker is cleared + // in the error arm so a later tick can retry — but there + // is no interval-0 fallback for the failing slot itself. + let next_slot = current_slot + 1; + if *last_prebuild_success.lock() != Some(next_slot) { + if let Some(next_proposer_idx) = vs.get_proposer_for_slot(Slot(next_slot)) { + info!( + slot = next_slot, + proposer = next_proposer_idx, + "Validator task: pre-building block one interval early" + ); + + let (tx, rx) = oneshot::channel(); + let send_result = + validator_chain_sender.send(ValidatorChainMessage::ProduceBlock { + slot: Slot(next_slot), + proposer_index: next_proposer_idx, + sender: tx, + }); + if send_result.is_ok() { + METRICS.get().map(|m| { + m.grandine_validator_chain_message_channel_depth.inc() + }); + } + if send_result.is_err() { + warn!("Validator task: chain channel closed, stopping"); + break; + } + + *last_prebuild_success.lock() = Some(next_slot); + + let vs_for_spawn = vs.clone(); + let chain_sender_for_spawn = chain_msg_sender_for_validator.clone(); + let log_inv_rate = chain_log_inv_rate; + let propose_slot = next_slot; + let genesis_millis_for_spawn = genesis_millis; + let last_prebuild_success_for_spawn = + Arc::clone(&last_prebuild_success); + + prebuild_handle = Some(tokio::spawn(async move { + let (block, signatures, validators, cached_post_state) = match rx + .await + { + Ok(Ok(tuple)) => tuple, + Ok(Err(e)) => { + warn!(slot = propose_slot, error = %e, "Validator task: chain failed to pre-build block"); + let mut guard = last_prebuild_success_for_spawn.lock(); + if *guard == Some(propose_slot) { + *guard = None; + } + return; + } + Err(_) => { + warn!( + slot = propose_slot, + "Validator task: no response to pre-build ProduceBlock" + ); + let mut guard = last_prebuild_success_for_spawn.lock(); + if *guard == Some(propose_slot) { + *guard = None; + } + return; + } + }; + + match vs_for_spawn + .sign_block_with_data( + block, + next_proposer_idx, + signatures, + validators, + log_inv_rate, + ) + .await + { + Ok(signed_block) => { + let block_root = signed_block.block.hash_tree_root(); + let target_slot_start_ms = genesis_millis_for_spawn + + propose_slot + * (MILLIS_PER_INTERVAL * INTERVALS_PER_SLOT); + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() + as u64; + if now_ms < target_slot_start_ms { + let sleep_ms = target_slot_start_ms - now_ms; + info!( + slot = propose_slot, + sleep_ms, + "Validator task: pre-built block ready, holding until slot boundary" + ); + tokio::time::sleep(Duration::from_millis(sleep_ms)) + .await; + } else { + warn!( + slot = propose_slot, + overrun_ms = now_ms - target_slot_start_ms, + "Pre-build overran target slot, publishing immediately" + ); + } + info!( + slot = propose_slot, + block_root = %format!("0x{:x}", block_root), + "Validator task: pre-built block signed, sending to chain" + ); + let send_result = chain_sender_for_spawn + .send(ChainMessage::ProcessBlock { + signed_block, + is_trusted: true, + should_gossip: true, + cached_post_state: Some(cached_post_state), + }) + .await; + if send_result.is_ok() { + METRICS.get().map(|m| { + m.grandine_chain_message_channel_depth.inc() + }); + } + if send_result.is_err() { + warn!( + slot = propose_slot, + "Validator task: chain message channel closed after pre-build" + ); + let mut guard = last_prebuild_success_for_spawn.lock(); + if *guard == Some(propose_slot) { + *guard = None; + } + } + } + Err(e) => { + warn!(slot = propose_slot, error = %e, "Validator task: failed to sign pre-built block"); + let mut guard = last_prebuild_success_for_spawn.lock(); + if *guard == Some(propose_slot) { + *guard = None; + } + } + } + })); + } + } + } _ => {} } } From ee1c5a020e5a3e1610fb8e881f4421e6275f7c85 Mon Sep 17 00:00:00 2001 From: bomanaps Date: Sun, 12 Jul 2026 12:50:43 +0100 Subject: [PATCH 2/4] metric scope, retention, single-flight --- lean_client/fork_choice/src/handlers.rs | 6 --- lean_client/metrics/src/metrics.rs | 13 +++++++ lean_client/src/main.rs | 14 +++++-- lean_client/validator/src/lib.rs | 50 ++++++++++++++++--------- 4 files changed, 57 insertions(+), 26 deletions(-) diff --git a/lean_client/fork_choice/src/handlers.rs b/lean_client/fork_choice/src/handlers.rs index 01e7ef25..61eea8c1 100644 --- a/lean_client/fork_choice/src/handlers.rs +++ b/lean_client/fork_choice/src/handlers.rs @@ -721,12 +721,6 @@ pub fn verify_and_transition( signed_block: SignedBlock, verify_signatures: bool, ) -> Result { - let _timer = METRICS.get().map(|metrics| { - metrics - .lean_fork_choice_block_processing_time_seconds - .start_timer() - }); - if verify_signatures { signed_block.verify_signatures(&parent_state)?; } diff --git a/lean_client/metrics/src/metrics.rs b/lean_client/metrics/src/metrics.rs index 51917685..18512de5 100644 --- a/lean_client/metrics/src/metrics.rs +++ b/lean_client/metrics/src/metrics.rs @@ -233,6 +233,11 @@ pub struct Metrics { /// of 1 mean the worker is continuously busy (XMSS slower than slot rate). pub lean_aggregation_in_flight_snapshots: IntGauge, + /// Per-job pre-SNARK setup wall-clock: children_arg construction, validator + /// index sort/dedup, participants bitfield, raw pubkey/sig cloning. + /// Observed once per job in `maybe_aggregate` right before the SNARK call. + pub lean_aggregation_job_setup_seconds: Histogram, + /// Time the chain task spends processing one `ChainMessage` (block, /// attestation, aggregated attestation, etc.). Captures total work done /// inside the `chain_message_receiver.recv() => { … }` select arm body @@ -662,6 +667,13 @@ impl Metrics { "lean_aggregation_in_flight_snapshots", "Snapshots currently held by the aggregation spawn_blocking worker (0 or 1 expected)", )?, + lean_aggregation_job_setup_seconds: Histogram::with_opts(histogram_opts!( + "lean_aggregation_job_setup_seconds", + "Per-job pre-SNARK setup wall-clock in maybe_aggregate before aggregate_with_children", + vec![ + 0.0001, 0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5 + ], + ))?, lean_chain_task_chain_message_seconds: Histogram::with_opts(histogram_opts!( "lean_chain_task_chain_message_seconds", "Wall-clock time the chain task spends inside one ChainMessage select-arm body", @@ -966,6 +978,7 @@ impl Metrics { self.lean_aggregation_snapshot_size_entries.clone(), ))?; default_registry.register(Box::new(self.lean_aggregation_in_flight_snapshots.clone()))?; + default_registry.register(Box::new(self.lean_aggregation_job_setup_seconds.clone()))?; default_registry.register(Box::new(self.lean_chain_task_chain_message_seconds.clone()))?; default_registry.register(Box::new(self.lean_chain_task_apply_seconds.clone()))?; default_registry.register(Box::new(self.grandine_store_blocks_size.clone()))?; diff --git a/lean_client/src/main.rs b/lean_client/src/main.rs index 657be947..44bef973 100644 --- a/lean_client/src/main.rs +++ b/lean_client/src/main.rs @@ -564,7 +564,7 @@ async fn main() -> Result<()> { // (output discarded). `DedicatedExecutor::Drop` joins worker threads, so no // verify thread leaks past process shutdown. let (verify_result_tx, mut verify_result_rx) = - mpsc::channel::<(H256, SignedBlock, bool, Result)>(1024); + mpsc::channel::<(H256, SignedBlock, bool, Result, Instant)>(1024); let (genesis_time, validators, genesis_log_inv_rate, genesis_attestation_committee_count) = if let Some(genesis_path) = &args.genesis { @@ -1365,7 +1365,7 @@ async fn main() -> Result<()> { // would pick it every iteration, and verified blocks would never land in // store.blocks (observed live: depth 200+ growing, store_blocks_size frozen). result = verify_result_rx.recv() => { - let Some((block_root, signed_block, should_gossip, outcome)) = result else { break }; + let Some((block_root, signed_block, should_gossip, outcome, block_processing_start)) = result else { break }; METRICS .get() .map(|m| m.grandine_verify_result_channel_depth.dec()); @@ -1416,6 +1416,11 @@ async fn main() -> Result<()> { block_root, ); + METRICS.get().map(|m| { + m.lean_fork_choice_block_processing_time_seconds + .observe(block_processing_start.elapsed().as_secs_f64()) + }); + match apply_result { Ok(()) => { info!("Block processed successfully"); @@ -1490,6 +1495,7 @@ async fn main() -> Result<()> { let parent_state_for_child = cascade_parent_state.clone(); let child_for_verify = child_block.clone(); + let cascade_block_processing_start = Instant::now(); METRICS.get().map(|m| { m.grandine_verify_result_channel_depth.inc() }); @@ -1511,6 +1517,7 @@ async fn main() -> Result<()> { child_for_send, false, // cascade: never re-gossip outcome, + cascade_block_processing_start, )) .await .is_err() @@ -1729,6 +1736,7 @@ async fn main() -> Result<()> { let result_tx = verify_result_tx.clone(); let signed_block_for_verify = signed_block.clone(); let verify_signatures = !is_trusted; + let block_processing_start = Instant::now(); METRICS .get() .map(|m| m.grandine_verify_result_channel_depth.inc()); @@ -1745,7 +1753,7 @@ async fn main() -> Result<()> { }) }; if result_tx - .send((block_root, signed_block_for_send, should_gossip, outcome)) + .send((block_root, signed_block_for_send, should_gossip, outcome, block_processing_start)) .await .is_err() { diff --git a/lean_client/validator/src/lib.rs b/lean_client/validator/src/lib.rs index 909f5ae7..3c504481 100644 --- a/lean_client/validator/src/lib.rs +++ b/lean_client/validator/src/lib.rs @@ -37,6 +37,11 @@ pub struct AggregationJob { pub struct AggregationSnapshot { pub jobs: Vec, + /// data_roots whose gossip raw signatures were dropped from job.raw_ids + /// because a stored child proof already covers them. Callers must remove + /// those gossip signatures from `store.gossip_signatures` alongside the + /// per-job consumed set to prevent unbounded pool growth. + pub settled_data_roots: HashSet, } pub const AGGREGATION_SLOT_LOOKBACK: u64 = 1; @@ -66,6 +71,7 @@ pub fn snapshot_aggregation_inputs(store: &Store) -> Option } let mut jobs: Vec = Vec::with_capacity(all_data_roots.len()); + let mut settled_data_roots: HashSet = HashSet::new(); for data_root in all_data_roots { let Some(attestation_data) = store.attestation_data_by_root.get(&data_root) else { @@ -97,15 +103,17 @@ pub fn snapshot_aggregation_inputs(store: &Store) -> Option &mut covered_by_children, ); - let mut entries: Vec<(u64, Signature)> = gossip_groups - .get(&data_root) - .cloned() - .unwrap_or_default() - .into_iter() + let raw_input = gossip_groups.get(&data_root).cloned().unwrap_or_default(); + let mut entries: Vec<(u64, Signature)> = raw_input + .iter() .filter(|(vid, _)| { !covered_by_children.contains(vid) && head_validators.get(*vid).is_ok() }) + .cloned() .collect(); + if !raw_input.is_empty() && entries.is_empty() { + settled_data_roots.insert(data_root); + } entries.sort_by_key(|(vid, _)| *vid); let mut raw_ids: Vec = Vec::with_capacity(entries.len()); @@ -160,11 +168,14 @@ pub fn snapshot_aggregation_inputs(store: &Store) -> Option }); } - if jobs.is_empty() { + if jobs.is_empty() && settled_data_roots.is_empty() { return None; } - Some(AggregationSnapshot { jobs }) + Some(AggregationSnapshot { + jobs, + settled_data_roots, + }) } /// Entry in an annotated_validators.yaml file. @@ -487,13 +498,20 @@ impl ValidatorService { return None; } - if snapshot.jobs.is_empty() { + if snapshot.jobs.is_empty() && snapshot.settled_data_roots.is_empty() { return None; } let mut aggregated_attestations: Vec = Vec::with_capacity(snapshot.jobs.len()); - let mut consumed_data_roots: HashSet = HashSet::with_capacity(snapshot.jobs.len()); + let mut consumed_data_roots: HashSet = + HashSet::with_capacity(snapshot.jobs.len() + snapshot.settled_data_roots.len()); + consumed_data_roots.extend(snapshot.settled_data_roots.iter().copied()); + + let _session_timer = METRICS.get().map(|m| { + m.lean_committee_signatures_aggregation_time_seconds + .start_timer() + }); for job in &snapshot.jobs { if cancel.load(Ordering::Relaxed) { @@ -512,11 +530,7 @@ impl ValidatorService { break; } - let timer = METRICS.get().map(|m| { - m.lean_committee_signatures_aggregation_time_seconds - .start_timer() - }); - + let setup_start = std::time::Instant::now(); let children_arg: Vec<(&[PublicKey], &AggregatedSignatureProof)> = job .children .iter() @@ -528,6 +542,10 @@ impl ValidatorService { all_validator_ids.sort(); all_validator_ids.dedup(); let all_participants = AggregationBits::from_validator_indices(&all_validator_ids); + METRICS.get().map(|m| { + m.lean_aggregation_job_setup_seconds + .observe(setup_start.elapsed().as_secs_f64()) + }); let type1_start = std::time::Instant::now(); let proof = match AggregatedSignatureProof::aggregate_with_children( @@ -540,7 +558,6 @@ impl ValidatorService { log_inv_rate, ) { Ok(p) => { - stop_and_record(timer); info!( slot = slot.0, raws = job.raw_ids.len(), @@ -551,7 +568,6 @@ impl ValidatorService { p } Err(e) => { - stop_and_discard(timer); warn!( error = %e, data_root = %format!("0x{:x}", job.data_root), @@ -579,7 +595,7 @@ impl ValidatorService { } } - if aggregated_attestations.is_empty() { + if aggregated_attestations.is_empty() && consumed_data_roots.is_empty() { return None; } From 018cf244a703806c1834ab34b48642f0bc8d4f06 Mon Sep 17 00:00:00 2001 From: bomanaps Date: Mon, 13 Jul 2026 22:47:34 +0100 Subject: [PATCH 3/4] add proposer head-lag guard --- lean_client/fork_choice/src/handlers.rs | 17 ++++++++ lean_client/networking/src/network/service.rs | 21 ++++++++- lean_client/networking/src/types.rs | 2 + lean_client/src/main.rs | 43 +++++++++++++++++-- 4 files changed, 78 insertions(+), 5 deletions(-) diff --git a/lean_client/fork_choice/src/handlers.rs b/lean_client/fork_choice/src/handlers.rs index 61eea8c1..778c38d4 100644 --- a/lean_client/fork_choice/src/handlers.rs +++ b/lean_client/fork_choice/src/handlers.rs @@ -19,6 +19,23 @@ use crate::store::{ tick_interval, update_head, }; +pub const BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS: u64 = 4; + +pub fn should_suppress_proposal_for_head_lag( + wall_head_lag_slots: u64, + max_proposal_head_lag_slots: u64, + latest_justified_slot: u64, + has_fresher_peer_near_wall: bool, +) -> bool { + if latest_justified_slot == 0 { + return false; + } + if !has_fresher_peer_near_wall { + return false; + } + wall_head_lag_slots > max_proposal_head_lag_slots +} + #[inline] pub fn on_tick(store: &mut Store, time_millis: u64, has_proposal: bool) { // Calculate target time in intervals using milliseconds (800ms intervals) diff --git a/lean_client/networking/src/network/service.rs b/lean_client/networking/src/network/service.rs index 90e69ed9..c3cd3fde 100644 --- a/lean_client/networking/src/network/service.rs +++ b/lean_client/networking/src/network/service.rs @@ -46,8 +46,8 @@ use crate::{ req_resp::{self, LeanRequest, ReqRespMessage}, types::{ CanonicalBlocksProvider, ChainMessage, ChainMessageSink, ConnectionState, - MAX_BLOCK_CACHE_SIZE, NetworkFinalizedSlot, OutboundP2pRequest, P2pRequestSource, - SignedBlockProvider, StatusProvider, + MAX_BLOCK_CACHE_SIZE, NetworkFinalizedSlot, NetworkHeadSlot, OutboundP2pRequest, + P2pRequestSource, SignedBlockProvider, StatusProvider, }, }; @@ -209,6 +209,7 @@ where /// Roots currently in-flight to deduplicate network-layer pipelining vs chain-side requests in_flight_roots: HashSet, network_finalized_slot: NetworkFinalizedSlot, + network_head_slot: NetworkHeadSlot, peer_finalized_slots: HashMap, /// Peer head slots reported via Status; used by the caller to decide /// when a backfill should switch from per-root BlocksByRoot to batched BlocksByRange. @@ -231,6 +232,7 @@ where status_provider: StatusProvider, blocks_by_range_provider: CanonicalBlocksProvider, network_finalized_slot: NetworkFinalizedSlot, + network_head_slot: NetworkHeadSlot, status_notify: Arc, ) -> Result { Self::new_with_peer_count( @@ -242,6 +244,7 @@ where status_provider, blocks_by_range_provider, network_finalized_slot, + network_head_slot, status_notify, ) .await @@ -256,6 +259,7 @@ where status_provider: StatusProvider, blocks_by_range_provider: CanonicalBlocksProvider, network_finalized_slot: NetworkFinalizedSlot, + network_head_slot: NetworkHeadSlot, status_notify: Arc, ) -> Result { let local_key = Keypair::generate_secp256k1(); @@ -269,6 +273,7 @@ where status_provider, blocks_by_range_provider, network_finalized_slot, + network_head_slot, status_notify, ) .await @@ -284,6 +289,7 @@ where status_provider: StatusProvider, blocks_by_range_provider: CanonicalBlocksProvider, network_finalized_slot: NetworkFinalizedSlot, + network_head_slot: NetworkHeadSlot, status_notify: Arc, ) -> Result { let behaviour = Self::build_behaviour(&local_key, &network_config)?; @@ -345,6 +351,7 @@ where pending_block_depths: HashMap::new(), in_flight_roots: HashSet::new(), network_finalized_slot, + network_head_slot, peer_finalized_slots: HashMap::new(), peer_head_slots: HashMap::new(), range_sync_state: None, @@ -533,6 +540,7 @@ where .retain(|_, (p, _, _)| *p != peer_id); self.inflight_range_keys.retain(|(p, _, _)| *p != peer_id); self.recompute_network_finalized_slot(); + self.recompute_network_head_slot(); info!(peer = %peer_id, ?cause, "Disconnected from peer (total: {})", connected); @@ -1276,6 +1284,14 @@ where } } + fn recompute_network_head_slot(&mut self) { + let max_head = self.peer_head_slots.values().copied().max(); + let mut slot_guard = self.network_head_slot.lock(); + if *slot_guard != max_head { + *slot_guard = max_head; + } + } + fn maybe_trigger_backfill( &mut self, peer: PeerId, @@ -1288,6 +1304,7 @@ where self.peer_finalized_slots.insert(peer, peer_finalized_slot); self.peer_head_slots.insert(peer, peer_head_slot); self.recompute_network_finalized_slot(); + self.recompute_network_head_slot(); if peer_head_slot > our_head_slot { let gap = peer_head_slot - our_head_slot; diff --git a/lean_client/networking/src/types.rs b/lean_client/networking/src/types.rs index 23d357f2..6d1877ca 100644 --- a/lean_client/networking/src/types.rs +++ b/lean_client/networking/src/types.rs @@ -35,6 +35,8 @@ pub type StatusProvider = Arc>; pub type NetworkFinalizedSlot = Arc>>; +pub type NetworkHeadSlot = Arc>>; + /// 1-byte domain for gossip message-id isolation of valid snappy messages. /// Per leanSpec, prepended to the message hash when decompression succeeds. pub const MESSAGE_DOMAIN_VALID_SNAPPY: &[u8; 1] = &[0x01]; diff --git a/lean_client/src/main.rs b/lean_client/src/main.rs index 44bef973..905ee26b 100644 --- a/lean_client/src/main.rs +++ b/lean_client/src/main.rs @@ -15,8 +15,9 @@ use features::Feature; use fork_choice::{ block_cache::BlockCache, handlers::{ - apply_verified_block, on_aggregated_attestation_async, on_attestation, - on_gossip_attestation, on_tick, verify_and_transition, + BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS, apply_verified_block, on_aggregated_attestation_async, + on_attestation, on_gossip_attestation, on_tick, should_suppress_proposal_for_head_lag, + verify_and_transition, }, store::{ INTERVALS_PER_SLOT, MILLIS_PER_INTERVAL, Store, execute_block_production, @@ -32,7 +33,8 @@ use networking::gossipsub::topic::{compute_subnet_id, get_subscription_topics}; use networking::network::{NetworkService, NetworkServiceConfig}; use networking::types::{ CanonicalBlocksProvider, ChainMessage, MAX_BLOCK_CACHE_SIZE, NetworkFinalizedSlot, - OutboundP2pRequest, SignedBlockProvider, StatusProvider, ValidatorChainMessage, + NetworkHeadSlot, OutboundP2pRequest, SignedBlockProvider, StatusProvider, + ValidatorChainMessage, }; use parking_lot::{Mutex, RwLock}; use ssz::{PersistentList, SszHash, SszReadDefault as _}; @@ -873,6 +875,9 @@ async fn main() -> Result<()> { let network_finalized_slot: NetworkFinalizedSlot = Arc::new(Mutex::new(None)); let network_finalized_slot_for_network = network_finalized_slot.clone(); + let network_head_slot: NetworkHeadSlot = Arc::new(Mutex::new(None)); + let network_head_slot_for_network = network_head_slot.clone(); + let canonical_blocks_provider: CanonicalBlocksProvider = { let block_provider = signed_block_provider.clone(); let status = status_provider.clone(); @@ -924,6 +929,7 @@ async fn main() -> Result<()> { status_provider_for_network, canonical_blocks_provider.clone(), network_finalized_slot_for_network, + network_head_slot_for_network, status_notify.clone(), ) .await @@ -940,6 +946,7 @@ async fn main() -> Result<()> { status_provider_for_network, canonical_blocks_provider.clone(), network_finalized_slot_for_network, + network_head_slot_for_network, status_notify.clone(), ) .await @@ -956,6 +963,7 @@ async fn main() -> Result<()> { status_provider_for_network, canonical_blocks_provider.clone(), network_finalized_slot_for_network, + network_head_slot_for_network, status_notify.clone(), ) .await @@ -1919,6 +1927,35 @@ async fn main() -> Result<()> { "produce_block dropped: previous build still in flight for slot {in_flight}" ))); } else { + let (head_slot, latest_justified_slot) = { + let s = store.read(); + let head_slot = s.blocks.get(&s.head).map(|b| b.slot.0).unwrap_or(0); + (head_slot, s.latest_justified.slot.0) + }; + let wall_head_lag = slot.0.saturating_sub(head_slot); + let peer_head = *network_head_slot.lock(); + let has_fresher_peer_near_wall = peer_head + .map(|h| h.saturating_add(BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS) >= slot.0) + .unwrap_or(false); + if should_suppress_proposal_for_head_lag( + wall_head_lag, + BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS, + latest_justified_slot, + has_fresher_peer_near_wall, + ) { + warn!( + slot = slot.0, + proposer_index, + wall_head_lag, + head_slot, + latest_justified_slot, + "Skipping block production: local head is {wall_head_lag} wall-clock slots behind" + ); + let _ = sender.send(Err(anyhow::anyhow!( + "produce_block suppressed: head {wall_head_lag} slots behind wall" + ))); + continue; + } let block_build_start = Instant::now(); let prepare_result = { let mut w = store.write(); From e0b8ee0e069609364cdd7a440d460182af188e9f Mon Sep 17 00:00:00 2001 From: bomanaps Date: Tue, 14 Jul 2026 13:47:51 +0100 Subject: [PATCH 4/4] advance xmss preparation --- lean_client/xmss/src/secret_key.rs | 32 +++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/lean_client/xmss/src/secret_key.rs b/lean_client/xmss/src/secret_key.rs index b5cae65d..f8cb3c49 100644 --- a/lean_client/xmss/src/secret_key.rs +++ b/lean_client/xmss/src/secret_key.rs @@ -1,11 +1,13 @@ +use std::sync::Mutex; + use anyhow::{Error, Result, anyhow}; use derive_more::Debug; use leansig::serialization::Serializable; -use leansig::signature::SignatureScheme; use leansig::signature::generalized_xmss::instantiations_aborting::lifetime_2_to_the_32::{ SIGAbortingTargetSumLifetime32Dim46Base8 as XmssScheme, SecretKeyAbortingTargetSumLifetime32Dim46Base8 as XmssSecretKey, }; +use leansig::signature::{SignatureScheme, SignatureSchemeSecretKey}; use rand::CryptoRng; use ssz::H256; @@ -17,11 +19,31 @@ use crate::{PublicKey, Signature}; // implement `Drop` here manually (zeroize the inner buffers via accessor). #[derive(Debug)] #[debug("[REDACTED]")] -pub struct SecretKey(XmssSecretKey); +pub struct SecretKey(Mutex); impl SecretKey { pub fn sign(&self, message: H256, epoch: u32) -> Result { - let sig = XmssScheme::sign(&self.0, epoch, message.as_fixed_bytes()) + let mut sk = self + .0 + .lock() + .map_err(|_| anyhow!("failed to acquire secret key lock"))?; + let target = epoch as u64; + + if !sk.get_activation_interval().contains(&target) { + return Err(anyhow!("epoch {epoch} outside key activation window")); + } + + while !sk.get_prepared_interval().contains(&target) { + let before = sk.get_prepared_interval(); + sk.advance_preparation(); + if sk.get_prepared_interval() == before { + return Err(anyhow!( + "advance_preparation made no progress for epoch {epoch}" + )); + } + } + + let sig = XmssScheme::sign(&sk, epoch, message.as_fixed_bytes()) .map_err(|_| anyhow!("failed to sign message"))?; Ok(Signature::from_lean(sig)) } @@ -33,7 +55,7 @@ impl SecretKey { ) -> (PublicKey, SecretKey) { let (pk, sk) = XmssScheme::key_gen(rng, activation_epoch as usize, num_active_epochs as usize); - (PublicKey::from_lean(pk), SecretKey(sk)) + (PublicKey::from_lean(pk), SecretKey(Mutex::new(sk))) } } @@ -43,6 +65,6 @@ impl TryFrom<&[u8]> for SecretKey { fn try_from(value: &[u8]) -> Result { let sk = XmssSecretKey::from_bytes(value) .map_err(|_| anyhow!("value is not valid secret key"))?; - Ok(Self(sk)) + Ok(Self(Mutex::new(sk))) } }