diff --git a/crates/blockchain/src/aggregation.rs b/crates/blockchain/src/aggregation.rs index 2a62b2a6..0d4a04a6 100644 --- a/crates/blockchain/src/aggregation.rs +++ b/crates/blockchain/src/aggregation.rs @@ -765,7 +765,7 @@ mod tests { use super::*; use ethlambda_storage::backend::InMemoryBackend; use ethlambda_types::{ - block::{Block, BlockBody, BlockHeader, MultiMessageAggregate, SignedBlock}, + block::{Block, BlockBody, BlockHeader, BlockProof, SignedBlock}, checkpoint::Checkpoint, state::{ChainConfig, JustificationValidators, JustifiedSlots, State}, }; @@ -862,7 +862,7 @@ mod tests { state_root: H256::ZERO, body: BlockBody::default(), }, - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), }; store .insert_signed_block(root, signed_block) diff --git a/crates/blockchain/src/block_builder.rs b/crates/blockchain/src/block_builder.rs index 787c55b4..f6db0b80 100644 --- a/crates/blockchain/src/block_builder.rs +++ b/crates/blockchain/src/block_builder.rs @@ -18,14 +18,14 @@ use std::{ use ethlambda_crypto::aggregate_proofs; use ethlambda_state_transition::{ attestation_data_matches_chain, justified_slots_ops, process_block, process_slots, - slot_is_justifiable_after, }; use ethlambda_types::{ ShortRoot, - attestation::{AggregatedAttestation, AggregationBits, AttestationData}, - block::{AggregatedAttestations, Block, BlockBody, SingleMessageAggregate}, + attestation::{AggregatedAttestation, AggregationBits, Attestation, AttestationData}, + block::{AggregatedAttestations, Block, BlockBody, HeartbeatVotes, SingleMessageAggregate}, checkpoint::Checkpoint, primitives::{H256, HashTreeRoot as _}, + signature::ValidatorSignature, state::{JustifiedSlots, State}, }; use tracing::{info, trace}; @@ -89,8 +89,17 @@ pub(crate) fn build_block( parent_root: H256, known_block_roots: &HashSet, aggregated_payloads: &HashMap)>, + heartbeat_votes: HashMap, config: ProposerConfig, -) -> Result<(Block, Vec, PostBlockCheckpoints), StoreError> { +) -> Result< + ( + Block, + Vec, + Vec, + PostBlockCheckpoints, + ), + StoreError, +> { info!(slot, proposer_index, "Building block"); let select_start = Instant::now(); @@ -125,6 +134,21 @@ pub(crate) fn build_block( let (aggregated_attestations, aggregated_signatures): (Vec<_>, Vec<_>) = compacted.into_iter().unzip(); + let (heartbeat_data, heartbeat_signatures): (Vec<_>, Vec<_>) = heartbeat_votes + .into_iter() + .map(|(vid, (data, sig))| { + ( + Attestation { + validator_id: vid, + data, + }, + sig, + ) + }) + .unzip(); + + let heartbeat: HeartbeatVotes = heartbeat_data.try_into().expect("max committee size"); + let attestations: AggregatedAttestations = aggregated_attestations .try_into() .expect("attestation count exceeds limit"); @@ -133,7 +157,10 @@ pub(crate) fn build_block( proposer_index, parent_root, state_root: H256::ZERO, - body: BlockBody { attestations }, + body: BlockBody { + heartbeat, + attestations, + }, }; let mut post_state = head_state.clone(); // ethlambda runs the STF once after selection (it projects justification @@ -154,7 +181,12 @@ pub(crate) fn build_block( finalized: post_state.latest_finalized, }; - Ok((final_block, aggregated_signatures, post_checkpoints)) + Ok(( + final_block, + heartbeat_signatures, + aggregated_signatures, + post_checkpoints, + )) } /// Tiered greedy attestation selection for block proposal. @@ -408,16 +440,14 @@ impl ProjectedState { let total = prior_count + new_voters.len(); let crosses_2_3 = 3 * total >= 2 * validator_count; - // 3SF-mini finalization requires the source to lie past the finalized - // boundary (a source at or behind it is already final and must not - // re-finalize) and no slot strictly between source.slot and target.slot - // to still be justifiable (so source and target are consecutive - // justified checkpoints in the projected post-state). Mirrors + // The simple BFT finality condition finalizes the source when it lies past + // the finalized boundary (a source at or behind it is already final and must + // not re-finalize) and the target is its immediate successor, so the two are + // consecutive justified checkpoints in the projected post-state. Mirrors // `try_finalize` in the state transition. let finalizes = crosses_2_3 && att_data.source.slot > self.finalized_slot - && (att_data.source.slot + 1..att_data.target.slot) - .all(|s| !slot_is_justifiable_after(s, self.finalized_slot)); + && att_data.source.slot + 1 == att_data.target.slot; let tier = if is_genesis_self_vote(att_data) || !crosses_2_3 { Tier::Build @@ -441,9 +471,8 @@ impl ProjectedState { /// /// Mirrors `state_transition::is_valid_vote`: the entry's head must be /// known, its source must be justified, its (source, target) must match - /// the candidate-block chain view, `target.slot > source.slot`, target - /// must not already be justified, and target must be a justifiable slot - /// relative to the projected finalized slot. The genesis self-vote + /// the candidate-block chain view, `target.slot > source.slot`, and target + /// must not already be justified. The genesis self-vote /// (source == target == slot 0) is exempt from the `target.slot > /// source.slot` and `target_already_justified` checks since fork-choice /// bootstrapping needs it; STF will silently drop it, but it carries @@ -480,11 +509,6 @@ impl ProjectedState { { return Err("target_already_justified"); } - if !is_genesis_self_vote - && !slot_is_justifiable_after(att_data.target.slot, self.finalized_slot) - { - return Err("target_not_justifiable"); - } Ok(()) } } @@ -877,8 +901,12 @@ fn trace_skipped_attestation(reason: &'static str, att: &AttestationData, data_r mod tests { use super::*; use ethlambda_types::{ - attestation::{AggregatedAttestation, AggregationBits, AttestationData}, - block::{ByteList512KiB, MultiMessageAggregate, SignedBlock, SingleMessageAggregate}, + attestation::{ + AggregatedAttestation, AggregationBits, AttestationData, blank_xmss_signature, + }, + block::{ + BlockProof, ByteList512KiB, MultiMessageAggregate, SignedBlock, SingleMessageAggregate, + }, checkpoint::Checkpoint, state::State, }; @@ -1087,11 +1115,16 @@ mod tests { ); // Substitute a worst-case-size proof to model what `propose_block` - // would attach. The actual SNARK can't be built without lean-multisig, - // but the size cap (`ByteList512KiB`) bounds the worst case. + // would attach: a 512 KiB attestation aggregate plus the fixed-size + // proposer signature. The actual SNARK can't be built without + // lean-multisig, but the size cap bounds the worst case. let _ = signatures; - let proof = MultiMessageAggregate::new( - ByteList512KiB::try_from(vec![0xAB; 512 * 1024]).expect("worst-case proof fits in cap"), + let proof = BlockProof::new( + blank_xmss_signature(), + MultiMessageAggregate::new( + ByteList512KiB::try_from(vec![0xAB; 512 * 1024]) + .expect("worst-case proof fits in cap"), + ), ); let signed_block = SignedBlock { message: block, diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index ffb63f46..03b65995 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -323,7 +323,7 @@ mod tests { use super::*; use ethlambda_storage::{ForkCheckpoints, backend::InMemoryBackend}; use ethlambda_types::{ - block::{Block, BlockBody, MultiMessageAggregate, SignedBlock}, + block::{Block, BlockBody, BlockProof, SignedBlock}, state::State, }; use std::sync::Arc; @@ -455,7 +455,7 @@ mod tests { state_root, body: BlockBody::default(), }, - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), }; store .insert_signed_block(root, signed_block) diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index b2929ab6..adfa47ec 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -2,15 +2,15 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::time::{Duration, Instant, SystemTime}; use ethlambda_network_api::{BlockChainToP2PRef, InitP2P}; -use ethlambda_state_transition::is_proposer; +use ethlambda_state_transition::{is_heartbeat_committee_member, is_proposer}; use ethlambda_storage::{ALL_TABLES, Store}; use ethlambda_types::{ ShortRoot, aggregator::AggregatorController, attestation::{SignedAggregatedAttestation, SignedAttestation}, - block::{ByteList512KiB, MultiMessageAggregate, SignedBlock}, + block::{BlockProof, ByteList512KiB, MultiMessageAggregate, SignedBlock}, primitives::{H256, HashTreeRoot as _}, - signature::{ValidatorPublicKey, ValidatorSignature}, + signature::ValidatorPublicKey, }; use crate::aggregation::{ @@ -666,9 +666,18 @@ impl BlockChainServer { // Publish to gossip network if let Some(ref p2p) = self.p2p { - let _ = p2p.publish_attestation(signed_attestation).inspect_err( - |err| error!(%slot, %validator_id, %err, "Failed to publish attestation"), - ); + let _ = p2p + .publish_attestation(signed_attestation.clone()) + .inspect_err( + |err| error!(%slot, %validator_id, %err, "Failed to publish attestation"), + ); + let head_state = self.store.head_state(); + let num_validators = head_state.validators.len() as u64; + if is_heartbeat_committee_member(validator_id, slot, num_validators) { + let _ = p2p.publish_heartbeat_attestation(signed_attestation).inspect_err( + |err| error!(%slot, %validator_id, %err, "Failed to publish attestation"), + ); + } info!(%slot, %validator_id, "Published attestation"); } } @@ -727,7 +736,7 @@ impl BlockChainServer { // the store to `slot`'s interval 0, so the head-recency gate uses `slot`. pre_build.diff_and_emit(&self.store, &self.events, slot); - let Ok((block, single_message_aggregates, _post_checkpoints)) = build_result else { + let Ok((block, signatures, single_message_aggregates, _post_checkpoints)) = build_result else { metrics::inc_block_building_failures(); return; }; @@ -749,102 +758,90 @@ impl BlockChainServer { return; }; - // Wrap the proposer's raw XMSS signature into a singleton - // single-message aggregate SNARK, then merge it with every attestation - // single-message aggregate into the single multi-message aggregate. + // Assemble SignedBlock: carry the proposer's raw XMSS signature as a + // standalone field, and aggregate the attestation single-message + // aggregates (only) into the block's attestation multi-message + // aggregate. The proposer no longer enters the aggregate, so a block + // with no attestations needs no prover work and the attestation + // aggregate can be built independently of the block root. let head_state = self.store.head_state(); let validators = &head_state.validators; - let Some(proposer_validator) = validators.get(validator_id as usize) else { + if validators.get(validator_id as usize).is_none() { error!(%slot, %validator_id, "Proposer index out of range when assembling block"); metrics::inc_block_building_failures(); return; - }; - - // Decode the proposer's proposal pubkey once and reuse it both for the - // singleton single-message aggregate wrap and for the multi-message - // aggregate merge inputs. - let Ok(proposer_pubkey) = proposer_validator.get_proposal_pubkey().inspect_err( - |err| error!(%slot, %validator_id, %err, "Failed to decode proposer proposal pubkey"), - ) else { - metrics::inc_block_building_failures(); - return; - }; + } - let Ok(proposer_validator_signature) = - ValidatorSignature::from_bytes(&proposer_signature).inspect_err(|err| { - error!(%slot, %validator_id, %err, "Failed to decode proposer signature bytes") - }) - else { - metrics::inc_block_building_failures(); - return; - }; - let Ok(proposer_proof_bytes) = ethlambda_crypto::aggregate_signatures( - vec![proposer_pubkey.clone()], - vec![proposer_validator_signature], - &block_root, - slot as u32, - ) - .inspect_err( - |err| error!(%slot, %validator_id, %err, "Failed to wrap proposer signature as single-message aggregate"), - ) else { - metrics::inc_block_building_failures(); - return; - }; + // `sign_block_root` already returns an `XmssSignature`, so the proposer + // signature is carried verbatim — no packing or prover work needed. - let mut merge_inputs: Vec<(Vec, ByteList512KiB)> = - Vec::with_capacity(single_message_aggregates.len() + 1); - let mut resolve_failed = false; - for sma in &single_message_aggregates { - let mut pubkeys = Vec::new(); - for vid in sma.participant_indices() { - let Some(validator) = validators.get(vid as usize) else { - error!(%slot, %validator_id, vid, "Participant out of range while resolving pubkeys"); - resolve_failed = true; - break; - }; - match validator.get_attestation_pubkey() { - Ok(pk) => pubkeys.push(pk), - Err(err) => { - error!(%slot, %validator_id, vid, %err, "Failed to decode attestation pubkey"); + // Aggregate the attestation single-message aggregates into a single + // multi-message aggregate. With no attestations the aggregate is empty: + // the proposer signature stands alone, mirroring `(prop-sig, empty-proof)`. + let attestation_proof = if single_message_aggregates.is_empty() { + MultiMessageAggregate::default() + } else { + let mut merge_inputs: Vec<(Vec, ByteList512KiB)> = + Vec::with_capacity(single_message_aggregates.len()); + let mut resolve_failed = false; + for sma in &single_message_aggregates { + let mut pubkeys = Vec::new(); + for vid in sma.participant_indices() { + let Some(validator) = validators.get(vid as usize) else { + error!(%slot, %validator_id, vid, "Participant out of range while resolving pubkeys"); resolve_failed = true; break; + }; + match validator.get_attestation_pubkey() { + Ok(pk) => pubkeys.push(pk), + Err(err) => { + error!(%slot, %validator_id, vid, %err, "Failed to decode attestation pubkey"); + resolve_failed = true; + break; + } } } + if resolve_failed { + break; + } + merge_inputs.push((pubkeys, sma.proof.clone())); } if resolve_failed { - break; - } - merge_inputs.push((pubkeys, sma.proof.clone())); - } - if resolve_failed { - metrics::inc_block_building_failures(); - return; - } - merge_inputs.push((vec![proposer_pubkey], proposer_proof_bytes)); - - // Merge yields raw lean-multisig type-2 bytes. Per-component - // participants are rederived at verify time from - // `block.body.attestations[i].aggregation_bits` plus - // `block.proposer_index`, so nothing else needs persisting. - let merged_bytes = match ethlambda_crypto::merge_type_1s_into_type_2(merge_inputs) { - Ok(bytes) => bytes, - Err(err) => { - error!(%slot, %validator_id, %err, "Failed to merge Type-1s into Type-2"); metrics::inc_block_building_failures(); return; } - }; - let proof = match MultiMessageAggregate::from_bytes(merged_bytes.iter().as_slice()) { - Ok(p) => p, - Err(err) => { - error!(%slot, %validator_id, %err, "Failed to build multi-message aggregate"); - metrics::inc_block_building_failures(); - return; + + // Merge yields raw lean-multisig multi-message aggregate bytes. + // Per-component participants are rederived at verify time from + // `block.body.attestations[i].aggregation_bits`, so nothing else + // needs persisting. + let merged_bytes = match ethlambda_crypto::merge_type_1s_into_type_2(merge_inputs) { + Ok(bytes) => bytes, + Err(err) => { + error!(%slot, %validator_id, %err, "Failed to merge single-message aggregates into multi-message aggregate"); + metrics::inc_block_building_failures(); + return; + } + }; + match MultiMessageAggregate::from_bytes(merged_bytes.iter().as_slice()) { + Ok(p) => p, + Err(err) => { + error!(%slot, %validator_id, %err, "Failed to build multi-message aggregate"); + metrics::inc_block_building_failures(); + return; + } } }; + // `single_message_aggregates` is no longer needed past this point. + drop(single_message_aggregates); + + let heartbeat_sigs = signatures + .try_into() + .expect("went over the committee limit"); + let signed_block = SignedBlock { message: block, - proof, + proof: BlockProof::new(proposer_signature, heartbeat_sigs, attestation_proof), }; // Stop timing here: the build is done, and the alignment wait below must diff --git a/crates/blockchain/src/reaggregate.rs b/crates/blockchain/src/reaggregate.rs index f47d6c51..ba127f47 100644 --- a/crates/blockchain/src/reaggregate.rs +++ b/crates/blockchain/src/reaggregate.rs @@ -71,11 +71,12 @@ pub fn reaggregate_from_block( let validators = &parent_state.validators; let num_validators = validators.len() as u64; - // Per-component pubkeys: one entry per body attestation in order, then - // the proposer entry. Layout is invariant per block, so it's resolved - // once and reused for every split call below. + // Per-component pubkeys: one entry per body attestation in order. The + // attestation aggregate no longer carries a proposer component (the + // proposer signature lives outside it), so the layout is attestations + // only. Resolved once and reused for every split call below. let mut pubkeys_per_component: Vec> = - Vec::with_capacity(attestations.len() + 1); + Vec::with_capacity(attestations.len()); for att in &attestations { let mut pubkeys = Vec::new(); for vid in validator_indices(&att.aggregation_bits) { @@ -91,14 +92,6 @@ pub fn reaggregate_from_block( } pubkeys_per_component.push(pubkeys); } - if block.proposer_index >= num_validators { - return Vec::new(); - } - let Ok(proposer_pubkey) = validators[block.proposer_index as usize].get_proposal_pubkey() - else { - return Vec::new(); - }; - pubkeys_per_component.push(vec![proposer_pubkey]); let candidates = select_candidates(store, &attestations); if candidates.is_empty() { @@ -120,8 +113,8 @@ pub fn reaggregate_from_block( }; // Step 1: SNARK-split this attestation's component out of the block's - // merged multi-message aggregate proof. - let merged_bytes = signed_block.proof.proof_bytes(); + // attestation Type-2 aggregate. + let merged_bytes = signed_block.proof.attestation_proof.proof_bytes(); let split_bytes = match ethlambda_crypto::split_type_2_by_message( merged_bytes, pubkeys_per_component.clone(), diff --git a/crates/blockchain/src/store.rs b/crates/blockchain/src/store.rs index 2bdf46b0..a491c444 100644 --- a/crates/blockchain/src/store.rs +++ b/crates/blockchain/src/store.rs @@ -1,6 +1,6 @@ use std::collections::{HashMap, HashSet}; -use ethlambda_state_transition::{is_proposer, slot_is_justifiable_after}; +use ethlambda_state_transition::{is_heartbeat_committee_member, is_proposer}; use ethlambda_storage::{ForkCheckpoints, Store}; use ethlambda_types::{ ShortRoot, @@ -66,14 +66,10 @@ pub fn update_head(store: &mut Store) -> HeadUpdate { let blocks = store .get_live_chain() .expect("get_live_chain should succeed"); - let attestations = store.extract_latest_known_attestations(); + let attestations = store.get_last_slot_votes(); let old_head = store.head().expect("head block exists"); - let latest_justified_root = store - .latest_justified() - .expect("latest justified checkpoint exists") - .root; let (new_head, weights) = ethlambda_fork_choice::compute_lmd_ghost_head( - latest_justified_root, + store.safe_target().expect("safe target exists"), &blocks, &attestations, 0, @@ -148,17 +144,13 @@ pub fn update_head(store: &mut Store) -> HeadUpdate { /// evidence even when live participation has collapsed: exactly the failure /// mode safe target is supposed to prevent. See leanSpec PR #680. fn update_safe_target(store: &mut Store) { - let head_state = store - .get_state(&store.head().unwrap()) - .expect("head state exists"); - let num_validators = head_state.unwrap().validators.len() as u64; - - let min_target_score = (num_validators * 2).div_ceil(3); - let blocks = store .get_live_chain() .expect("get_live_chain should succeed"); - let attestations = store.extract_latest_new_attestations(); + let attestations = store.get_last_period_votes(); + // Use a 2/3 threshold of the number of voting validators + let min_target_score = (attestations.len() as u64 * 2).div_ceil(3); + let (safe_target, _weights) = ethlambda_fork_choice::compute_lmd_ghost_head( store .latest_justified() @@ -436,6 +428,13 @@ pub fn on_gossip_attestation( } metrics::inc_pq_sig_attestation_signatures_valid(); + let num_validators = target_state.validators.len() as u64; + // If the validator is in the heartbeat committee, persist the vote and + // signature for fork choice usage. + if is_heartbeat_committee_member(validator_id, attestation.data.slot, num_validators) { + store.insert_heartbeat_vote(attestation, signature.clone()); + } + // Only aggregators persist the signature for later aggregation at // interval 2. Non-aggregators drop the validated attestation — they // still participate in the mesh so peers see the message propagate. @@ -746,7 +745,10 @@ pub fn get_attestation_target(store: &Store) -> Checkpoint { pub fn get_attestation_target_with_checkpoints( store: &Store, justified: Checkpoint, - finalized: Checkpoint, + // Unused under the simple BFT finality condition (every slot is justifiable, + // so the target no longer needs a justifiability walk-back). Kept on the + // signature pending the finality redesign. + _finalized: Checkpoint, ) -> Checkpoint { // Start from current head let mut target_block_root = store.head().unwrap(); @@ -777,21 +779,6 @@ pub fn get_attestation_target_with_checkpoints( } } - let finalized_slot = finalized.slot; - - // Ensure target is in justifiable slot range - // - // Walk back until we find a slot that satisfies justifiability rules - // relative to the latest finalized checkpoint. - while target_header.slot > finalized_slot - && !slot_is_justifiable_after(target_header.slot, finalized_slot) - { - target_block_root = target_header.parent_root; - target_header = store - .get_block_header(&target_block_root) - .expect("parent block exists") - .unwrap(); - } // Guard: clamp target to justified (not in the spec). // // The spec's walk-back has no lower bound, so it can produce attestations @@ -903,7 +890,15 @@ pub fn produce_block_with_signatures( slot: u64, validator_index: u64, config: ProposerConfig, -) -> Result<(Block, Vec, PostBlockCheckpoints), StoreError> { +) -> Result< + ( + Block, + Vec, + Vec, + PostBlockCheckpoints, + ), + StoreError, +> { // Get parent block and state to build upon let head_root = get_proposal_head(store, slot); let head_state = store @@ -925,10 +920,11 @@ pub fn produce_block_with_signatures( // Get known aggregated payloads: data_root -> (AttestationData, Vec) let aggregated_payloads = store.known_aggregated_payloads(); + let heartbeat_votes = store.get_last_slot_votes(); let known_block_roots = store.get_block_roots().unwrap(); - let (block, signatures, post_checkpoints) = { + let (block, heartbeat_sigs, att_sigs, post_checkpoints) = { let _timing = metrics::time_block_building_payload_aggregation(); build_block( &head_state, @@ -937,6 +933,7 @@ pub fn produce_block_with_signatures( head_root, &known_block_roots, &aggregated_payloads, + heartbeat_votes, config, )? }; @@ -963,9 +960,9 @@ pub fn produce_block_with_signatures( ); } - metrics::observe_block_aggregated_payloads(signatures.len()); + metrics::observe_block_aggregated_payloads(att_sigs.len()); - Ok((block, signatures, post_checkpoints)) + Ok((block, heartbeat_sigs, att_sigs, post_checkpoints)) } /// Errors that can occur during Store operations. @@ -986,6 +983,9 @@ pub enum StoreError { #[error("Validator signature verification failed")] SignatureVerificationFailed, + #[error("Block carries an attestation proof but has no attestations")] + UnexpectedAttestationProof, + #[error("Block slot {0} exceeds u32 range")] SlotOutOfRange(u64), @@ -1085,13 +1085,18 @@ pub enum StoreError { BlockTooFarInFuture { block_slot: u64, current_slot: u64 }, } -/// Full verification of a signed block's merged multi-message aggregate proof. +/// Full verification of a signed block's proof. /// -/// Structural pre-checks (fast fail) ensure the merged proof's `info` list lines -/// up with the block body (one entry per attestation plus a trailing proposer -/// entry; messages, slots, and participants match what the body declares). -/// On success, the lean-multisig devnet5 `verify_type_2` primitive runs the -/// SNARK verifier over the merged proof bytes against the resolved pubkey set. +/// The proof has two independent parts: +/// +/// 1. The proposer's raw XMSS signature over the block root, verified directly +/// against the proposer's `proposal_pubkey` with the hash-based verifier. +/// 2. The attestation aggregate: a lean-multisig Type-2 over the body +/// attestations only. Structural pre-checks (fast fail) ensure its `info` +/// list lines up with the block body (one entry per attestation; messages, +/// slots, and participants match what the body declares), then the +/// `verify_type_2` SNARK verifier runs over the proof bytes. A block with no +/// attestations carries no aggregate. /// /// Exposed publicly so RPC handlers (notably the Hive test-driver /// `verify_signatures/run` endpoint) can run the exact same verification path @@ -1127,52 +1132,70 @@ pub fn verify_block_signatures( let block_root = block.hash_tree_root(); let structural_elapsed = total_start.elapsed(); - // Resolve pubkeys per multi-message aggregate component for verify_type_2 and rederive the - // expected (message, slot) bindings from the block body. Attestation - // components use each participant's attestation_pubkey; the trailing - // proposer component uses the proposal_pubkey of `block.proposer_index`. - let expected_components = attestations.len() + 1; - let mut pubkeys_per_component: Vec> = - Vec::with_capacity(expected_components); - let mut expected_bindings: Vec<(H256, u32)> = Vec::with_capacity(expected_components); - - for attestation in attestations.iter() { - let mut pubkeys = Vec::new(); - for vid in validator_indices(&attestation.aggregation_bits) { - let validator = validators - .get(vid as usize) - .ok_or(StoreError::InvalidValidatorIndex)?; - let pk = validator - .get_attestation_pubkey() - .map_err(|_| StoreError::PubkeyDecodingFailed(vid))?; - pubkeys.push(pk); - } - pubkeys_per_component.push(pubkeys); - let slot_u32 = u32::try_from(attestation.data.slot) - .map_err(|_| StoreError::SlotOutOfRange(attestation.data.slot))?; - expected_bindings.push((attestation.data.hash_tree_root(), slot_u32)); - } + let block_slot_u32 = + u32::try_from(block.slot).map_err(|_| StoreError::SlotOutOfRange(block.slot))?; + // 1. Verify the proposer's raw XMSS signature over the block root. It is + // carried outside the attestation aggregate, so it is checked directly + // against the proposer's proposal pubkey with the hash-based verifier. let proposer_validator = validators .get(block.proposer_index as usize) .ok_or(StoreError::InvalidValidatorIndex)?; let proposer_pubkey = proposer_validator .get_proposal_pubkey() .map_err(|_| StoreError::PubkeyDecodingFailed(block.proposer_index))?; - pubkeys_per_component.push(vec![proposer_pubkey]); - let block_slot_u32 = - u32::try_from(block.slot).map_err(|_| StoreError::SlotOutOfRange(block.slot))?; - expected_bindings.push((block_root, block_slot_u32)); - - let merged_bytes = signed_block.proof.proof_bytes(); + let proposer_signature = ValidatorSignature::from_bytes(&signed_block.proof.proposer_signature) + .map_err(|_| StoreError::SignatureDecodingFailed)?; + if !proposer_signature.is_valid(&proposer_pubkey, block_slot_u32, &block_root) { + return Err(StoreError::SignatureVerificationFailed); + } + // 2. Verify the attestation aggregate (Type-2 over the body attestations + // only). A block with no attestations carries no aggregate; reject a + // stray proof rather than silently ignoring it. let crypto_start = std::time::Instant::now(); - ethlambda_crypto::verify_type_2_signature( - merged_bytes, - pubkeys_per_component, - &expected_bindings, - ) - .map_err(StoreError::AggregateVerificationFailed)?; + if attestations.is_empty() { + if !signed_block + .proof + .attestation_proof + .proof_bytes() + .is_empty() + { + return Err(StoreError::UnexpectedAttestationProof); + } + } else { + // Resolve pubkeys per Type-2 component and rederive the expected + // (message, slot) bindings from the block body. Each component uses its + // participants' attestation_pubkeys. + let mut pubkeys_per_component: Vec> = + Vec::with_capacity(attestations.len()); + let mut expected_bindings: Vec<(H256, u32)> = Vec::with_capacity(attestations.len()); + + for attestation in attestations.iter() { + let mut pubkeys = Vec::new(); + for vid in validator_indices(&attestation.aggregation_bits) { + let validator = validators + .get(vid as usize) + .ok_or(StoreError::InvalidValidatorIndex)?; + let pk = validator + .get_attestation_pubkey() + .map_err(|_| StoreError::PubkeyDecodingFailed(vid))?; + pubkeys.push(pk); + } + pubkeys_per_component.push(pubkeys); + let slot_u32 = u32::try_from(attestation.data.slot) + .map_err(|_| StoreError::SlotOutOfRange(attestation.data.slot))?; + expected_bindings.push((attestation.data.hash_tree_root(), slot_u32)); + } + + let merged_bytes = signed_block.proof.attestation_proof.proof_bytes(); + ethlambda_crypto::verify_type_2_signature( + merged_bytes, + pubkeys_per_component, + &expected_bindings, + ) + .map_err(StoreError::AggregateVerificationFailed)?; + } let crypto_elapsed = crypto_start.elapsed(); let total_elapsed = total_start.elapsed(); @@ -1182,7 +1205,7 @@ pub fn verify_block_signatures( ?structural_elapsed, ?crypto_elapsed, ?total_elapsed, - "Block multi-message aggregate proof verified" + "Block proof verified" ); Ok(()) @@ -1244,24 +1267,24 @@ mod tests { use ethlambda_types::{ attestation::{AggregatedAttestation, AggregationBits, AttestationData}, block::{ - AggregatedAttestations, BlockBody, MultiMessageAggregate, SignedBlock, - SingleMessageAggregate, + AggregatedAttestations, BlockBody, BlockProof, SignedBlock, SingleMessageAggregate, }, checkpoint::Checkpoint, state::State, }; - /// Test helper: placeholder block proof bytes. + /// Test helper: placeholder block proof. /// - /// In production the merged proof is the raw `compress_without_pubkeys()` - /// output of `merge_many_type_1`, which can only be built by the - /// lean-multisig prover. Tests that don't go through - /// `verify_block_signatures` use an empty blob. + /// In production the attestation aggregate is the raw + /// `compress_without_pubkeys()` output of `merge_many_type_1`, which can + /// only be built by the lean-multisig prover, and the proposer signature is + /// a real XMSS signature. Tests that don't go through + /// `verify_block_signatures` use an empty proof. fn make_signed_block_proof( _proposer_index: u64, _attestation_proofs: Vec, - ) -> MultiMessageAggregate { - MultiMessageAggregate::default() + ) -> BlockProof { + BlockProof::default() } fn make_bits(indices: &[usize]) -> AggregationBits { @@ -1737,7 +1760,7 @@ mod tests { }; let signed_block = SignedBlock { message: block, - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), }; let result = on_block_without_verification(&mut store, signed_block); @@ -1778,7 +1801,7 @@ mod tests { }; let signed_block = SignedBlock { message: block, - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), }; let result = on_block_without_verification(&mut store, signed_block); diff --git a/crates/blockchain/state_transition/src/lib.rs b/crates/blockchain/state_transition/src/lib.rs index 403b170a..7d19dd2f 100644 --- a/crates/blockchain/state_transition/src/lib.rs +++ b/crates/blockchain/state_transition/src/lib.rs @@ -13,6 +13,8 @@ use tracing::{info, warn}; pub mod justified_slots_ops; pub mod metrics; +pub const HEARTBEAT_COMMITTEE_SIZE: usize = 4; + #[derive(Debug, thiserror::Error)] pub enum Error { #[error("target slot {target_slot} is in the past (current is {current_slot})")] @@ -235,6 +237,21 @@ pub fn is_proposer(validator_index: u64, slot: u64, num_validators: u64) -> bool current_proposer(slot, num_validators) == Some(validator_index) } +/// Check if a validator is part of the heartbeat committee for a given slot. +/// +/// The heartbeat committee is formed by the proposer and the next N validators. +pub fn is_heartbeat_committee_member(validator_index: u64, slot: u64, num_validators: u64) -> bool { + let Some(proposer) = current_proposer(slot, num_validators) else { + return false; + }; + for i in 0..HEARTBEAT_COMMITTEE_SIZE as u64 { + if validator_index == (proposer + i) % num_validators { + return true; + } + } + false +} + /// Apply attestations and update justification/finalization /// according to the Lean Consensus 3SF-mini rules. fn process_attestations( @@ -426,11 +443,6 @@ fn is_valid_vote(state: &State, data: &AttestationData) -> bool { return false; } - // Ensure the target falls on a slot that can be justified after the finalized one. - if !slot_is_justifiable_after(target.slot, state.latest_finalized.slot) { - return false; - } - true } @@ -456,9 +468,7 @@ fn try_finalize( } // Consider whether finalization can advance. - if ((source.slot + 1)..target.slot) - .any(|slot| slot_is_justifiable_after(slot, state.latest_finalized.slot)) - { + if source.slot + 1 != target.slot { metrics::inc_finalizations("error"); return; } @@ -567,47 +577,6 @@ pub fn attestation_data_matches_chain( && historical_block_hashes[head_slot] == data.head.root } -/// Checks if the slot is a valid candidate for justification after a given finalized slot. -/// -/// According to the 3SF-mini specification, a slot is justifiable if its -/// distance (`delta`) from the last finalized slot is: -/// 1. Less than or equal to 5. -/// 2. A perfect square (e.g., 9, 16, 25...). -/// 3. A pronic number (of the form x^2 + x, e.g., 6, 12, 20...). -/// -/// See https://github.com/ethereum/research/blob/c003fe1c1a785797e7b53e3cbf9569b989be6e93/3sf-mini/consensus.py#L52-L54 -/// for the 3SF-mini reference. -/// -/// For why we have unjustifiable slots, consider that in high-latency -/// scenarios, validators may vote for many different slots, making none of them -/// reach the supermajority threshold. By having unjustifiable slots, we can -/// funnel votes towards only some slots, increasing finalization chances. -pub fn slot_is_justifiable_after(slot: u64, finalized_slot: u64) -> bool { - let Some(delta) = slot.checked_sub(finalized_slot) else { - // Candidate slot must not be before finalized slot - return false; - }; - // Rule 1: The first 5 slots after finalization are always justifiable. - // - // Examples: delta = 0, 1, 2, 3, 4, 5 - delta <= 5 - // Rule 2: Slots at perfect square distances are justifiable. - // - // Examples: delta = 1, 4, 9, 16, 25, 36, 49, 64, ... - // Check: integer square root squared equals delta - || delta.isqrt().pow(2) == delta - // Rule 3: Slots at pronic number distances are justifiable. - // - // Pronic numbers have the form n(n+1): 2, 6, 12, 20, 30, 42, 56, ... - // Mathematical insight: For pronic delta = n(n+1), we have: - // 4*delta + 1 = 4n(n+1) + 1 = (2n+1)^2 - // Check: 4*delta+1 is an odd perfect square - || delta - .checked_mul(4) - .and_then(|v| v.checked_add(1)) - .is_some_and(|val| val.isqrt().pow(2) == val && val % 2 == 1) -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/common/test-fixtures/src/fork_choice.rs b/crates/common/test-fixtures/src/fork_choice.rs index a6d2952f..889dff2d 100644 --- a/crates/common/test-fixtures/src/fork_choice.rs +++ b/crates/common/test-fixtures/src/fork_choice.rs @@ -8,7 +8,7 @@ use crate::{ deser_xmss_hex, }; use ethlambda_types::attestation::XmssSignature; -use ethlambda_types::block::{MultiMessageAggregate, SignedBlock}; +use ethlambda_types::block::{BlockProof, SignedBlock}; use ethlambda_types::primitives::H256; use serde::{Deserialize, Deserializer}; use std::collections::HashMap; @@ -209,7 +209,7 @@ impl BlockStepData { pub fn to_blank_signed_block(&self) -> SignedBlock { SignedBlock { message: self.to_block(), - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), } } } diff --git a/crates/common/test-fixtures/src/verify_signatures.rs b/crates/common/test-fixtures/src/verify_signatures.rs index 038ee5d7..54fbaeef 100644 --- a/crates/common/test-fixtures/src/verify_signatures.rs +++ b/crates/common/test-fixtures/src/verify_signatures.rs @@ -11,7 +11,8 @@ //! proof: { proof: { data: "0x" } } use crate::{Block, TestInfo, TestState}; -use ethlambda_types::block::{MultiMessageAggregate, SignedBlock}; +use ethlambda_types::attestation::blank_xmss_signature; +use ethlambda_types::block::{BlockProof, MultiMessageAggregate, SignedBlock}; use serde::Deserialize; use std::collections::HashMap; use std::fmt; @@ -139,17 +140,23 @@ impl TestSignedBlock { /// /// The container carries the raw lean-multisig wire in the /// `MultiMessageAggregate` stored by `SignedBlock.proof`. + /// + /// NOTE: these fixtures use the leanSpec #799 layout (proposer folded into + /// one merged Type-2). This client now carries the proposer signature + /// outside the attestation aggregate, so the merged bytes land in + /// `attestation_proof` with an empty proposer signature. The verify spec + /// tests therefore fail against these fixtures until they are regenerated. pub fn try_into_signed_block_with_proofs(self) -> Result { let bytes = self .proof .decode() .map_err(|err| SignedBlockConvertError::InvalidProofHex(err.to_string()))?; let len = bytes.len(); - let proof = MultiMessageAggregate::from_bytes(&bytes) + let attestation_proof = MultiMessageAggregate::from_bytes(&bytes) .map_err(|_| SignedBlockConvertError::ProofTooLarge(len))?; Ok(SignedBlock { message: self.block.into(), - proof, + proof: BlockProof::new(blank_xmss_signature(), attestation_proof), }) } } diff --git a/crates/common/types/src/attestation.rs b/crates/common/types/src/attestation.rs index 0d68addb..7715c58c 100644 --- a/crates/common/types/src/attestation.rs +++ b/crates/common/types/src/attestation.rs @@ -11,7 +11,7 @@ use crate::{ }; /// Validator specific attestation wrapping shared attestation data. -#[derive(Debug, Clone, SszEncode, SszDecode, HashTreeRoot)] +#[derive(Debug, Clone, Serialize, SszEncode, SszDecode, HashTreeRoot)] pub struct Attestation { /// The index of the validator making the attestation. pub validator_id: u64, diff --git a/crates/common/types/src/block.rs b/crates/common/types/src/block.rs index 5c5508a2..9770ace5 100644 --- a/crates/common/types/src/block.rs +++ b/crates/common/types/src/block.rs @@ -4,22 +4,27 @@ use libssz_derive::{HashTreeRoot, SszDecode, SszEncode}; use libssz_types::SszList; use crate::{ - attestation::{AggregatedAttestation, AggregationBits, validator_indices}, + attestation::{ + AggregatedAttestation, AggregationBits, Attestation, XmssSignature, blank_xmss_signature, + validator_indices, + }, primitives::{self, ByteList, H256}, }; // Convenience trait for calling hash_tree_root() without a hasher argument use primitives::HashTreeRoot as _; -/// Envelope carrying a block and the single merged proof binding every -/// signature it depends on. +/// Envelope carrying a block and its [`BlockProof`]. +/// +/// The proof keeps the proposer's raw signature separate from the attestation +/// aggregate (see [`BlockProof`]). /// ///
/// /// `HashTreeRoot` is intentionally not derived: consumers never hash a /// `SignedBlock` directly — they always hash the inner `Block`. Keeping the /// envelope structurally minimal also means the on-chain root is independent -/// of how the merged proof is serialised. +/// of how the proof is serialised. /// ///
#[derive(Clone, SszEncode, SszDecode)] @@ -27,16 +32,23 @@ pub struct SignedBlock { /// The block being signed. pub message: Block, - /// Single full-block proof covering attestations and the proposer signature. - pub proof: MultiMessageAggregate, + /// Full-block proof: proposer signature + attestation aggregate. + pub proof: BlockProof, } -// Manual Debug impl because the merged proof bytes are large and opaque. +// Manual Debug impl because the proof bytes are large and opaque. impl core::fmt::Debug for SignedBlock { fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { f.debug_struct("SignedBlock") .field("message", &self.message) - .field("proof", &format_args!("<{} bytes>", self.proof.proof.len())) + .field( + "proposer_signature", + &format_args!("<{} bytes>", self.proof.proposer_signature.len()), + ) + .field( + "attestation_proof", + &format_args!("<{} bytes>", self.proof.attestation_proof.proof.len()), + ) .finish() } } @@ -88,6 +100,77 @@ pub enum MultiMessageAggregateError { ProofTooLarge(usize), } +// ============================================================================ +// Block proof (proposer signature outside the attestation aggregate) +// ============================================================================ + +/// A full-block proof: the proposer's raw signature plus the attestation +/// aggregate, carried as two independent fields. +/// +/// ```text +/// attestations = [att0, att1] -> (proposer_signature, aggregate([att0, att1])) +/// attestations = [] -> (proposer_signature, empty-proof) +/// ``` +/// +/// `proposer_signature` is the proposer's raw XMSS signature over the block +/// root — the same fixed-size [`XmssSignature`] wire type carried by +/// `SignedAttestation`. It is verified directly against the proposer's +/// `proposal_pubkey` with the hash-based XMSS verifier, so it never enters the +/// lean-multisig prover/verifier. +/// +/// `attestation_proof` is the lean-multisig Type-2 over the block body's +/// attestations *only* — the proposer is no longer one of its components, so +/// it is empty when the block carries no attestations. +/// +///
+/// +/// `HashTreeRoot` is intentionally not derived (as on `SignedAttestation`): +/// `XmssSignature` is a fixed-size byte vector here, but the spec Merkleizes +/// the signature as a container, so a derived root would diverge. Nothing +/// hashes a `BlockProof`. +/// +///
+#[derive(Debug, Clone, PartialEq, Eq, SszEncode, SszDecode)] +pub struct BlockProof { + /// The proposer's raw XMSS signature over the block root. + pub proposer_signature: XmssSignature, + /// The heartbeat committee signatures over their votes. + pub heartbeat_signatures: HeartbeatSignatures, + /// Type-2 aggregate over the body attestations (empty if there are none). + pub attestation_proof: MultiMessageAggregate, +} + +/// List of heartbeat votes included in a block. +pub type HeartbeatSignatures = SszList; + +impl BlockProof { + /// Build a proof from a proposer signature and an attestation aggregate. + pub fn new( + proposer_signature: XmssSignature, + heartbeat_signatures: HeartbeatSignatures, + attestation_proof: MultiMessageAggregate, + ) -> Self { + Self { + proposer_signature, + heartbeat_signatures, + attestation_proof, + } + } +} + +impl Default for BlockProof { + /// A blank proof: the structurally-valid all-zero XMSS placeholder used by + /// genesis-style anchor blocks (see [`blank_xmss_signature`]) plus an empty + /// attestation aggregate. `XmssSignature` is fixed-size and has no empty + /// form, so the blank doubles as the genesis placeholder. + fn default() -> Self { + Self { + proposer_signature: blank_xmss_signature(), + attestation_proof: MultiMessageAggregate::default(), + } + } +} + // ============================================================================ // Single-message aggregate // ============================================================================ @@ -95,13 +178,12 @@ pub enum MultiMessageAggregateError { // Wire format mirrors leanSpec PR #717: `SingleMessageAggregate` is a flat // `{ participants, proof }` pair. The signed `message` and `slot` are NOT // carried on the envelope — verifiers rederive each component's binding -// from the surrounding block body (attestation `data` + slot for body -// components, block root + slot for the proposer component). +// from the surrounding block body (attestation `data` + slot). // -// `MultiMessageAggregate` carries the raw lean-multisig type-2 bytes. -// Component participant bitfields come from -// `block.body.attestations[i].aggregation_bits` (and `block.proposer_index` for -// the trailing proposer entry). +// `MultiMessageAggregate` carries the raw lean-multisig type-2 bytes for the +// body attestations only; the proposer signature is carried separately in +// `BlockProof::proposer_signature`. Component participant bitfields come from +// `block.body.attestations[i].aggregation_bits`. /// Maximum number of distinct `AttestationData` entries permitted in a single /// block. Canonical home for the cap shared across `ethlambda-blockchain`, @@ -245,6 +327,12 @@ impl Block { /// packaged into blocks. #[derive(Debug, Default, Clone, Serialize, SszEncode, SszDecode, HashTreeRoot)] pub struct BlockBody { + /// Plain validator attestations carried in the block body. + /// + /// Individual signatures live in the aggregated block signature list, so + /// these entries contain only attestation data without per-attestation signatures. + #[serde(serialize_with = "serialize_heartbeat")] + pub heartbeat: HeartbeatVotes, /// Plain validator attestations carried in the block body. /// /// Individual signatures live in the aggregated block signature list, so @@ -253,9 +341,22 @@ pub struct BlockBody { pub attestations: AggregatedAttestations, } +/// List of heartbeat votes included in a block. +pub type HeartbeatVotes = SszList; /// List of aggregated attestations included in a block. pub type AggregatedAttestations = SszList; +fn serialize_heartbeat(attestations: &HeartbeatVotes, serializer: S) -> Result +where + S: Serializer, +{ + let mut seq = serializer.serialize_seq(Some(attestations.len()))?; + for attestation in attestations.iter() { + seq.serialize_element(attestation)?; + } + seq.end() +} + fn serialize_attestations( attestations: &AggregatedAttestations, serializer: S, @@ -307,11 +408,13 @@ mod tests { }; let signed = SignedBlock { message: block, - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), }; let bytes = signed.to_ssz(); let decoded = SignedBlock::from_ssz_bytes(&bytes).expect("decode"); - assert_eq!(decoded.proof.proof.len(), 0); + // Default proof: empty attestation aggregate + the blank XMSS placeholder. + assert_eq!(decoded.proof.attestation_proof.proof.len(), 0); + assert_eq!(decoded.proof.proposer_signature, blank_xmss_signature()); assert_eq!(decoded.message.slot, signed.message.slot); assert_eq!( decoded.message.proposer_index, @@ -330,4 +433,42 @@ mod tests { assert_eq!(&encoded[4..], proof_bytes); assert_eq!(aggregate.proof_bytes(), proof_bytes); } + + #[test] + fn signed_block_ssz_round_trip_with_proposer_signature() { + let block = Block { + slot: 9, + proposer_index: 2, + parent_root: H256::ZERO, + state_root: H256::ZERO, + body: BlockBody::default(), + }; + // A distinctive, full-size XMSS signature blob (fixed `SIGNATURE_SIZE`). + let proposer_bytes: Vec = (0..crate::signature::SIGNATURE_SIZE) + .map(|i| (i % 251) as u8) + .collect(); + let proposer_signature = XmssSignature::try_from(proposer_bytes.clone()).unwrap(); + let attestation_bytes: Vec = (0..64).collect(); + let signed = SignedBlock { + message: block, + proof: BlockProof::new( + proposer_signature, + MultiMessageAggregate::from_bytes(&attestation_bytes).unwrap(), + ), + }; + + let bytes = signed.to_ssz(); + let decoded = SignedBlock::from_ssz_bytes(&bytes).expect("decode"); + + assert_eq!( + &*decoded.proof.proposer_signature, + proposer_bytes.as_slice() + ); + assert_eq!( + decoded.proof.attestation_proof.proof_bytes(), + attestation_bytes + ); + assert_eq!(decoded.message.slot, 9); + assert_eq!(decoded.message.proposer_index, 2); + } } diff --git a/crates/net/api/src/lib.rs b/crates/net/api/src/lib.rs index 9cdbcdd0..4f00dfd2 100644 --- a/crates/net/api/src/lib.rs +++ b/crates/net/api/src/lib.rs @@ -13,6 +13,10 @@ use spawned_concurrency::protocol; pub trait BlockChainToP2P: Send + Sync { fn publish_block(&self, block: SignedBlock) -> Result<(), ActorError>; fn publish_attestation(&self, attestation: SignedAttestation) -> Result<(), ActorError>; + fn publish_heartbeat_attestation( + &self, + attestation: SignedAttestation, + ) -> Result<(), ActorError>; fn publish_aggregated_attestation( &self, attestation: SignedAggregatedAttestation, diff --git a/crates/net/p2p/src/gossipsub/handler.rs b/crates/net/p2p/src/gossipsub/handler.rs index c257006b..0825fe1b 100644 --- a/crates/net/p2p/src/gossipsub/handler.rs +++ b/crates/net/p2p/src/gossipsub/handler.rs @@ -15,7 +15,7 @@ use super::{ attestation_subnet_topic, }, }; -use crate::{P2PServer, metrics}; +use crate::{P2PServer, gossipsub::messages::HEARTBEAT_TOPIC_KIND, metrics}; pub async fn handle_gossipsub_message(server: &mut P2PServer, event: Event) { let Event::Message { @@ -95,7 +95,10 @@ pub async fn handle_gossipsub_message(server: &mut P2PServer, event: Event) { ); } } - Some(kind) if kind.starts_with(ATTESTATION_SUBNET_TOPIC_PREFIX) => { + Some(kind) + if kind.starts_with(ATTESTATION_SUBNET_TOPIC_PREFIX) + || kind == HEARTBEAT_TOPIC_KIND => + { info!(kind = "attestation", peer_count, "P2P message received"); let compressed_len = message.data.len(); let Ok(uncompressed_data) = decompress_message(&message.data) @@ -196,6 +199,27 @@ pub async fn publish_block(server: &mut P2PServer, signed_block: SignedBlock) { ); } +pub async fn publish_heartbeat_attestation(server: &mut P2PServer, attestation: SignedAttestation) { + let slot = attestation.data.slot; + let validator = attestation.validator_id; + + // Encode to SSZ + let ssz_bytes = attestation.to_ssz(); + + // Compress with raw snappy + let compressed = compress_message(&ssz_bytes); + + // Publish to gossipsub + server + .swarm_handle + .publish(server.heartbeat_topic.clone(), compressed); + info!( + %slot, + validator, + "Published heartbeat attestation to gossipsub" + ); +} + pub async fn publish_aggregated_attestation( server: &mut P2PServer, attestation: SignedAggregatedAttestation, diff --git a/crates/net/p2p/src/gossipsub/messages.rs b/crates/net/p2p/src/gossipsub/messages.rs index 11664750..993a93dd 100644 --- a/crates/net/p2p/src/gossipsub/messages.rs +++ b/crates/net/p2p/src/gossipsub/messages.rs @@ -2,6 +2,10 @@ pub use ethlambda_types::constants::FORK_DIGEST; /// Topic kind for block gossip pub const BLOCK_TOPIC_KIND: &str = "block"; + +/// Topic kind for heartbeat gossip +pub const HEARTBEAT_TOPIC_KIND: &str = "heartbeat"; + /// Topic kind prefix for per-committee attestation subnets. /// /// Full topic format: `/leanconsensus/{FORK_DIGEST}/attestation_{subnet_id}/ssz_snappy` @@ -31,3 +35,10 @@ pub fn attestation_subnet_topic(subnet_id: u64) -> libp2p::gossipsub::IdentTopic "/leanconsensus/{FORK_DIGEST}/{ATTESTATION_SUBNET_TOPIC_PREFIX}_{subnet_id}/ssz_snappy" )) } + +/// Build a heartbeat gossipsub topic. +pub fn heartbeat_topic() -> libp2p::gossipsub::IdentTopic { + libp2p::gossipsub::IdentTopic::new(format!( + "/leanconsensus/{FORK_DIGEST}/{HEARTBEAT_TOPIC_KIND}/ssz_snappy" + )) +} diff --git a/crates/net/p2p/src/gossipsub/mod.rs b/crates/net/p2p/src/gossipsub/mod.rs index b50ea4fd..2ad00233 100644 --- a/crates/net/p2p/src/gossipsub/mod.rs +++ b/crates/net/p2p/src/gossipsub/mod.rs @@ -5,5 +5,6 @@ mod messages; pub use encoding::decompress_message; pub use handler::{ handle_gossipsub_message, publish_aggregated_attestation, publish_attestation, publish_block, + publish_heartbeat_attestation, }; -pub use messages::{aggregation_topic, attestation_subnet_topic, block_topic}; +pub use messages::{aggregation_topic, attestation_subnet_topic, block_topic, heartbeat_topic}; diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index 4726ce25..34e4068f 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -9,6 +9,7 @@ use ethlambda_network_api::{ InitBlockChain, P2PToBlockChainRef, block_chain_to_p2p::{ FetchBlock, PublishAggregatedAttestation, PublishAttestation, PublishBlock, + PublishHeartbeatAttestation, }, }; use ethlambda_storage::Store; @@ -37,8 +38,9 @@ use tracing::{info, trace, warn}; use crate::{ gossipsub::{ - aggregation_topic, attestation_subnet_topic, block_topic, publish_aggregated_attestation, - publish_attestation, publish_block, + aggregation_topic, attestation_subnet_topic, block_topic, heartbeat_topic, + publish_aggregated_attestation, publish_attestation, publish_block, + publish_heartbeat_attestation, }, req_resp::{ BLOCKS_BY_RANGE_PROTOCOL_V1, BLOCKS_BY_ROOT_PROTOCOL_V1, Codec, @@ -212,6 +214,7 @@ pub struct BuiltSwarm { pub(crate) attestation_topics: HashMap, pub(crate) attestation_committee_count: u64, pub(crate) block_topic: libp2p::gossipsub::IdentTopic, + pub(crate) heartbeat_topic: libp2p::gossipsub::IdentTopic, pub(crate) aggregation_topic: libp2p::gossipsub::IdentTopic, pub(crate) bootnode_addrs: HashMap, } @@ -327,6 +330,14 @@ pub fn build_swarm( .subscribe(&block_topic) .unwrap(); + // Subscribe to heartbeat topic (all nodes) + let heartbeat_topic = heartbeat_topic(); + swarm + .behaviour_mut() + .gossipsub + .subscribe(&heartbeat_topic) + .unwrap(); + // Subscribe to aggregation topic (all validators) let aggregation_topic = aggregation_topic(); swarm @@ -361,6 +372,7 @@ pub fn build_swarm( attestation_topics, attestation_committee_count: config.attestation_committee_count, block_topic, + heartbeat_topic, aggregation_topic, bootnode_addrs, }) @@ -386,6 +398,7 @@ impl P2P { attestation_topics: built.attestation_topics, attestation_committee_count: built.attestation_committee_count, block_topic: built.block_topic, + heartbeat_topic: built.heartbeat_topic, aggregation_topic: built.aggregation_topic, connected_peers: HashSet::new(), pending_root_requests: HashMap::new(), @@ -422,6 +435,7 @@ pub struct P2PServer { pub(crate) attestation_topics: HashMap, pub(crate) attestation_committee_count: u64, pub(crate) block_topic: libp2p::gossipsub::IdentTopic, + pub(crate) heartbeat_topic: libp2p::gossipsub::IdentTopic, pub(crate) aggregation_topic: libp2p::gossipsub::IdentTopic, pub(crate) connected_peers: HashSet, @@ -516,6 +530,12 @@ impl Handler for P2PServer { } } +impl Handler for P2PServer { + async fn handle(&mut self, msg: PublishHeartbeatAttestation, _ctx: &Context) { + publish_heartbeat_attestation(self, msg.attestation).await; + } +} + impl Handler for P2PServer { async fn handle(&mut self, msg: PublishAggregatedAttestation, _ctx: &Context) { publish_aggregated_attestation(self, msg.attestation).await; diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 810e658b..172975af 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -589,7 +589,7 @@ mod tests { use super::*; use ethlambda_storage::{ForkCheckpoints, backend::InMemoryBackend}; use ethlambda_types::{ - block::{Block, BlockBody, MultiMessageAggregate}, + block::{Block, BlockBody, BlockProof}, state::State, }; use std::sync::Arc; @@ -603,7 +603,7 @@ mod tests { state_root: H256::ZERO, body: BlockBody::default(), }, - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), } } diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs index 217543dd..6895ada6 100644 --- a/crates/net/rpc/src/lib.rs +++ b/crates/net/rpc/src/lib.rs @@ -460,7 +460,7 @@ mod tests { #[tokio::test] async fn test_get_latest_finalized_block() { use ethlambda_types::{ - block::{Block, BlockBody, MultiMessageAggregate, SignedBlock}, + block::{Block, BlockBody, BlockProof, SignedBlock}, checkpoint::Checkpoint, primitives::{H256, HashTreeRoot as _}, }; @@ -484,7 +484,7 @@ mod tests { let block_root = block.header().hash_tree_root(); let signed_block = SignedBlock { message: block, - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), }; // Persist the signed block and mark it as the latest finalized checkpoint. @@ -528,7 +528,7 @@ mod tests { #[tokio::test] async fn test_get_latest_finalized_block_serves_genesis_with_placeholder_proof() { - use ethlambda_types::block::{MultiMessageAggregate, SignedBlock}; + use ethlambda_types::block::{BlockProof, SignedBlock}; use libssz::SszEncode; // Genesis-anchored store: `init_store` writes the header + state but no @@ -553,7 +553,7 @@ mod tests { .unwrap(); let expected = SignedBlock { message: genesis_block.message.clone(), - proof: MultiMessageAggregate::default(), + proof: BlockProof::default(), }; let expected_ssz = expected.to_ssz(); diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 83f1e8fc..5a8f7cf5 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::num::NonZeroUsize; use std::sync::{Arc, LazyLock, Mutex}; +use ethlambda_types::attestation::Attestation; use lru::LruCache; use crate::api::{StorageBackend, StorageReadView, StorageWriteBatch, Table}; @@ -9,9 +10,7 @@ use crate::error::Error; use ethlambda_types::{ attestation::{AggregationBits, AttestationData, HashedAttestationData, bits_is_subset}, - block::{ - Block, BlockBody, BlockHeader, MultiMessageAggregate, SignedBlock, SingleMessageAggregate, - }, + block::{Block, BlockBody, BlockHeader, BlockProof, SignedBlock, SingleMessageAggregate}, checkpoint::Checkpoint, primitives::{H256, HashTreeRoot as _}, signature::ValidatorSignature, @@ -42,6 +41,10 @@ pub enum GetForkchoiceStoreError { /// allowing us to skip storing empty bodies and reconstruct them on read. static EMPTY_BODY_ROOT: LazyLock = LazyLock::new(|| BlockBody::default().hash_tree_root()); +const INTERVALS_PER_SLOT: u64 = 5; + +const RLMD_LOOKBACK_LIMIT: u64 = 8; + /// Checkpoints to update in the forkchoice store. /// /// Used with `Store::update_checkpoints` to update head and optionally @@ -551,6 +554,8 @@ fn encode_block_root_key(slot: u64) -> Vec { #[derive(Clone)] pub struct Store { backend: Arc, + /// List of votes observed in the last [`RLMD_LOOKBACK_LIMIT`] slots. + votes_per_slot: Arc>>>, new_payloads: Arc>, known_payloads: Arc>, /// In-memory gossip signatures, consumed at interval 2 aggregation. @@ -636,6 +641,7 @@ impl Store { } let store = Self { backend, + votes_per_slot: Arc::new(Mutex::new(BTreeMap::new())), new_payloads: Arc::new(Mutex::new(PayloadBuffer::new(NEW_PAYLOAD_CAP))), known_payloads: Arc::new(Mutex::new(PayloadBuffer::new(AGGREGATED_PAYLOAD_CAP))), gossip_signatures: Arc::new(Mutex::new(GossipSignatureBuffer::new( @@ -748,6 +754,7 @@ impl Store { Ok(Self { backend, + votes_per_slot: Arc::new(Mutex::new(BTreeMap::new())), new_payloads: Arc::new(Mutex::new(PayloadBuffer::new(NEW_PAYLOAD_CAP))), known_payloads: Arc::new(Mutex::new(PayloadBuffer::new(AGGREGATED_PAYLOAD_CAP))), gossip_signatures: Arc::new(Mutex::new(GossipSignatureBuffer::new( @@ -878,13 +885,17 @@ impl Store { .prune_live_chain(finalized.slot) .expect("prune live chain"); let pruned_sigs = self.prune_gossip_signatures(finalized.slot); - + let pruned_votes = self.prune_heartbeat_votes(finalized.slot); let pruned_payloads = self.prune_stale_aggregated_payloads(finalized.slot); if pruned_chain > 0 || pruned_sigs > 0 || pruned_payloads > 0 { info!( finalized_slot = finalized.slot, - pruned_chain, pruned_sigs, pruned_payloads, "Pruned finalized data" + pruned_chain, + pruned_sigs, + pruned_votes, + pruned_payloads, + "Pruned finalized data" ); } } @@ -1051,6 +1062,16 @@ impl Store { gossip.prune(finalized_slot) } + /// Prune heartbeat votes for slots <= finalized_slot. + /// + /// Returns the number of entries pruned. + pub fn prune_heartbeat_votes(&mut self, finalized_slot: u64) -> usize { + let mut votes_per_slot = self.votes_per_slot.lock().unwrap(); + let initial_len = votes_per_slot.len(); + votes_per_slot.retain(|slot, _| *slot >= finalized_slot); + initial_len - votes_per_slot.len() + } + /// Prune aggregated payload buffers (new + known) whose target slot is at or below /// `finalized_slot`. /// @@ -1233,12 +1254,12 @@ impl Store { let sig_key = encode_slot_root_key(header.slot, root); let proof = match view.get(Table::BlockSignatures, &sig_key).expect("get") { Some(proof_bytes) => { - MultiMessageAggregate::from_ssz_bytes(&proof_bytes).expect("valid block proof") + BlockProof::from_ssz_bytes(&proof_bytes).expect("valid block proof") } // Synthesis only covers the genesis-style anchor (slot 0). For any // other slot a missing proof (pruned finalized block, or genuine // corruption) surfaces as `None` rather than a fabricated block. - None if header.slot == 0 => MultiMessageAggregate::default(), + None if header.slot == 0 => BlockProof::default(), None => return None, }; @@ -1447,6 +1468,29 @@ impl Store { .extract_latest_attestations() } + pub fn get_last_period_votes(&self) -> HashMap { + let current_slot = self.time().unwrap() / INTERVALS_PER_SLOT; + let period_start_slot = current_slot.saturating_sub(RLMD_LOOKBACK_LIMIT); + // Deduplicate entries by keeping the latest attestation per validator + self.votes_per_slot + .lock() + .unwrap() + .range(period_start_slot..current_slot) + .flat_map(|(_, votes)| votes) + .map(|(x, (y, _))| (*x, y.clone())) + .collect() + } + + pub fn get_last_slot_votes(&self) -> HashMap { + let current_slot = self.time().unwrap() / INTERVALS_PER_SLOT; + self.votes_per_slot + .lock() + .unwrap() + .get(¤t_slot) + .cloned() + .unwrap_or_default() + } + // ============ Known Aggregated Payloads ============ // // "Known" aggregated payloads are active in fork choice weight calculations. @@ -1639,6 +1683,15 @@ impl Store { gossip.insert(hashed, validator_id, signature); } + pub fn insert_heartbeat_vote(&self, attestation: Attestation, signature: ValidatorSignature) { + self.votes_per_slot + .lock() + .unwrap() + .entry(attestation.data.slot) + .or_default() + .insert(attestation.validator_id, (attestation.data, signature)); + } + // ============ Derived Accessors ============ /// Returns the slot of the current head block. @@ -1805,6 +1858,7 @@ mod tests { let backend = Arc::new(InMemoryBackend::new()); Self { backend, + votes_per_slot: Arc::new(Mutex::new(BTreeMap::new())), new_payloads: Arc::new(Mutex::new(PayloadBuffer::new(NEW_PAYLOAD_CAP))), known_payloads: Arc::new(Mutex::new(PayloadBuffer::new(AGGREGATED_PAYLOAD_CAP))), gossip_signatures: Arc::new(Mutex::new(GossipSignatureBuffer::new( @@ -1819,6 +1873,7 @@ mod tests { fn test_store_with_backend(backend: Arc) -> Self { Self { backend, + votes_per_slot: Arc::new(Mutex::new(BTreeMap::new())), new_payloads: Arc::new(Mutex::new(PayloadBuffer::new(NEW_PAYLOAD_CAP))), known_payloads: Arc::new(Mutex::new(PayloadBuffer::new(AGGREGATED_PAYLOAD_CAP))), gossip_signatures: Arc::new(Mutex::new(GossipSignatureBuffer::new( @@ -2889,7 +2944,7 @@ mod tests { .expect("genesis block must be retrievable with synthetic proof"); assert_eq!(signed.message.slot, 0); - assert_eq!(signed.proof, MultiMessageAggregate::default()); + assert_eq!(signed.proof, BlockProof::default()); } /// The synthesis branch must be confined to the slot-0 anchor: a