Resolve merge conflicts

beacon_node/beacon_processor/src/lib.rs

@@ -421,6 +421,7 @@ pub enum Work<E: EthSpec> {
         process_fn: AsyncFn,
     },
     RpcCustodyColumn(AsyncFn),
+    RpcEnvelope(AsyncFn),
     ColumnReconstruction(AsyncFn),
     IgnoredRpcBlock {
         process_fn: BlockingFn,
@@ -434,6 +435,7 @@ pub enum Work<E: EthSpec> {
     Status(BlockingFn),
     BlocksByRangeRequest(AsyncFn),
     BlocksByRootsRequest(AsyncFn),
+    BlocksByHeadRequest(AsyncFn),
     PayloadEnvelopesByRangeRequest(AsyncFn),
     PayloadEnvelopesByRootRequest(AsyncFn),
     BlobsByRangeRequest(BlockingFn),
@@ -488,6 +490,7 @@ pub enum WorkType {
     RpcBlobs,
     RpcPayloadEnvelope,
     RpcCustodyColumn,
+    RpcEnvelope,
     ColumnReconstruction,
     IgnoredRpcBlock,
     ChainSegment,
@@ -495,6 +498,7 @@ pub enum WorkType {
     Status,
     BlocksByRangeRequest,
     BlocksByRootsRequest,
+    BlocksByHeadRequest,
     PayloadEnvelopesByRangeRequest,
     PayloadEnvelopesByRootRequest,
     BlobsByRangeRequest,
@@ -551,6 +555,7 @@ impl<E: EthSpec> Work<E> {
             Work::RpcBlobs { .. } => WorkType::RpcBlobs,
             Work::RpcPayloadEnvelope { .. } => WorkType::RpcPayloadEnvelope,
             Work::RpcCustodyColumn { .. } => WorkType::RpcCustodyColumn,
+            Work::RpcEnvelope(_) => WorkType::RpcEnvelope,
             Work::ColumnReconstruction(_) => WorkType::ColumnReconstruction,
             Work::IgnoredRpcBlock { .. } => WorkType::IgnoredRpcBlock,
             Work::ChainSegment { .. } => WorkType::ChainSegment,
@@ -558,6 +563,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,
@@ -827,6 +833,8 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             Some(item)
                         } else if let Some(item) = work_queues.rpc_custody_column_queue.pop() {
                             Some(item)
+                        } else if let Some(item) = work_queues.rpc_envelope_queue.pop() {
+                            Some(item)
                         // Check delayed blocks before gossip blocks, the gossip blocks might rely
                         // on the delayed ones.
                         } else if let Some(item) = work_queues.delayed_block_queue.pop() {
@@ -1005,6 +1013,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() {
@@ -1194,6 +1204,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                                 work_queues.rpc_block_queue.push(work, work_id)
                             }
                             Work::RpcBlobs { .. } => work_queues.rpc_blob_queue.push(work, work_id),
+                            Work::RpcEnvelope(_) => {
+                                work_queues.rpc_envelope_queue.push(work, work_id)
+                            }
                             Work::RpcCustodyColumn { .. } => {
                                 work_queues.rpc_custody_column_queue.push(work, work_id)
                             }
@@ -1213,6 +1226,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),
@@ -1331,6 +1347,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         WorkType::RpcBlobs | WorkType::IgnoredRpcBlock => {
                             work_queues.rpc_blob_queue.len()
                         }
+                        WorkType::RpcEnvelope => work_queues.rpc_envelope_queue.len(),
                         WorkType::RpcCustodyColumn => work_queues.rpc_custody_column_queue.len(),
                         WorkType::ColumnReconstruction => {
                             work_queues.column_reconstruction_queue.len()
@@ -1340,6 +1357,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()
                         }
@@ -1524,6 +1542,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
             | Work::RpcBlobs { process_fn }
             | Work::RpcPayloadEnvelope { process_fn }
             | Work::RpcCustodyColumn(process_fn)
+            | Work::RpcEnvelope(process_fn)
             | Work::ColumnReconstruction(process_fn) => task_spawner.spawn_async(process_fn),
             Work::IgnoredRpcBlock { process_fn } => task_spawner.spawn_blocking(process_fn),
             Work::GossipBlock(work)
@@ -1541,6 +1560,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

@@ -120,6 +120,7 @@ pub struct BeaconProcessorQueueLengths {
     rpc_block_queue: usize,
     rpc_blob_queue: usize,
     rpc_custody_column_queue: usize,
+    rpc_envelope_queue: usize,
     column_reconstruction_queue: usize,
     chain_segment_queue: usize,
     backfill_chain_segment: usize,
@@ -132,6 +133,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,
@@ -194,6 +196,8 @@ impl BeaconProcessorQueueLengths {
             // We don't request more than `PARENT_DEPTH_TOLERANCE` (32) lookups, so we can limit
             // this queue size. With 48 max blobs per block, each column sidecar list could be up to 12MB.
             rpc_custody_column_queue: 64,
+            // Bounded by `PARENT_DEPTH_TOLERANCE`; one envelope per Gloas block.
+            rpc_envelope_queue: 1024,
             column_reconstruction_queue: 1,
             chain_segment_queue: 64,
             backfill_chain_segment: 64,
@@ -206,6 +210,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,
@@ -251,6 +256,7 @@ pub struct WorkQueues<E: EthSpec> {
     pub rpc_block_queue: FifoQueue<Work<E>>,
     pub rpc_blob_queue: FifoQueue<Work<E>>,
     pub rpc_custody_column_queue: FifoQueue<Work<E>>,
+    pub rpc_envelope_queue: FifoQueue<Work<E>>,
     pub column_reconstruction_queue: LifoQueue<Work<E>>,
     pub chain_segment_queue: FifoQueue<Work<E>>,
     pub backfill_chain_segment: FifoQueue<Work<E>>,
@@ -263,6 +269,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>>,
@@ -320,6 +327,7 @@ impl<E: EthSpec> WorkQueues<E> {
         let rpc_block_queue = FifoQueue::new(queue_lengths.rpc_block_queue);
         let rpc_blob_queue = FifoQueue::new(queue_lengths.rpc_blob_queue);
         let rpc_custody_column_queue = FifoQueue::new(queue_lengths.rpc_custody_column_queue);
+        let rpc_envelope_queue = FifoQueue::new(queue_lengths.rpc_envelope_queue);
         let column_reconstruction_queue = LifoQueue::new(queue_lengths.column_reconstruction_queue);
         let chain_segment_queue = FifoQueue::new(queue_lengths.chain_segment_queue);
         let backfill_chain_segment = FifoQueue::new(queue_lengths.backfill_chain_segment);
@@ -334,6 +342,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);
@@ -387,6 +396,7 @@ impl<E: EthSpec> WorkQueues<E> {
             rpc_block_queue,
             rpc_blob_queue,
             rpc_custody_column_queue,
+            rpc_envelope_queue,
             chain_segment_queue,
             column_reconstruction_queue,
             backfill_chain_segment,
@@ -399,6 +409,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,