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

crates/networking/p2p/sync_manager.rs

@@ -15,6 +15,7 @@ use crate::{
     sync::{SyncMode, Syncer},
 };
 
+#[derive(Debug)]
 pub enum SyncStatus {
     Active(SyncMode),
     Inactive,

crates/networking/rpc/engine/fork_choice.rs

@@ -5,7 +5,6 @@ use ethrex_blockchain::{
     payload::{create_payload, BuildPayloadArgs},
 };
 use ethrex_common::types::BlockHeader;
-use ethrex_p2p::sync_manager::SyncStatus;
 use serde_json::Value;
 use tracing::{debug, info, warn};
 
@@ -211,58 +210,46 @@ 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)?
     {
-        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 check_parent = context
-            .storage
-            .get_block_header_by_hash(fork_choice_state.head_block_hash)?
-            .and_then(|head_block| {
-                debug!(
-                    "Checking parent for invalid ancestor {}",
-                    head_block.parent_hash
-                );
-                context
-                    .storage
-                    .get_latest_valid_ancestor(head_block.parent_hash)
-                    .ok()?
-            });
+        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)?
+        {
+            return Ok((
+                None,
+                ForkChoiceResponse::from(PayloadStatus::invalid_with(
+                    latest_valid_hash,
+                    InvalidForkChoice::InvalidAncestor(latest_valid_hash).to_string(),
+                )),
+            ));
         }
-    };
+    }
 
-    match fork_choice_res {
+    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)
@@ -307,6 +294,7 @@ async fn handle_forkchoice(
                         .update_sync_status(false)
                         .await
                         .map_err(|e| RpcErr::Internal(e.to_string()))?;
+                    context.syncer.set_head(fork_choice_state.head_block_hash);
                     context.syncer.start_sync();
                     ForkChoiceResponse::from(PayloadStatus::syncing())
                 }

crates/networking/rpc/engine/payload.rs

@@ -583,7 +583,7 @@ async fn handle_new_payload_v1_v2(
     }
 
     // All checks passed, execute payload
-    let payload_status = execute_payload(&block, &context).await?;
+    let payload_status = try_execute_payload(&block, &context).await?;
     serde_json::to_value(payload_status).map_err(|error| RpcErr::Internal(error.to_string()))
 }
 
@@ -593,13 +593,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));
@@ -624,8 +617,7 @@ async fn handle_new_payload_v3(
         ));
     }
 
-    // All checks passed, execute payload
-    execute_payload(&block, &context).await
+    try_execute_payload(&block, &context).await
 }
 
 // Elements of the list MUST be ordered by request_type in ascending order.
@@ -671,7 +663,10 @@ fn validate_block_hash(payload: &ExecutionPayload, block: &Block) -> Result<(),
     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.
@@ -690,7 +685,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

@@ -294,7 +294,7 @@ pub trait StoreEngine: Debug + Send + Sync + RefUnwindSafe {
 
     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

@@ -979,8 +979,8 @@ impl Store {
     pub fn is_synced(&self) -> Result<bool, StoreError> {
         self.engine.is_synced()
     }
-    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

@@ -510,8 +510,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

@@ -659,8 +659,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
     }
 

crates/storage/store_db/redb.rs

@@ -887,11 +887,11 @@ impl StoreEngine for RedBStore {
         }
     }
 
-    async fn update_sync_status(&self, status: bool) -> Result<(), StoreError> {
+    async fn update_sync_status(&self, is_synced: bool) -> Result<(), StoreError> {
         self.write(
             CHAIN_DATA_TABLE,
             ChainDataIndex::IsSynced,
-            status.encode_to_vec(),
+            is_synced.encode_to_vec(),
         )
         .await
     }