Implement PTC reprocess queue

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,

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),
     /// An unaggregated attestation (`index == 1`) whose block's execution payload envelope has not
     /// been seen yet.
     UnknownPayloadUnaggregate(QueuedUnaggregate),
@@ -150,6 +154,7 @@ pub enum ReadyWork {
     IgnoredRpcBlock(IgnoredRpcBlock),
     Unaggregate(QueuedUnaggregate),
     Aggregate(QueuedAggregate),
+    PayloadAttestation(QueuedPayloadAttestation),
     LightClientUpdate(QueuedLightClientUpdate),
     BackfillSync(QueuedBackfillBatch),
     ColumnReconstruction(QueuedColumnReconstruction),
@@ -170,6 +175,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 {
@@ -300,7 +312,9 @@ struct ReprocessQueue<S> {
     queued_aggregates: FnvHashMap<usize, (QueuedAggregate, DelayKey)>,
     /// Queued attestations.
     queued_unaggregates: FnvHashMap<usize, (QueuedUnaggregate, DelayKey)>,
-    /// Attestations (aggregated and unaggregated) per root.
+    /// Queued payload attestation messages.
+    queued_payload_attestations: FnvHashMap<usize, (QueuedPayloadAttestation, DelayKey)>,
+    /// Attestations (aggregated, unaggregated and payload attestation messages) per root.
     awaiting_attestations_per_root: HashMap<Hash256, Vec<QueuedAttestationId>>,
     /// Attestations (aggregated and unaggregated) awaiting a block's execution payload envelope,
     /// keyed by block root. Released on `PayloadEnvelopeImported`.
@@ -338,12 +352,15 @@ pub type QueuedLightClientUpdateId = usize;
 enum QueuedAttestationId {
     Aggregate(usize),
     Unaggregate(usize),
+    PayloadAttestation(usize),
 }
 
-/// An attestation queued for re-processing, of either aggregation kind.
+/// An attestation queued for re-processing, of either aggregation kind, or a payload
+/// attestation message.
 enum QueuedAttestation {
     Aggregate(QueuedAggregate),
     Unaggregate(QueuedUnaggregate),
+    PayloadAttestation(QueuedPayloadAttestation),
 }
 
 /// The component an attestation is waiting on before it can be re-processed.
@@ -366,6 +383,12 @@ impl QueuedUnaggregate {
     }
 }
 
+impl QueuedPayloadAttestation {
+    pub fn beacon_block_root(&self) -> &Hash256 {
+        &self.beacon_block_root
+    }
+}
+
 impl<S: SlotClock> Stream for ReprocessQueue<S> {
     type Item = InboundEvent;
 
@@ -516,6 +539,7 @@ impl<S: SlotClock> ReprocessQueue<S> {
             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_attestations_per_payload: HashMap::new(),
             awaiting_lc_updates_per_parent_root: HashMap::new(),
@@ -564,6 +588,10 @@ impl<S: SlotClock> ReprocessQueue<S> {
             QueuedAttestation::Unaggregate(u) => {
                 (QueuedAttestationId::Unaggregate(id), *u.beacon_block_root())
             }
+            QueuedAttestation::PayloadAttestation(p) => (
+                QueuedAttestationId::PayloadAttestation(id),
+                *p.beacon_block_root(),
+            ),
         };
 
         // Register the delay.
@@ -590,6 +618,10 @@ impl<S: SlotClock> ReprocessQueue<S> {
                 self.queued_unaggregates
                     .insert(id, (queued_unaggregate, delay_key));
             }
+            QueuedAttestation::PayloadAttestation(queued_payload_attestation) => {
+                self.queued_payload_attestations
+                    .insert(id, (queued_payload_attestation, delay_key));
+            }
         }
 
         self.next_attestation += 1;
@@ -747,6 +779,11 @@ impl<S: SlotClock> ReprocessQueue<S> {
                     QueuedAttestation::Unaggregate(queued_unaggregate),
                     AwaitingComponent::Block,
                 ),
+            InboundEvent::Msg(UnknownBlockPayloadAttestation(queued_payload_attestation)) => self
+                .queue_awaiting_attestation(
+                    QueuedAttestation::PayloadAttestation(queued_payload_attestation),
+                    AwaitingComponent::Block,
+                ),
             InboundEvent::Msg(UnknownPayloadAggregate(queued_aggregate)) => self
                 .queue_awaiting_attestation(
                     QueuedAttestation::Aggregate(queued_aggregate),
@@ -865,6 +902,15 @@ impl<S: SlotClock> ReprocessQueue<S> {
                                 .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);
@@ -940,6 +986,8 @@ impl<S: SlotClock> ReprocessQueue<S> {
                                 .map(|(unaggregate, delay_key)| {
                                     (ReadyWork::Unaggregate(unaggregate), delay_key)
                                 }),
+                            // Payload attestations are only ever queued awaiting a block.
+                            QueuedAttestationId::PayloadAttestation(_) => None,
                         } {
                             // Remove the delay.
                             self.attestations_delay_queue.remove(&delay_key);
@@ -1112,6 +1160,15 @@ impl<S: SlotClock> ReprocessQueue<S> {
                                 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!(
@@ -1507,6 +1564,89 @@ 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);
+
+        // 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,
+        };
+        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());
+    }
+
     // Regression test for the same memory leak as `prune_awaiting_attestations_per_root`, but for
     // attestations awaiting a block's execution payload envelope.
     #[tokio::test]