fix(l1): validate incoming payloads even when the node is syncing. (#2426)

**Motivation**
We should be able to do payload validations even when the node is in a
sync process (except if it's snap sync).

**Description**
- Refactored some code to make it flatter
- Removed early return when the node is syncing
- minor renames for clarity sake.

Author: mpaulucci

.github/workflows/pr-main_l1.yaml

@@ -160,11 +160,11 @@ jobs:
             ethrex_flags: ""
           - 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: *|Invalid Missing Ancestor Syncing ReOrg|Request Blob Pooled Transactions"
+            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 Ancestor Syncing ReOrg|Request Blob Pooled Transactions|Invalid NewPayload, Incomplete Transactions|Invalid P9|Invalid P10|Re-Org Back to Canonical Chain*|Invalid PayloadAttributes*|Invalid NewPayload, VersionedHashes|Invalid NewPayload, Incomplete VersionedHashes|Invalid NewPayload, Extra VersionedHashes|Bad Hash on NewPayload|Unknown HeadBlockHash|In-Order Consecutive Payload Execution|Valid NewPayload->ForkchoiceUpdated|Invalid NewPayload, ParentHash|Syncing=False|Payload Build after New Invalid Payload"
             ethrex_flags: ""
           - 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"
+            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|Invalid PayloadAttributes|Unknown HeadBlockHash|Valid NewPayload->ForkchoiceUpdated|Invalid NewPayload, ParentHash|Bad Hash on NewPayload|Invalid P9|In-Order Consecutive Payload Execution|Re-Org Back to Canonical Chain"
             ethrex_flags: ""
           - name: "Engine withdrawal tests"
             simulation: ethereum/engine

crates/networking/p2p/sync_manager.rs

@@ -18,6 +18,7 @@ use crate::{
     sync::{SyncMode, Syncer},
 };
 
+#[derive(Debug)]
 pub enum SyncStatus {
     Active(SyncMode),
     Inactive,
@@ -62,7 +63,7 @@ impl SyncManager {
             .await
             .is_ok_and(|res| res.is_some())
         {
-            sync_manager.start_sync().await;
+            sync_manager.start_sync();
         }
         sync_manager
     }
@@ -100,15 +101,17 @@ impl SyncManager {
     /// Attempts to sync to the last received fcu head
     /// Will do nothing if the syncer is already involved in a sync process
     /// If the sync process would require multiple sync cycles (such as snap sync), starts all required sync cycles until the sync is complete
-    pub async fn start_sync(&self) {
+    pub fn start_sync(&self) {
         let syncer = self.syncer.clone();
         let store = self.store.clone();
-        let Ok(Some(current_head)) = self.store.get_latest_canonical_block_hash().await else {
-            tracing::error!("Failed to fecth latest canonical block, unable to sync");
-            return;
-        };
         let sync_head = self.last_fcu_head.clone();
+
         tokio::spawn(async move {
+            let Ok(Some(current_head)) = store.get_latest_canonical_block_hash().await else {
+                tracing::error!("Failed to fecth latest canonical block, unable to sync");
+                return;
+            };
+
             // If we can't get hold of the syncer, then it means that there is an active sync in process
             let Ok(mut syncer) = syncer.try_lock() else {
                 return;
@@ -147,7 +150,7 @@ impl SyncManager {
     }
 
     /// Returns the syncer's current syncmode (either snap or full)
-    fn sync_mode(&self) -> SyncMode {
+    pub fn sync_mode(&self) -> SyncMode {
         if self.snap_enabled.load(Ordering::Relaxed) {
             SyncMode::Snap
         } else {

crates/networking/rpc/engine/fork_choice.rs

@@ -5,7 +5,7 @@ use ethrex_blockchain::{
     payload::{create_payload, BuildPayloadArgs},
 };
 use ethrex_common::types::BlockHeader;
-use ethrex_p2p::sync_manager::SyncStatus;
+use ethrex_p2p::sync::SyncMode;
 use serde_json::Value;
 use tracing::{debug, info, warn};
 
@@ -211,63 +211,53 @@ async fn handle_forkchoice(
         fork_choice_state.finalized_block_hash
     );
 
-    // Update fcu head in syncer
-    context.syncer.set_head(fork_choice_state.head_block_hash);
-
-    let fork_choice_res = if let Some(latest_valid_hash) = context
+    if let Some(latest_valid_hash) = context
         .storage
         .get_latest_valid_ancestor(fork_choice_state.head_block_hash)
         .await?
     {
-        warn!(
-            "Invalid fork choice state. Reason: Invalid ancestor {:#x}",
-            latest_valid_hash
-        );
-        Err(InvalidForkChoice::InvalidAncestor(latest_valid_hash))
-    } else {
-        // Check parent block hash in invalid_ancestors (if head block exists)
-        let head_block = context
-            .storage
-            .get_block_header_by_hash(fork_choice_state.head_block_hash)?;
-        let check_parent = if let Some(head_block) = head_block {
-            debug!(
-                "Checking parent for invalid ancestor {}",
-                head_block.parent_hash
-            );
-            context
-                .storage
-                .get_latest_valid_ancestor(head_block.parent_hash)
-                .await
-                .ok()
-                .flatten()
-        } else {
-            None
-        };
+        return Ok((
+            None,
+            ForkChoiceResponse::from(PayloadStatus::invalid_with(
+                latest_valid_hash,
+                InvalidForkChoice::InvalidAncestor(latest_valid_hash).to_string(),
+            )),
+        ));
+    }
 
-        // Check head block hash in invalid_ancestors
-        if let Some(latest_valid_hash) = check_parent {
-            Err(InvalidForkChoice::InvalidAncestor(latest_valid_hash))
-        } else {
-            // Check if there is an ongoing sync before applying the forkchoice
-            match context.syncer.status()? {
-                // Apply current fork choice
-                SyncStatus::Inactive => {
-                    // All checks passed, apply fork choice
-                    apply_fork_choice(
-                        &context.storage,
-                        fork_choice_state.head_block_hash,
-                        fork_choice_state.safe_block_hash,
-                        fork_choice_state.finalized_block_hash,
-                    )
-                    .await
-                }
-                // Restart sync if needed
-                _ => Err(InvalidForkChoice::Syncing),
-            }
+    // Check parent block hash in invalid_ancestors (if head block exists)
+    if let Some(head_block) = context
+        .storage
+        .get_block_header_by_hash(fork_choice_state.head_block_hash)?
+    {
+        if let Some(latest_valid_hash) = context
+            .storage
+            .get_latest_valid_ancestor(head_block.parent_hash)
+            .await?
+        {
+            return Ok((
+                None,
+                ForkChoiceResponse::from(PayloadStatus::invalid_with(
+                    latest_valid_hash,
+                    InvalidForkChoice::InvalidAncestor(latest_valid_hash).to_string(),
+                )),
+            ));
         }
-    };
+    }
 
-    match fork_choice_res {
+    if context.syncer.sync_mode() == SyncMode::Snap {
+        context.syncer.set_head(fork_choice_state.head_block_hash);
+        return Ok((None, PayloadStatus::syncing().into()));
+    }
+
+    match apply_fork_choice(
+        &context.storage,
+        fork_choice_state.head_block_hash,
+        fork_choice_state.safe_block_hash,
+        fork_choice_state.finalized_block_hash,
+    )
+    .await
+    {
         Ok(head) => {
             // Remove included transactions from the mempool after we accept the fork choice
             // TODO(#797): The remove of transactions from the mempool could be incomplete (i.e. REORGS)
@@ -318,7 +308,8 @@ async fn handle_forkchoice(
                         .update_sync_status(false)
                         .await
                         .map_err(|e| RpcErr::Internal(e.to_string()))?;
-                    context.syncer.start_sync().await;
+                    context.syncer.set_head(fork_choice_state.head_block_hash);
+                    context.syncer.start_sync();
                     ForkChoiceResponse::from(PayloadStatus::syncing())
                 }
                 InvalidForkChoice::Disconnected(_, _) | InvalidForkChoice::ElementNotFound(_) => {

crates/networking/rpc/engine/payload.rs

@@ -4,9 +4,9 @@ use ethrex_common::types::payload::PayloadBundle;
 use ethrex_common::types::requests::{compute_requests_hash, EncodedRequests};
 use ethrex_common::types::{Block, BlockBody, BlockHash, BlockNumber, Fork};
 use ethrex_common::{H256, U256};
-use ethrex_p2p::sync_manager::SyncStatus;
+use ethrex_p2p::sync::SyncMode;
 use serde_json::Value;
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, warn};
 
 use crate::rpc::{RpcApiContext, RpcHandler};
 use crate::types::payload::{
@@ -34,7 +34,9 @@ impl RpcHandler for NewPayloadV1Request {
 
     async fn handle(&self, context: RpcApiContext) -> Result<Value, RpcErr> {
         validate_execution_payload_v1(&self.payload)?;
-        handle_new_payload_v1_v2(&self.payload, context).await
+        let block = get_block_from_payload(&self.payload, None, None)?;
+        let payload_status = handle_new_payload_v1_v2(&self.payload, block, context).await?;
+        serde_json::to_value(payload_status).map_err(|error| RpcErr::Internal(error.to_string()))
     }
 }
 
@@ -58,7 +60,9 @@ impl RpcHandler for NewPayloadV2Request {
             validate_execution_payload_v1(&self.payload)?;
         }
 
-        handle_new_payload_v1_v2(&self.payload, context).await
+        let block = get_block_from_payload(&self.payload, None, None)?;
+        let payload_status = handle_new_payload_v1_v2(&self.payload, block, context).await?;
+        serde_json::to_value(payload_status).map_err(|error| RpcErr::Internal(error.to_string()))
     }
 }
 
@@ -572,36 +576,29 @@ async fn validate_ancestors(
     Ok(None)
 }
 
-// TODO: We need to check why we return a Result<Value, RpcErr> here instead of a Result<PayloadStatus, RpcErr> as in v3.
 async fn handle_new_payload_v1_v2(
     payload: &ExecutionPayload,
+    block: Block,
     context: RpcApiContext,
-) -> Result<Value, RpcErr> {
-    let block = get_block_from_payload(payload, None, None)?;
-
-    // Check sync status
-    match context.syncer.status()? {
-        SyncStatus::Active(_) => {
-            return serde_json::to_value(PayloadStatus::syncing())
-                .map_err(|error| RpcErr::Internal(error.to_string()));
-        }
-        SyncStatus::Inactive => {}
-    }
-
+) -> Result<PayloadStatus, RpcErr> {
     // Validate block hash
     if let Err(RpcErr::Internal(error_msg)) = validate_block_hash(payload, &block) {
-        return serde_json::to_value(PayloadStatus::invalid_with_err(&error_msg))
-            .map_err(|error| RpcErr::Internal(error.to_string()));
+        return Ok(PayloadStatus::invalid_with_err(&error_msg));
     }
 
     // Check for invalid ancestors
     if let Some(status) = validate_ancestors(&block, &context).await? {
-        return serde_json::to_value(status).map_err(|error| RpcErr::Internal(error.to_string()));
+        return Ok(status);
+    }
+
+    if context.syncer.sync_mode() == SyncMode::Snap {
+        warn!("Snap sync in progress, skipping new payload validation");
+        return Ok(PayloadStatus::syncing());
     }
 
     // All checks passed, execute payload
-    let payload_status = execute_payload(&block, &context).await?;
-    serde_json::to_value(payload_status).map_err(|error| RpcErr::Internal(error.to_string()))
+    let payload_status = try_execute_payload(&block, &context).await?;
+    Ok(payload_status)
 }
 
 async fn handle_new_payload_v3(
@@ -610,23 +607,6 @@ async fn handle_new_payload_v3(
     block: Block,
     expected_blob_versioned_hashes: Vec<H256>,
 ) -> Result<PayloadStatus, RpcErr> {
-    // Ignore incoming
-    // Check sync status
-    match context.syncer.status()? {
-        SyncStatus::Active(_) => return Ok(PayloadStatus::syncing()),
-        SyncStatus::Inactive => {}
-    }
-
-    // Validate block hash
-    if let Err(RpcErr::Internal(error_msg)) = validate_block_hash(payload, &block) {
-        return Ok(PayloadStatus::invalid_with_err(&error_msg));
-    }
-
-    // Check for invalid ancestors
-    if let Some(status) = validate_ancestors(&block, &context).await? {
-        return Ok(status);
-    }
-
     // V3 specific: validate blob hashes
     let blob_versioned_hashes: Vec<H256> = block
         .body
@@ -641,8 +621,7 @@ async fn handle_new_payload_v3(
         ));
     }
 
-    // All checks passed, execute payload
-    execute_payload(&block, &context).await
+    handle_new_payload_v1_v2(payload, block, context).await
 }
 
 // Elements of the list MUST be ordered by request_type in ascending order.
@@ -684,11 +663,13 @@ fn validate_block_hash(payload: &ExecutionPayload, block: &Block) -> Result<(),
             "Invalid block hash. Expected {actual_block_hash:#x}, got {block_hash:#x}"
         )));
     }
-    debug!("Block hash {block_hash} is valid");
     Ok(())
 }
 
-async fn execute_payload(block: &Block, context: &RpcApiContext) -> Result<PayloadStatus, RpcErr> {
+async fn try_execute_payload(
+    block: &Block,
+    context: &RpcApiContext,
+) -> Result<PayloadStatus, RpcErr> {
     let block_hash = block.hash();
     let storage = &context.storage;
     // Return the valid message directly if we have it.
@@ -698,6 +679,9 @@ async fn execute_payload(block: &Block, context: &RpcApiContext) -> Result<Paylo
 
     // Execute and store the block
     info!("Executing payload with block hash: {block_hash:#x}");
+
+    // TODO: this is not correct, the block being validated it no necesarily a descendant
+    // of the latest canonical block
     let latest_valid_hash = context
         .storage
         .get_latest_canonical_block_hash()
@@ -707,7 +691,18 @@ async fn execute_payload(block: &Block, context: &RpcApiContext) -> Result<Paylo
         ))?;
 
     match context.blockchain.add_block(block).await {
-        Err(ChainError::ParentNotFound) => Ok(PayloadStatus::syncing()),
+        Err(ChainError::ParentNotFound) => {
+            // Start sync
+            context
+                .storage
+                .update_sync_status(false)
+                .await
+                .map_err(|e| RpcErr::Internal(e.to_string()))?;
+            context.syncer.set_head(block_hash);
+            context.syncer.start_sync();
+
+            Ok(PayloadStatus::syncing())
+        }
         // Under the current implementation this is not possible: we always calculate the state
         // transition of any new payload as long as the parent is present. If we received the
         // parent payload but it was stashed, then new payload would stash this one too, with a

crates/storage/api.rs

@@ -318,7 +318,7 @@ pub trait StoreEngine: Debug + Send + Sync + RefUnwindSafe {
 
     async fn is_synced(&self) -> Result<bool, StoreError>;
 
-    async fn update_sync_status(&self, status: bool) -> Result<(), StoreError>;
+    async fn update_sync_status(&self, is_synced: bool) -> Result<(), StoreError>;
 
     /// Write an account batch into the current state snapshot
     async fn write_snapshot_account_batch(

crates/storage/store.rs

@@ -1006,8 +1006,8 @@ impl Store {
     pub async fn is_synced(&self) -> Result<bool, StoreError> {
         self.engine.is_synced().await
     }
-    pub async fn update_sync_status(&self, status: bool) -> Result<(), StoreError> {
-        self.engine.update_sync_status(status).await
+    pub async fn update_sync_status(&self, is_synced: bool) -> Result<(), StoreError> {
+        self.engine.update_sync_status(is_synced).await
     }
 
     /// Write an account batch into the current state snapshot

crates/storage/store_db/in_memory.rs

@@ -557,8 +557,8 @@ impl StoreEngine for Store {
         Ok(self.inner().chain_data.is_synced)
     }
 
-    async fn update_sync_status(&self, status: bool) -> Result<(), StoreError> {
-        self.inner().chain_data.is_synced = status;
+    async fn update_sync_status(&self, is_synced: bool) -> Result<(), StoreError> {
+        self.inner().chain_data.is_synced = is_synced;
         Ok(())
     }
 

crates/storage/store_db/libmdbx.rs

@@ -768,8 +768,8 @@ impl StoreEngine for Store {
         }
     }
 
-    async fn update_sync_status(&self, status: bool) -> Result<(), StoreError> {
-        self.write::<ChainData>(ChainDataIndex::IsSynced, status.encode_to_vec())
+    async fn update_sync_status(&self, is_synced: bool) -> Result<(), StoreError> {
+        self.write::<ChainData>(ChainDataIndex::IsSynced, is_synced.encode_to_vec())
             .await
     }