consensus: Improve error handling for failed missing block requests

Improve error handling for failed missing block requests in the block
queue by removing from the buffer the block successor chain if a
request for a missing block failed.

consensus/src/sync/live/block_queue/block_request_component.rs

@@ -66,6 +66,15 @@ pub enum MissingBlockError {
     Response(ResponseBlocksError),
 }
 
+#[derive(Debug, Error)]
+#[error("Sync queue error. Target block hash: {target_block_hash}, target block number {target_block_number}")]
+pub struct SyncQueueError {
+    /// Target block hash
+    pub target_block_hash: Blake2bHash,
+    /// Target block number
+    pub target_block_number: u32,
+}
+
 /// Peer Tracking & Block Request Component.
 /// We use this component to request missing blocks from peers.
 ///
@@ -321,33 +330,35 @@ impl<N: Network> BlockRequestComponent<N> {
 }
 
 impl<N: Network> Stream for BlockRequestComponent<N> {
-    type Item = BlockRequestResult<N>;
+    type Item = Result<BlockRequestResult<N>, SyncQueueError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         // Poll self.sync_queue, return results.
-        while let Poll::Ready(Some(result)) = self.sync_queue.poll_next_unpin(cx) {
+        while let Poll::Ready(result) = self.sync_queue.poll_next_unpin(cx) {
             match result {
-                Ok(response) => {
+                Some(Ok(response)) => {
                     self.pending_requests.remove(&response.target_block_hash);
-                    return Poll::Ready(Some(BlockRequestResult {
+                    return Poll::Ready(Some(Ok(BlockRequestResult {
                         target_block_number: response.target_block_number,
                         target_block_hash: response.target_block_hash,
                         epoch_validators: response.epoch_validators,
                         blocks: response.blocks,
                         sender: response.sender,
-                    }));
+                    })));
                 }
-                Err(request) => {
+                Some(Err(request)) => {
                     self.pending_requests.remove(&request.target_block_hash);
                     debug!(
                         request.target_block_number,
                         ?request.target_block_hash,
                         "Failed to retrieve missing blocks"
                     );
-                    // TODO: Do we need to do anything else?
-                    //  We might want to delete the target hash from our buffer
-                    //  since none of our peers is sending us a good response.
+                    return Poll::Ready(Some(Err(SyncQueueError {
+                        target_block_hash: request.target_block_hash,
+                        target_block_number: request.target_block_number,
+                    })));
                 }
+                None => {}
             }
         }
 

consensus/src/sync/live/block_queue/queue.rs

@@ -822,7 +822,7 @@ impl<N: Network> Stream for BlockQueue<N> {
         loop {
             let poll_res = self.request_component.poll_next_unpin(cx);
             match poll_res {
-                Poll::Ready(Some(result)) => {
+                Poll::Ready(Some(Ok(result))) => {
                     if let Some(block) = self.handle_missing_blocks(
                         result.target_block_number,
                         result.target_block_hash,
@@ -833,6 +833,26 @@ impl<N: Network> Stream for BlockQueue<N> {
                         return Poll::Ready(Some(block));
                     }
                 }
+                Poll::Ready(Some(Err(syn_queue_error))) => {
+                    // The request failed, remove from the buffer all blocks from the target block chain that errored.
+                    // Do this by locating the successor of the block if it exists and see if there is a block whose
+                    // parent hash is the block in question.
+                    let mut block_number = syn_queue_error.target_block_number;
+                    let mut block_hash = syn_queue_error.target_block_hash;
+                    while let Some(blocks) = self.buffer.get_mut(&(block_number + 1)) {
+                        let Some((hash, (block, _source))) = blocks
+                            .iter()
+                            .find(|(_hash, (block, _source))| *block.parent_hash() == block_hash)
+                        else {
+                            // No block found whose parent block is the one we're looking: we should have already removed
+                            // all blocks from the target block chain that errored.
+                            break;
+                        };
+                        block_number = block.block_number();
+                        block_hash = hash.clone();
+                        blocks.remove(&block_hash);
+                    }
+                }
                 Poll::Ready(None) => return Poll::Ready(None),
                 Poll::Pending => break,
             }