add BatchHeaderFieldsProvider

Author: eval-exec

Cargo.lock

@@ -183,7 +183,11 @@ impl StoreTransaction {
         // Also insert into COLUMN_HEADER_INDEX for fast number→hash lookup during IBD
         // This allows get_ancestor to use O(1) lookups instead of walking parent chain
         let block_number: packed::Uint64 = header.number().into();
-        self.insert_raw(COLUMN_HEADER_INDEX, block_number.as_slice(), hash.as_slice())
+        self.insert_raw(
+            COLUMN_HEADER_INDEX,
+            block_number.as_slice(),
+            hash.as_slice(),
+        )
     }
 
     /// TODO(doc): @quake

store/src/transaction.rs

@@ -6,12 +6,62 @@ use ckb_error::Error;
 use ckb_logger::{Level, debug, log_enabled, warn};
 use ckb_network::{CKBProtocolContext, PeerIndex};
 use ckb_shared::block_status::BlockStatus;
-use ckb_traits::HeaderFieldsProvider;
+use ckb_traits::{HeaderFields, HeaderFieldsProvider};
+use ckb_types::core::HeaderView;
 use ckb_types::{core, packed, prelude::*};
-use ckb_verification::{HeaderError, HeaderVerifier};
+use ckb_verification::{HeaderError, HeaderVerifier, UnknownParentError};
 use ckb_verification_traits::Verifier;
+use std::collections::HashMap;
 use std::sync::Arc;
 
+/// HeaderFieldsProvider that checks in-memory headers first, then falls back to database
+/// This allows batch verification of continuous headers without inserting them into DB first
+pub struct BatchHeaderFieldsProvider<'a, DL> {
+    data_loader: &'a DL,
+    headers_map: HashMap<packed::Byte32, HeaderFields>,
+}
+
+impl<'a, DL: HeaderFieldsProvider> BatchHeaderFieldsProvider<'a, DL> {
+    pub fn new(data_loader: &'a DL, headers: &[HeaderView]) -> Self {
+        let mut headers_map = HashMap::new();
+        for header in headers {
+            let fields = HeaderFields {
+                hash: header.hash(),
+                number: header.number(),
+                epoch: header.epoch(),
+                timestamp: header.timestamp(),
+                parent_hash: header.parent_hash(),
+            };
+            headers_map.insert(header.hash(), fields);
+        }
+        BatchHeaderFieldsProvider {
+            data_loader,
+            headers_map,
+        }
+    }
+}
+
+impl<'a, DL: HeaderFieldsProvider> HeaderFieldsProvider for BatchHeaderFieldsProvider<'a, DL> {
+    fn get_header_fields(&self, hash: &packed::Byte32) -> Option<HeaderFields> {
+        // Check in-memory batch first
+        if let Some(fields) = self.headers_map.get(hash) {
+            return Some(HeaderFields {
+                hash: fields.hash.clone(),
+                number: fields.number,
+                epoch: fields.epoch,
+                timestamp: fields.timestamp,
+                parent_hash: fields.parent_hash.clone(),
+            });
+        }
+        // Fall back to database
+        self.data_loader.get_header_fields(hash)
+    }
+
+    // Note: block_median_time() uses the default trait implementation,
+    // which internally calls get_header_fields() above, so it automatically
+    // checks the in-memory batch first, then falls back to database.
+}
+
 pub struct HeadersProcess<'a> {
     message: packed::SendHeadersReader<'a>,
     synchronizer: &'a Synchronizer,
@@ -156,33 +206,60 @@ impl<'a> HeadersProcess<'a> {
             }
         };
 
-        for header in headers.iter().skip(1) {
-            let verifier = HeaderVerifier::new(shared, consensus);
+        // Batch verify all headers using BatchHeaderFieldsProvider
+        // This allows headers to reference each other without being in DB yet
+        let batch_provider = BatchHeaderFieldsProvider::new(shared, &headers[1..]);
+        let mut headers_to_insert = Vec::new();
+
+        for (idx, header) in headers.iter().enumerate().skip(1) {
+            // Check if already valid (skip re-validation)
+            let status = self.active_chain.get_block_status(&header.hash());
+            if status.contains(BlockStatus::HEADER_VALID) {
+                continue;
+            }
+
+            let verifier = HeaderVerifier::new(&batch_provider, consensus);
             let acceptor =
                 HeaderAcceptor::new(header, self.peer, verifier, self.active_chain.clone());
-            let result = acceptor.accept();
+            let (result, already_valid) = acceptor.validate();
+
             match result.state {
                 ValidationState::Invalid => {
+                    // Mark all remaining headers as invalid (invalid parent)
+                    for remaining_header in headers.iter().skip(idx + 1) {
+                        shared.shared().insert_block_status(
+                            remaining_header.hash(),
+                            BlockStatus::BLOCK_INVALID,
+                        );
+                    }
                     debug!(
-                        "HeadersProcess accept result is invalid, error = {:?}, header = {:?}",
-                        result.error, headers,
+                        "HeadersProcess validation failed, error = {:?}, header = {:?}",
+                        result.error, header,
                     );
                     return StatusCode::HeadersIsInvalid
-                        .with_context(format!("accept header {header:?}"));
+                        .with_context(format!("validate header {header:?}"));
                 }
                 ValidationState::TemporaryInvalid => {
                     debug!(
-                        "HeadersProcess accept result is temporarily invalid, header = {:?}",
+                        "HeadersProcess validation temporarily invalid, header = {:?}",
                         header
                     );
                     return Status::ok();
                 }
                 ValidationState::Valid => {
-                    // Valid, do nothing
+                    // Collect valid headers for batch insertion
+                    if !already_valid {
+                        headers_to_insert.push(header.clone());
+                    }
                 }
             };
         }
 
+        // Batch insert all valid headers in ONE transaction
+        if !headers_to_insert.is_empty() {
+            shared.insert_valid_headers_batch(self.peer, &headers_to_insert);
+        }
+
         self.debug();
 
         if headers.len() == MAX_HEADERS_LEN {
@@ -273,6 +350,10 @@ impl<'a, DL: HeaderFieldsProvider> HeaderAcceptor<'a, DL> {
                     state.invalid(Some(ValidationError::Verify(error)));
                     true
                 }
+            } else if error.downcast_ref::<UnknownParentError>().is_some() {
+                // UnknownParent is temporary - we just don't have the parent header yet
+                state.temporary_invalid(Some(ValidationError::Verify(error)));
+                false
             } else {
                 state.invalid(Some(ValidationError::Verify(error)));
                 true
@@ -289,14 +370,15 @@ impl<'a, DL: HeaderFieldsProvider> HeaderAcceptor<'a, DL> {
         }
     }
 
-    pub fn accept(&self) -> ValidationResult {
+    /// Validate header without inserting it to database
+    /// Returns ValidationResult and whether the header was already valid
+    pub fn validate(&self) -> (ValidationResult, bool) {
         let mut result = ValidationResult::default();
         let sync_shared = self.active_chain.sync_shared();
         let state = self.active_chain.state();
         let shared = sync_shared.shared();
 
-        // FIXME If status == BLOCK_INVALID then return early. But which error
-        // type should we return?
+        // Check if already valid
         let status = self.active_chain.get_block_status(&self.header.hash());
         if status.contains(BlockStatus::HEADER_VALID) {
             let header_index = sync_shared
@@ -311,7 +393,7 @@ impl<'a, DL: HeaderFieldsProvider> HeaderAcceptor<'a, DL> {
             state
                 .peers()
                 .may_set_best_known_header(self.peer, header_index);
-            return result;
+            return (result, true);
         }
 
         if self.prev_block_check(&mut result).is_err() {
@@ -321,7 +403,7 @@ impl<'a, DL: HeaderFieldsProvider> HeaderAcceptor<'a, DL> {
                 self.header.hash(),
             );
             shared.insert_block_status(self.header.hash(), BlockStatus::BLOCK_INVALID);
-            return result;
+            return (result, false);
         }
 
         if let Some(is_invalid) = self.non_contextual_check(&mut result).err() {
@@ -333,7 +415,7 @@ impl<'a, DL: HeaderFieldsProvider> HeaderAcceptor<'a, DL> {
             if is_invalid {
                 shared.insert_block_status(self.header.hash(), BlockStatus::BLOCK_INVALID);
             }
-            return result;
+            return (result, false);
         }
 
         if self.version_check(&mut result).is_err() {
@@ -343,10 +425,21 @@ impl<'a, DL: HeaderFieldsProvider> HeaderAcceptor<'a, DL> {
                 self.header.hash(),
             );
             shared.insert_block_status(self.header.hash(), BlockStatus::BLOCK_INVALID);
-            return result;
+            return (result, false);
+        }
+
+        (result, false)
+    }
+
+    pub fn accept(&self) -> ValidationResult {
+        let (result, already_valid) = self.validate();
+
+        // If validation passed and header wasn't already valid, insert it
+        if result.state == ValidationState::Valid && !already_valid {
+            let sync_shared = self.active_chain.sync_shared();
+            sync_shared.insert_valid_header(self.peer, self.header);
         }
 
-        sync_shared.insert_valid_header(self.peer, self.header);
         result
     }
 }

sync/src/synchronizer/headers_process.rs

@@ -1147,6 +1147,74 @@ impl SyncShared {
         self.state.may_set_shared_best_header(header_index);
     }
 
+    /// Batch insert multiple valid headers in a single transaction
+    /// Headers must be continuous (parent-child chain)
+    /// Returns the last header's HeaderIndex
+    pub fn insert_valid_headers_batch(
+        &self,
+        peer: PeerIndex,
+        headers: &[core::HeaderView],
+    ) -> Option<HeaderIndex> {
+        if headers.is_empty() {
+            return None;
+        }
+
+        let store = self.store();
+
+        // Start with the first header's parent total_difficulty
+        let first_parent_hash = headers[0].parent_hash();
+        let mut current_td = {
+            let shared_best = self.state.shared_best_header_ref();
+            if shared_best.hash() == first_parent_hash {
+                shared_best.total_difficulty().clone()
+            } else {
+                drop(shared_best);
+                let parent_header = store
+                    .get_block_header(&first_parent_hash)
+                    .expect("parent should exist");
+                self.state
+                    .get_header_total_difficulty(store, &first_parent_hash, &parent_header)
+                    .expect("parent total_difficulty should be available")
+            }
+        };
+
+        // Create ONE transaction for all headers
+        let db_txn = store.begin_transaction();
+
+        let mut last_header_index = None;
+
+        // Insert all headers in sequence
+        for header in headers {
+            // Compute total_difficulty incrementally
+            current_td = current_td + header.difficulty();
+
+            // Insert header to DB
+            db_txn
+                .insert_header(header)
+                .expect("insert header should be ok");
+
+            // Cache the total_difficulty
+            let hash = header.hash();
+            self.state
+                .cache_header_difficulty(hash.clone(), current_td.clone());
+
+            // Track the last header index
+            let header_index = HeaderIndex::new(header.number(), hash, current_td.clone());
+            last_header_index = Some(header_index.clone());
+
+            // Update peer's best known header
+            self.state
+                .peers()
+                .may_set_best_known_header(peer, header_index.clone());
+            self.state.may_set_shared_best_header(header_index);
+        }
+
+        // Commit ONCE - single fsync for all headers
+        db_txn.commit().expect("commit should be ok");
+
+        last_header_index
+    }
+
     /// Get HeaderIndex for a given hash
     /// Returns HeaderIndex with number, hash, and total_difficulty
     pub(crate) fn get_header_index(&self, hash: &Byte32) -> Option<HeaderIndex> {