merge queue: checking unstable (81d5769) and [#9607 + #9458 + #9608] together

beacon_node/beacon_chain/src/validator_monitor.rs

@@ -1893,7 +1893,7 @@ impl<E: EthSpec> ValidatorMonitor<E> {
                 epoch - 2
             };
 
-            for (_, validator) in self.validators.iter() {
+            for validator in self.validators.values() {
                 let id = &validator.id;
                 let summaries = validator.summaries.read();
 

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<E: EthSpec> From<ReadyWork> for WorkEvent<E> {
                 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<E: EthSpec> {
     UnknownBlockAggregate {
         process_fn: BlockingFn,
     },
+    UnknownBlockPayloadAttestation {
+        process_fn: BlockingFn,
+    },
     UnknownLightClientOptimisticUpdate {
         parent_root: Hash256,
         process_fn: BlockingFn,
@@ -474,6 +484,7 @@ pub enum WorkType {
     GossipAttestationBatch,
     GossipAggregate,
     UnknownBlockAggregate,
+    UnknownBlockPayloadAttestation,
     UnknownLightClientOptimisticUpdate,
     GossipAggregateBatch,
     GossipBlock,
@@ -578,6 +589,7 @@ impl<E: EthSpec> Work<E> {
             Work::UnknownBlockAttestation { .. } => WorkType::UnknownBlockAttestation,
             Work::UnknownBlockDataColumn { .. } => WorkType::UnknownBlockDataColumn,
             Work::UnknownBlockAggregate { .. } => WorkType::UnknownBlockAggregate,
+            Work::UnknownBlockPayloadAttestation { .. } => WorkType::UnknownBlockPayloadAttestation,
             Work::UnknownLightClientOptimisticUpdate { .. } => {
                 WorkType::UnknownLightClientOptimisticUpdate
             }
@@ -989,6 +1001,10 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         } 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
@@ -1252,6 +1268,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             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),
@@ -1304,6 +1323,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         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()
                         }
@@ -1515,6 +1537,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
             }),
             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)

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,8 @@ 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)
+            unknown_block_payload_attestation_queue: 1536,
             sync_message_queue: 2048,
             sync_contribution_queue: 1024,
             gossip_voluntary_exit_queue: 4096,
@@ -247,6 +250,7 @@ pub struct WorkQueues<E: EthSpec> {
     pub attestation_debounce: TimeLatch,
     pub unknown_block_aggregate_queue: LifoQueue<Work<E>>,
     pub unknown_block_attestation_queue: LifoQueue<Work<E>>,
+    pub unknown_block_payload_attestation_queue: LifoQueue<Work<E>>,
     pub unknown_block_data_column_queue: FifoQueue<Work<E>>,
     pub sync_message_queue: LifoQueue<Work<E>>,
     pub sync_contribution_queue: LifoQueue<Work<E>>,
@@ -304,6 +308,8 @@ impl<E: EthSpec> WorkQueues<E> {
             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);
 
@@ -386,6 +392,7 @@ impl<E: EthSpec> WorkQueues<E> {
             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,