feat(l1): download all headers from new to old as first step of full 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 are_block_headers_chained(&block_headers, &order)
+                            && !block_headers.is_empty()
+                        {
                             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,46 @@ 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 || {
+            db.drop_cf(CF_FULLSYNC_HEADERS).map_err(StoreError::from)
+        })
+        .await
+        .map_err(|e| StoreError::Custom(format!("Task panicked: {}", e)))?
+    }
 }
 
 /// Open column families