Resolve merge conflicts

beacon_node/beacon_processor/src/lib.rs

@@ -431,6 +431,7 @@ pub enum Work<E: EthSpec> {
     Status(BlockingFn),
     BlocksByRangeRequest(AsyncFn),
     BlocksByRootsRequest(AsyncFn),
+    BlocksByHeadRequest(AsyncFn),
     PayloadEnvelopesByRangeRequest(AsyncFn),
     PayloadEnvelopesByRootRequest(AsyncFn),
     BlobsByRangeRequest(BlockingFn),
@@ -492,6 +493,7 @@ pub enum WorkType {
     Status,
     BlocksByRangeRequest,
     BlocksByRootsRequest,
+    BlocksByHeadRequest,
     PayloadEnvelopesByRangeRequest,
     PayloadEnvelopesByRootRequest,
     BlobsByRangeRequest,
@@ -555,6 +557,7 @@ impl<E: EthSpec> Work<E> {
             Work::Status(_) => WorkType::Status,
             Work::BlocksByRangeRequest(_) => WorkType::BlocksByRangeRequest,
             Work::BlocksByRootsRequest(_) => WorkType::BlocksByRootsRequest,
+            Work::BlocksByHeadRequest(_) => WorkType::BlocksByHeadRequest,
             Work::PayloadEnvelopesByRangeRequest(_) => WorkType::PayloadEnvelopesByRangeRequest,
             Work::PayloadEnvelopesByRootRequest(_) => WorkType::PayloadEnvelopesByRootRequest,
             Work::BlobsByRangeRequest(_) => WorkType::BlobsByRangeRequest,
@@ -1006,6 +1009,8 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             Some(item)
                         } else if let Some(item) = work_queues.block_broots_queue.pop() {
                             Some(item)
+                        } else if let Some(item) = work_queues.block_bhead_queue.pop() {
+                            Some(item)
                         } else if let Some(item) = work_queues.blob_brange_queue.pop() {
                             Some(item)
                         } else if let Some(item) = work_queues.blob_broots_queue.pop() {
@@ -1215,6 +1220,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             Work::BlocksByRootsRequest { .. } => {
                                 work_queues.block_broots_queue.push(work, work_id)
                             }
+                            Work::BlocksByHeadRequest { .. } => {
+                                work_queues.block_bhead_queue.push(work, work_id)
+                            }
                             Work::PayloadEnvelopesByRangeRequest { .. } => work_queues
                                 .payload_envelopes_brange_queue
                                 .push(work, work_id),
@@ -1340,6 +1348,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         WorkType::Status => work_queues.status_queue.len(),
                         WorkType::BlocksByRangeRequest => work_queues.block_brange_queue.len(),
                         WorkType::BlocksByRootsRequest => work_queues.block_broots_queue.len(),
+                        WorkType::BlocksByHeadRequest => work_queues.block_bhead_queue.len(),
                         WorkType::PayloadEnvelopesByRangeRequest => {
                             work_queues.payload_envelopes_brange_queue.len()
                         }
@@ -1543,6 +1552,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
             }
             Work::BlocksByRangeRequest(work)
             | Work::BlocksByRootsRequest(work)
+            | Work::BlocksByHeadRequest(work)
             | Work::PayloadEnvelopesByRangeRequest(work)
             | Work::PayloadEnvelopesByRootRequest(work) => task_spawner.spawn_async(work),
             Work::ChainSegmentBackfill(process_fn) => {

beacon_node/beacon_processor/src/scheduler/work_queue.rs

@@ -132,6 +132,7 @@ pub struct BeaconProcessorQueueLengths {
     status_queue: usize,
     block_brange_queue: usize,
     block_broots_queue: usize,
+    block_bhead_queue: usize,
     blob_broots_queue: usize,
     blob_brange_queue: usize,
     dcbroots_queue: usize,
@@ -208,6 +209,7 @@ impl BeaconProcessorQueueLengths {
             status_queue: 1024,
             block_brange_queue: 1024,
             block_broots_queue: 1024,
+            block_bhead_queue: 1024,
             blob_broots_queue: 1024,
             blob_brange_queue: 1024,
             dcbroots_queue: 1024,
@@ -266,6 +268,7 @@ pub struct WorkQueues<E: EthSpec> {
     pub status_queue: FifoQueue<Work<E>>,
     pub block_brange_queue: FifoQueue<Work<E>>,
     pub block_broots_queue: FifoQueue<Work<E>>,
+    pub block_bhead_queue: FifoQueue<Work<E>>,
     pub payload_envelopes_brange_queue: FifoQueue<Work<E>>,
     pub payload_envelopes_broots_queue: FifoQueue<Work<E>>,
     pub blob_broots_queue: FifoQueue<Work<E>>,
@@ -338,6 +341,7 @@ impl<E: EthSpec> WorkQueues<E> {
         let status_queue = FifoQueue::new(queue_lengths.status_queue);
         let block_brange_queue = FifoQueue::new(queue_lengths.block_brange_queue);
         let block_broots_queue = FifoQueue::new(queue_lengths.block_broots_queue);
+        let block_bhead_queue = FifoQueue::new(queue_lengths.block_bhead_queue);
         let blob_broots_queue = FifoQueue::new(queue_lengths.blob_broots_queue);
         let blob_brange_queue = FifoQueue::new(queue_lengths.blob_brange_queue);
         let dcbroots_queue = FifoQueue::new(queue_lengths.dcbroots_queue);
@@ -404,6 +408,7 @@ impl<E: EthSpec> WorkQueues<E> {
             status_queue,
             block_brange_queue,
             block_broots_queue,
+            block_bhead_queue,
             blob_broots_queue,
             blob_brange_queue,
             dcbroots_queue,

beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs

@@ -280,8 +280,8 @@ struct ReprocessQueue<S> {
     queued_lc_updates: FnvHashMap<usize, (QueuedLightClientUpdate, DelayKey)>,
     /// Light Client Updates per parent_root.
     awaiting_lc_updates_per_parent_root: HashMap<Hash256, Vec<QueuedLightClientUpdateId>>,
-    /// Column reconstruction per block root.
-    queued_column_reconstructions: HashMap<Hash256, DelayKey>,
+    /// Column reconstruction per block root. `None` means reconstruction was already dispatched.
+    queued_column_reconstructions: HashMap<Hash256, Option<DelayKey>>,
     /// Queued backfill batches
     queued_backfill_batches: Vec<QueuedBackfillBatch>,
 
@@ -865,20 +865,20 @@ impl<S: SlotClock> ReprocessQueue<S> {
                     && duration_from_current_slot >= reconstruction_deadline
                     && current_slot == request.slot
                 {
-                    // If we are at least `reconstruction_deadline` seconds into the current slot,
-                    // and the reconstruction request is for the current slot, process reconstruction immediately.
                     reconstruction_delay = Duration::from_secs(0);
                 }
                 match self.queued_column_reconstructions.entry(request.block_root) {
-                    Entry::Occupied(key) => {
-                        self.column_reconstructions_delay_queue
-                            .reset(key.get(), reconstruction_delay);
+                    Entry::Occupied(entry) => {
+                        if let Some(delay_key) = entry.get() {
+                            self.column_reconstructions_delay_queue
+                                .reset(delay_key, reconstruction_delay);
+                        }
                     }
                     Entry::Vacant(vacant) => {
                         let delay_key = self
                             .column_reconstructions_delay_queue
                             .insert(request, reconstruction_delay);
-                        vacant.insert(delay_key);
+                        vacant.insert(Some(delay_key));
                     }
                 }
             }
@@ -1039,7 +1039,9 @@ impl<S: SlotClock> ReprocessQueue<S> {
             }
             InboundEvent::ReadyColumnReconstruction(column_reconstruction) => {
                 self.queued_column_reconstructions
-                    .remove(&column_reconstruction.block_root);
+                    .retain(|_, v| v.is_some());
+                self.queued_column_reconstructions
+                    .insert(column_reconstruction.block_root, None);
                 if self
                     .ready_work_tx
                     .try_send(ReadyWork::ColumnReconstruction(column_reconstruction))
@@ -1398,7 +1400,10 @@ mod tests {
             queue.handle_message(InboundEvent::ReadyColumnReconstruction(reconstruction));
         }
 
-        assert!(queue.queued_column_reconstructions.is_empty());
+        assert_eq!(
+            queue.queued_column_reconstructions.get(&block_root),
+            Some(&None)
+        );
     }
 
     /// Tests that column reconstruction queued after the deadline is triggered immediately