fix(l1): hive invalid ancestor (#1926)

**Motivation**
Fixes hive invalid ancestor tests under `engine-cancun` with the pattern
`engine-cancun/Invalid Missing .* .*CanonicalReOrg=False`.

**Description**
The following was implemented to fix the tests:
- Because `forkchoice_update` and `new_payload` require us to send the
latest valid hash, we now cache the last valid hash of the chain
corresponding to bad blocks, these are the `invalid_ancestors`.
- The bad blocks are cached when:
   1. When failing to add a block during syncing
   2.  When the block execution of `new_payload` fails.
- The latest valid hash was fixed on `new_payload`; this was a known
issue see #982.
- Finally, when syncing, if we have a pending block from a `new_payload`
request, we attach it if the parent hash of the block corresponds to the
last header parent hash.

Closes #982
Advances on #1285

Author: MarcosNicolau

.github/workflows/ci_l1.yaml

@@ -168,7 +168,7 @@ jobs:
             test_pattern: engine-(auth|exchange-capabilities)/
           - name: "Cancun Engine tests"
             simulation: ethereum/engine
-            test_pattern: "engine-cancun/Blob Transactions On Block 1|Blob Transaction Ordering, Single|Blob Transaction Ordering, Multiple Accounts|Replace Blob Transactions|Parallel Blob Transactions|ForkchoiceUpdatedV3|ForkchoiceUpdatedV2|ForkchoiceUpdated Version|GetPayload|NewPayloadV3 After Cancun|NewPayloadV3 Before Cancun|NewPayloadV3 Versioned Hashes|Incorrect BlobGasUsed|ParentHash equals BlockHash|RPC:|in ForkchoiceState|Unknown SafeBlockHash|Unknown FinalizedBlockHash|Unique|Re-Execute Payload|Multiple New Payloads|NewPayload with|Build Payload with|Re-org to Previously|Safe Re-Org to Side Chain|Transaction Re-Org|Re-Org Back into Canonical Chain, Depth=5|Suggested Fee Recipient Test|PrevRandao Opcode|Fork ID: *"
+            test_pattern: "engine-cancun/Blob Transactions On Block 1|Blob Transaction Ordering, Single|Blob Transaction Ordering, Multiple Accounts|Replace Blob Transactions|Parallel Blob Transactions|ForkchoiceUpdatedV3|ForkchoiceUpdatedV2|ForkchoiceUpdated Version|GetPayload|NewPayloadV3 After Cancun|NewPayloadV3 Before Cancun|NewPayloadV3 Versioned Hashes|Incorrect BlobGasUsed|ParentHash equals BlockHash|RPC:|in ForkchoiceState|Unknown SafeBlockHash|Unknown FinalizedBlockHash|Unique|Re-Execute Payload|Multiple New Payloads|NewPayload with|Build Payload with|Re-org to Previously|Safe Re-Org to Side Chain|Transaction Re-Org|Re-Org Back into Canonical Chain, Depth=5|Suggested Fee Recipient Test|PrevRandao Opcode|Fork ID: *|Invalid Missing .* .*CanonicalReOrg=False"
           - name: "Paris Engine tests"
             simulation: ethereum/engine
             test_pattern: "engine-api/RPC|Re-org to Previously Validated Sidechain Payload|Re-Org Back into Canonical Chain, Depth=5|Safe Re-Org|Transaction Re-Org|Inconsistent|Suggested Fee|PrevRandao Opcode Transactions|Fork ID|Unknown SafeBlockHash|Unknown FinalizedBlockHash|Unique Payload ID|Re-Execute Payload|Multiple New Payloads|NewPayload with|Payload Build|Build Payload"

crates/blockchain/error.rs

@@ -1,4 +1,4 @@
-use ethrex_common::types::{BlobsBundleError, InvalidBlockHeaderError};
+use ethrex_common::types::{BlobsBundleError, BlockHash, InvalidBlockHeaderError};
 use ethrex_storage::error::StoreError;
 use ethrex_vm::EvmError;
 
@@ -101,4 +101,6 @@ pub enum InvalidForkChoice {
     Disconnected(ForkChoiceElement, ForkChoiceElement),
     #[error("Requested head is an invalid block.")]
     InvalidHead,
+    #[error("Previously rejected block.")]
+    InvalidAncestor(BlockHash),
 }

crates/blockchain/fork_choice.rs

@@ -28,9 +28,6 @@ pub fn apply_fork_choice(
         return Err(InvalidForkChoice::InvalidHeadHash);
     }
 
-    // We get the block bodies even if we only use headers them so we check that they are
-    // stored too.
-
     let finalized_res = if !finalized_hash.is_zero() {
         store.get_block_by_hash(finalized_hash)?
     } else {

crates/networking/p2p/sync.rs

@@ -16,7 +16,7 @@ use ethrex_storage::{error::StoreError, Store, STATE_TRIE_SEGMENTS};
 use ethrex_trie::{Nibbles, Node, TrieError, TrieState};
 use state_healing::heal_state_trie;
 use state_sync::state_sync;
-use std::{array, sync::Arc};
+use std::{array, collections::HashMap, sync::Arc};
 use storage_healing::storage_healer;
 use tokio::{
     sync::{
@@ -26,7 +26,7 @@ use tokio::{
     time::{Duration, Instant},
 };
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, info, warn};
+use tracing::{debug, error, info, warn};
 use trie_rebuild::TrieRebuilder;
 
 use crate::{
@@ -78,6 +78,14 @@ pub struct SyncManager {
     /// Syncing beyond this pivot should re-enable snap-sync (as we will not have that state stored)
     /// TODO: Reorgs
     last_snap_pivot: u64,
+    /// The `forkchoice_update` and `new_payload` methods require the `latest_valid_hash`
+    /// when processing an invalid payload. To provide this, we must track invalid chains.
+    ///
+    /// We only store the last known valid head upon encountering a bad block,
+    /// rather than tracking every subsequent invalid block.
+    ///
+    /// This map stores the bad block hash with and latest valid block hash of the chain corresponding to the bad block
+    pub invalid_ancestors: HashMap<BlockHash, BlockHash>,
     trie_rebuilder: Option<TrieRebuilder>,
     // Used for cancelling long-living tasks upon shutdown
     cancel_token: CancellationToken,
@@ -93,6 +101,7 @@ impl SyncManager {
             sync_mode,
             peers: PeerHandler::new(peer_table),
             last_snap_pivot: 0,
+            invalid_ancestors: HashMap::new(),
             trie_rebuilder: None,
             cancel_token,
         }
@@ -106,6 +115,7 @@ impl SyncManager {
             sync_mode: SyncMode::Full,
             peers: PeerHandler::new(dummy_peer_table),
             last_snap_pivot: 0,
+            invalid_ancestors: HashMap::new(),
             trie_rebuilder: None,
             // This won't be used
             cancel_token: CancellationToken::new(),
@@ -155,6 +165,12 @@ impl SyncManager {
                 current_head = last_header;
             }
         }
+
+        let pending_block = match store.get_pending_block(sync_head) {
+            Ok(res) => res,
+            Err(e) => return Err(e.into()),
+        };
+
         loop {
             debug!("Requesting Block Headers from {current_head}");
             // Request Block Headers from Peer
@@ -167,12 +183,23 @@ impl SyncManager {
                     debug!(
                         "Received {} block headers| Last Number: {}",
                         block_headers.len(),
-                        block_headers.last().as_ref().unwrap().number
+                        block_headers.last().as_ref().unwrap().number,
                     );
                     let mut block_hashes = block_headers
                         .iter()
                         .map(|header| header.compute_block_hash())
                         .collect::<Vec<_>>();
+                    let last_header = block_headers.last().unwrap().clone();
+
+                    // If we have a pending block from new_payload request
+                    // attach it to the end if it matches the parent_hash of the latest received header
+                    if let Some(ref block) = pending_block {
+                        if block.header.parent_hash == last_header.compute_block_hash() {
+                            block_hashes.push(block.hash());
+                            block_headers.push(block.header.clone());
+                        }
+                    }
+
                     // Check if we already found the sync head
                     let sync_head_found = block_hashes.contains(&sync_head);
                     // Update current fetch head if needed
@@ -185,9 +212,8 @@ impl SyncManager {
                             store.set_header_download_checkpoint(current_head)?;
                         } else {
                             // If the sync head is less than 64 blocks away from our current head switch to full-sync
-                            let last_header_number = block_headers.last().unwrap().number;
                             let latest_block_number = store.get_latest_block_number()?;
-                            if last_header_number.saturating_sub(latest_block_number)
+                            if last_header.number.saturating_sub(latest_block_number)
                                 < MIN_FULL_BLOCKS as u64
                             {
                                 // Too few blocks for a snap sync, switching to full sync
@@ -262,7 +288,13 @@ impl SyncManager {
             }
             SyncMode::Full => {
                 // full-sync: Fetch all block bodies and execute them sequentially to build the state
-                download_and_run_blocks(all_block_hashes, self.peers.clone(), store.clone()).await?
+                download_and_run_blocks(
+                    all_block_hashes,
+                    self.peers.clone(),
+                    store.clone(),
+                    &mut self.invalid_ancestors,
+                )
+                .await?
             }
         }
         Ok(())
@@ -275,7 +307,9 @@ async fn download_and_run_blocks(
     mut block_hashes: Vec<BlockHash>,
     peers: PeerHandler,
     store: Store,
+    invalid_ancestors: &mut HashMap<BlockHash, BlockHash>,
 ) -> Result<(), SyncError> {
+    let mut last_valid_hash = H256::default();
     loop {
         debug!("Requesting Block Bodies ");
         if let Some(block_bodies) = peers.request_block_bodies(block_hashes.clone()).await {
@@ -292,11 +326,12 @@ async fn download_and_run_blocks(
                 let number = header.number;
                 let block = Block::new(header, body);
                 if let Err(error) = ethrex_blockchain::add_block(&block, &store) {
-                    warn!("Failed to add block during FullSync: {error}");
+                    invalid_ancestors.insert(hash, last_valid_hash);
                     return Err(error.into());
                 }
                 store.set_canonical_block(number, hash)?;
                 store.update_latest_block_number(number)?;
+                last_valid_hash = hash;
             }
             info!("Executed & stored {} blocks", block_bodies_len);
             // Check if we need to ask for another batch

crates/networking/rpc/engine/fork_choice.rs

@@ -173,12 +173,28 @@ fn handle_forkchoice(
     // Check if there is an ongoing sync before applying the forkchoice
     let fork_choice_res = match context.sync_status()? {
         // Apply current fork choice
-        SyncStatus::Inactive => apply_fork_choice(
-            &context.storage,
-            fork_choice_state.head_block_hash,
-            fork_choice_state.safe_block_hash,
-            fork_choice_state.finalized_block_hash,
-        ),
+        SyncStatus::Inactive => {
+            let invalid_ancestors = {
+                let lock = context.syncer.try_lock();
+                match lock {
+                    Ok(sync) => sync.invalid_ancestors.clone(),
+                    Err(_) => return Err(RpcErr::Internal("Internal error".into())),
+                }
+            };
+
+            // Check if the block has already been invalidated
+            match invalid_ancestors.get(&fork_choice_state.head_block_hash) {
+                Some(latest_valid_hash) => {
+                    Err(InvalidForkChoice::InvalidAncestor(*latest_valid_hash))
+                }
+                None => apply_fork_choice(
+                    &context.storage,
+                    fork_choice_state.head_block_hash,
+                    fork_choice_state.safe_block_hash,
+                    fork_choice_state.finalized_block_hash,
+                ),
+            }
+        }
         // Restart sync if needed
         _ => Err(InvalidForkChoice::Syncing),
     };
@@ -203,14 +219,9 @@ fn handle_forkchoice(
                         .storage
                         .update_sync_status(false)
                         .map_err(|e| RpcErr::Internal(e.to_string()))?;
-                    let current_number = context.storage.get_latest_block_number()?;
-                    let Some(current_head) =
-                        context.storage.get_canonical_block_hash(current_number)?
-                    else {
-                        return Err(RpcErr::Internal(
-                            "Missing latest canonical block".to_owned(),
-                        ));
-                    };
+                    let current_head = context.storage.get_latest_canonical_block_hash()?.ok_or(
+                        RpcErr::Internal("Missing latest canonical block".to_owned()),
+                    )?;
                     let sync_head = fork_choice_state.head_block_hash;
                     tokio::spawn(async move {
                         // If we can't get hold of the syncer, then it means that there is an active sync in process
@@ -222,9 +233,29 @@ fn handle_forkchoice(
                     });
                     ForkChoiceResponse::from(PayloadStatus::syncing())
                 }
+                InvalidForkChoice::Disconnected(_, _) | InvalidForkChoice::ElementNotFound(_) => {
+                    warn!("Invalid fork choice state. Reason: {:?}", forkchoice_error);
+                    return Err(RpcErr::InvalidForkChoiceState(forkchoice_error.to_string()));
+                }
+                InvalidForkChoice::InvalidAncestor(last_valid_hash) => {
+                    ForkChoiceResponse::from(PayloadStatus::invalid_with(
+                        last_valid_hash,
+                        InvalidForkChoice::InvalidAncestor(last_valid_hash).to_string(),
+                    ))
+                }
                 reason => {
-                    warn!("Invalid fork choice state. Reason: {:#?}", reason);
-                    return Err(RpcErr::InvalidForkChoiceState(reason.to_string()));
+                    warn!(
+                        "Invalid fork choice payload. Reason: {}",
+                        reason.to_string()
+                    );
+                    let latest_valid_hash =
+                        context.storage.get_latest_canonical_block_hash()?.ok_or(
+                            RpcErr::Internal("Missing latest canonical block".to_owned()),
+                        )?;
+                    ForkChoiceResponse::from(PayloadStatus::invalid_with(
+                        latest_valid_hash,
+                        reason.to_string(),
+                    ))
                 }
             };
             Ok((None, forkchoice_response))

crates/networking/rpc/engine/payload.rs

@@ -354,7 +354,17 @@ fn handle_new_payload_v1_v2(
     let payload_status = match context.sync_status()? {
         SyncStatus::Active | SyncStatus::Pending => PayloadStatus::syncing(),
         SyncStatus::Inactive => {
-            if let Err(RpcErr::Internal(error_msg)) = validate_block_hash(payload, &block) {
+            // Check if the block has already been invalidated
+            let invalid_ancestors = match context.syncer.try_lock() {
+                Ok(syncer) => syncer.invalid_ancestors.clone(),
+                Err(_) => return Err(RpcErr::Internal("Internal error".into())),
+            };
+            if let Some(latest_valid_hash) = invalid_ancestors.get(&block.hash()) {
+                PayloadStatus::invalid_with(
+                    *latest_valid_hash,
+                    "Header has been previously invalidated.".into(),
+                )
+            } else if let Err(RpcErr::Internal(error_msg)) = validate_block_hash(payload, &block) {
                 PayloadStatus::invalid_with_err(&error_msg)
             } else {
                 execute_payload(&block, &context)?
@@ -393,12 +403,30 @@ fn execute_payload(block: &Block, context: &RpcApiContext) -> Result<PayloadStat
     let block_hash = block.hash();
     let storage = &context.storage;
     // Return the valid message directly if we have it.
-    if storage.get_block_header_by_hash(block_hash)?.is_some() {
+    if storage.get_block_by_hash(block_hash)?.is_some() {
         return Ok(PayloadStatus::valid_with_hash(block_hash));
     }
 
     // Execute and store the block
     info!("Executing payload with block hash: {block_hash:#x}");
+    let latest_valid_hash =
+        context
+            .storage
+            .get_latest_canonical_block_hash()?
+            .ok_or(RpcErr::Internal(
+                "Missing latest canonical block".to_owned(),
+            ))?;
+
+    // adds a bad block as a bad ancestor so we can catch it on fork_choice as well
+    let add_block_to_invalid_ancestor = || {
+        let lock = context.syncer.try_lock();
+        if let Ok(mut syncer) = lock {
+            syncer
+                .invalid_ancestors
+                .insert(block_hash, latest_valid_hash);
+        };
+    };
+
     match add_block(block, storage) {
         Err(ChainError::ParentNotFound) => Ok(PayloadStatus::syncing()),
         // Under the current implementation this is not possible: we always calculate the state
@@ -412,16 +440,17 @@ fn execute_payload(block: &Block, context: &RpcApiContext) -> Result<PayloadStat
         }
         Err(ChainError::InvalidBlock(error)) => {
             warn!("Error adding block: {error}");
-            // TODO(#982): this is only valid for the cases where the parent was found, but fully invalid ones may also happen.
+            add_block_to_invalid_ancestor();
             Ok(PayloadStatus::invalid_with(
-                block.header.parent_hash,
+                latest_valid_hash,
                 error.to_string(),
             ))
         }
         Err(ChainError::EvmError(error)) => {
             warn!("Error executing block: {error}");
+            add_block_to_invalid_ancestor();
             Ok(PayloadStatus::invalid_with(
-                block.header.parent_hash,
+                latest_valid_hash,
                 error.to_string(),
             ))
         }

crates/networking/rpc/rpc.rs

@@ -82,6 +82,7 @@ pub struct RpcApiContext {
 /// Inactive: There is no active sync process
 /// Active: The client is currently syncing
 /// Pending: The previous sync process became stale, awaiting restart
+#[derive(Debug)]
 pub enum SyncStatus {
     Inactive,
     Active,

crates/storage/store/storage.rs

@@ -733,6 +733,14 @@ impl Store {
         self.engine.get_canonical_block_hash(block_number)
     }
 
+    pub fn get_latest_canonical_block_hash(&self) -> Result<Option<BlockHash>, StoreError> {
+        let latest_block_number = match self.engine.get_latest_block_number() {
+            Ok(n) => n.ok_or(StoreError::MissingLatestBlockNumber)?,
+            Err(e) => return Err(e),
+        };
+        self.get_canonical_block_hash(latest_block_number)
+    }
+
     /// Marks a block number as not having any canonical blocks associated with it.
     /// Used for reorgs.
     /// Note: Should we also remove all others up to the head here?