fix(l1): fcu not triggering sync if snap is enabled + re-enable snap sync hive test (#2605)

**Motivation**
PR #2426 changed how fork choice & new payload interact with the syncer
and also introduced a bug. If snap sync is enabled, then fork choice
update will never attempt to trigger a sync, so the sync process never
gets started.
This PR fixes the bug and also refactors the sync manager api to better
suit the new use cases
<!-- Why does this pull request exist? What are its goals? -->
* Combine commonly used together `SyncManager` methods `set_head` &
`start_sync` into `sync_to_head`
* Remove unused `SyncManager` method `status` and associated struct
* Make sure sync is triggered during fcu when needed even if snap sync
is enabled
* Re-enable snap sync hive test suite
**Description**

<!-- A clear and concise general description of the changes this PR
introduces -->

<!-- Link to issues: Resolves #111, Resolves #222 -->

Closes #2521

Author: fmoletta

.github/workflows/pr-main_l1.yaml

@@ -169,11 +169,10 @@ jobs:
             simulation: ethereum/sync
             test_pattern: ""
             ethrex_flags: "--syncmode full"
-          # Flaky test, reenable when fixed
-          # - name: "Sync snap"
-          #   simulation: ethereum/sync
-          #   test_pattern: ""
-          #   ethrex_flags: "--syncmode snap"
+          - name: "Sync snap"
+            simulation: ethereum/sync
+            test_pattern: ""
+            ethrex_flags: "--syncmode snap"
     steps:
       - name: Checkout sources
         uses: actions/checkout@v4

crates/networking/p2p/sync_manager.rs

@@ -5,7 +5,7 @@ use std::sync::{
 
 use ethrex_blockchain::Blockchain;
 use ethrex_common::H256;
-use ethrex_storage::{error::StoreError, Store};
+use ethrex_storage::Store;
 use tokio::{
     sync::Mutex,
     time::{sleep, Duration},
@@ -18,12 +18,6 @@ use crate::{
     sync::{SyncMode, Syncer},
 };
 
-#[derive(Debug)]
-pub enum SyncStatus {
-    Active(SyncMode),
-    Inactive,
-}
-
 /// Abstraction to interact with the active sync process without disturbing it
 #[derive(Debug)]
 pub struct SyncManager {
@@ -80,35 +74,48 @@ impl SyncManager {
         }
     }
 
+    /// Sets the latest fcu head and starts the next sync cycle if the syncer is currently inactive
+    pub fn sync_to_head(&self, fcu_head: H256) {
+        self.set_head(fcu_head);
+        if !self.is_active() {
+            self.start_sync();
+        }
+    }
+
+    /// Returns the syncer's current syncmode (either snap or full)
+    pub fn sync_mode(&self) -> SyncMode {
+        if self.snap_enabled.load(Ordering::Relaxed) {
+            SyncMode::Snap
+        } else {
+            SyncMode::Full
+        }
+    }
+
     /// Updates the last fcu head. This may be used on the next sync cycle if needed
-    pub fn set_head(&self, fcu_head: H256) {
+    fn set_head(&self, fcu_head: H256) {
         if let Ok(mut latest_fcu_head) = self.last_fcu_head.try_lock() {
             *latest_fcu_head = fcu_head;
         } else {
             warn!("Failed to update latest fcu head for syncing")
         }
     }
 
-    /// Returns the current sync status, either active or inactive and what the current syncmode is in the case of active
-    pub fn status(&self) -> Result<SyncStatus, StoreError> {
-        Ok(if self.syncer.try_lock().is_err() {
-            SyncStatus::Active(self.sync_mode())
-        } else {
-            SyncStatus::Inactive
-        })
+    /// Returns true is the syncer is active
+    fn is_active(&self) -> bool {
+        self.syncer.try_lock().is_err()
     }
 
     /// 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 fn start_sync(&self) {
+    fn start_sync(&self) {
         let syncer = self.syncer.clone();
         let store = self.store.clone();
         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");
+                tracing::error!("Failed to fetch latest canonical block, unable to sync");
                 return;
             };
 
@@ -148,13 +155,4 @@ impl SyncManager {
             }
         });
     }
-
-    /// Returns the syncer's current syncmode (either snap or full)
-    pub fn sync_mode(&self) -> SyncMode {
-        if self.snap_enabled.load(Ordering::Relaxed) {
-            SyncMode::Snap
-        } else {
-            SyncMode::Full
-        }
-    }
 }

crates/networking/rpc/engine/fork_choice.rs

@@ -250,7 +250,9 @@ async fn handle_forkchoice(
     }
 
     if context.syncer.sync_mode() == SyncMode::Snap {
-        context.syncer.set_head(fork_choice_state.head_block_hash);
+        context
+            .syncer
+            .sync_to_head(fork_choice_state.head_block_hash);
         return Ok((None, PayloadStatus::syncing().into()));
     }
 
@@ -312,8 +314,9 @@ 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();
+                    context
+                        .syncer
+                        .sync_to_head(fork_choice_state.head_block_hash);
                     ForkChoiceResponse::from(PayloadStatus::syncing())
                 }
                 InvalidForkChoice::Disconnected(_, _) | InvalidForkChoice::ElementNotFound(_) => {

crates/networking/rpc/engine/payload.rs

@@ -698,9 +698,7 @@ async fn try_execute_payload(
                 .update_sync_status(false)
                 .await
                 .map_err(|e| RpcErr::Internal(e.to_string()))?;
-            context.syncer.set_head(block_hash);
-            context.syncer.start_sync();
-
+            context.syncer.sync_to_head(block_hash);
             Ok(PayloadStatus::syncing())
         }
         // Under the current implementation this is not possible: we always calculate the state