From 1543db8b87ed2c2e4e81d8288b84f7d46763f6c9 Mon Sep 17 00:00:00 2001 From: Jimmy Chen Date: Mon, 27 Apr 2026 11:32:25 +0200 Subject: [PATCH] Add payload attestation pool and block packing Store gossip-verified PayloadAttestationMessages in the operation pool keyed by PayloadAttestationData. At block production time, aggregate them into PayloadAttestations using PTC position mapping and include in the block body. Wire pool insertion into the gossip handler after fork choice. --- beacon_node/beacon_chain/src/beacon_chain.rs | 12 + .../src/block_production/gloas.rs | 6 +- .../gossip_methods.rs | 8 + beacon_node/operation_pool/src/lib.rs | 319 +++++++++++++++++- beacon_node/operation_pool/src/persistence.rs | 1 + 5 files changed, 343 insertions(+), 3 deletions(-) diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index cf5afb089a..9da64888c2 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -61,6 +61,7 @@ use crate::observed_data_sidecars::ObservedDataSidecars; use crate::observed_operations::{ObservationOutcome, ObservedOperations}; use crate::observed_slashable::ObservedSlashable; use crate::partial_data_column_assembler::PartialMergeResult; +use crate::payload_attestation_verification::VerifiedPayloadAttestationMessage; use crate::payload_bid_verification::payload_bid_cache::GossipVerifiedPayloadBidCache; #[cfg(not(test))] use crate::payload_envelope_streamer::{EnvelopeRequestSource, launch_payload_envelope_stream}; @@ -2328,6 +2329,17 @@ impl BeaconChain { .map_err(Into::into) } + /// Add a verified payload attestation message to the operation pool for block inclusion. + pub fn add_payload_attestation_to_pool( + &self, + verified: &VerifiedPayloadAttestationMessage, + ) -> Result<(), Error> { + self.op_pool + .insert_payload_attestation_message(verified.payload_attestation_message().clone()) + .map_err(Error::OpPoolError)?; + Ok(()) + } + /// Accepts some `SyncCommitteeMessage` from the network and attempts to verify it, returning `Ok(_)` if /// it is valid to be (re)broadcast on the gossip network. pub fn verify_sync_committee_message_for_gossip( diff --git a/beacon_node/beacon_chain/src/block_production/gloas.rs b/beacon_node/beacon_chain/src/block_production/gloas.rs index 9b3fc2806e..6e5dd456f0 100644 --- a/beacon_node/beacon_chain/src/block_production/gloas.rs +++ b/beacon_node/beacon_chain/src/block_production/gloas.rs @@ -434,8 +434,10 @@ impl BeaconChain { deposits, voluntary_exits, sync_aggregate, - // TODO(gloas) need to implement payload attestations - payload_attestations: vec![], + payload_attestations: self + .op_pool + .get_payload_attestations(&state, parent_root, &self.spec) + .map_err(BlockProductionError::OpPoolError)?, bls_to_execution_changes, }, state, diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs index 4083b1a3af..676ea1c69e 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -4156,6 +4156,14 @@ impl NetworkBeaconProcessor { ), } } + + if let Err(e) = self.chain.add_payload_attestation_to_pool(&verified) { + warn!( + reason = ?e, + %peer_id, + "Failed to add payload attestation to pool" + ); + } } Err(RejectedPayloadAttestation { payload_attestation_message, diff --git a/beacon_node/operation_pool/src/lib.rs b/beacon_node/operation_pool/src/lib.rs index 4b815704d9..5f1a4cb95f 100644 --- a/beacon_node/operation_pool/src/lib.rs +++ b/beacon_node/operation_pool/src/lib.rs @@ -23,10 +23,12 @@ use crate::attestation_storage::{AttestationMap, CheckpointKey}; use crate::bls_to_execution_changes::BlsToExecutionChanges; use crate::sync_aggregate_id::SyncAggregateId; use attester_slashing::AttesterSlashingMaxCover; +use bls::AggregateSignature; use max_cover::maximum_cover; use parking_lot::{RwLock, RwLockWriteGuard}; use rand::rng; use rand::seq::SliceRandom; +use ssz::BitVector; use state_processing::per_block_processing::errors::AttestationValidationError; use state_processing::per_block_processing::{ VerifySignatures, get_slashable_indices_modular, verify_exit, @@ -38,7 +40,8 @@ use std::ptr; use typenum::Unsigned; use types::{ AbstractExecPayload, Attestation, AttestationData, AttesterSlashing, BeaconState, - BeaconStateError, ChainSpec, Epoch, EthSpec, ProposerSlashing, SignedBeaconBlock, + BeaconStateError, ChainSpec, Epoch, EthSpec, Hash256, PayloadAttestation, + PayloadAttestationData, PayloadAttestationMessage, ProposerSlashing, SignedBeaconBlock, SignedBlsToExecutionChange, SignedVoluntaryExit, Slot, SyncAggregate, SyncAggregateError, SyncCommitteeContribution, Validator, }; @@ -59,6 +62,9 @@ pub struct OperationPool { voluntary_exits: RwLock>>, /// Map from credential changing validator to their position in the queue. bls_to_execution_changes: RwLock>, + /// Map from payload attestation data to individual messages for aggregation at block production. + payload_attestation_messages: + RwLock>>, /// Reward cache for accelerating attestation packing. reward_cache: RwLock, _phantom: PhantomData, @@ -78,6 +84,8 @@ pub enum OpPoolError { IncorrectOpPoolVariant, EpochCacheNotInitialized, EpochCacheError(EpochCacheError), + GetPtcError(BeaconStateError), + PayloadAttestationBitError, } #[derive(Default)] @@ -193,6 +201,92 @@ impl OperationPool { }); } + /// Insert a validated `PayloadAttestationMessage` into the pool. + pub fn insert_payload_attestation_message( + &self, + message: PayloadAttestationMessage, + ) -> Result<(), OpPoolError> { + let mut messages = self.payload_attestation_messages.write(); + let entry = messages.entry(message.data.clone()).or_default(); + if !entry + .iter() + .any(|m| m.validator_index == message.validator_index) + { + entry.push(message); + } + Ok(()) + } + + /// Build aggregated `PayloadAttestation`s from stored messages for block production. + /// + /// `parent_block_root` is the root of the parent block (the block PTC members attested to). + /// Returns up to `MAX_PAYLOAD_ATTESTATIONS` attestations for the previous slot. + pub fn get_payload_attestations( + &self, + state: &BeaconState, + parent_block_root: Hash256, + spec: &ChainSpec, + ) -> Result>, OpPoolError> { + let target_slot = state.slot().saturating_sub(1u64); + + let ptc = state + .get_ptc(target_slot, spec) + .map_err(OpPoolError::GetPtcError)?; + + let messages = self.payload_attestation_messages.read(); + let mut result = Vec::new(); + + for (data, msgs) in messages.iter() { + if data.slot != target_slot || data.beacon_block_root != parent_block_root { + continue; + } + + let mut aggregation_bits = BitVector::new(); + let mut aggregate_sig = AggregateSignature::infinity(); + + for msg in msgs { + if let Some(pos) = ptc + .0 + .iter() + .position(|&idx| idx == msg.validator_index as usize) + { + if !aggregation_bits.get(pos).unwrap_or(false) { + aggregation_bits + .set(pos, true) + .map_err(|_| OpPoolError::PayloadAttestationBitError)?; + aggregate_sig.add_assign(&msg.signature); + } + } + } + + if aggregation_bits.num_set_bits() > 0 { + result.push(PayloadAttestation { + aggregation_bits, + data: data.clone(), + signature: aggregate_sig, + }); + } + } + + Ok(result) + } + + /// Remove payload attestation messages that are too old for block inclusion. + pub fn prune_payload_attestation_messages(&self, current_slot: Slot) { + self.payload_attestation_messages + .write() + .retain(|data, _| current_slot <= data.slot.saturating_add(Slot::new(1))); + } + + /// Total number of payload attestation messages in the pool. + pub fn num_payload_attestation_messages(&self) -> usize { + self.payload_attestation_messages + .read() + .values() + .map(|msgs| msgs.len()) + .sum() + } + /// Insert an attestation into the pool, aggregating it with existing attestations if possible. /// /// ## Note @@ -646,6 +740,7 @@ impl OperationPool { ) { self.prune_attestations(current_epoch); self.prune_sync_contributions(head_state.slot()); + self.prune_payload_attestation_messages(head_state.slot()); self.prune_proposer_slashings(finalized_state); self.prune_attester_slashings(finalized_state); self.prune_voluntary_exits(finalized_state, spec); @@ -2075,4 +2170,226 @@ mod release_tests { op_pool.prune_attester_slashings(&electra_head.beacon_state); assert_eq!(op_pool.attester_slashings.read().len(), 1); } + + fn make_payload_attestation_message( + slot: Slot, + validator_index: u64, + beacon_block_root: Hash256, + ) -> PayloadAttestationMessage { + PayloadAttestationMessage { + validator_index, + data: PayloadAttestationData { + beacon_block_root, + slot, + payload_present: true, + blob_data_available: true, + }, + signature: bls::Signature::empty(), + } + } + + #[test] + fn payload_attestation_insert_and_dedup() { + let op_pool = OperationPool::::new(); + let root = Hash256::repeat_byte(0xaa); + let slot = Slot::new(1); + + let msg1 = make_payload_attestation_message(slot, 0, root); + let msg2 = make_payload_attestation_message(slot, 1, root); + let msg1_dup = make_payload_attestation_message(slot, 0, root); + + op_pool.insert_payload_attestation_message(msg1).unwrap(); + op_pool.insert_payload_attestation_message(msg2).unwrap(); + op_pool + .insert_payload_attestation_message(msg1_dup) + .unwrap(); + + assert_eq!(op_pool.num_payload_attestation_messages(), 2); + } + + #[test] + fn payload_attestation_prune() { + let op_pool = OperationPool::::new(); + let root = Hash256::repeat_byte(0xaa); + + let msg_slot1 = make_payload_attestation_message(Slot::new(1), 0, root); + let msg_slot2 = make_payload_attestation_message(Slot::new(2), 1, root); + let msg_slot3 = make_payload_attestation_message(Slot::new(3), 2, root); + + op_pool + .insert_payload_attestation_message(msg_slot1) + .unwrap(); + op_pool + .insert_payload_attestation_message(msg_slot2) + .unwrap(); + op_pool + .insert_payload_attestation_message(msg_slot3) + .unwrap(); + + assert_eq!(op_pool.num_payload_attestation_messages(), 3); + + op_pool.prune_payload_attestation_messages(Slot::new(3)); + assert_eq!(op_pool.num_payload_attestation_messages(), 2); + + op_pool.prune_payload_attestation_messages(Slot::new(4)); + assert_eq!(op_pool.num_payload_attestation_messages(), 1); + + op_pool.prune_payload_attestation_messages(Slot::new(5)); + assert_eq!(op_pool.num_payload_attestation_messages(), 0); + } + + #[tokio::test] + async fn payload_attestation_packs_bits_from_ptc_positions() { + let spec = test_spec::(); + if !spec.gloas_fork_epoch.is_some_and(|e| e == Epoch::new(0)) { + return; + } + + let num_validators = 64; + let harness = get_harness::(num_validators, Some(spec.clone())); + + harness + .add_attested_blocks_at_slots( + harness.get_current_state(), + Hash256::zero(), + &[Slot::new(1)], + (0..num_validators).collect::>().as_slice(), + ) + .await; + + let head = harness.chain.canonical_head.cached_head(); + let state = &head.snapshot.beacon_state; + assert_eq!(state.slot(), Slot::new(1)); + + let target_slot = Slot::new(1); + let parent_root = head.head_block_root(); + let ptc = state.get_ptc(target_slot, &spec).unwrap(); + let ptc_member_0 = ptc.0[0] as u64; + let ptc_member_1 = ptc.0[1] as u64; + + let op_pool = OperationPool::::new(); + + let msg0 = make_payload_attestation_message(target_slot, ptc_member_0, parent_root); + let msg1 = make_payload_attestation_message(target_slot, ptc_member_1, parent_root); + op_pool.insert_payload_attestation_message(msg0).unwrap(); + op_pool.insert_payload_attestation_message(msg1).unwrap(); + + // Advance state to slot 2 so get_payload_attestations looks at slot 1. + let mut advanced_state = state.clone(); + state_processing::state_advance::complete_state_advance( + &mut advanced_state, + None, + Slot::new(2), + &spec, + ) + .unwrap(); + + let attestations = op_pool + .get_payload_attestations(&advanced_state, parent_root, &spec) + .unwrap(); + + assert_eq!(attestations.len(), 1); + assert_eq!(attestations[0].aggregation_bits.num_set_bits(), 2); + assert!(attestations[0].aggregation_bits.get(0).unwrap()); + assert!(attestations[0].aggregation_bits.get(1).unwrap()); + assert!(attestations[0].data.payload_present); + } + + #[tokio::test] + async fn payload_attestation_filters_wrong_root() { + let spec = test_spec::(); + if !spec.gloas_fork_epoch.is_some_and(|e| e == Epoch::new(0)) { + return; + } + + let num_validators = 64; + let harness = get_harness::(num_validators, Some(spec.clone())); + + harness + .add_attested_blocks_at_slots( + harness.get_current_state(), + Hash256::zero(), + &[Slot::new(1)], + (0..num_validators).collect::>().as_slice(), + ) + .await; + + let head = harness.chain.canonical_head.cached_head(); + let state = &head.snapshot.beacon_state; + let target_slot = Slot::new(1); + let ptc = state.get_ptc(target_slot, &spec).unwrap(); + let ptc_member = ptc.0[0] as u64; + + let op_pool = OperationPool::::new(); + let wrong_root = Hash256::repeat_byte(0xff); + let msg = make_payload_attestation_message(target_slot, ptc_member, wrong_root); + op_pool.insert_payload_attestation_message(msg).unwrap(); + + let mut advanced_state = state.clone(); + state_processing::state_advance::complete_state_advance( + &mut advanced_state, + None, + Slot::new(2), + &spec, + ) + .unwrap(); + + let parent_root = head.head_block_root(); + let attestations = op_pool + .get_payload_attestations(&advanced_state, parent_root, &spec) + .unwrap(); + + assert!(attestations.is_empty()); + } + + #[tokio::test] + async fn payload_attestation_non_ptc_validator_skipped() { + let spec = test_spec::(); + if !spec.gloas_fork_epoch.is_some_and(|e| e == Epoch::new(0)) { + return; + } + + let num_validators = 64; + let harness = get_harness::(num_validators, Some(spec.clone())); + + harness + .add_attested_blocks_at_slots( + harness.get_current_state(), + Hash256::zero(), + &[Slot::new(1)], + (0..num_validators).collect::>().as_slice(), + ) + .await; + + let head = harness.chain.canonical_head.cached_head(); + let state = &head.snapshot.beacon_state; + let target_slot = Slot::new(1); + let parent_root = head.head_block_root(); + let ptc = state.get_ptc(target_slot, &spec).unwrap(); + + // Find a validator NOT in the PTC. + let non_ptc_validator = (0..num_validators as u64) + .find(|&i| !ptc.0.contains(&(i as usize))) + .expect("should find non-PTC validator"); + + let op_pool = OperationPool::::new(); + let msg = make_payload_attestation_message(target_slot, non_ptc_validator, parent_root); + op_pool.insert_payload_attestation_message(msg).unwrap(); + + let mut advanced_state = state.clone(); + state_processing::state_advance::complete_state_advance( + &mut advanced_state, + None, + Slot::new(2), + &spec, + ) + .unwrap(); + + let attestations = op_pool + .get_payload_attestations(&advanced_state, parent_root, &spec) + .unwrap(); + + // Message was in the pool but validator isn't in PTC, so no attestation produced. + assert!(attestations.is_empty()); + } } diff --git a/beacon_node/operation_pool/src/persistence.rs b/beacon_node/operation_pool/src/persistence.rs index 241b5fec53..56aafc27fe 100644 --- a/beacon_node/operation_pool/src/persistence.rs +++ b/beacon_node/operation_pool/src/persistence.rs @@ -209,6 +209,7 @@ impl PersistedOperationPool { proposer_slashings, voluntary_exits, bls_to_execution_changes: RwLock::new(bls_to_execution_changes), + payload_attestation_messages: Default::default(), reward_cache: Default::default(), _phantom: Default::default(), };