Merge branch 'main' into refactor/use-genserver-instead-sync-threads

Author: ElFantasma

.github/workflows/pr-main_l1.yaml

@@ -176,6 +176,8 @@ jobs:
     runs-on: ubuntu-latest
     needs: [detect-changes, docker_build]
     if: ${{ needs.detect-changes.outputs.run_tests == 'true' && github.event_name != 'merge_group' }}
+    env:
+      DOCKERHUB_TOKEN: ${{ secrets.DOCKERHUB_TOKEN }}
     strategy:
       fail-fast: false
       matrix:
@@ -252,10 +254,11 @@ jobs:
           echo "flags=$FLAGS" >> "$GITHUB_OUTPUT"
 
       - name: Log in to the Container registry
+        if: ${{ env.DOCKERHUB_TOKEN != '' }}
         uses: docker/login-action@v3
         with:
           username: ${{ vars.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          password: ${{ env.DOCKERHUB_TOKEN }}
 
       - name: Run Hive Simulation
         id: run-hive-action

crates/blockchain/error.rs

@@ -175,4 +175,8 @@ pub enum InvalidForkChoice {
     InvalidAncestor(BlockHash),
     #[error("Cannot find link between Head and the canonical chain")]
     UnlinkedHead,
+
+    // TODO(#5564): handle arbitrary reorgs
+    #[error("State root of the new head is not reachable from the database")]
+    StateNotReachable,
 }

crates/blockchain/fork_choice.rs

@@ -4,6 +4,7 @@ use ethrex_common::{
 };
 use ethrex_metrics::metrics;
 use ethrex_storage::{Store, error::StoreError};
+use tracing::{error, warn};
 
 use crate::{
     error::{self, InvalidForkChoice},
@@ -67,9 +68,9 @@ pub async fn apply_fork_choice(
         return Err(InvalidForkChoice::UnlinkedHead);
     };
 
-    let link_block_number = match new_canonical_blocks.last() {
-        Some((number, _)) => *number,
-        None => head.number,
+    let (link_block_number, link_block_hash) = match new_canonical_blocks.last() {
+        Some((number, hash)) => (*number, *hash),
+        None => (head.number, head_hash),
     };
 
     // Check that finalized and safe blocks are part of the new canonical chain.
@@ -97,6 +98,24 @@ pub async fn apply_fork_choice(
         ));
     }
 
+    let Some(link_header) = store.get_block_header_by_hash(link_block_hash)? else {
+        // Probably unreachable, but we return this error just in case.
+        error!("Link block not found although it was just retrieved from the DB");
+        return Err(InvalidForkChoice::UnlinkedHead);
+    };
+
+    // If the state can't be constructed from the DB, we ignore it and log a warning.
+    // TODO(#5564): handle arbitrary reorgs
+    if !store.has_state_root(link_header.state_root)? {
+        warn!(
+            link_block=%link_block_hash,
+            link_number=%link_header.number,
+            head_number=%head.number,
+            "FCU head state not reachable from DB state. Ignoring fork choice update. This is expected if the consensus client is currently syncing. Otherwise, if consensus is synced and this is a consistent message it can be fixed by removing the DB and re-syncing the execution client."
+        );
+        return Err(InvalidForkChoice::StateNotReachable);
+    }
+
     // Finished all validations.
 
     store

crates/networking/p2p/discv4/server.rs

@@ -182,7 +182,8 @@ impl DiscoveryServer {
                     return Ok(());
                 }
 
-                self.handle_find_node(sender_public_key, from).await?;
+                self.handle_find_node(sender_public_key, find_node_message.target, from)
+                    .await?;
             }
             Message::Neighbors(neighbors_message) => {
                 trace!(received = "Neighbors", msg = ?neighbors_message, from = %format!("{sender_public_key:#x}"));
@@ -425,14 +426,18 @@ impl DiscoveryServer {
     async fn handle_find_node(
         &mut self,
         sender_public_key: H512,
+        target: H512,
         from: SocketAddr,
     ) -> Result<(), DiscoveryServerError> {
-        let node_id = node_id(&sender_public_key);
+        let sender_id = node_id(&sender_public_key);
         if let Ok(contact) = self
-            .validate_contact(sender_public_key, node_id, from, "FindNode")
+            .validate_contact(sender_public_key, sender_id, from, "FindNode")
             .await
         {
-            let neighbors = self.peer_table.get_closest_nodes(&node_id).await?;
+            // According to https://github.com/ethereum/devp2p/blob/master/discv4.md#findnode-packet-0x03
+            // reply closest 16 nodes to target
+            let target_id = node_id(&target);
+            let neighbors = self.peer_table.get_closest_nodes(&target_id).await?;
 
             // A single node encodes to at most 89B, so 8 of them are at most 712B plus
             // recursive length and expiration time, well within bound of 1280B per packet.

crates/networking/rpc/engine/fork_choice.rs

@@ -220,6 +220,10 @@ async fn handle_forkchoice(
         ));
     }
 
+    // Ignore any FCU during snap-sync.
+    // Processing the FCU while snap-syncing can result in reading inconsistent data
+    // from the DB, and the later head update can overwrite changes made by the syncer
+    // process, corrupting the forkchoice state (see #5547)
     if syncer.sync_mode() == SyncMode::Snap {
         syncer.sync_to_head(fork_choice_state.head_block_hash);
         return Ok((None, PayloadStatus::syncing().into()));
@@ -275,6 +279,11 @@ async fn handle_forkchoice(
                     syncer.sync_to_head(fork_choice_state.head_block_hash);
                     ForkChoiceResponse::from(PayloadStatus::syncing())
                 }
+                // TODO(#5564): handle arbitrary reorgs
+                InvalidForkChoice::StateNotReachable => {
+                    // Ignore the FCU
+                    ForkChoiceResponse::from(PayloadStatus::syncing())
+                }
                 InvalidForkChoice::Disconnected(_, _) | InvalidForkChoice::ElementNotFound(_) => {
                     warn!("Invalid fork choice state. Reason: {:?}", forkchoice_error);
                     return Err(RpcErr::InvalidForkChoiceState(forkchoice_error.to_string()));