diff --git a/crates/beacon_state/tile/src/bls.rs b/crates/beacon_state/tile/src/bls.rs index 6b141d14..55b8d014 100644 --- a/crates/beacon_state/tile/src/bls.rs +++ b/crates/beacon_state/tile/src/bls.rs @@ -35,6 +35,8 @@ pub const DOMAIN_VOLUNTARY_EXIT: u32 = 0x0000_0004; pub const DOMAIN_SELECTION_PROOF: u32 = 0x0000_0005; pub const DOMAIN_AGGREGATE_AND_PROOF: u32 = 0x0000_0006; pub const DOMAIN_SYNC_COMMITTEE: u32 = 0x0000_0007; +pub const DOMAIN_SYNC_COMMITTEE_SELECTION_PROOF: u32 = 0x0000_0008; +pub const DOMAIN_CONTRIBUTION_AND_PROOF: u32 = 0x0000_0009; pub const DOMAIN_BLS_TO_EXECUTION_CHANGE: u32 = 0x0000_000a; // Gloas pub const DOMAIN_BEACON_BUILDER: u32 = 0x0000_000b; diff --git a/crates/beacon_state/tile/src/counters.rs b/crates/beacon_state/tile/src/counters.rs index 733d8fe3..fa370952 100644 --- a/crates/beacon_state/tile/src/counters.rs +++ b/crates/beacon_state/tile/src/counters.rs @@ -21,10 +21,14 @@ silver_common::declare_counters! { // aggregates inflate Miss without ever becoming votes. AttestationRootMemoHit, AttestationRootMemoMiss, - // gossip-attestation batching: size of the latest flush (gauge) and - // batch-verify failures that fell back to per-attestation verifies - // (non-zero = someone is feeding us invalid signatures). - AttestationBatchSize, - AttestationBatchFallback, + // gossip vote batching (attestations + sync messages + PTC): size of + // the latest flush (gauge) and batch-verify failures that fell back + // to per-message verifies (non-zero = someone is feeding us invalid + // signatures). + VoteBatchSize, + VoteBatchFallback, + // Valid sync messages are still accepted and relayed when this is + // full; only creation of a local contribution is skipped. + SyncContributionPoolFull, } } diff --git a/crates/beacon_state/tile/src/fork_choice/node.rs b/crates/beacon_state/tile/src/fork_choice/node.rs index b805367f..28db4159 100644 --- a/crates/beacon_state/tile/src/fork_choice/node.rs +++ b/crates/beacon_state/tile/src/fork_choice/node.rs @@ -106,6 +106,23 @@ impl PtcVotes { } } + #[inline] + pub(super) fn record_mask(&mut self, positions: &[u64; 8], present: bool, da: bool) { + for (word, &positions) in self.voted.iter_mut().zip(positions) { + *word |= positions; + } + if present { + for (word, &positions) in self.present.iter_mut().zip(positions) { + *word |= positions; + } + } + if da { + for (word, &positions) in self.da.iter_mut().zip(positions) { + *word |= positions; + } + } + } + #[inline] pub(super) fn present_count(&self) -> usize { popcount(&self.present) @@ -116,18 +133,18 @@ impl PtcVotes { popcount(&self.da) } - #[cfg(feature = "ef_tests")] + #[cfg(any(test, feature = "ef_tests"))] pub(super) fn timeliness(&self) -> [Option; PTC_SIZE] { self.optional(&self.present) } /// Data-availability votes (see `timeliness`). - #[cfg(feature = "ef_tests")] + #[cfg(any(test, feature = "ef_tests"))] pub(super) fn availability(&self) -> [Option; PTC_SIZE] { self.optional(&self.da) } - #[cfg(feature = "ef_tests")] + #[cfg(any(test, feature = "ef_tests"))] fn optional(&self, value: &[u64; 8]) -> [Option; PTC_SIZE] { let mut out = [None; PTC_SIZE]; for (i, slot) in out.iter_mut().enumerate() { diff --git a/crates/beacon_state/tile/src/fork_choice/payload.rs b/crates/beacon_state/tile/src/fork_choice/payload.rs index 4cc8d82d..43837f02 100644 --- a/crates/beacon_state/tile/src/fork_choice/payload.rs +++ b/crates/beacon_state/tile/src/fork_choice/payload.rs @@ -106,7 +106,21 @@ impl ForkChoice { self.head_moved = true; } - #[cfg(feature = "ef_tests")] + pub fn record_ptc_votes( + &mut self, + block_root: &B256, + positions: &[u64; PTC_SIZE / 64], + present: bool, + da: bool, + ) { + let Some(idx) = self.find_node_idx(block_root) else { + return; + }; + self.nodes[idx].ptc.record_mask(positions, present, da); + self.head_moved = true; + } + + #[cfg(any(test, feature = "ef_tests"))] pub fn ptc_timeliness_votes(&self, block_root: &B256) -> [Option; PTC_SIZE] { match self.find_node_idx(block_root) { Some(idx) => self.nodes[idx].ptc.timeliness(), @@ -114,7 +128,7 @@ impl ForkChoice { } } - #[cfg(feature = "ef_tests")] + #[cfg(any(test, feature = "ef_tests"))] pub fn ptc_data_availability_votes(&self, block_root: &B256) -> [Option; PTC_SIZE] { match self.find_node_idx(block_root) { Some(idx) => self.nodes[idx].ptc.availability(), diff --git a/crates/beacon_state/tile/src/ssz_hash.rs b/crates/beacon_state/tile/src/ssz_hash.rs index fac00852..d8b981ba 100644 --- a/crates/beacon_state/tile/src/ssz_hash.rs +++ b/crates/beacon_state/tile/src/ssz_hash.rs @@ -20,6 +20,40 @@ pub fn hash_tree_root_block_header(hdr: &BeaconBlockHeader) -> B256 { merkleize(&chunks) } +pub fn hash_tree_root_sync_contribution( + slot: u64, + beacon_block_root: &B256, + subcommittee_index: u64, + aggregation_bits: &[u8; 16], + signature: &[u8; 96], +) -> B256 { + let mut bits_chunk = [0u8; 32]; + bits_chunk[..16].copy_from_slice(aggregation_bits); + merkleize(&[ + uint64_chunk(slot), + *beacon_block_root, + uint64_chunk(subcommittee_index), + bits_chunk, + hash_fixed_bytes(signature), + ]) +} + +pub fn hash_tree_root_contribution_and_proof( + aggregator_index: u64, + contribution_root: &B256, + selection_proof: &[u8; 96], +) -> B256 { + merkleize(&[ + uint64_chunk(aggregator_index), + *contribution_root, + hash_fixed_bytes(selection_proof), + ]) +} + +pub fn hash_tree_root_sync_selection_data(slot: u64, subcommittee_index: u64) -> B256 { + hash_concat(&uint64_chunk(slot), &uint64_chunk(subcommittee_index)) +} + pub fn hash_checkpoint(cp: &Checkpoint) -> B256 { hash_concat(&uint64_chunk(cp.epoch), &cp.root) } diff --git a/crates/beacon_state/tile/src/stf/gloas/mod.rs b/crates/beacon_state/tile/src/stf/gloas/mod.rs index b0f01c85..0a68edb4 100644 --- a/crates/beacon_state/tile/src/stf/gloas/mod.rs +++ b/crates/beacon_state/tile/src/stf/gloas/mod.rs @@ -16,5 +16,6 @@ pub use committee::process_ptc_window; pub(crate) use committee::{fill_epoch_ptc, get_ptc}; pub use envelope::verify_execution_payload_envelope; pub use parent_payload::process_parent_execution_payload; +pub(crate) use payload_attestation::hash_payload_attestation_data; pub use payload_attestation::{collect_sigs_payload_attestations, process_payload_attestations}; pub use withdrawals::process_withdrawals as process_withdrawals_gloas; diff --git a/crates/beacon_state/tile/src/stf/gloas/payload_attestation.rs b/crates/beacon_state/tile/src/stf/gloas/payload_attestation.rs index f06a8bf2..06780038 100644 --- a/crates/beacon_state/tile/src/stf/gloas/payload_attestation.rs +++ b/crates/beacon_state/tile/src/stf/gloas/payload_attestation.rs @@ -102,7 +102,7 @@ fn collect_sigs_payload_attestation( Ok(()) } -fn hash_payload_attestation_data(data: &[u8; 42]) -> silver_beacon_state_data::B256 { +pub(crate) fn hash_payload_attestation_data(data: &[u8; 42]) -> silver_beacon_state_data::B256 { let bool_chunk = |b: u8| { let mut c = [0u8; 32]; c[0] = b; diff --git a/crates/beacon_state/tile/src/stf/mod.rs b/crates/beacon_state/tile/src/stf/mod.rs index 3d810db7..87dc73f3 100644 --- a/crates/beacon_state/tile/src/stf/mod.rs +++ b/crates/beacon_state/tile/src/stf/mod.rs @@ -36,7 +36,6 @@ pub use epoch::{ }; pub use epoch_shuffling::{EpochShuffling, ShufflingRef}; pub use fork_transition::upgrade_to_gloas; -pub(crate) use gloas::get_ptc; pub use gloas::{ collect_sigs_execution_payload_bid, collect_sigs_payload_attestations, get_builder_payment_quorum_threshold, process_builder_deposit_request, @@ -44,6 +43,7 @@ pub use gloas::{ process_parent_execution_payload, process_payload_attestations, process_ptc_window, process_withdrawals_gloas, verify_execution_payload_envelope, }; +pub(crate) use gloas::{get_ptc, hash_payload_attestation_data}; pub(crate) use operations::process_execution_requests; pub use operations::{ collect_sigs_bls_to_execution_changes, collect_sigs_voluntary_exits, diff --git a/crates/beacon_state/tile/src/test_signing.rs b/crates/beacon_state/tile/src/test_signing.rs index 749e6176..2cc92575 100644 --- a/crates/beacon_state/tile/src/test_signing.rs +++ b/crates/beacon_state/tile/src/test_signing.rs @@ -9,10 +9,10 @@ use silver_beacon_state_data::{ B256, BLSPubkey, BeaconBlockHeader, Fork, Immutable, SLOTS_PER_EPOCH, }; use silver_common::ssz_view::{ - ATTESTATION_FIXED, AttestationView, IndexedAttestationView, PROPOSER_SLASHING_SIZE, - ProposerSlashingView, SIGNED_AGG_PROOF_MIN, SIGNED_BLS_CHANGE_SIZE, SIGNED_VOLUNTARY_EXIT_SIZE, - SINGLE_ATT_SIZE, SignedAggregateAndProofView, SignedBlsToExecutionChangeView, - SignedVoluntaryExitView, SingleAttestationView, + ATTESTATION_FIXED, AttestationView, IndexedAttestationView, PAYLOAD_ATTESTATION_MESSAGE_SIZE, + PROPOSER_SLASHING_SIZE, ProposerSlashingView, SIGNED_AGG_PROOF_MIN, SIGNED_BLS_CHANGE_SIZE, + SIGNED_VOLUNTARY_EXIT_SIZE, SINGLE_ATT_SIZE, SignedAggregateAndProofView, + SignedBlsToExecutionChangeView, SignedVoluntaryExitView, SingleAttestationView, }; use crate::{ @@ -264,6 +264,117 @@ pub fn resign_single_attestation(sk_idx: usize, buf: &mut [u8; SINGLE_ATT_SIZE], debug_assert_eq!(SingleAttestationView::signature(buf), &sig); } +pub fn sign_sync_committee_message( + sk_idx: usize, + vi: u64, + slot: u64, + beacon_block_root: B256, + imm: &Immutable, +) -> [u8; 144] { + let mut buf = [0u8; 144]; + buf[0..8].copy_from_slice(&slot.to_le_bytes()); + buf[8..40].copy_from_slice(&beacon_block_root); + buf[40..48].copy_from_slice(&vi.to_le_bytes()); + + let fv = test_fork_version(slot / SLOTS_PER_EPOCH); + let domain = bls::compute_domain(bls::DOMAIN_SYNC_COMMITTEE, fv, &imm.genesis_validators_root); + let sig = sign(sk_idx, &bls::compute_signing_root(&beacon_block_root, &domain)); + buf[48..144].copy_from_slice(&sig); + buf +} + +pub fn sign_payload_attestation_message( + sk_idx: usize, + validator_index: u64, + slot: u64, + beacon_block_root: B256, + payload_present: u8, + blob_data_available: u8, + imm: &Immutable, +) -> [u8; PAYLOAD_ATTESTATION_MESSAGE_SIZE] { + let mut buf = [0u8; PAYLOAD_ATTESTATION_MESSAGE_SIZE]; + buf[0..8].copy_from_slice(&validator_index.to_le_bytes()); + buf[8..40].copy_from_slice(&beacon_block_root); + buf[40..48].copy_from_slice(&slot.to_le_bytes()); + buf[48] = payload_present; + buf[49] = blob_data_available; + + let data: &[u8; 42] = (&buf[8..50]).try_into().unwrap(); + let fv = test_fork_version(slot / SLOTS_PER_EPOCH); + let domain = bls::compute_domain(bls::DOMAIN_PTC_ATTESTER, fv, &imm.genesis_validators_root); + let signing_root = + bls::compute_signing_root(&crate::stf::hash_payload_attestation_data(data), &domain); + let signature = sign(sk_idx, &signing_root); + buf[50..146].copy_from_slice(&signature); + buf +} + +pub fn sync_selection_proof( + sk_idx: usize, + slot: u64, + subcommittee: u64, + imm: &Immutable, +) -> [u8; 96] { + let fv = test_fork_version(slot / SLOTS_PER_EPOCH); + let domain = bls::compute_domain( + bls::DOMAIN_SYNC_COMMITTEE_SELECTION_PROOF, + fv, + &imm.genesis_validators_root, + ); + let root = ssz_hash::hash_tree_root_sync_selection_data(slot, subcommittee); + sign(sk_idx, &bls::compute_signing_root(&root, &domain)) +} + +pub fn sign_contribution_and_proof( + sk_agg: usize, + aggregator_index: u64, + slot: u64, + subcommittee: u64, + participant_pos: usize, + sk_part: usize, + beacon_block_root: B256, + imm: &Immutable, +) -> [u8; 360] { + let fv = test_fork_version(slot / SLOTS_PER_EPOCH); + let domain = |ty| bls::compute_domain(ty, fv, &imm.genesis_validators_root); + + let mut bits = [0u8; 16]; + bits[participant_pos / 8] |= 1 << (participant_pos % 8); + let contrib_sig = sign( + sk_part, + &bls::compute_signing_root(&beacon_block_root, &domain(bls::DOMAIN_SYNC_COMMITTEE)), + ); + let selection_proof = sync_selection_proof(sk_agg, slot, subcommittee, imm); + + let contribution_root = ssz_hash::hash_tree_root_sync_contribution( + slot, + &beacon_block_root, + subcommittee, + &bits, + &contrib_sig, + ); + let cap_root = ssz_hash::hash_tree_root_contribution_and_proof( + aggregator_index, + &contribution_root, + &selection_proof, + ); + let outer = sign( + sk_agg, + &bls::compute_signing_root(&cap_root, &domain(bls::DOMAIN_CONTRIBUTION_AND_PROOF)), + ); + + let mut buf = [0u8; 360]; + buf[0..8].copy_from_slice(&aggregator_index.to_le_bytes()); + buf[8..16].copy_from_slice(&slot.to_le_bytes()); + buf[16..48].copy_from_slice(&beacon_block_root); + buf[48..56].copy_from_slice(&subcommittee.to_le_bytes()); + buf[56..72].copy_from_slice(&bits); + buf[72..168].copy_from_slice(&contrib_sig); + buf[168..264].copy_from_slice(&selection_proof); + buf[264..360].copy_from_slice(&outer); + buf +} + /// Build a single-signer `Attestation`: `committee_index` is the (single) /// bit in `committee_bits`; the bit at `participant_pos` is the only set bit /// in `aggregation_bits` — that participant signs the aggregate. diff --git a/crates/beacon_state/tile/src/tile.rs b/crates/beacon_state/tile/src/tile.rs index 011c5f25..b92920b0 100644 --- a/crates/beacon_state/tile/src/tile.rs +++ b/crates/beacon_state/tile/src/tile.rs @@ -1,4 +1,4 @@ -use std::{fmt::Debug, sync::Arc}; +use std::{fmt::Debug, sync::Arc, time::Duration}; use flux::{ spine::{FluxSpine, SpineAdapter, SpineProducers}, @@ -26,6 +26,7 @@ use crate::{ attestation_pool::AttestationPool, attestation_root_memo::AttestationRootMemo, fork_data_roots::ForkDataRoots, orphan_pool::PendingBlock, seen_aggregates::SeenAggregates, seen_validators::SeenValidators, shuffling_cache::ShufflingCache, + sync_contribution_pool::SyncContributionPool, }, weak_subjectivity::{weak_subjectivity_period_fulu, weak_subjectivity_period_gloas}, }; @@ -42,6 +43,10 @@ mod orphan_pool; mod seen_aggregates; mod seen_validators; mod shuffling_cache; +mod sync_contribution_pool; + +/// Consensus-spec clock-skew allowance for slot-scoped gossip validation. +const MAXIMUM_GOSSIP_CLOCK_DISPARITY: Duration = Duration::from_millis(500); #[derive(Clone, Copy, PartialEq, Eq)] pub enum Feedback { @@ -121,9 +126,13 @@ pub struct BeaconStateTile { seen_aggregates: SeenAggregates, attestation_pool: AttestationPool, attestation_root_memo: AttestationRootMemo, - att_batch: Vec, - att_pending: Vec<(NewGossipMsg, gossip::PreparedAttestation)>, - att_sig_batch: bls::SigBatch, + vote_batch: Vec, + vote_pending: Vec<(NewGossipMsg, gossip::PreparedVote)>, + vote_sig_batch: bls::SigBatch, + seen_sync_msgs: [SeenValidators; silver_common::SYNC_COMMITTEE_SUBNETS], + sync_contribution_pool: SyncContributionPool, + seen_contribution_aggregators: [SeenValidators; silver_common::SYNC_COMMITTEE_SUBNETS], + seen_ptc: SeenValidators, fork_data_roots: ForkDataRoots, /// Canonical in-process state: finalized base + per-fork per-tier rings. @@ -215,9 +224,13 @@ impl BeaconStateTile { seen_aggregators: SeenValidators::new(val_cap), seen_aggregates: SeenAggregates::new(), attestation_pool: AttestationPool::new(), - att_batch: Vec::with_capacity(gossip::ATT_BATCH_CAP), - att_pending: Vec::with_capacity(gossip::ATT_BATCH_CAP), - att_sig_batch: bls::SigBatch::new(), + vote_batch: Vec::with_capacity(gossip::VOTE_BATCH_CAP), + vote_pending: Vec::with_capacity(gossip::VOTE_BATCH_CAP), + vote_sig_batch: bls::SigBatch::new(), + seen_sync_msgs: std::array::from_fn(|_| SeenValidators::new(val_cap)), + sync_contribution_pool: SyncContributionPool::new(), + seen_contribution_aggregators: std::array::from_fn(|_| SeenValidators::new(val_cap)), + seen_ptc: SeenValidators::new(val_cap), attestation_root_memo: AttestationRootMemo::default(), fork_data_roots: ForkDataRoots::default(), last_applied: anchor, @@ -540,6 +553,7 @@ impl BeaconStateTile { self.fork_choice_tick(); let floor = slot.saturating_sub(1); self.attestation_pool.prune_before(floor); + self.sync_contribution_pool.prune_before(floor); self.seen_aggregates.prune_before(floor); self.attestation_root_memo.prune_before(floor); advanced @@ -602,18 +616,25 @@ impl BeaconStateTile { } adapter.consume(|m: NewGossipMsg, producers| self.on_gossip(m, producers)); - self.flush_attestations(&mut adapter.producers); + self.flush_votes(&mut adapter.producers); self.gossip_consumer.free(); } - /// Attestations are deferred into the batch; anything else flushes it - /// first, so the queue order the batch reorders is restored here. + /// Per-validator votes (attestations, sync committee messages, PTC + /// attestations) are deferred into the shared batch; anything else + /// flushes it first, so the queue order the batch reorders is restored + /// here. fn on_gossip(&mut self, m: NewGossipMsg, producers: &mut Producers) { - if matches!(m.topic, GossipTopic::BeaconAttestation(_)) { - self.defer_attestation(m, producers); + if matches!( + m.topic, + GossipTopic::BeaconAttestation(_) | + GossipTopic::SyncCommittee(_) | + GossipTopic::PayloadAttestationMessage + ) { + self.defer_vote(m, producers); return; } - self.flush_attestations(producers); + self.flush_votes(producers); self.handle_gossip(m.ssz, m, true, false, producers); } @@ -751,7 +772,14 @@ impl BeaconStateTile { } pub fn ef_apply_payload_attestation(&mut self, ssz: &[u8]) -> bool { - matches!(self.handle_payload_attestation(ssz), Feedback::Accept(_)) + match self.prepare_ptc(ssz) { + Ok(p) => { + self.commit_ptc(&p); + self.recompute_head(); + true + } + Err(_) => false, + } } pub fn ef_payload_verdict( diff --git a/crates/beacon_state/tile/src/tile/attestation_pool.rs b/crates/beacon_state/tile/src/tile/attestation_pool.rs index c85bfef7..fccad862 100644 --- a/crates/beacon_state/tile/src/tile/attestation_pool.rs +++ b/crates/beacon_state/tile/src/tile/attestation_pool.rs @@ -40,7 +40,7 @@ pub(super) enum InsertOutcome { /// Slot below the retention floor — valid vote, just not aggregable. Stale, Full, - /// Position ≥ committee length, or length differs from the entry's. + /// Participation metadata is invalid or conflicts with the entry's. Inconsistent, } diff --git a/crates/beacon_state/tile/src/tile/fork_choice.rs b/crates/beacon_state/tile/src/tile/fork_choice.rs index 9c807131..a64f7426 100644 --- a/crates/beacon_state/tile/src/tile/fork_choice.rs +++ b/crates/beacon_state/tile/src/tile/fork_choice.rs @@ -12,8 +12,12 @@ use silver_common::{ }; use silver_ssz::ssz_view::PAYLOAD_ATTESTATION_MESSAGE_SIZE; -use super::{BeaconStateTile, Producers}; -use crate::{stf, tile::Feedback}; +use super::{BeaconStateTile, MAXIMUM_GOSSIP_CLOCK_DISPARITY, Producers}; +use crate::{ + bls::{self, CheckedSignature}, + stf, + tile::{Feedback, gossip::PreparedPtc}, +}; impl BeaconStateTile { /// Rebuild fork choice's justified-balance snapshot when its justified @@ -140,41 +144,90 @@ impl BeaconStateTile { } } - pub(super) fn handle_payload_attestation(&mut self, ssz: &[u8]) -> Feedback { - if ssz.len() < PAYLOAD_ATTESTATION_MESSAGE_SIZE { - return Feedback::Reject(None); + pub(super) fn prepare_ptc(&mut self, ssz: &[u8]) -> Result { + if ssz.len() != PAYLOAD_ATTESTATION_MESSAGE_SIZE { + return Err(Feedback::Reject(None)); } let buf: &[u8; PAYLOAD_ATTESTATION_MESSAGE_SIZE] = ssz[..PAYLOAD_ATTESTATION_MESSAGE_SIZE].try_into().unwrap(); let validator_index = PayloadAttestationMessage::validator_index(buf); let data = PayloadAttestationMessage::data(buf); + // SSZ bools have exactly two canonical encodings. The view accessors + // intentionally expose bool semantics, so reject non-canonical bytes + // before converting them. + if data[40] > 1 || data[41] > 1 { + return Err(Feedback::Reject(None)); + } let block_root = *PayloadAttestationData::beacon_block_root(data); let slot = PayloadAttestationData::slot(data); let present = PayloadAttestationData::payload_present(data); let da = PayloadAttestationData::blob_data_available(data); + if !self.ticker.is_current_slot_with_disparity(slot, MAXIMUM_GOSSIP_CLOCK_DISPARITY) { + return Err(Feedback::Ignore); + } + self.seen_ptc + .rotate_to(self.ticker.latest_slot_with_disparity(MAXIMUM_GOSSIP_CLOCK_DISPARITY)); + if self.seen_ptc.contains(slot, validator_index as usize) { + return Err(Feedback::Ignore); + } + let Some(idx) = self.fork_choice.find_node_idx(&block_root) else { - return Feedback::Ignore; + return Err(Feedback::Ignore); }; - let state_id = self.fork_choice.node(idx).state_id; + let node = self.fork_choice.node(idx); + if node.slot != slot { + return Err(Feedback::Ignore); + } + let state_id = node.state_id; - let ptc_idx = { - let rv = self.state.read_view(state_id); - let state_epoch = rv.slot.slot_number() / SLOTS_PER_EPOCH; - let Some(ptc) = stf::get_ptc(&rv.epoch, state_epoch, slot) else { - return Feedback::Ignore; - }; - match ptc.iter().position(|&v| v == validator_index) { - Some(p) => p, - None => return Feedback::Ignore, + let rv = self.state.read_view(state_id); + let state_epoch = rv.slot.slot_number() / SLOTS_PER_EPOCH; + let Some(ptc) = stf::get_ptc(&rv.epoch, state_epoch, slot) else { + return Err(Feedback::Ignore); + }; + let mut ptc_positions = [0u64; crate::tile::gossip::PTC_MASK_WORDS]; + for (position, &member) in ptc.iter().enumerate() { + if member == validator_index { + ptc_positions[position / 64] |= 1 << (position % 64); } + } + if ptc_positions.iter().all(|&word| word == 0) { + return Err(Feedback::Ignore); }; + if validator_index as usize >= rv.validators.count() { + return Err(Feedback::Reject(None)); + } - self.fork_choice.record_ptc_vote(&block_root, ptc_idx, present, da); - self.recompute_head(); + let fork_version = rv.epoch.fork_version_at(slot / SLOTS_PER_EPOCH); + let domain = bls::domain_from_fork_data( + bls::DOMAIN_PTC_ATTESTER, + &self.fork_data_roots.root(fork_version, &rv.imm.genesis_validators_root), + ); + let signing_root = + bls::compute_signing_root(&stf::hash_payload_attestation_data(data), &domain); + let Some(signature) = CheckedSignature::parse(PayloadAttestationMessage::signature(buf)) + else { + return Err(Feedback::Reject(None)); + }; + + Ok(PreparedPtc { + block_root, + slot, + validator: validator_index, + ptc_positions, + present, + da, + pubkey: *rv.validators.pubkey_decompressed(validator_index as usize), + signing_root, + signature, + }) + } - Feedback::Accept(Some(block_root)) + pub(super) fn commit_ptc(&mut self, p: &PreparedPtc) { + self.fork_choice.record_ptc_votes(&p.block_root, &p.ptc_positions, p.present, p.da); + self.seen_ptc.mark(p.slot, p.validator as usize); } pub(super) fn notify_ptc_from_block(&mut self, block_data: &[u8]) { diff --git a/crates/beacon_state/tile/src/tile/gossip.rs b/crates/beacon_state/tile/src/tile/gossip.rs index 161028f4..3d713691 100644 --- a/crates/beacon_state/tile/src/tile/gossip.rs +++ b/crates/beacon_state/tile/src/tile/gossip.rs @@ -1,6 +1,7 @@ use flux::spine::SpineProducers; use silver_beacon_state_data::{ - B256, ParsedAggregateAndProof, SLOTS_PER_EPOCH, Slot, StateId, StateReadView, + B256, BLSPubkey, EPOCHS_PER_SYNC_COMMITTEE_PERIOD, ParsedAggregateAndProof, SLOTS_PER_EPOCH, + SYNC_COMMITTEE_SIZE, Slot, StateId, StateReadView, ValidatorsView, gloas::PTC_SIZE, }; use silver_common::{ ATTESTATION_SUBNETS, BeaconStateEvent, BlockSource, DataKind, EngineNewPayloadEnvelopeReq, @@ -10,17 +11,21 @@ use silver_common::{ ssz_view::{ AttestationDataView, AttesterSlashingView, ExecutionPayloadEnvelopeView as Envelope, PROPOSER_SLASHING_SIZE, ProposerSlashingView, SIGNED_BLS_CHANGE_SIZE, - SIGNED_VOLUNTARY_EXIT_SIZE, SINGLE_ATT_SIZE, SignedBlsToExecutionChangeView, + SIGNED_CONTRIBUTION_AND_PROOF_SIZE, SIGNED_VOLUNTARY_EXIT_SIZE, SINGLE_ATT_SIZE, + SYNC_COMMITTEE_MSG_SIZE, SignedBlsToExecutionChangeView, + SignedContributionAndProofView as ContributionView, SignedExecutionPayloadEnvelopeView as SignedPayload, SignedVoluntaryExitView, - SingleAttestationView, + SingleAttestationView, SyncCommitteeView, }, }; use super::{ - ATTESTATION_PROPAGATION_SLOT_RANGE, BeaconStateTile, Feedback, Producers, + ATTESTATION_PROPAGATION_SLOT_RANGE, BeaconStateTile, Feedback, MAXIMUM_GOSSIP_CLOCK_DISPARITY, + Producers, attestation_pool::InsertOutcome, orphan_pool::{PendingBlock, has_room}, seen_aggregates::Coverage, + sync_contribution_pool::SYNC_SUBCOMMITTEE_MASK_WORDS, }; use crate::{ bls::{self, CheckedSignature, PublicKey, VerifiedSingleAttestation}, @@ -28,7 +33,10 @@ use crate::{ merkle, ssz_hash, stf, validate, }; -pub(super) const ATT_BATCH_CAP: usize = 1024; +pub(super) const VOTE_BATCH_CAP: usize = 1024; + +const SYNC_SUBCOMMITTEE_SIZE: usize = SYNC_COMMITTEE_SIZE / silver_common::SYNC_COMMITTEE_SUBNETS; +pub(super) const PTC_MASK_WORDS: usize = PTC_SIZE.div_ceil(64); pub(super) struct PreparedAttestation { buf: [u8; SINGLE_ATT_SIZE], @@ -41,6 +49,144 @@ pub(super) struct PreparedAttestation { vote: stf::AttestationVote, } +pub(super) struct PreparedSyncMessage { + slot: Slot, + subnet: u64, + validator: u64, + block_root: B256, + positions: [u64; SYNC_SUBCOMMITTEE_MASK_WORDS], + pubkey: PublicKey, + signing_root: B256, + signature: CheckedSignature, +} + +pub(crate) struct PreparedPtc { + pub block_root: B256, + pub slot: Slot, + pub validator: u64, + pub ptc_positions: [u64; PTC_MASK_WORDS], + pub present: bool, + pub da: bool, + pub pubkey: PublicKey, + pub signing_root: B256, + pub signature: CheckedSignature, +} + +#[allow(clippy::large_enum_variant)] +pub(super) enum PreparedVote { + Attestation(PreparedAttestation), + SyncMessage(PreparedSyncMessage), + Ptc(PreparedPtc), +} + +impl PreparedVote { + fn sig_parts(&self) -> (&PublicKey, CheckedSignature, &B256) { + match self { + Self::Attestation(p) => (&p.pubkey, p.signature, &p.signing_root), + Self::SyncMessage(p) => (&p.pubkey, p.signature, &p.signing_root), + Self::Ptc(p) => (&p.pubkey, p.signature, &p.signing_root), + } + } + + fn is_seen(&self, tile: &BeaconStateTile) -> bool { + match self { + Self::Attestation(p) => { + tile.seen_attesters.contains(p.vote.target_epoch, p.vote.validator as usize) + } + Self::SyncMessage(p) => { + tile.seen_sync_msgs[p.subnet as usize].contains(p.slot, p.validator as usize) + } + Self::Ptc(p) => tile.seen_ptc.contains(p.slot, p.validator as usize), + } + } + + fn dedup_key(&self) -> (u8, u64, u64, u64) { + match self { + Self::Attestation(p) => (0, p.vote.validator as u64, p.vote.target_epoch, 0), + Self::SyncMessage(p) => (1, p.validator, p.slot, p.subnet), + Self::Ptc(p) => (2, p.validator, p.slot, 0), + } + } +} + +enum SyncSubcommittee<'a> { + /// Current committee indices are cached in the state. + Current(&'a [u32]), + /// The next committee has no index cache; resolve only the positions that + /// validation needs from its committed pubkeys. + Next(&'a [BLSPubkey]), +} + +impl SyncSubcommittee<'_> { + fn contains(&self, validator: usize, validators: &ValidatorsView<'_>) -> bool { + match self { + Self::Current(indices) => indices.iter().any(|&v| v as usize == validator), + Self::Next(pubkeys) => { + validator < validators.count() && pubkeys.contains(validators.pubkey(validator)) + } + } + } + + fn positions( + &self, + validator: usize, + validators: &ValidatorsView<'_>, + ) -> [u64; SYNC_SUBCOMMITTEE_MASK_WORDS] { + let mut positions = [0u64; SYNC_SUBCOMMITTEE_MASK_WORDS]; + match self { + Self::Current(indices) => { + for (position, &member) in indices.iter().enumerate() { + if member as usize == validator { + positions[position / 64] |= 1 << (position % 64); + } + } + } + Self::Next(pubkeys) if validator < validators.count() => { + let validator_pubkey = validators.pubkey(validator); + for (position, member) in pubkeys.iter().enumerate() { + if member == validator_pubkey { + positions[position / 64] |= 1 << (position % 64); + } + } + } + Self::Next(_) => {} + } + positions + } + + fn validator_at(&self, position: usize, validators: &ValidatorsView<'_>) -> Option { + match self { + Self::Current(indices) => { + let validator = *indices.get(position)? as usize; + (validator < validators.count()).then_some(validator) + } + Self::Next(pubkeys) => validators + .find_by_pubkey(pubkeys.get(position)?) + .map(|validator| validator as usize), + } + } +} + +#[inline] +pub(super) fn uses_next_sync_committee(slot: Slot) -> bool { + let epoch = slot / SLOTS_PER_EPOCH; + let next_slot_epoch = slot.saturating_add(1) / SLOTS_PER_EPOCH; + epoch / EPOCHS_PER_SYNC_COMMITTEE_PERIOD != next_slot_epoch / EPOCHS_PER_SYNC_COMMITTEE_PERIOD +} + +fn sync_subcommittee<'a>(view: &StateReadView<'a>, subcommittee: usize) -> SyncSubcommittee<'a> { + let base = subcommittee * SYNC_SUBCOMMITTEE_SIZE; + let end = base + SYNC_SUBCOMMITTEE_SIZE; + let committees = view.longtail.sync_committees(); + // The spec selects from the state at `state.slot + 1`, not from the + // message slot. They differ during the clock-disparity window. + if uses_next_sync_committee(view.slot.slot_number()) { + SyncSubcommittee::Next(&committees.next().pubkeys[base..end]) + } else { + SyncSubcommittee::Current(&committees.indices()[base..end]) + } +} + pub(super) enum EnvelopeCheck { Ready { block_root: B256, state_id: StateId }, AwaitBlock(B256), @@ -192,48 +338,61 @@ impl BeaconStateTile { self.seen_attesters.mark(p.vote.target_epoch, p.vote.validator as usize); } - pub(super) fn defer_attestation(&mut self, m: NewGossipMsg, producers: &mut Producers) { - self.att_batch.push(m); - if self.att_batch.len() >= ATT_BATCH_CAP { - self.flush_attestations(producers); + pub(super) fn defer_vote(&mut self, m: NewGossipMsg, producers: &mut Producers) { + self.vote_batch.push(m); + if self.vote_batch.len() >= VOTE_BATCH_CAP { + self.flush_votes(producers); } } - pub(super) fn flush_attestations(&mut self, producers: &mut Producers) { - if !self.att_batch.is_empty() { - self.verify_and_commit_attestations(producers); + pub(super) fn flush_votes(&mut self, producers: &mut Producers) { + if !self.vote_batch.is_empty() { + self.verify_and_commit_votes(producers); } } - /// Apply the deferred attestations: sequential cheap validation (in - /// arrival order, so intra-batch duplicate attesters dedup naturally), - /// then one multi-pairing verify for the survivors. A failed batch falls - /// back to per-attestation verification so only the culprits are - /// rejected. + /// Apply the deferred votes (attestations, sync committee messages, PTC + /// attestations): sequential cheap validation in arrival order, then one + /// multi-pairing verify for the survivors. A failed batch falls back to + /// per-message verification so only the culprits are rejected. #[timed] - fn verify_and_commit_attestations(&mut self, producers: &mut Producers) { - debug_assert!(!self.att_batch.is_empty()); - BeaconStateCounters::AttestationBatchSize.set(self.att_batch.len() as u64); + fn verify_and_commit_votes(&mut self, producers: &mut Producers) { + debug_assert!(!self.vote_batch.is_empty()); + BeaconStateCounters::VoteBatchSize.set(self.vote_batch.len() as u64); - self.att_sig_batch.clear(); - debug_assert!(self.att_pending.is_empty()); + self.vote_sig_batch.clear(); + debug_assert!(self.vote_pending.is_empty()); - while !self.att_batch.is_empty() { - let m = self.att_batch.swap_remove(0); - let GossipTopic::BeaconAttestation(subnet) = m.topic else { continue }; + // Drain from the back after one in-place reversal: this preserves + // arrival order without O(n) front-removes or another allocation. + self.vote_batch.reverse(); + while let Some(m) = self.vote_batch.pop() { let acquired = self.gossip_consumer.acquire(m.ssz); let Some(data) = acquired.buffer().ok().map(|(d, _)| d) else { continue }; - match self.prepare_attestation(data, subnet) { + let prepared = match m.topic { + GossipTopic::BeaconAttestation(subnet) => { + self.prepare_attestation(data, subnet).map(PreparedVote::Attestation) + } + GossipTopic::SyncCommittee(subnet) => { + self.prepare_sync_message(data, subnet).map(PreparedVote::SyncMessage) + } + GossipTopic::PayloadAttestationMessage => { + self.prepare_ptc(data).map(PreparedVote::Ptc) + } + _ => continue, + }; + match prepared { Ok(p) => { - let duplicate = self.att_pending.iter().any(|(_, q)| { - q.vote.validator == p.vote.validator && - q.vote.target_epoch == p.vote.target_epoch - }); - if duplicate { - continue; + // Pair only the first candidate for each dedup key, but + // retain later candidates. If that representative makes + // the batch fail, fallback verification can still find a + // later valid candidate for the same key. + let key = p.dedup_key(); + if !self.vote_pending.iter().any(|(_, q)| q.dedup_key() == key) { + let (pk, sig, root) = p.sig_parts(); + self.vote_sig_batch.push_parsed(pk, sig, *root); } - self.att_sig_batch.push_parsed(&p.pubkey, p.signature, p.signing_root); - self.att_pending.push((m, p)); + self.vote_pending.push((m, p)); } Err(Feedback::Reject(_)) => producers.produce(PeerEvent::P2pGossipInvalidMsg { p2p_peer: m.stream_id.peer(), @@ -244,18 +403,32 @@ impl BeaconStateTile { } } - let batch_ok = self.att_sig_batch.verify_all(); - if !batch_ok && !self.att_pending.is_empty() { - BeaconStateCounters::AttestationBatchFallback.inc(); + let batch_ok = self.vote_sig_batch.verify_all(); + if !batch_ok && !self.vote_pending.is_empty() { + BeaconStateCounters::VoteBatchFallback.inc(); } let mut accepted = false; - while !self.att_pending.is_empty() { - let (m, p) = self.att_pending.swap_remove(0); - let valid = - batch_ok || bls::verify_one_checked(&p.pubkey, &p.signature, &p.signing_root); + let mut committed_ptc = false; + self.vote_pending.reverse(); + while let Some((m, p)) = self.vote_pending.pop() { + // Deduplicate only against votes whose signatures have already + // verified and been committed. An invalid earlier arrival with + // the same key must not suppress a later valid vote. + if p.is_seen(self) { + continue; + } + let (pk, sig, root) = p.sig_parts(); + let valid = batch_ok || bls::verify_one_checked(pk, &sig, root); if valid { - self.commit_attestation(&p); + match &p { + PreparedVote::Attestation(p) => self.commit_attestation(p), + PreparedVote::SyncMessage(p) => self.commit_sync_message(p), + PreparedVote::Ptc(p) => { + self.commit_ptc(p); + committed_ptc = true; + } + } Self::relay_gossip(&m, producers); accepted = true; } else { @@ -267,11 +440,199 @@ impl BeaconStateTile { } } + // PTC votes all dirty the same fork-choice structure; fold the whole + // flush in one head recomputation rather than once per message. + if committed_ptc { + self.recompute_head(); + } + if accepted { self.on_accept(None, producers); } } + pub(super) fn prepare_sync_message( + &mut self, + data: &[u8], + subnet: u64, + ) -> Result { + if data.len() != SYNC_COMMITTEE_MSG_SIZE { + return Err(Feedback::Reject(None)); + } + let buf: &[u8; SYNC_COMMITTEE_MSG_SIZE] = + data[..SYNC_COMMITTEE_MSG_SIZE].try_into().unwrap(); + let slot = SyncCommitteeView::slot(buf); + let validator = SyncCommitteeView::validator_index(buf); + + if subnet >= silver_common::SYNC_COMMITTEE_SUBNETS as u64 { + return Err(Feedback::Reject(None)); + } + if !self.ticker.is_current_slot_with_disparity(slot, MAXIMUM_GOSSIP_CLOCK_DISPARITY) { + return Err(Feedback::Ignore); + } + + let seen = &mut self.seen_sync_msgs[subnet as usize]; + seen.rotate_to(self.ticker.latest_slot_with_disparity(MAXIMUM_GOSSIP_CLOCK_DISPARITY)); + if seen.contains(slot, validator as usize) { + return Err(Feedback::Ignore); + } + + let canon_id = self.canonical_state_id(); + let view = self.state.read_view(canon_id); + if validator as usize >= view.validators.count() { + return Err(Feedback::Reject(None)); + } + + let committee = sync_subcommittee(&view, subnet as usize); + let positions = committee.positions(validator as usize, &view.validators); + if positions.iter().all(|&word| word == 0) { + return Err(Feedback::Reject(None)); + } + + let block_root = *SyncCommitteeView::beacon_block_root(buf); + let fork_version = view.epoch.fork_version_at(slot / SLOTS_PER_EPOCH); + let domain = bls::domain_from_fork_data( + bls::DOMAIN_SYNC_COMMITTEE, + &self.fork_data_roots.root(fork_version, &view.imm.genesis_validators_root), + ); + let signing_root = bls::compute_signing_root(&block_root, &domain); + let Some(signature) = CheckedSignature::parse(SyncCommitteeView::signature(buf)) else { + return Err(Feedback::Reject(None)); + }; + + Ok(PreparedSyncMessage { + slot, + subnet, + validator, + block_root, + positions, + pubkey: *view.validators.pubkey_decompressed(validator as usize), + signing_root, + signature, + }) + } + + fn commit_sync_message(&mut self, p: &PreparedSyncMessage) { + let outcome = self.sync_contribution_pool.insert_verified( + p.slot, + p.subnet, + p.block_root, + &p.positions, + p.signature.as_sig(), + ); + debug_assert!(outcome != InsertOutcome::Inconsistent); + if outcome == InsertOutcome::Full { + BeaconStateCounters::SyncContributionPoolFull.inc(); + tracing::debug!( + slot = p.slot, + subcommittee = p.subnet, + block = hex32(&p.block_root), + "sync contribution pool full" + ); + } + self.seen_sync_msgs[p.subnet as usize].mark(p.slot, p.validator as usize); + } + + #[timed] + pub(super) fn handle_sync_contribution(&mut self, data: &[u8]) -> Feedback { + if data.len() != SIGNED_CONTRIBUTION_AND_PROOF_SIZE { + return Feedback::Reject(None); + } + let buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE] = + data[..SIGNED_CONTRIBUTION_AND_PROOF_SIZE].try_into().unwrap(); + let slot = ContributionView::slot(buf); + let subcommittee = ContributionView::subcommittee_index(buf); + let aggregator = ContributionView::aggregator_index(buf); + let block_root = *ContributionView::beacon_block_root(buf); + let bits = ContributionView::aggregation_bits(buf); + + if subcommittee >= silver_common::SYNC_COMMITTEE_SUBNETS as u64 { + return Feedback::Reject(None); + } + if !self.ticker.is_current_slot_with_disparity(slot, MAXIMUM_GOSSIP_CLOCK_DISPARITY) { + return Feedback::Ignore; + } + + let seen = &mut self.seen_contribution_aggregators[subcommittee as usize]; + seen.rotate_to(self.ticker.latest_slot_with_disparity(MAXIMUM_GOSSIP_CLOCK_DISPARITY)); + if seen.contains(slot, aggregator as usize) { + return Feedback::Ignore; + } + let coverage = self.seen_aggregates.coverage(slot, subcommittee, block_root, bits); + if coverage == Coverage::BySuperset { + return Feedback::Ignore; + } + + if !is_sync_aggregator(ContributionView::selection_proof(buf)) { + return Feedback::Reject(None); + } + + let canon_id = self.canonical_state_id(); + let view = self.state.read_view(canon_id); + let count = view.validators.count(); + if aggregator as usize >= count { + return Feedback::Ignore; + } + let committee = sync_subcommittee(&view, subcommittee as usize); + if !committee.contains(aggregator as usize, &view.validators) { + return Feedback::Reject(None); + } + + let fv = view.epoch.fork_version_at(slot / SLOTS_PER_EPOCH); + let fork_data_root = + ssz_hash::hash_tree_root_fork_data(fv, &view.imm.genesis_validators_root); + let domain = |ty| bls::domain_from_fork_data(ty, &fork_data_root); + + let sr_sp = bls::compute_signing_root( + &ssz_hash::hash_tree_root_sync_selection_data(slot, subcommittee), + &domain(bls::DOMAIN_SYNC_COMMITTEE_SELECTION_PROOF), + ); + let contribution_root = ssz_hash::hash_tree_root_sync_contribution( + slot, + &block_root, + subcommittee, + bits, + ContributionView::contribution_signature(buf), + ); + let cap_root = ssz_hash::hash_tree_root_contribution_and_proof( + aggregator, + &contribution_root, + ContributionView::selection_proof(buf), + ); + let sr_outer = + bls::compute_signing_root(&cap_root, &domain(bls::DOMAIN_CONTRIBUTION_AND_PROOF)); + let sr_agg = bls::compute_signing_root(&block_root, &domain(bls::DOMAIN_SYNC_COMMITTEE)); + + let mut participants = 0usize; + let mut unknown = false; + self.sig_batch.clear(); + let aggregator_pk = view.validators.pubkey_decompressed(aggregator as usize); + self.sig_batch.push_one(aggregator_pk, ContributionView::selection_proof(buf), sr_sp); + self.sig_batch.push_one(aggregator_pk, ContributionView::signature(buf), sr_outer); + self.sig_batch.push_aggregate( + (0..SYNC_SUBCOMMITTEE_SIZE).filter_map(|i| { + if bits[i / 8] & (1 << (i % 8)) == 0 { + return None; + } + participants += 1; + let Some(vi) = committee.validator_at(i, &view.validators) else { + unknown = true; + return None; + }; + Some(view.validators.pubkey_decompressed(vi)) + }), + ContributionView::contribution_signature(buf), + sr_agg, + ); + if unknown || participants == 0 || !self.sig_batch.verify_all() { + return Feedback::Reject(None); + } + + self.seen_aggregates.record(slot, subcommittee, block_root, bits); + self.seen_contribution_aggregators[subcommittee as usize].mark(slot, aggregator as usize); + Feedback::Accept(None) + } + /// EF `fork_choice` vector path only: production gossip reaches the same /// work through `handle_attestation` / `handle_aggregate_and_proof`, which /// have already resolved the committee by the time votes are recorded. @@ -846,7 +1207,7 @@ impl BeaconStateTile { BlockSource::Gossip, producers, ), - GossipTopic::PayloadAttestationMessage => self.handle_payload_attestation(data), + GossipTopic::SyncCommitteeContributionAndProof => self.handle_sync_contribution(data), _ => return, }; match feedback { @@ -901,6 +1262,13 @@ pub(super) fn is_aggregator(committee_len: usize, selection_proof: &[u8; 96]) -> u64::from_le_bytes(h[0..8].try_into().unwrap()) % modulo == 0 } +pub(super) fn is_sync_aggregator(selection_proof: &[u8; 96]) -> bool { + const TARGET_AGGREGATORS_PER_SYNC_SUBCOMMITTEE: u64 = 16; + let modulo = (SYNC_SUBCOMMITTEE_SIZE as u64 / TARGET_AGGREGATORS_PER_SYNC_SUBCOMMITTEE).max(1); + let h = merkle::sha256(selection_proof); + u64::from_le_bytes(h[0..8].try_into().unwrap()) % modulo == 0 +} + pub(super) fn compute_subnet_for_attestation( committees_per_slot: usize, slot: Slot, diff --git a/crates/beacon_state/tile/src/tile/sync_contribution_pool.rs b/crates/beacon_state/tile/src/tile/sync_contribution_pool.rs new file mode 100644 index 00000000..be8fe4f7 --- /dev/null +++ b/crates/beacon_state/tile/src/tile/sync_contribution_pool.rs @@ -0,0 +1,279 @@ +use blst::min_pk::{AggregateSignature, Signature}; +use rustc_hash::FxHashMap; +use silver_beacon_state_data::{B256, SYNC_COMMITTEE_SIZE, Slot}; +use silver_common::{ + SYNC_COMMITTEE_SUBNETS, metrics::timed, ssz_view::SYNC_COMMITTEE_CONTRIBUTION_SIZE, +}; + +use super::attestation_pool::InsertOutcome; + +const SYNC_SUBCOMMITTEE_SIZE: usize = SYNC_COMMITTEE_SIZE / SYNC_COMMITTEE_SUBNETS; +pub(super) const SYNC_SUBCOMMITTEE_MASK_WORDS: usize = SYNC_SUBCOMMITTEE_SIZE.div_ceil(64); +const AGGREGATION_BITS_BYTES: usize = SYNC_SUBCOMMITTEE_SIZE.div_ceil(8); + +/// Retention is two slots (current + previous), with four subcommittees +/// per slot. The x4 leaves room for competing beacon-block roots; each +/// additional entry requires a valid sync-committee member signature. +const MAX_ENTRIES: usize = 4 * 2 * SYNC_COMMITTEE_SUBNETS; + +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +struct ContributionKey { + slot: Slot, + subcommittee_index: u64, + beacon_block_root: B256, +} + +struct ContributionEntry { + aggregation_bits: [u64; SYNC_SUBCOMMITTEE_MASK_WORDS], + signature: AggregateSignature, +} + +pub(super) struct SyncContributionPool { + entries: FxHashMap, + floor: Slot, +} + +impl SyncContributionPool { + pub(super) fn new() -> Self { + Self { + entries: FxHashMap::with_capacity_and_hasher(MAX_ENTRIES, Default::default()), + floor: 0, + } + } + + /// Adds an already-verified `SyncCommitteeMessage`. `positions` contains + /// every occurrence of its validator in this subcommittee. The same + /// signature is deliberately aggregated once per new position, as + /// required when a validator occurs more than once in a sync committee. + #[timed] + pub(super) fn insert_verified( + &mut self, + slot: Slot, + subcommittee_index: u64, + beacon_block_root: B256, + positions: &[u64; SYNC_SUBCOMMITTEE_MASK_WORDS], + signature: &Signature, + ) -> InsertOutcome { + if subcommittee_index >= SYNC_COMMITTEE_SUBNETS as u64 || + positions.iter().all(|&word| word == 0) + { + return InsertOutcome::Inconsistent; + } + if slot < self.floor { + return InsertOutcome::Stale; + } + + let key = ContributionKey { slot, subcommittee_index, beacon_block_root }; + if let Some(entry) = self.entries.get_mut(&key) { + return entry.add(positions, signature); + } + if self.entries.len() >= MAX_ENTRIES { + return InsertOutcome::Full; + } + + self.entries.insert(key, ContributionEntry::new(*positions, signature)); + InsertOutcome::Inserted + } + + /// Materializes the unsigned `SyncCommitteeContribution` that a selected + /// validator will wrap in `ContributionAndProof` and sign. The surrounding + /// selection proof and validator signatures intentionally remain outside + /// this pool. + #[allow(dead_code)] // retrieval interface for the local-validator milestone + #[timed] + pub(super) fn contribution_ssz( + &self, + slot: Slot, + subcommittee_index: u64, + beacon_block_root: B256, + ) -> Option<[u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE]> { + let entry = + self.entries.get(&ContributionKey { slot, subcommittee_index, beacon_block_root })?; + let mut out = [0u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE]; + out[0..8].copy_from_slice(&slot.to_le_bytes()); + out[8..40].copy_from_slice(&beacon_block_root); + out[40..48].copy_from_slice(&subcommittee_index.to_le_bytes()); + for (i, word) in entry.aggregation_bits.iter().enumerate() { + let start = 48 + i * 8; + out[start..start + 8].copy_from_slice(&word.to_le_bytes()); + } + let signature_offset = 48 + AGGREGATION_BITS_BYTES; + out[signature_offset..].copy_from_slice(&entry.signature.to_signature().to_bytes()); + Some(out) + } + + #[timed] + pub(super) fn prune_before(&mut self, floor: Slot) { + self.floor = floor; + self.entries.retain(|key, _| key.slot >= floor); + } +} + +impl ContributionEntry { + fn new(aggregation_bits: [u64; SYNC_SUBCOMMITTEE_MASK_WORDS], signature: &Signature) -> Self { + let copies = aggregation_bits.iter().map(|word| word.count_ones()).sum::(); + debug_assert!(copies > 0); + let mut aggregate = AggregateSignature::from_signature(signature); + for _ in 1..copies { + aggregate.add_signature(signature, false).expect("infallible without groupcheck"); + } + Self { aggregation_bits, signature: aggregate } + } + + #[timed] + fn add( + &mut self, + positions: &[u64; SYNC_SUBCOMMITTEE_MASK_WORDS], + signature: &Signature, + ) -> InsertOutcome { + let mut new_positions = [0u64; SYNC_SUBCOMMITTEE_MASK_WORDS]; + let mut copies = 0u32; + for ((new, &incoming), &existing) in + new_positions.iter_mut().zip(positions).zip(&self.aggregation_bits) + { + *new = incoming & !existing; + copies += new.count_ones(); + } + if copies == 0 { + return InsertOutcome::Duplicate; + } + + // The caller only supplies a subgroup-checked, successfully verified + // signature. BLS addition is not idempotent, so only positions not + // already represented above may add another signature copy. + for _ in 0..copies { + self.signature.add_signature(signature, false).expect("infallible without groupcheck"); + } + for (bits, new) in self.aggregation_bits.iter_mut().zip(new_positions) { + *bits |= new; + } + InsertOutcome::Inserted + } +} + +#[cfg(test)] +mod tests { + use blst::BLST_ERROR; + use silver_common::ssz_view::SyncCommitteeContributionView; + + use super::*; + use crate::{bls, test_signing}; + + const SLOT: Slot = 3; + const SUBCOMMITTEE: u64 = 1; + const BLOCK_ROOT: B256 = [0xAB; 32]; + const SIGNING_ROOT: B256 = [0xCD; 32]; + + fn signature(sk_idx: usize) -> Signature { + Signature::from_bytes(&test_signing::sign(sk_idx, &SIGNING_ROOT)).unwrap() + } + + fn positions(indices: &[usize]) -> [u64; SYNC_SUBCOMMITTEE_MASK_WORDS] { + let mut mask = [0u64; SYNC_SUBCOMMITTEE_MASK_WORDS]; + for &position in indices { + mask[position / 64] |= 1 << (position % 64); + } + mask + } + + #[test] + fn messages_aggregate_to_bits_and_verifying_signature() { + let mut pool = SyncContributionPool::new(); + assert_eq!( + pool.insert_verified(SLOT, SUBCOMMITTEE, BLOCK_ROOT, &positions(&[1]), &signature(0)), + InsertOutcome::Inserted + ); + assert_eq!( + pool.insert_verified(SLOT, SUBCOMMITTEE, BLOCK_ROOT, &positions(&[65]), &signature(1)), + InsertOutcome::Inserted + ); + + let out = pool.contribution_ssz(SLOT, SUBCOMMITTEE, BLOCK_ROOT).unwrap(); + assert_eq!(SyncCommitteeContributionView::slot(&out), SLOT); + assert_eq!(SyncCommitteeContributionView::beacon_block_root(&out), &BLOCK_ROOT); + assert_eq!(SyncCommitteeContributionView::subcommittee_index(&out), SUBCOMMITTEE); + let bits = SyncCommitteeContributionView::aggregation_bits(&out); + assert_eq!(bits[0], 0b0000_0010); + assert_eq!(bits[8], 0b0000_0010); + + let sig = Signature::from_bytes(SyncCommitteeContributionView::signature(&out)).unwrap(); + let pks = [&test_signing::pubkey_pk(0), &test_signing::pubkey_pk(1)]; + assert_eq!( + sig.fast_aggregate_verify(true, &SIGNING_ROOT, bls::DST, &pks), + BLST_ERROR::BLST_SUCCESS + ); + } + + #[test] + fn repeated_validator_positions_repeat_its_signature() { + let mut pool = SyncContributionPool::new(); + assert_eq!( + pool.insert_verified( + SLOT, + SUBCOMMITTEE, + BLOCK_ROOT, + &positions(&[2, 70]), + &signature(0), + ), + InsertOutcome::Inserted + ); + + let out = pool.contribution_ssz(SLOT, SUBCOMMITTEE, BLOCK_ROOT).unwrap(); + let sig = Signature::from_bytes(SyncCommitteeContributionView::signature(&out)).unwrap(); + let pk = test_signing::pubkey_pk(0); + assert_eq!( + sig.fast_aggregate_verify(true, &SIGNING_ROOT, bls::DST, &[&pk, &pk]), + BLST_ERROR::BLST_SUCCESS + ); + assert_ne!( + sig.fast_aggregate_verify(true, &SIGNING_ROOT, bls::DST, &[&pk]), + BLST_ERROR::BLST_SUCCESS + ); + } + + #[test] + fn duplicate_positions_do_not_change_signature() { + let mut pool = SyncContributionPool::new(); + let mask = positions(&[4, 68]); + assert_eq!( + pool.insert_verified(SLOT, SUBCOMMITTEE, BLOCK_ROOT, &mask, &signature(0)), + InsertOutcome::Inserted + ); + let before = pool.contribution_ssz(SLOT, SUBCOMMITTEE, BLOCK_ROOT).unwrap(); + + assert_eq!( + pool.insert_verified(SLOT, SUBCOMMITTEE, BLOCK_ROOT, &mask, &signature(0)), + InsertOutcome::Duplicate + ); + assert_eq!(pool.contribution_ssz(SLOT, SUBCOMMITTEE, BLOCK_ROOT).unwrap(), before); + } + + #[test] + fn roots_and_subcommittees_are_separate_and_old_slots_are_pruned() { + let mut pool = SyncContributionPool::new(); + let other_root = [0xBC; 32]; + let mask = positions(&[0]); + assert_eq!( + pool.insert_verified(SLOT, SUBCOMMITTEE, BLOCK_ROOT, &mask, &signature(0)), + InsertOutcome::Inserted + ); + assert_eq!( + pool.insert_verified(SLOT + 1, SUBCOMMITTEE, other_root, &mask, &signature(0)), + InsertOutcome::Inserted + ); + assert_eq!( + pool.insert_verified(SLOT + 1, SUBCOMMITTEE + 1, BLOCK_ROOT, &mask, &signature(0)), + InsertOutcome::Inserted + ); + + pool.prune_before(SLOT + 1); + + assert_eq!(pool.contribution_ssz(SLOT, SUBCOMMITTEE, BLOCK_ROOT), None); + assert!(pool.contribution_ssz(SLOT + 1, SUBCOMMITTEE, other_root).is_some()); + assert!(pool.contribution_ssz(SLOT + 1, SUBCOMMITTEE + 1, BLOCK_ROOT).is_some()); + assert_eq!( + pool.insert_verified(SLOT, SUBCOMMITTEE, BLOCK_ROOT, &mask, &signature(0)), + InsertOutcome::Stale + ); + } +} diff --git a/crates/beacon_state/tile/src/tile/tests.rs b/crates/beacon_state/tile/src/tile/tests.rs index faae7729..fca16c4d 100644 --- a/crates/beacon_state/tile/src/tile/tests.rs +++ b/crates/beacon_state/tile/src/tile/tests.rs @@ -2,9 +2,10 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use flux::timing::Nanos; use silver_beacon_state_data::{ - BLSPubkey, BeaconBlockHeader, BeaconState, BlockRootsId, ColumnGroup, ColumnSpec, EpochState, - EpochStateFinalized, Eth1Data, HistoricalSummary, Id, Immutable, PROPOSER_LOOKAHEAD_SIZE, - PendingDeposit, SLOTS_PER_HISTORICAL_ROOT, SlotStateId, StateReadView, ValSeed, Withdrawals, + BLSPubkey, BeaconBlockHeader, BeaconState, BlockRootsId, ColumnGroup, ColumnSpec, + EPOCHS_PER_SYNC_COMMITTEE_PERIOD, EpochState, EpochStateFinalized, Eth1Data, HistoricalSummary, + Id, Immutable, PROPOSER_LOOKAHEAD_SIZE, PendingDeposit, SLOTS_PER_HISTORICAL_ROOT, SlotStateId, + StateReadView, ValSeed, Withdrawals, }; use silver_common::{ GossipTopic, MessageId, P2pStreamId, StreamProtocol, TCache, TCacheProducer, TProducer, @@ -12,6 +13,7 @@ use silver_common::{ ATTESTATION_DATA_SIZE, AttestationView, PROPOSER_SLASHING_SIZE, SIGNED_AGG_PROOF_MIN, SIGNED_BLS_CHANGE_SIZE, SIGNED_EXECUTION_PAYLOAD_ENVELOPE_MIN, SIGNED_VOLUNTARY_EXIT_SIZE, SINGLE_ATT_SIZE, SignedAggregateAndProofView, SingleAttestationView, StatusView, + SyncCommitteeContributionView, }, }; use silver_ssz::ssz_view::EXECUTION_PAYLOAD_ENVELOPE_MIN; @@ -936,6 +938,22 @@ fn attestation_updates_vote_tracker() { assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_epoch, want_epoch); } +fn gossip_msg(producer: &mut TProducer, bytes: &[u8], topic: GossipTopic) -> NewGossipMsg { + let mut r = producer.reserve(bytes.len(), true).expect("reserve"); + r.buffer().unwrap()[..bytes.len()].copy_from_slice(bytes); + r.increment_offset(bytes.len()); + let read = r.read(); + producer.publish_head(); + NewGossipMsg { + stream_id: P2pStreamId::new(0, 0, StreamProtocol::Unset, false), + topic, + msg_hash: MessageId { id: [0u8; 20] }, + recv_ts: Nanos(0), + ssz: read, + protobuf: read, + } +} + fn gossip_att_msg( producer: &mut TProducer, att: &[u8; SINGLE_ATT_SIZE], @@ -989,15 +1007,15 @@ fn attestation_batch_flush_applies_all() { for vi in [0u32, 1] { let (buf, subnet) = batched_att(&tile, vi as usize, vi); let m = gossip_att_msg(&mut gp, &buf, subnet); - tile.defer_attestation(m, &mut adapter.producers); + tile.defer_vote(m, &mut adapter.producers); } assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, [0u8; 32], "vote before flush"); - tile.flush_attestations(&mut adapter.producers); + tile.flush_votes(&mut adapter.producers); assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, bbr); assert_eq!(tile.fork_choice.vote_tracker.votes[1].latest_root, bbr); - assert!(tile.att_batch.is_empty()); - assert!(tile.att_pending.is_empty()); + assert!(tile.vote_batch.is_empty()); + assert!(tile.vote_pending.is_empty()); } /// A forged signature (valid G2 point, wrong key) fails the batch verify; @@ -1011,15 +1029,33 @@ fn attestation_batch_fallback_rejects_only_forged() { let (good, good_subnet) = batched_att(&tile, 0, 0); let (forged, forged_subnet) = batched_att(&tile, 2, 1); let m = gossip_att_msg(&mut gp, &good, good_subnet); - tile.defer_attestation(m, &mut adapter.producers); + tile.defer_vote(m, &mut adapter.producers); let m = gossip_att_msg(&mut gp, &forged, forged_subnet); - tile.defer_attestation(m, &mut adapter.producers); + tile.defer_vote(m, &mut adapter.producers); - tile.flush_attestations(&mut adapter.producers); + tile.flush_votes(&mut adapter.producers); assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, bbr); assert_eq!(tile.fork_choice.vote_tracker.votes[1].latest_root, [0u8; 32], "forged vote"); } +#[test] +fn invalid_vote_does_not_deduplicate_later_valid_vote() { + let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(31); + seed_tile_with_keys(&mut tile, 128, 0); + let bbr = tile.last_applied_block_root; + + // Both messages have the same gossip dedup key. The first is a valid BLS + // point signed by the wrong key; only the second may establish "seen". + let (forged, subnet) = batched_att(&tile, 1, 0); + let (valid, valid_subnet) = batched_att(&tile, 0, 0); + assert_eq!(subnet, valid_subnet); + tile.defer_vote(gossip_att_msg(&mut gp, &forged, subnet), &mut adapter.producers); + tile.defer_vote(gossip_att_msg(&mut gp, &valid, subnet), &mut adapter.producers); + + tile.flush_votes(&mut adapter.producers); + assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, bbr); +} + /// A non-attestation gossip message flushes the pending batch first, so /// queue order is preserved. #[test] @@ -1030,7 +1066,7 @@ fn attestation_batch_flushed_before_other_gossip() { let (buf, subnet) = batched_att(&tile, 0, 0); let m = gossip_att_msg(&mut gp, &buf, subnet); - tile.defer_attestation(m, &mut adapter.producers); + tile.defer_vote(m, &mut adapter.producers); assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, [0u8; 32]); let mut exit = gossip_att_msg(&mut gp, &[0u8; SINGLE_ATT_SIZE], 0); @@ -1039,6 +1075,289 @@ fn attestation_batch_flushed_before_other_gossip() { assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, bbr); } +#[test] +fn sync_message_batch_applies_and_marks_seen() { + let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(31); + seed_tile_with_keys(&mut tile, 128, 0); + let imm = seed_immutable(&tile); + let bbr = tile.last_applied_block_root; + let wall = tile.ticker.current_slot(); + + let msg = test_signing::sign_sync_committee_message(0, 0, wall, bbr, &imm); + let m = gossip_msg(&mut gp, &msg, GossipTopic::SyncCommittee(1)); + tile.defer_vote(m, &mut adapter.producers); + assert!(!tile.seen_sync_msgs[1].contains(wall, 0), "not applied before flush"); + assert_eq!(tile.sync_contribution_pool.contribution_ssz(wall, 1, bbr), None); + + tile.flush_votes(&mut adapter.producers); + assert!(tile.seen_sync_msgs[1].contains(wall, 0)); + let contribution = + tile.sync_contribution_pool.contribution_ssz(wall, 1, bbr).expect("pooled contribution"); + // The test state's default current committee repeats validator 0 in all + // 128 positions of each subcommittee. + assert_eq!(SyncCommitteeContributionView::aggregation_bits(&contribution), &[0xff; 16]); + assert!(tile.vote_batch.is_empty() && tile.vote_pending.is_empty()); +} + +#[test] +fn sync_message_uses_gossip_clock_disparity() { + let wall = 31; + let mut tile = make_tile_at_wall_slot(wall); + seed_tile_with_keys(&mut tile, 128, 0); + let imm = seed_immutable(&tile); + let bbr = tile.last_applied_block_root; + + // During the final 500 ms, the next slot is already admissible. + tile.ticker.set_since_genesis_ms(wall * 12_000 + 11_750); + let next = test_signing::sign_sync_committee_message(0, 0, wall + 1, bbr, &imm); + assert!(tile.prepare_sync_message(&next, 0).is_ok()); + + // Outside that window it is still from the future. + tile.ticker.set_since_genesis_ms(wall * 12_000 + 10_000); + assert!(matches!(tile.prepare_sync_message(&next, 0), Err(Feedback::Ignore))); +} + +#[test] +fn sync_message_uses_next_committee_at_period_handoff() { + let period_slots = EPOCHS_PER_SYNC_COMMITTEE_PERIOD * SLOTS_PER_EPOCH; + let handoff_slot = period_slots - 1; + assert!(!super::gossip::uses_next_sync_committee(handoff_slot - 1)); + assert!(super::gossip::uses_next_sync_committee(handoff_slot)); + + // The test state's cached current committee contains validator 0, while + // its default next-committee pubkeys do not. The same current member is + // therefore accepted one slot before handoff and rejected at handoff. + let imm = Immutable::default(); + let mut current = make_tile_at_wall_slot(handoff_slot - 1); + seed_tile_with_keys(&mut current, 128, handoff_slot - 1); + let msg = test_signing::sign_sync_committee_message( + 0, + 0, + handoff_slot - 1, + current.last_applied_block_root, + &imm, + ); + assert!(current.prepare_sync_message(&msg, 0).is_ok()); + + let mut handoff = make_tile_at_wall_slot(handoff_slot); + seed_tile_with_keys(&mut handoff, 128, handoff_slot); + let msg = test_signing::sign_sync_committee_message( + 0, + 0, + handoff_slot, + handoff.last_applied_block_root, + &imm, + ); + assert!(matches!(handoff.prepare_sync_message(&msg, 0), Err(Feedback::Reject(None)))); +} + +#[test] +fn sync_message_from_non_member_is_rejected() { + let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(31); + seed_tile_with_keys(&mut tile, 128, 0); + let imm = seed_immutable(&tile); + let wall = tile.ticker.current_slot(); + + let msg = + test_signing::sign_sync_committee_message(2, 5, wall, tile.last_applied_block_root, &imm); + let m = gossip_msg(&mut gp, &msg, GossipTopic::SyncCommittee(0)); + tile.defer_vote(m, &mut adapter.producers); + tile.flush_votes(&mut adapter.producers); + assert!(!tile.seen_sync_msgs[0].contains(wall, 5)); +} + +#[test] +fn sync_message_forged_signature_rejected_by_fallback() { + let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(31); + seed_tile_with_keys(&mut tile, 128, 0); + let imm = seed_immutable(&tile); + let bbr = tile.last_applied_block_root; + let wall = tile.ticker.current_slot(); + + let good = test_signing::sign_sync_committee_message(0, 0, wall, bbr, &imm); + let forged = test_signing::sign_sync_committee_message(1, 0, wall, bbr, &imm); + let m = gossip_msg(&mut gp, &good, GossipTopic::SyncCommittee(0)); + tile.defer_vote(m, &mut adapter.producers); + let m = gossip_msg(&mut gp, &forged, GossipTopic::SyncCommittee(2)); + tile.defer_vote(m, &mut adapter.producers); + + tile.flush_votes(&mut adapter.producers); + assert!(tile.seen_sync_msgs[0].contains(wall, 0), "honest message applied"); + assert!(!tile.seen_sync_msgs[2].contains(wall, 0), "forgery rejected"); + assert!(tile.sync_contribution_pool.contribution_ssz(wall, 0, bbr).is_some()); + assert_eq!(tile.sync_contribution_pool.contribution_ssz(wall, 2, bbr), None); +} + +#[test] +fn mixed_vote_batch_applies_all_kinds() { + let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(31); + seed_tile_with_keys(&mut tile, 128, 0); + let imm = seed_immutable(&tile); + let bbr = tile.last_applied_block_root; + let wall = tile.ticker.current_slot(); + + let (att, subnet) = batched_att(&tile, 0, 0); + let m = gossip_att_msg(&mut gp, &att, subnet); + tile.defer_vote(m, &mut adapter.producers); + let msg = test_signing::sign_sync_committee_message(0, 0, wall, bbr, &imm); + let m = gossip_msg(&mut gp, &msg, GossipTopic::SyncCommittee(3)); + tile.defer_vote(m, &mut adapter.producers); + + tile.flush_votes(&mut adapter.producers); + assert_eq!(tile.fork_choice.vote_tracker.votes[0].latest_root, bbr); + assert!(tile.seen_sync_msgs[3].contains(wall, 0)); +} + +fn sync_aggregator_slot(imm: &Immutable, qualify: bool) -> (u64, u64) { + for slot in 31..4096 { + for sub in 0..4u64 { + let proof = test_signing::sync_selection_proof(0, slot, sub, imm); + if super::gossip::is_sync_aggregator(&proof) == qualify { + return (slot, sub); + } + } + } + panic!("no qualifying (slot, subcommittee) found"); +} + +fn sync_aggregator_slot_for_two_subcommittees(imm: &Immutable) -> (u64, u64, u64) { + for slot in 31..4096 { + let mut first = None; + for sub in 0..4u64 { + let proof = test_signing::sync_selection_proof(0, slot, sub, imm); + if super::gossip::is_sync_aggregator(&proof) { + if let Some(other) = first { + return (slot, other, sub); + } + first = Some(sub); + } + } + } + panic!("no slot qualifying for two subcommittees found"); +} + +#[test] +fn sync_contribution_accepted_then_superset_ignored() { + let imm = Immutable::default(); + let (slot, sub) = sync_aggregator_slot(&imm, true); + let mut tile = make_tile_at_wall_slot(slot); + seed_tile_with_keys(&mut tile, 128, slot); + let bbr = tile.last_applied_block_root; + + let buf = test_signing::sign_contribution_and_proof(0, 0, slot, sub, 3, 0, bbr, &imm); + assert!(matches!(tile.handle_sync_contribution(&buf), Feedback::Accept(None))); + assert!(matches!(tile.handle_sync_contribution(&buf), Feedback::Ignore)); +} + +#[test] +fn sync_contribution_dedup_is_per_subcommittee() { + let imm = Immutable::default(); + let (slot, first_sub, second_sub) = sync_aggregator_slot_for_two_subcommittees(&imm); + let mut tile = make_tile_at_wall_slot(slot); + seed_tile_with_keys(&mut tile, 128, slot); + let bbr = tile.last_applied_block_root; + + let first = test_signing::sign_contribution_and_proof(0, 0, slot, first_sub, 3, 0, bbr, &imm); + let second = test_signing::sign_contribution_and_proof(0, 0, slot, second_sub, 3, 0, bbr, &imm); + assert!(matches!(tile.handle_sync_contribution(&first), Feedback::Accept(None))); + assert!(matches!(tile.handle_sync_contribution(&second), Feedback::Accept(None))); +} + +#[test] +fn sync_contribution_non_aggregator_rejected() { + let imm = Immutable::default(); + let (slot, sub) = sync_aggregator_slot(&imm, false); + let mut tile = make_tile_at_wall_slot(slot); + seed_tile_with_keys(&mut tile, 128, slot); + let bbr = tile.last_applied_block_root; + + let buf = test_signing::sign_contribution_and_proof(0, 0, slot, sub, 3, 0, bbr, &imm); + assert!(matches!(tile.handle_sync_contribution(&buf), Feedback::Reject(None))); +} + +#[test] +fn sync_contribution_forged_outer_signature_rejected() { + let imm = Immutable::default(); + let (slot, sub) = sync_aggregator_slot(&imm, true); + let mut tile = make_tile_at_wall_slot(slot); + seed_tile_with_keys(&mut tile, 128, slot); + let bbr = tile.last_applied_block_root; + + let mut buf = test_signing::sign_contribution_and_proof(0, 0, slot, sub, 3, 0, bbr, &imm); + buf[300] ^= 0x01; + assert!(matches!(tile.handle_sync_contribution(&buf), Feedback::Reject(None))); +} + +#[test] +fn ptc_rejects_non_canonical_bool_bytes() { + let slot = 31; + let mut tile = make_tile_at_wall_slot(slot); + seed_tile_with_keys(&mut tile, 128, slot); + let msg = test_signing::sign_payload_attestation_message( + 0, + 0, + slot, + tile.last_applied_block_root, + 2, + 1, + &seed_immutable(&tile), + ); + + assert!(matches!(tile.prepare_ptc(&msg), Err(Feedback::Reject(None)))); +} + +#[test] +fn ptc_requires_referenced_block_at_message_slot() { + let message_slot = 31; + let mut tile = make_tile_at_wall_slot(message_slot); + seed_tile_with_keys(&mut tile, 128, message_slot - 1); + let msg = test_signing::sign_payload_attestation_message( + 0, + 0, + message_slot, + tile.last_applied_block_root, + 1, + 1, + &seed_immutable(&tile), + ); + + assert!(matches!(tile.prepare_ptc(&msg), Err(Feedback::Ignore))); +} + +#[test] +fn ptc_vote_records_every_matching_committee_position() { + let slot = 31; + let (mut tile, mut gp, _rp, _spine, mut adapter) = tile_with_producers(slot); + seed_tile_with_keys(&mut tile, 128, slot); + let root = tile.last_applied_block_root; + let msg = test_signing::sign_payload_attestation_message( + 0, + 0, + slot, + root, + 1, + 1, + &seed_immutable(&tile), + ); + + let prepared = tile.prepare_ptc(&msg).expect("valid PTC message"); + assert_eq!( + prepared.ptc_positions.iter().map(|word| word.count_ones()).sum::(), + 512, + "the test PTC repeats validator 0 in every position", + ); + drop(prepared); + + let gossip = gossip_msg(&mut gp, &msg, GossipTopic::PayloadAttestationMessage); + tile.defer_vote(gossip, &mut adapter.producers); + tile.flush_votes(&mut adapter.producers); + assert!(tile.seen_ptc.contains(slot, 0)); + assert!(tile.fork_choice.ptc_timeliness_votes(&root).iter().all(|vote| *vote == Some(true))); + assert!( + tile.fork_choice.ptc_data_availability_votes(&root).iter().all(|vote| *vote == Some(true)) + ); +} + /// Spec `validate_on_attestation`: a single attestation for a block we /// don't hold is dropped (Ignore), self-healing on the validator's next /// vote. diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index f44b5cf4..0a18c43e 100644 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -227,6 +227,10 @@ fn main() -> Result<(), Box> { control_tile.set_pending_subnet_topics( silver_common::attnet_subnets(subnets) .map(silver_common::GossipTopic::BeaconAttestation) + .chain( + (0..silver_common::SYNC_COMMITTEE_SUBNETS as u64) + .map(silver_common::GossipTopic::SyncCommittee), + ) .collect(), ); diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index cc4e5e82..8694d93c 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -6,7 +6,8 @@ pub use crate::{ ATTESTATION_SUBNETS, GOSSIP_TOPIC_COUNTER_SLOTS, GossipTopic, MAX_GOSSIP_COMPRESSED_PAYLOAD_SIZE, MAX_GOSSIP_FRAME_SIZE, MAX_GOSSIP_UNCOMPRESSED_PAYLOAD_SIZE, MESSAGE_ID_LEN, MessageId, MessageIdHasher, - gossip_topic_for_counter_slot, msg_id_invalid_snappy, msg_id_valid_snappy, + SYNC_COMMITTEE_SUBNETS, gossip_topic_for_counter_slot, msg_id_invalid_snappy, + msg_id_valid_snappy, }, id::{Keypair, PeerId, decode_protobuf_pubkey, encode_secp256k1_protobuf}, identity::{ diff --git a/crates/common/src/ticker.rs b/crates/common/src/ticker.rs index 0b7cbfb7..080fce29 100644 --- a/crates/common/src/ticker.rs +++ b/crates/common/src/ticker.rs @@ -109,6 +109,26 @@ impl SlotTicker { self.millis_since_genesis() / self.slot_ms } + /// Whether `slot` is current after extending both ends of its wall-clock + /// interval by `disparity`. Consensus gossip validation uses this for its + /// permitted clock skew around slot boundaries. + pub fn is_current_slot_with_disparity(&self, slot: Slot, disparity: Duration) -> bool { + let now = self.millis_since_genesis(); + let disparity_ms = u64::try_from(disparity.as_millis()).unwrap_or(u64::MAX); + let start = slot.saturating_mul(self.slot_ms); + let end = slot.saturating_add(1).saturating_mul(self.slot_ms); + now.saturating_add(disparity_ms) >= start && now <= end.saturating_add(disparity_ms) + } + + /// The newest slot admitted by [`Self::is_current_slot_with_disparity`]. + /// A two-slot seen-cache can rotate to this value and retain every slot + /// that is valid at the current instant, including across the boundary. + pub fn latest_slot_with_disparity(&self, disparity: Duration) -> Slot { + let wall = self.current_slot(); + let next = wall.saturating_add(1); + if self.is_current_slot_with_disparity(next, disparity) { next } else { wall } + } + pub fn is_before_attesting_interval(&self, is_gloas: bool) -> bool { let fraction = if is_gloas { 4 } else { 3 }; self.millis_since_genesis() % self.slot_ms < self.slot_ms / fraction @@ -222,4 +242,25 @@ mod tests { let ev = t.tick(); assert!(matches!(ev, TickEvent::SlotStart(3))); } + + #[test] + fn current_slot_disparity_covers_both_boundary_sides() { + let mut t = SlotTicker::new(0, Duration::from_secs(12), Duration::from_secs(4)); + let disparity = Duration::from_millis(500); + + t.set_since_genesis_ms(12_000 + 499); + assert!(t.is_current_slot_with_disparity(0, disparity)); + assert!(t.is_current_slot_with_disparity(1, disparity)); + assert!(!t.is_current_slot_with_disparity(2, disparity)); + assert_eq!(t.latest_slot_with_disparity(disparity), 1); + + t.set_since_genesis_ms(24_000 - 500); + assert!(t.is_current_slot_with_disparity(1, disparity)); + assert!(t.is_current_slot_with_disparity(2, disparity)); + assert_eq!(t.latest_slot_with_disparity(disparity), 2); + + t.set_since_genesis_ms(24_000 - 501); + assert!(!t.is_current_slot_with_disparity(2, disparity)); + assert_eq!(t.latest_slot_with_disparity(disparity), 1); + } } diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index acdd43fc..7eb234eb 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize}; use silver_chain_spec::ForkName; use silver_common::{ Enr, Error, GossipTopic, Identify, Keypair, NodeId, PeerId, SAMPLES_PER_SLOT, SLOTS_PER_EPOCH, - SUBNETS_PER_NODE, StreamProtocol, + SUBNETS_PER_NODE, SYNC_COMMITTEE_SUBNETS, StreamProtocol, }; pub use syncing_config::{PendingBounds, SyncingConfig}; @@ -88,6 +88,7 @@ fn default_gossip_topics() -> Vec { GossipTopic::BlsToExecutionChange.to_string(), GossipTopic::ExecutionPayload.to_string(), GossipTopic::PayloadAttestationMessage.to_string(), + GossipTopic::SyncCommitteeContributionAndProof.to_string(), ] } @@ -390,6 +391,7 @@ impl Config { builder.eth2(eth2); // Floor at SAMPLES_PER_SLOT: custody set must cover the sample set. builder.cgc(self.data_column_custody_group_count.max(SAMPLES_PER_SLOT) as u64); + builder.syncnets((1u8 << SYNC_COMMITTEE_SUBNETS) - 1); if let Some(ip) = self.external_ip_v4 { builder.ip4(ip); @@ -554,7 +556,7 @@ mod tests { assert_eq!(cfg.fork_digest(), [0x8c, 0x9f, 0x62, 0xfe]); assert_eq!(cfg.next_fork_epoch, u64::MAX); assert_eq!(cfg.supported_protocols().unwrap().len(), 11); - assert_eq!(cfg.gossip_topics().unwrap().len(), 8); + assert_eq!(cfg.gossip_topics().unwrap().len(), 9); assert_eq!(cfg.beacon_api_bind(), ["0.0.0.0:5051"]); assert_eq!(cfg.beacon_api_max_connections(), 64); assert_eq!(cfg.beacon_api_idle_timeout(), Duration::from_secs(75)); diff --git a/crates/ssz/src/ssz_view.rs b/crates/ssz/src/ssz_view.rs index 6469b86a..b180799b 100644 --- a/crates/ssz/src/ssz_view.rs +++ b/crates/ssz/src/ssz_view.rs @@ -375,6 +375,44 @@ impl SyncCommitteeView { } } +// -- SyncCommitteeContribution -------------------------------------- +// +// All fixed, exactly 160B. +// [0..8) slot +// [8..40) beacon_block_root +// [40..48) subcommittee_index +// [48..64) aggregation_bits (Bitvector[128]) +// [64..160) signature + +pub const SYNC_COMMITTEE_CONTRIBUTION_SIZE: usize = 160; + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +#[repr(C)] +pub struct SyncCommitteeContributionView; + +impl SyncCommitteeContributionView { + #[inline] + pub fn slot(buf: &[u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE]) -> u64 { + u64_le(buf, 0) + } + #[inline] + pub fn beacon_block_root(buf: &[u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE]) -> &[u8; 32] { + fixed(buf, 8) + } + #[inline] + pub fn subcommittee_index(buf: &[u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE]) -> u64 { + u64_le(buf, 40) + } + #[inline] + pub fn aggregation_bits(buf: &[u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE]) -> &[u8; 16] { + fixed(buf, 48) + } + #[inline] + pub fn signature(buf: &[u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE]) -> &[u8; 96] { + fixed(buf, 64) + } +} + // -- SignedContributionAndProof (sync_committee_contribution_and_proof) // // All fixed, exactly 360B. @@ -405,24 +443,30 @@ impl SignedContributionAndProofView { u64_le(buf, 0) } #[inline] + pub fn contribution( + buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE], + ) -> &[u8; SYNC_COMMITTEE_CONTRIBUTION_SIZE] { + fixed(buf, 8) + } + #[inline] pub fn slot(buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE]) -> u64 { - u64_le(buf, 8) + SyncCommitteeContributionView::slot(Self::contribution(buf)) } #[inline] pub fn beacon_block_root(buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE]) -> &[u8; 32] { - fixed(buf, 16) + SyncCommitteeContributionView::beacon_block_root(Self::contribution(buf)) } #[inline] pub fn subcommittee_index(buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE]) -> u64 { - u64_le(buf, 48) + SyncCommitteeContributionView::subcommittee_index(Self::contribution(buf)) } #[inline] pub fn aggregation_bits(buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE]) -> &[u8; 16] { - fixed(buf, 56) + SyncCommitteeContributionView::aggregation_bits(Self::contribution(buf)) } #[inline] pub fn contribution_signature(buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE]) -> &[u8; 96] { - fixed(buf, 72) + SyncCommitteeContributionView::signature(Self::contribution(buf)) } #[inline] pub fn selection_proof(buf: &[u8; SIGNED_CONTRIBUTION_AND_PROOF_SIZE]) -> &[u8; 96] {