From aa371fbb911ca16e36bba39f7fc96fe988eb6eb9 Mon Sep 17 00:00:00 2001 From: conache Date: Thu, 11 Jun 2026 19:15:12 +0300 Subject: [PATCH 1/9] Setup payload attestation reprocess queue --- .../gossip_verified_payload_attestation.rs | 3 + beacon_node/beacon_processor/src/lib.rs | 27 ++- .../src/scheduler/work_queue.rs | 8 + .../src/scheduler/work_reprocessing_queue.rs | 167 +++++++++++++++++- 4 files changed, 199 insertions(+), 6 deletions(-) diff --git a/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs b/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs index 3e9f9e4b60e..c80982830a9 100644 --- a/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs +++ b/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs @@ -76,6 +76,9 @@ impl VerifiedPayloadAttestationMessage { .get_block(&beacon_block_root) .is_none() { + // TODO(gloas): add to reporcessing queue - + // re-process the attestation message once we have the + // actual block data (until the slot end) return Err(Error::UnknownHeadBlock { beacon_block_root }); } diff --git a/beacon_node/beacon_processor/src/lib.rs b/beacon_node/beacon_processor/src/lib.rs index d6233ebaf92..580f7710ab5 100644 --- a/beacon_node/beacon_processor/src/lib.rs +++ b/beacon_node/beacon_processor/src/lib.rs @@ -68,8 +68,8 @@ use tracing::{debug, error, trace, warn}; use types::{EthSpec, Hash256, SignedAggregateAndProof, SingleAttestation, Slot, SubnetId}; use work_reprocessing_queue::IgnoredRpcBlock; use work_reprocessing_queue::{ - QueuedAggregate, QueuedLightClientUpdate, QueuedRpcBlock, QueuedUnaggregate, ReadyWork, - spawn_reprocess_scheduler, + QueuedAggregate, QueuedLightClientUpdate, QueuedPayloadAttestation, QueuedRpcBlock, + QueuedUnaggregate, ReadyWork, spawn_reprocess_scheduler, }; mod metrics; @@ -284,6 +284,13 @@ impl From for WorkEvent { drop_during_sync: true, work: Work::UnknownBlockAggregate { process_fn }, }, + ReadyWork::PayloadAttestation(QueuedPayloadAttestation { + beacon_block_root: _, + process_fn, + }) => Self { + drop_during_sync: true, + work: Work::UnknownBlockPayloadAttestation { process_fn }, + }, ReadyWork::LightClientUpdate(QueuedLightClientUpdate { parent_root, process_fn, @@ -388,6 +395,9 @@ pub enum Work { UnknownBlockAggregate { process_fn: BlockingFn, }, + UnknownBlockPayloadAttestation { + process_fn: BlockingFn, + }, UnknownLightClientOptimisticUpdate { parent_root: Hash256, process_fn: BlockingFn, @@ -475,6 +485,7 @@ pub enum WorkType { GossipAttestationBatch, GossipAggregate, UnknownBlockAggregate, + UnknownBlockPayloadAttestation, UnknownLightClientOptimisticUpdate, GossipAggregateBatch, GossipBlock, @@ -579,6 +590,7 @@ impl Work { Work::UnknownBlockAttestation { .. } => WorkType::UnknownBlockAttestation, Work::UnknownBlockDataColumn { .. } => WorkType::UnknownBlockDataColumn, Work::UnknownBlockAggregate { .. } => WorkType::UnknownBlockAggregate, + Work::UnknownBlockPayloadAttestation { .. } => WorkType::UnknownBlockPayloadAttestation, Work::UnknownLightClientOptimisticUpdate { .. } => { WorkType::UnknownLightClientOptimisticUpdate } @@ -993,6 +1005,10 @@ impl BeaconProcessor { } else if let Some(item) = work_queues.unknown_block_aggregate_queue.pop() { Some(item) } else if let Some(item) = work_queues.unknown_block_attestation_queue.pop() + { + Some(item) + } else if let Some(item) = + work_queues.unknown_block_payload_attestation_queue.pop() { Some(item) // Check execution payload bids. Most proposers will request bids directly from builders @@ -1256,6 +1272,9 @@ impl BeaconProcessor { Work::UnknownBlockAggregate { .. } => { work_queues.unknown_block_aggregate_queue.push(work) } + Work::UnknownBlockPayloadAttestation { .. } => work_queues + .unknown_block_payload_attestation_queue + .push(work), Work::GossipBlsToExecutionChange { .. } => work_queues .gossip_bls_to_execution_change_queue .push(work, work_id), @@ -1311,6 +1330,9 @@ impl BeaconProcessor { WorkType::UnknownBlockAggregate => { work_queues.unknown_block_aggregate_queue.len() } + WorkType::UnknownBlockPayloadAttestation => { + work_queues.unknown_block_payload_attestation_queue.len() + } WorkType::UnknownLightClientOptimisticUpdate => { work_queues.unknown_light_client_update_queue.len() } @@ -1522,6 +1544,7 @@ impl BeaconProcessor { }), Work::UnknownBlockAttestation { process_fn } | Work::UnknownBlockAggregate { process_fn } + | Work::UnknownBlockPayloadAttestation { process_fn } | Work::UnknownBlockDataColumn { process_fn } | Work::UnknownLightClientOptimisticUpdate { process_fn, .. } => { task_spawner.spawn_blocking(process_fn) diff --git a/beacon_node/beacon_processor/src/scheduler/work_queue.rs b/beacon_node/beacon_processor/src/scheduler/work_queue.rs index cc03feac51d..8f0da26cd1d 100644 --- a/beacon_node/beacon_processor/src/scheduler/work_queue.rs +++ b/beacon_node/beacon_processor/src/scheduler/work_queue.rs @@ -111,6 +111,7 @@ pub struct BeaconProcessorQueueLengths { attestation_queue: usize, unknown_block_aggregate_queue: usize, unknown_block_attestation_queue: usize, + unknown_block_payload_attestation_queue: usize, unknown_block_data_column_queue: usize, sync_message_queue: usize, sync_contribution_queue: usize, @@ -187,6 +188,9 @@ impl BeaconProcessorQueueLengths { active_validator_count / slots_per_epoch, MIN_QUEUE_LEN, ), + // PTC size ~512 per slot, buffer 2-3 slots for reorgs and processing delays (512 * 3 = 1536) + // TODO(EIP-7732): verify if this is preferable queue length or otherwise + unknown_block_payload_attestation_queue: 1536, sync_message_queue: 2048, sync_contribution_queue: 1024, gossip_voluntary_exit_queue: 4096, @@ -248,6 +252,7 @@ pub struct WorkQueues { pub attestation_debounce: TimeLatch, pub unknown_block_aggregate_queue: LifoQueue>, pub unknown_block_attestation_queue: LifoQueue>, + pub unknown_block_payload_attestation_queue: LifoQueue>, pub unknown_block_data_column_queue: FifoQueue>, pub sync_message_queue: LifoQueue>, pub sync_contribution_queue: LifoQueue>, @@ -306,6 +311,8 @@ impl WorkQueues { LifoQueue::new(queue_lengths.unknown_block_aggregate_queue); let unknown_block_attestation_queue = LifoQueue::new(queue_lengths.unknown_block_attestation_queue); + let unknown_block_payload_attestation_queue = + LifoQueue::new(queue_lengths.unknown_block_payload_attestation_queue); let unknown_block_data_column_queue = FifoQueue::new(queue_lengths.unknown_block_data_column_queue); @@ -389,6 +396,7 @@ impl WorkQueues { attestation_debounce, unknown_block_aggregate_queue, unknown_block_attestation_queue, + unknown_block_payload_attestation_queue, unknown_block_data_column_queue, sync_message_queue, sync_contribution_queue, diff --git a/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs b/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs index 62ed86fbad0..1f3790f125f 100644 --- a/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs +++ b/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs @@ -8,8 +8,9 @@ //! There is the edge-case where the slot arrives before this queue manages to process it. In that //! case, the block will be sent off for immediate processing (skipping the `DelayQueue`). //! -//! Aggregated and unaggregated attestations that failed verification due to referencing an unknown -//! block will be re-queued until their block is imported, or until they expire. +//! Aggregated and unaggregated attestations, as well as payload attestation messages, that failed +//! verification due to referencing an unknown block will be re-queued until their block is +//! imported, or until they expire. use crate::metrics; use crate::{AsyncFn, BlockingFn, Work, WorkEvent}; use fnv::FnvHashMap; @@ -46,7 +47,8 @@ const LIGHT_CLIENT_UPDATES_PER_PARENT_ROOT: &str = "lc_updates_per_parent_root"; /// This is to account for any slight drift in the system clock. pub const ADDITIONAL_QUEUED_BLOCK_DELAY: Duration = Duration::from_millis(5); -/// For how long to queue aggregated and unaggregated attestations for re-processing. +/// For how long to queue aggregated and unaggregated attestations, as well as payload attestation +/// messages, for re-processing. pub const QUEUED_ATTESTATION_DELAY: Duration = Duration::from_secs(12); /// For how long to queue light client updates for re-processing. @@ -126,6 +128,8 @@ pub enum ReprocessQueueMessage { UnknownBlockUnaggregate(QueuedUnaggregate), /// An aggregated attestation that references an unknown block. UnknownBlockAggregate(QueuedAggregate), + /// A payload attestation message that references an unknown block. + UnknownBlockPayloadAttestation(QueuedPayloadAttestation), /// A light client optimistic update that references a parent root that has not been seen as a parent. UnknownLightClientOptimisticUpdate(QueuedLightClientUpdate), /// A new backfill batch that needs to be scheduled for processing. @@ -144,6 +148,7 @@ pub enum ReadyWork { IgnoredRpcBlock(IgnoredRpcBlock), Unaggregate(QueuedUnaggregate), Aggregate(QueuedAggregate), + PayloadAttestation(QueuedPayloadAttestation), LightClientUpdate(QueuedLightClientUpdate), BackfillSync(QueuedBackfillBatch), ColumnReconstruction(QueuedColumnReconstruction), @@ -164,6 +169,13 @@ pub struct QueuedAggregate { pub process_fn: BlockingFn, } +/// A payload attestation message for which the corresponding block was not seen while processing, +/// queued for later. +pub struct QueuedPayloadAttestation { + pub beacon_block_root: Hash256, + pub process_fn: BlockingFn, +} + /// A light client update for which the corresponding parent block was not seen while processing, /// queued for later. pub struct QueuedLightClientUpdate { @@ -294,7 +306,9 @@ struct ReprocessQueue { queued_aggregates: FnvHashMap, /// Queued attestations. queued_unaggregates: FnvHashMap, - /// Attestations (aggregated and unaggregated) per root. + /// Queued payload attestation messages. + queued_payload_attestations: FnvHashMap, + /// Attestations (aggregated, unaggregated and payload attestation messages) per root. awaiting_attestations_per_root: HashMap>, /// Queued Light Client Updates. queued_lc_updates: FnvHashMap, @@ -329,6 +343,7 @@ pub type QueuedLightClientUpdateId = usize; enum QueuedAttestationId { Aggregate(usize), Unaggregate(usize), + PayloadAttestation(usize), } impl QueuedAggregate { @@ -343,6 +358,12 @@ impl QueuedUnaggregate { } } +impl QueuedPayloadAttestation { + pub fn beacon_block_root(&self) -> &Hash256 { + &self.beacon_block_root + } +} + impl Stream for ReprocessQueue { type Item = InboundEvent; @@ -493,6 +514,7 @@ impl ReprocessQueue { queued_lc_updates: FnvHashMap::default(), queued_aggregates: FnvHashMap::default(), queued_unaggregates: FnvHashMap::default(), + queued_payload_attestations: FnvHashMap::default(), awaiting_attestations_per_root: HashMap::new(), awaiting_lc_updates_per_parent_root: HashMap::new(), queued_backfill_batches: Vec::new(), @@ -718,6 +740,40 @@ impl ReprocessQueue { self.next_attestation += 1; } + InboundEvent::Msg(UnknownBlockPayloadAttestation(queued_payload_attestation)) => { + if self.attestations_delay_queue.len() >= MAXIMUM_QUEUED_ATTESTATIONS { + if self.attestation_delay_debounce.elapsed() { + error!( + queue_size = MAXIMUM_QUEUED_ATTESTATIONS, + msg = "system resources may be saturated", + "Payload attestation delay queue is full" + ); + } + // Drop the payload attestation. + return; + } + + let att_id = QueuedAttestationId::PayloadAttestation(self.next_attestation); + + // Register the delay. + let delay_key = self + .attestations_delay_queue + .insert(att_id, QUEUED_ATTESTATION_DELAY); + + // Register this payload attestation for the corresponding root. + self.awaiting_attestations_per_root + .entry(*queued_payload_attestation.beacon_block_root()) + .or_default() + .push(att_id); + + // Store the payload attestation and its info. + self.queued_payload_attestations.insert( + self.next_attestation, + (queued_payload_attestation, delay_key), + ); + + self.next_attestation += 1; + } InboundEvent::Msg(UnknownBlockDataColumn(queued_data_column)) => { let block_root = queued_data_column.beacon_block_root; @@ -829,6 +885,15 @@ impl ReprocessQueue { .map(|(unaggregate, delay_key)| { (ReadyWork::Unaggregate(unaggregate), delay_key) }), + QueuedAttestationId::PayloadAttestation(id) => self + .queued_payload_attestations + .remove(&id) + .map(|(payload_attestation, delay_key)| { + ( + ReadyWork::PayloadAttestation(payload_attestation), + delay_key, + ) + }), } { // Remove the delay. self.attestations_delay_queue.remove(&delay_key); @@ -1024,6 +1089,15 @@ impl ReprocessQueue { ReadyWork::Unaggregate(unaggregate), ) }), + QueuedAttestationId::PayloadAttestation(id) => self + .queued_payload_attestations + .remove(&id) + .map(|(payload_attestation, _delay_key)| { + ( + *payload_attestation.beacon_block_root(), + ReadyWork::PayloadAttestation(payload_attestation), + ) + }), } { if self.ready_work_tx.try_send(work).is_err() { error!( @@ -1412,6 +1486,91 @@ mod tests { assert!(queue.awaiting_attestations_per_root.is_empty()); } + /// Tests that a queued payload attestation message is released when its block is imported. + #[tokio::test] + async fn payload_attestation_released_on_block_imported() { + create_test_tracing_subscriber(); + + let mut queue = test_queue(); + + // Pause time so it only advances manually + tokio::time::pause(); + + let beacon_block_root = Hash256::repeat_byte(0xaf); + let parent_root = Hash256::repeat_byte(0xab); + + // Insert a payload attestation. + let msg = ReprocessQueueMessage::UnknownBlockPayloadAttestation(QueuedPayloadAttestation { + beacon_block_root, + process_fn: Box::new(|| {}), + }); + + // Process the event to enter it into the delay queue. + queue.handle_message(InboundEvent::Msg(msg)); + + // Check that it is queued. + assert_eq!(queue.queued_payload_attestations.len(), 1); + assert!( + queue + .awaiting_attestations_per_root + .contains_key(&beacon_block_root) + ); + + // Simulate block import. + let imported = ReprocessQueueMessage::BlockImported { + block_root: beacon_block_root, + parent_root, + }; + queue.handle_message(InboundEvent::Msg(imported)); + + // The entry for the block root should be gone. + assert!(queue.queued_payload_attestations.is_empty()); + assert!(queue.awaiting_attestations_per_root.is_empty()); + // Delay queue entry should also be cancelled. + assert_eq!(queue.attestations_delay_queue.len(), 0); + } + + /// Tests that an expired payload attestation message is pruned from + /// `awaiting_attestations_per_root`. + #[tokio::test] + async fn prune_awaiting_payload_attestations_per_root() { + create_test_tracing_subscriber(); + + let mut queue = test_queue(); + + // Pause time so it only advances manually + tokio::time::pause(); + + let beacon_block_root = Hash256::repeat_byte(0xaf); + + // Insert a payload attestation. + let msg = ReprocessQueueMessage::UnknownBlockPayloadAttestation(QueuedPayloadAttestation { + beacon_block_root, + process_fn: Box::new(|| {}), + }); + + // Process the event to enter it into the delay queue. + queue.handle_message(InboundEvent::Msg(msg)); + + // Check that it is queued. + assert_eq!(queue.awaiting_attestations_per_root.len(), 1); + assert!( + queue + .awaiting_attestations_per_root + .contains_key(&beacon_block_root) + ); + + // Advance time to expire the payload attestation. + advance_time(&queue.slot_clock, 2 * QUEUED_ATTESTATION_DELAY).await; + let ready_msg = queue.next().await.unwrap(); + assert!(matches!(ready_msg, InboundEvent::ReadyAttestation(_))); + queue.handle_message(ready_msg); + + // The entry for the block root should be gone. + assert!(queue.queued_payload_attestations.is_empty()); + assert!(queue.awaiting_attestations_per_root.is_empty()); + } + // This is a regression test for a memory leak in `awaiting_lc_updates_per_parent_root`. // See: https://github.com/sigp/lighthouse/pull/8065 #[tokio::test] From 085dcb8b33add398a036f372ec3078f7ffee8613 Mon Sep 17 00:00:00 2001 From: conache Date: Thu, 11 Jun 2026 20:43:09 +0300 Subject: [PATCH 2/9] Trigger payload attestations reprocessing for unknwon blocks --- .../gossip_methods.rs | 73 ++++++++++++++++--- .../src/network_beacon_processor/mod.rs | 1 + 2 files changed, 62 insertions(+), 12 deletions(-) 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 98c143eaeba..b44b34f6437 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -62,7 +62,8 @@ use beacon_processor::{ DuplicateCache, GossipAggregatePackage, GossipAttestationBatch, work_reprocessing_queue::{ QueuedAggregate, QueuedGossipBlock, QueuedGossipDataColumn, QueuedGossipEnvelope, - QueuedLightClientUpdate, QueuedUnaggregate, ReprocessQueueMessage, + QueuedLightClientUpdate, QueuedPayloadAttestation, QueuedUnaggregate, + ReprocessQueueMessage, }, }; @@ -3915,13 +3916,21 @@ impl NetworkBeaconProcessor { message_id: MessageId, peer_id: PeerId, payload_attestation_message: Box, + allow_reprocess: bool, ) { - let message_slot = payload_attestation_message.data.slot; - let result = self - .chain - .verify_payload_attestation_message_for_gossip(*payload_attestation_message); + // Clone the message for verification, retaining the original so that it can be + // re-queued if it references a block we haven't seen yet. + let result = self.chain.verify_payload_attestation_message_for_gossip( + payload_attestation_message.as_ref().clone(), + ); - self.process_gossip_payload_attestation_result(result, message_id, peer_id, message_slot); + self.process_gossip_payload_attestation_result( + result, + message_id, + peer_id, + payload_attestation_message, + allow_reprocess, + ); } fn process_gossip_payload_attestation_result( @@ -3929,7 +3938,8 @@ impl NetworkBeaconProcessor { result: Result, PayloadAttestationError>, message_id: MessageId, peer_id: PeerId, - message_slot: Slot, + payload_attestation_message: Box, + allow_reprocess: bool, ) { match result { Ok(verified) => { @@ -3970,19 +3980,22 @@ impl NetworkBeaconProcessor { peer_id, message_id, error, - message_slot, + payload_attestation_message, + allow_reprocess, ); } } } fn handle_payload_attestation_verification_failure( - &self, + self: &Arc, peer_id: PeerId, message_id: MessageId, error: PayloadAttestationError, - message_slot: Slot, + payload_attestation_message: Box, + allow_reprocess: bool, ) { + let message_slot = payload_attestation_message.data.slot; match &error { PayloadAttestationError::FutureSlot { .. } => { self.gossip_penalize_peer( @@ -4002,11 +4015,47 @@ impl NetworkBeaconProcessor { %message_slot, "Payload attestation references unknown block" ); - self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Ignore); + // We don't know the block yet, get the sync manager to handle the block lookup self.send_sync_message(SyncMessage::UnknownBlockHashFromAttestation( peer_id, *beacon_block_root, - )) + )); + + if allow_reprocess { + // Queue the payload attestation for re-processing + let processor = self.clone(); + let msg = ReprocessQueueMessage::UnknownBlockPayloadAttestation( + QueuedPayloadAttestation { + beacon_block_root: *beacon_block_root, + process_fn: Box::new(move || { + processor.process_gossip_payload_attestation( + message_id, + peer_id, + payload_attestation_message, + false, + ) + }), + }, + ); + + if let Err(e) = self.beacon_processor_send.try_send(WorkEvent { + drop_during_sync: false, + work: Work::Reprocess(msg), + }) { + error!( + error = %e, + ?beacon_block_root, + %message_slot, + "Failed to send payload attestation for re-processing" + ) + } + } else { + self.propagate_validation_result( + message_id, + peer_id, + MessageAcceptance::Ignore, + ); + } } PayloadAttestationError::NotInPTC { .. } => { self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Reject); diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index a9579caaeb6..23331e56d1d 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -485,6 +485,7 @@ impl NetworkBeaconProcessor { message_id, peer_id, payload_attestation_message, + true, ) }; From 73ef69bad9189b58a1d961dd5b365e6d38786c5f Mon Sep 17 00:00:00 2001 From: conache Date: Fri, 12 Jun 2026 11:00:36 +0300 Subject: [PATCH 3/9] Add e2e network tests --- .../src/network_beacon_processor/tests.rs | 154 +++++++++++++++++- 1 file changed, 151 insertions(+), 3 deletions(-) diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index 6b7c623230f..ffdc7ea8c34 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -43,10 +43,11 @@ use std::time::Duration; use tokio::sync::mpsc; use types::data::BlobIdentifier; use types::{ - AttesterSlashing, ChainSpec, DataColumnSidecarList, DataColumnSubnetId, Epoch, EthSpec, + AttesterSlashing, ChainSpec, DataColumnSidecarList, DataColumnSubnetId, Domain, Epoch, EthSpec, ExecutionPayloadEnvelope, ExecutionPayloadGloas, ExecutionRequests, Hash256, MainnetEthSpec, - ProposerSlashing, SignedAggregateAndProof, SignedBeaconBlock, SignedExecutionPayloadEnvelope, - SignedVoluntaryExit, SingleAttestation, Slot, SubnetId, + PayloadAttestationData, PayloadAttestationMessage, ProposerSlashing, SignedAggregateAndProof, + SignedBeaconBlock, SignedExecutionPayloadEnvelope, SignedRoot, SignedVoluntaryExit, + SingleAttestation, Slot, SubnetId, }; type E = MainnetEthSpec; @@ -603,6 +604,48 @@ impl TestRig { .unwrap(); } + /// Enqueue a valid payload attestation message for `next_block`, signed by the first + /// member of the PTC for its slot. + pub fn enqueue_next_block_payload_attestation(&self) { + let slot = self.next_block.slot(); + let beacon_block_root = self.next_block.canonical_root(); + let head = self.chain.canonical_head.cached_head(); + let state = &head.snapshot.beacon_state; + + let ptc = state + .get_ptc(slot, &self.chain.spec) + .expect("should get PTC"); + let validator_index = *ptc.0.first().expect("PTC should not be empty") as u64; + + let data = PayloadAttestationData { + beacon_block_root, + slot, + payload_present: true, + blob_data_available: true, + }; + let domain = self.chain.spec.get_domain( + slot.epoch(E::slots_per_epoch()), + Domain::PTCAttester, + &state.fork(), + state.genesis_validators_root(), + ); + let signature = self._harness.validator_keypairs[validator_index as usize] + .sk + .sign(data.signing_root(domain)); + + self.network_beacon_processor + .send_gossip_payload_attestation( + junk_message_id(), + junk_peer_id(), + Box::new(PayloadAttestationMessage { + validator_index, + data, + signature, + }), + ) + .unwrap(); + } + /// Assert that the `BeaconProcessor` doesn't produce any events in the given `duration`. pub async fn assert_no_events_for(&mut self, duration: Duration) { tokio::select! { @@ -1353,6 +1396,111 @@ async fn aggregate_attestation_to_unknown_block_processed_after_rpc_block() { aggregate_attestation_to_unknown_block(BlockImportMethod::Rpc).await } +async fn payload_attestation_to_unknown_block_processed(import_method: BlockImportMethod) { + // Only test when the Gloas fork is scheduled + if test_spec::().gloas_fork_epoch.is_none() { + return; + } + + let mut rig = TestRig::new(SMALL_CHAIN).await; + + // Send the payload attestation but not the block, and check that it was not imported. + + let initial_messages = rig.chain.op_pool.num_payload_attestation_messages(); + + rig.enqueue_next_block_payload_attestation(); + + rig.assert_event_journal_completes(&[WorkType::GossipPayloadAttestation]) + .await; + + assert_eq!( + rig.chain.op_pool.num_payload_attestation_messages(), + initial_messages, + "Payload attestation should not have been included." + ); + + // The gossipsub validation result should be withheld while the payload attestation is + // queued for re-processing. + assert!( + rig.receive_network_messages_with_timeout(Duration::from_millis(100), None) + .await + .is_none(), + "no validation result should be sent while the payload attestation is queued" + ); + + // Send the block and ensure that the payload attestation is received back and imported. + let num_data_columns = rig.next_data_columns.as_ref().map(|c| c.len()).unwrap_or(0); + let mut events = vec![]; + match import_method { + BlockImportMethod::Gossip => { + rig.enqueue_gossip_block(); + events.push(WorkType::GossipBlock); + for i in 0..num_data_columns { + rig.enqueue_gossip_data_columns(i); + events.push(WorkType::GossipDataColumnSidecar); + } + } + BlockImportMethod::Rpc => { + rig.enqueue_lookup_block(); + events.push(WorkType::RpcBlock); + if num_data_columns > 0 { + rig.enqueue_single_lookup_rpc_data_columns(); + events.push(WorkType::RpcCustodyColumn); + } + } + }; + + events.push(WorkType::UnknownBlockPayloadAttestation); + + rig.assert_event_journal_contains_ordered(&events).await; + + assert_eq!( + rig.chain.op_pool.num_payload_attestation_messages(), + initial_messages + 1, + "Payload attestation should have been included." + ); + + // The re-processed payload attestation should have been propagated with an `Accept` + // validation result. + let messages = rig + .receive_network_messages_with_timeout(Duration::from_millis(100), None) + .await + .expect("should receive validation results after block import"); + let accepts_count = messages + .iter() + .filter(|msg| { + matches!( + msg, + NetworkMessage::ValidationResult { + validation_result: MessageAcceptance::Accept, + .. + } + ) + }) + .count(); + let expected_accepts_count = match import_method { + // The gossip block import also propagates an `Accept` for the block itself. + BlockImportMethod::Gossip => 2, + // RPC blocks never touch gossipsub, so the only `Accept` is the re-processed + // payload attestation's. + BlockImportMethod::Rpc => 1, + }; + assert_eq!( + accepts_count, expected_accepts_count, + "re-processed payload attestation should be propagated" + ); +} + +#[tokio::test] +async fn payload_attestation_to_unknown_block_processed_after_gossip_block() { + payload_attestation_to_unknown_block_processed(BlockImportMethod::Gossip).await +} + +#[tokio::test] +async fn payload_attestation_to_unknown_block_processed_after_rpc_block() { + payload_attestation_to_unknown_block_processed(BlockImportMethod::Rpc).await +} + /// Ensure that attestations that reference an unknown block get properly re-queued and re-processed /// when the block is not seen. #[tokio::test] From b6b68933725ddf888e0317870d817b380f025ae2 Mon Sep 17 00:00:00 2001 From: conache Date: Fri, 12 Jun 2026 11:09:48 +0300 Subject: [PATCH 4/9] Remove misleading comment --- .../gossip_verified_payload_attestation.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs b/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs index c80982830a9..3e9f9e4b60e 100644 --- a/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs +++ b/beacon_node/beacon_chain/src/payload_attestation_verification/gossip_verified_payload_attestation.rs @@ -76,9 +76,6 @@ impl VerifiedPayloadAttestationMessage { .get_block(&beacon_block_root) .is_none() { - // TODO(gloas): add to reporcessing queue - - // re-process the attestation message once we have the - // actual block data (until the slot end) return Err(Error::UnknownHeadBlock { beacon_block_root }); } From 44599ef1427525dfabb18b45eb5f40c995815660 Mon Sep 17 00:00:00 2001 From: conache Date: Wed, 8 Jul 2026 10:11:05 +0300 Subject: [PATCH 5/9] Remove unknown block ptc queue todo --- beacon_node/beacon_processor/src/scheduler/work_queue.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/beacon_node/beacon_processor/src/scheduler/work_queue.rs b/beacon_node/beacon_processor/src/scheduler/work_queue.rs index 6d7eabfe60b..9f51d21497b 100644 --- a/beacon_node/beacon_processor/src/scheduler/work_queue.rs +++ b/beacon_node/beacon_processor/src/scheduler/work_queue.rs @@ -189,7 +189,6 @@ impl BeaconProcessorQueueLengths { MIN_QUEUE_LEN, ), // PTC size ~512 per slot, buffer 2-3 slots for reorgs and processing delays (512 * 3 = 1536) - // TODO(EIP-7732): verify if this is preferable queue length or otherwise unknown_block_payload_attestation_queue: 1536, sync_message_queue: 2048, sync_contribution_queue: 1024, From 577bc98e848442288d0b8c684d7d5995694782e5 Mon Sep 17 00:00:00 2001 From: conache Date: Wed, 8 Jul 2026 10:26:20 +0300 Subject: [PATCH 6/9] Flag unknown block only if reprocess is allowed --- .../src/network_beacon_processor/gossip_methods.rs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) 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 edbcb2715fb..b22a76de1f5 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -4209,13 +4209,13 @@ impl NetworkBeaconProcessor { %message_slot, "Payload attestation references unknown block" ); - // We don't know the block yet, get the sync manager to handle the block lookup - self.send_sync_message(SyncMessage::UnknownBlockHashFromAttestation( - peer_id, - *beacon_block_root, - )); - if allow_reprocess { + // We don't know the block yet, get the sync manager to handle the block lookup + self.send_sync_message(SyncMessage::UnknownBlockHashFromAttestation( + peer_id, + *beacon_block_root, + )); + // Queue the payload attestation for re-processing let processor = self.clone(); let msg = ReprocessQueueMessage::UnknownBlockPayloadAttestation( From 918ea83f7773335ce3cd73ceb4b05977848c2c3f Mon Sep 17 00:00:00 2001 From: conache Date: Wed, 8 Jul 2026 10:47:31 +0300 Subject: [PATCH 7/9] Use ReprocessAllowance with a new BlockOnly variant instead of allow_reporcess bool flag --- .../gossip_methods.rs | 33 ++++++++++++------- .../src/network_beacon_processor/mod.rs | 2 +- 2 files changed, 23 insertions(+), 12 deletions(-) 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 b22a76de1f5..58e4fe135df 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -81,6 +81,8 @@ const STRICT_LATE_MESSAGE_PENALTIES: bool = false; pub enum ReprocessAllowance { /// Re-queue for either an unknown block or an unknown payload envelope. BlockAndPayload, + /// Re-queue only for an unknown block + BlockOnly, /// Re-queue only for an unknown payload envelope (already re-queued once for the block). PayloadOnly, /// Do not re-queue again. @@ -90,7 +92,10 @@ pub enum ReprocessAllowance { impl ReprocessAllowance { /// Whether the attestation may be re-queued for an unknown block. fn allows_block(self) -> bool { - matches!(self, ReprocessAllowance::BlockAndPayload) + matches!( + self, + ReprocessAllowance::BlockAndPayload | ReprocessAllowance::BlockOnly + ) } /// Whether the attestation may be re-queued for an unknown payload envelope. @@ -105,7 +110,9 @@ impl ReprocessAllowance { fn next_requeue(self) -> Self { match self { ReprocessAllowance::BlockAndPayload => ReprocessAllowance::PayloadOnly, - ReprocessAllowance::PayloadOnly | ReprocessAllowance::None => ReprocessAllowance::None, + ReprocessAllowance::BlockOnly + | ReprocessAllowance::PayloadOnly + | ReprocessAllowance::None => ReprocessAllowance::None, } } } @@ -4110,7 +4117,7 @@ impl NetworkBeaconProcessor { message_id: MessageId, peer_id: PeerId, payload_attestation_message: Box, - allow_reprocess: bool, + reprocess_allowance: ReprocessAllowance, ) { // Clone the message for verification, retaining the original so that it can be // re-queued if it references a block we haven't seen yet. @@ -4123,7 +4130,7 @@ impl NetworkBeaconProcessor { message_id, peer_id, payload_attestation_message, - allow_reprocess, + reprocess_allowance, ); } @@ -4133,7 +4140,7 @@ impl NetworkBeaconProcessor { message_id: MessageId, peer_id: PeerId, payload_attestation_message: Box, - allow_reprocess: bool, + reprocess_allowance: ReprocessAllowance, ) { match result { Ok(verified) => { @@ -4175,7 +4182,7 @@ impl NetworkBeaconProcessor { message_id, error, payload_attestation_message, - allow_reprocess, + reprocess_allowance, ); } } @@ -4187,7 +4194,7 @@ impl NetworkBeaconProcessor { message_id: MessageId, error: PayloadAttestationError, payload_attestation_message: Box, - allow_reprocess: bool, + reprocess_allowance: ReprocessAllowance, ) { let message_slot = payload_attestation_message.data.slot; match &error { @@ -4209,7 +4216,7 @@ impl NetworkBeaconProcessor { %message_slot, "Payload attestation references unknown block" ); - if allow_reprocess { + if reprocess_allowance.allows_block() { // We don't know the block yet, get the sync manager to handle the block lookup self.send_sync_message(SyncMessage::UnknownBlockHashFromAttestation( peer_id, @@ -4226,7 +4233,7 @@ impl NetworkBeaconProcessor { message_id, peer_id, payload_attestation_message, - false, + reprocess_allowance.next_requeue(), ) }), }, @@ -4298,18 +4305,21 @@ impl NetworkBeaconProcessor { #[cfg(test)] mod tests { - use super::ReprocessAllowance::{BlockAndPayload, None, PayloadOnly}; + use super::ReprocessAllowance::{BlockAndPayload, BlockOnly, None, PayloadOnly}; #[test] fn reprocess_allowance_gates() { // A block re-queue is only permitted for a freshly received attestation. assert!(BlockAndPayload.allows_block()); + assert!(BlockOnly.allows_block()); assert!(!PayloadOnly.allows_block()); assert!(!None.allows_block()); // A payload-envelope re-queue is permitted until we've already re-queued for it. assert!(BlockAndPayload.allows_payload()); assert!(PayloadOnly.allows_payload()); + // `BlockOnly` never waits on a payload envelope (e.g. payload attestations). + assert!(!BlockOnly.allows_payload()); assert!(!None.allows_payload()); } @@ -4317,6 +4327,7 @@ mod tests { fn reprocess_allowance_progression() { // Each re-queue narrows the allowance to the next variant in the progression. assert_eq!(BlockAndPayload.next_requeue(), PayloadOnly); + assert_eq!(BlockOnly.next_requeue(), None); assert_eq!(PayloadOnly.next_requeue(), None); assert_eq!(None.next_requeue(), None); } @@ -4325,7 +4336,7 @@ mod tests { fn reprocess_allowance_is_bounded() { // Safety property: from any starting state, re-queuing twice reaches the terminal `None`, // so an attestation can never loop indefinitely. - for start in [BlockAndPayload, PayloadOnly, None] { + for start in [BlockAndPayload, BlockOnly, PayloadOnly, None] { assert_eq!( start.next_requeue().next_requeue(), None, diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index 2251ccb0382..0d2fa8468b5 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -495,7 +495,7 @@ impl NetworkBeaconProcessor { message_id, peer_id, payload_attestation_message, - true, + ReprocessAllowance::BlockOnly, ) }; From 6e44d89f4458ef7578320045cd507184eae7365a Mon Sep 17 00:00:00 2001 From: conache Date: Wed, 8 Jul 2026 11:06:30 +0300 Subject: [PATCH 8/9] Cover payload attestation timeout case without block import --- .../gossip_methods.rs | 1 - .../src/network_beacon_processor/tests.rs | 47 +++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) 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 58e4fe135df..499ca34cc65 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -4318,7 +4318,6 @@ mod tests { // A payload-envelope re-queue is permitted until we've already re-queued for it. assert!(BlockAndPayload.allows_payload()); assert!(PayloadOnly.allows_payload()); - // `BlockOnly` never waits on a payload envelope (e.g. payload attestations). assert!(!BlockOnly.allows_payload()); assert!(!None.allows_payload()); } diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index e3e30e73b3e..a38f61e75e6 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -1498,6 +1498,53 @@ async fn payload_attestation_to_unknown_block_processed_after_rpc_block() { payload_attestation_to_unknown_block_processed(BlockImportMethod::Rpc).await } +/// Ensure that a payload attestation referencing an unknown block gets re-queued and, when the +/// block is never seen, is received back on timeout without being imported. +#[tokio::test] +async fn requeue_unknown_block_gossip_payload_attestation_without_import() { + // Only test when the Gloas fork is scheduled + if test_spec::().gloas_fork_epoch.is_none() { + return; + } + + let mut rig = TestRig::new(SMALL_CHAIN).await; + + // Send the payload attestation but not the block, and check that it was not imported. + + let initial_messages = rig.chain.op_pool.num_payload_attestation_messages(); + + rig.enqueue_next_block_payload_attestation(); + + rig.assert_event_journal_completes(&[WorkType::GossipPayloadAttestation]) + .await; + + assert_eq!( + rig.chain.op_pool.num_payload_attestation_messages(), + initial_messages, + "Payload attestation should not have been included." + ); + + // Ensure that the payload attestation is received back on timeout but not imported. + + rig.assert_event_journal_with_timeout( + &[ + WorkType::UnknownBlockPayloadAttestation.into(), + WORKER_FREED, + NOTHING_TO_DO, + ], + Duration::from_secs(1) + QUEUED_ATTESTATION_DELAY, + false, + false, + ) + .await; + + assert_eq!( + rig.chain.op_pool.num_payload_attestation_messages(), + initial_messages, + "Payload attestation should not have been included." + ); +} + /// Ensure that attestations that reference an unknown block get properly re-queued and re-processed /// when the block is not seen. #[tokio::test] From bc49d70551037e1f7b28e81952812d8d1277304c Mon Sep 17 00:00:00 2001 From: conache Date: Wed, 8 Jul 2026 11:27:31 +0300 Subject: [PATCH 9/9] Add payload attestation requeued counter metric --- beacon_node/network/src/metrics.rs | 7 +++++++ .../network/src/network_beacon_processor/gossip_methods.rs | 3 +++ 2 files changed, 10 insertions(+) diff --git a/beacon_node/network/src/metrics.rs b/beacon_node/network/src/metrics.rs index add2f1c966f..6200273a86c 100644 --- a/beacon_node/network/src/metrics.rs +++ b/beacon_node/network/src/metrics.rs @@ -317,6 +317,13 @@ pub static BEACON_PROCESSOR_AGGREGATED_ATTESTATION_REQUEUED_TOTAL: LazyLock> = + LazyLock::new(|| { + try_create_int_counter( + "beacon_processor_payload_attestation_requeued_total", + "Total number of payload attestations that referenced an unknown block and were re-queued.", + ) + }); // Sync committee messages. pub static BEACON_PROCESSOR_SYNC_MESSAGE_VERIFIED_TOTAL: LazyLock> = LazyLock::new(|| { 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 499ca34cc65..8e871ae1778 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -4224,6 +4224,9 @@ impl NetworkBeaconProcessor { )); // Queue the payload attestation for re-processing + metrics::inc_counter( + &metrics::BEACON_PROCESSOR_PAYLOAD_ATTESTATION_REQUEUED_TOTAL, + ); let processor = self.clone(); let msg = ReprocessQueueMessage::UnknownBlockPayloadAttestation( QueuedPayloadAttestation {