feat(l1): download all headers from new to old as first step of full sync

.github/workflows/pr-main_l1.yaml

@@ -167,7 +167,7 @@ jobs:
             artifact_prefix: engine_paris
           - name: "Engine withdrawal tests"
             simulation: ethereum/engine
-            limit: "engine-withdrawals/Corrupted Block Hash Payload|Empty Withdrawals|engine-withdrawals test loader|GetPayloadBodies|GetPayloadV2 Block Value|Max Initcode Size|Sync after 2 blocks - Withdrawals on Genesis|Withdraw many accounts|Withdraw to a single account|Withdraw to two accounts|Withdraw zero amount|Withdraw many accounts|Withdrawals Fork on Block 1 - 1 Block Re-Org|Withdrawals Fork on Block 1 - 8 Block Re-Org NewPayload|Withdrawals Fork on Block 2|Withdrawals Fork on Block 3|Withdrawals Fork on Block 8 - 10 Block Re-Org NewPayload|Withdrawals Fork on Canonical Block 8 / Side Block 7 - 10 Block Re-Org [^S]|Withdrawals Fork on Canonical Block 8 / Side Block 9 - 10 Block Re-Org [^S]"
+            limit: "engine-withdrawals"
             artifact_prefix: engine_withdrawals
           # Investigate this test
           # - name: "Sync"

crates/networking/p2p/peer_handler.rs

@@ -19,7 +19,7 @@ use crate::{
         },
     },
     snap::encodable_to_proof,
-    sync::{AccountStorageRoots, BlockSyncState, block_is_stale, update_pivot},
+    sync::{AccountStorageRoots, SnapBlockSyncState, block_is_stale, update_pivot},
     utils::{
         AccountsWithStorage, dump_accounts_to_file, dump_storages_to_file,
         get_account_state_snapshot_file, get_account_storages_snapshot_file,
@@ -469,11 +469,13 @@ impl PeerHandler {
                     )
                     .await
                     {
-                        if are_block_headers_chained(&block_headers, &order) {
+                        if !block_headers.is_empty()
+                            && are_block_headers_chained(&block_headers, &order)
+                        {
                             return Ok(Some(block_headers));
                         } else {
                             warn!(
-                                "[SYNCING] Received invalid headers from peer, penalizing peer {peer_id}"
+                                "[SYNCING] Received empty/invalid headers from peer, penalizing peer {peer_id}"
                             );
                         }
                     }
@@ -678,7 +680,7 @@ impl PeerHandler {
         limit: H256,
         account_state_snapshots_dir: &Path,
         pivot_header: &mut BlockHeader,
-        block_sync_state: &mut BlockSyncState,
+        block_sync_state: &mut SnapBlockSyncState,
     ) -> Result<(), PeerHandlerError> {
         METRICS
             .current_step

crates/networking/rpc/engine/fork_choice.rs

@@ -201,6 +201,11 @@ async fn handle_forkchoice(
             .get_latest_valid_ancestor(head_block.parent_hash)
             .await?
     {
+        // Invalidate the child too
+        context
+            .storage
+            .set_latest_valid_ancestor(head_block.hash(), latest_valid_hash)
+            .await?;
         return Ok((
             None,
             ForkChoiceResponse::from(PayloadStatus::invalid_with(

crates/networking/rpc/engine/payload.rs

@@ -571,6 +571,11 @@ async fn validate_ancestors(
         .get_latest_valid_ancestor(block.header.parent_hash)
         .await?
     {
+        // Invalidate child too
+        context
+            .storage
+            .set_latest_valid_ancestor(block.header.hash(), latest_valid_hash)
+            .await?;
         return Ok(Some(PayloadStatus::invalid_with(
             latest_valid_hash,
             "Parent header has been previously invalidated.".into(),

crates/storage/api.rs

@@ -362,4 +362,17 @@ pub trait StoreEngine: Debug + Send + Sync + RefUnwindSafe {
         &self,
         account_codes: Vec<(H256, Bytes)>,
     ) -> Result<(), StoreError>;
+
+    /// Add a batch of headers downloaded during fullsync
+    async fn add_fullsync_batch(&self, headers: Vec<BlockHeader>) -> Result<(), StoreError>;
+
+    /// Read a batch of headers downloaded during fullsync
+    async fn read_fullsync_batch(
+        &self,
+        start: BlockNumber,
+        limit: u64,
+    ) -> Result<Vec<BlockHeader>, StoreError>;
+
+    /// Clear all headers downloaded during fullsync
+    async fn clear_fullsync_headers(&self) -> Result<(), StoreError>;
 }

crates/storage/store.rs

@@ -1333,6 +1333,25 @@ impl Store {
     ) -> Result<(), StoreError> {
         self.engine.write_account_code_batch(account_codes).await
     }
+
+    /// Add a batch of headers downloaded during fullsync
+    pub async fn add_fullsync_batch(&self, headers: Vec<BlockHeader>) -> Result<(), StoreError> {
+        self.engine.add_fullsync_batch(headers).await
+    }
+
+    /// Read a batch of headers downloaded during fullsync
+    pub async fn read_fullsync_batch(
+        &self,
+        start: BlockNumber,
+        limit: u64,
+    ) -> Result<Vec<BlockHeader>, StoreError> {
+        self.engine.read_fullsync_batch(start, limit).await
+    }
+
+    /// Clear all headers downloaded during fullsync
+    pub async fn clear_fullsync_headers(&self) -> Result<(), StoreError> {
+        self.engine.clear_fullsync_headers().await
+    }
 }
 
 pub struct AccountProof {

crates/storage/store_db/in_memory.rs

@@ -47,6 +47,8 @@ pub struct StoreInner {
     invalid_ancestors: HashMap<BlockHash, BlockHash>,
     // Stores current Snap State
     snap_state: SnapState,
+    // Stores fetched headers during a fullsync
+    fullsync_headers: HashMap<BlockNumber, BlockHeader>,
 }
 
 #[derive(Default, Debug)]
@@ -695,6 +697,37 @@ impl StoreEngine for Store {
 
         Ok(())
     }
+
+    async fn add_fullsync_batch(&self, headers: Vec<BlockHeader>) -> Result<(), StoreError> {
+        self.inner()?
+            .fullsync_headers
+            .extend(headers.into_iter().map(|h| (h.number, h)));
+        Ok(())
+    }
+
+    async fn read_fullsync_batch(
+        &self,
+        start: BlockNumber,
+        limit: u64,
+    ) -> Result<Vec<BlockHeader>, StoreError> {
+        let store = self.inner()?;
+        (start..start + limit)
+            .map(|ref n| {
+                store
+                    .fullsync_headers
+                    .get(n)
+                    .cloned()
+                    .ok_or(StoreError::Custom(format!(
+                        "Missing fullsync header for block {n}"
+                    )))
+            })
+            .collect::<Result<Vec<_>, _>>()
+    }
+
+    async fn clear_fullsync_headers(&self) -> Result<(), StoreError> {
+        self.inner()?.fullsync_headers.clear();
+        Ok(())
+    }
 }
 
 impl Debug for Store {

crates/storage/store_db/rocksdb.rs

@@ -104,6 +104,11 @@ const CF_PENDING_BLOCKS: &str = "pending_blocks";
 /// - [`Vec<u8>`] = `BlockHashRLP::from(latest_valid).bytes().clone()`
 const CF_INVALID_ANCESTORS: &str = "invalid_ancestors";
 
+/// Block headers downloaded during fullsync column family: [`u8;_`] => [`Vec<u8>`]
+/// - [`u8;_`] = `block_number.to_le_bytes()`
+/// - [`Vec<u8>`] = `BlockHeaderRLP::from(block.header.clone()).bytes().clone()`
+const CF_FULLSYNC_HEADERS: &str = "fullsync_headers";
+
 #[derive(Debug)]
 pub struct Store {
     db: Arc<OptimisticTransactionDB<MultiThreaded>>,
@@ -166,6 +171,7 @@ impl Store {
             CF_TRIE_NODES,
             CF_PENDING_BLOCKS,
             CF_INVALID_ANCESTORS,
+            CF_FULLSYNC_HEADERS,
         ];
 
         // Get existing column families to know which ones to drop later
@@ -1479,6 +1485,58 @@ impl StoreEngine for Store {
 
         self.write_batch_async(batch_ops).await
     }
+
+    async fn add_fullsync_batch(&self, headers: Vec<BlockHeader>) -> Result<(), StoreError> {
+        let mut batch_ops = Vec::new();
+
+        for header in headers {
+            let number_value = header.number.to_le_bytes().to_vec();
+            let header_value = BlockHeaderRLP::from(header).bytes().clone();
+
+            batch_ops.push((CF_FULLSYNC_HEADERS.to_string(), number_value, header_value));
+        }
+
+        self.write_batch_async(batch_ops).await
+    }
+
+    async fn read_fullsync_batch(
+        &self,
+        start: BlockNumber,
+        limit: u64,
+    ) -> Result<Vec<BlockHeader>, StoreError> {
+        self.read_bulk_async(
+            CF_FULLSYNC_HEADERS,
+            (start..start + limit).map(|n| n.to_le_bytes()).collect(),
+            |bytes| {
+                BlockHeaderRLP::from_bytes(bytes)
+                    .to()
+                    .map_err(StoreError::from)
+            },
+        )
+        .await
+    }
+
+    async fn clear_fullsync_headers(&self) -> Result<(), StoreError> {
+        let db = self.db.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let cf = db
+                .cf_handle(CF_FULLSYNC_HEADERS)
+                .ok_or_else(|| StoreError::Custom("Column family not found".to_string()))?;
+
+            let mut iter = db.iterator_cf(&cf, rocksdb::IteratorMode::Start);
+            let mut batch = WriteBatchWithTransaction::default();
+
+            while let Some(Ok((key, _))) = iter.next() {
+                batch.delete_cf(&cf, key);
+            }
+
+            db.write(batch)
+                .map_err(|e| StoreError::Custom(format!("RocksDB batch write error: {}", e)))
+        })
+        .await
+        .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
+    }
 }
 
 /// Open column families

tooling/reorgs/src/main.rs

@@ -33,19 +33,18 @@ async fn main() {
     info!("");
 
     run_test(&cmd_path, no_reorgs_full_sync_smoke_test).await;
+    run_test(&cmd_path, test_reorg_back_to_base).await;
 
-    // TODO: uncomment once #4676 is fixed
-    // run_test(&cmd_path, test_reorg_back_to_base).await;
-    // // This test is flaky 50% of the time, check that it runs correctly 30 times in a row
-    // // TODO(#4775): make it deterministic
-    // for _ in 0..30 {
-    //     run_test(&cmd_path, test_chain_split).await;
-    // }
-    // run_test(&cmd_path, test_one_block_reorg_and_back).await;
-    // run_test(&cmd_path, test_reorg_back_to_base_with_common_ancestor).await;
-    // run_test(&cmd_path, test_storage_slots_reorg).await;
-
-    // run_test(&cmd_path, test_many_blocks_reorg).await;
+    // This test is flaky 50% of the time, check that it runs correctly multiple times in a row
+    // TODO(#4775): make it deterministic
+    for _ in 0..10 {
+        run_test(&cmd_path, test_chain_split).await;
+    }
+
+    run_test(&cmd_path, test_one_block_reorg_and_back).await;
+    run_test(&cmd_path, test_reorg_back_to_base_with_common_ancestor).await;
+    run_test(&cmd_path, test_storage_slots_reorg).await;
+    run_test(&cmd_path, test_many_blocks_reorg).await;
 }
 
 async fn get_ethrex_version(cmd_path: &Path) -> String {
@@ -104,7 +103,6 @@ async fn no_reorgs_full_sync_smoke_test(simulator: Arc<Mutex<Simulator>>) {
     node1.update_forkchoice(&base_chain).await;
 }
 
-#[expect(unused)]
 async fn test_reorg_back_to_base(simulator: Arc<Mutex<Simulator>>) {
     let mut simulator = simulator.lock().await;
 
@@ -124,7 +122,6 @@ async fn test_reorg_back_to_base(simulator: Arc<Mutex<Simulator>>) {
     node0.update_forkchoice(&base_chain).await;
 }
 
-#[expect(unused)]
 async fn test_reorg_back_to_base_with_common_ancestor(simulator: Arc<Mutex<Simulator>>) {
     let mut simulator = simulator.lock().await;
 
@@ -149,7 +146,6 @@ async fn test_reorg_back_to_base_with_common_ancestor(simulator: Arc<Mutex<Simul
     node0.update_forkchoice(&base_chain).await;
 }
 
-#[expect(unused)]
 async fn test_chain_split(simulator: Arc<Mutex<Simulator>>) {
     let mut simulator = simulator.lock().await;
 
@@ -172,7 +168,6 @@ async fn test_chain_split(simulator: Arc<Mutex<Simulator>>) {
     node0.update_forkchoice(&base_chain).await;
 }
 
-#[expect(unused)]
 async fn test_one_block_reorg_and_back(simulator: Arc<Mutex<Simulator>>) {
     let mut simulator = simulator.lock().await;
     let signer: Signer = LocalSigner::new(
@@ -243,7 +238,6 @@ async fn test_one_block_reorg_and_back(simulator: Arc<Mutex<Simulator>>) {
     assert_eq!(new_balance, initial_balance);
 }
 
-#[expect(unused)]
 async fn test_many_blocks_reorg(simulator: Arc<Mutex<Simulator>>) {
     let mut simulator = simulator.lock().await;
     let signer: Signer = LocalSigner::new(
@@ -315,7 +309,6 @@ async fn test_many_blocks_reorg(simulator: Arc<Mutex<Simulator>>) {
     assert_eq!(new_balance, initial_balance + transfer_amount);
 }
 
-#[expect(unused)]
 async fn test_storage_slots_reorg(simulator: Arc<Mutex<Simulator>>) {
     let mut simulator = simulator.lock().await;
     // Initcode for deploying a contract that receives two `bytes32` parameters and sets `storage[param0] = param1`