feat: cancel in progress full blocks (paradigmxyz#2813)

crates/consensus/beacon/src/engine/mod.rs

@@ -624,6 +624,7 @@ where
             Ok(block) => block,
             Err(status) => return Ok(status),
         };
+        let block_hash = block.hash();
 
         let res = if self.sync.is_pipeline_idle() {
             // we can only insert new payloads if the pipeline is _not_ running, because it holds
@@ -634,7 +635,14 @@ where
         };
 
         let status = match res {
-            Ok(status) => Ok(status),
+            Ok(status) => {
+                if status.is_valid() {
+                    // block was successfully inserted, so we can cancel the full block request, if
+                    // any exists
+                    self.sync.cancel_full_block_request(block_hash);
+                }
+                Ok(status)
+            }
             Err(error) => {
                 debug!(target: "consensus::engine", ?error, "Error while processing payload");
                 self.map_insert_error(error)

crates/consensus/beacon/src/engine/sync.rs

@@ -85,6 +85,11 @@ where
         self.inflight_full_block_requests.clear();
     }
 
+    /// Cancels the full block request with the given hash.
+    pub(crate) fn cancel_full_block_request(&mut self, hash: H256) {
+        self.inflight_full_block_requests.retain(|req| *req.hash() != hash);
+    }
+
     /// Returns `true` if the pipeline is idle.
     pub(crate) fn is_pipeline_idle(&self) -> bool {
         self.pipeline_state.is_idle()