Parallelise block/transaction changes creation in Importer

.changes/changed/2618.md

@@ -0,0 +1 @@
+Parallelize block/transaction changes creation in Importer

crates/fuel-core/src/service/adapters.rs

@@ -394,8 +394,7 @@ pub struct BlockProducerAdapter {
 
 #[derive(Clone)]
 pub struct BlockImporterAdapter {
-    pub block_importer:
-        Arc<fuel_core_importer::Importer<Database, ExecutorAdapter, VerifierAdapter>>,
+    pub block_importer: Arc<fuel_core_importer::Importer>,
 }
 
 impl BlockImporterAdapter {

crates/fuel-core/src/service/adapters/block_importer.rs

@@ -1,5 +1,8 @@
 use crate::{
-    database::Database,
+    database::{
+        commit_changes_with_height_update,
+        Database,
+    },
     service::adapters::{
         BlockImporterAdapter,
         ExecutorAdapter,
@@ -27,7 +30,10 @@ use fuel_core_storage::{
         },
         FuelBlocks,
     },
-    transactional::Changes,
+    transactional::{
+        Changes,
+        StorageChanges,
+    },
     MerkleRoot,
     Result as StorageResult,
     StorageAsRef,
@@ -52,6 +58,7 @@ use fuel_core_types::{
         UncommittedValidationResult,
     },
 };
+use itertools::Itertools;
 use std::sync::Arc;
 
 impl BlockImporterAdapter {
@@ -62,11 +69,7 @@ impl BlockImporterAdapter {
         executor: ExecutorAdapter,
         verifier: VerifierAdapter,
     ) -> Self {
-        let metrics = config.metrics;
         let importer = Importer::new(chain_id, config, database, executor, verifier);
-        if metrics {
-            importer.init_metrics();
-        }
         Self {
             block_importer: Arc::new(importer),
         }
@@ -104,6 +107,13 @@ impl ImporterDatabase for Database {
             .get(&DenseMetadataKey::Latest)?
             .map(|cow| *cow.root()))
     }
+
+    fn commit_changes(&mut self, changes: StorageChanges) -> StorageResult<()> {
+        commit_changes_with_height_update(self, changes, |iter| {
+            iter.iter_all_keys::<FuelBlocks>(Some(IterDirection::Reverse))
+                .try_collect()
+        })
+    }
 }
 
 impl Validator for ExecutorAdapter {

crates/fuel-core/src/service/genesis.rs

@@ -206,6 +206,11 @@ pub async fn execute_and_commit_genesis_block(
     config: &Config,
     db: &CombinedDatabase,
 ) -> anyhow::Result<()> {
+    use fuel_core_importer::ports::{
+        MockBlockVerifier,
+        MockValidator,
+    };
+
     let result = execute_genesis_block(StateWatcher::default(), config, db).await?;
     let importer = fuel_core_importer::Importer::new(
         config
@@ -215,8 +220,8 @@ pub async fn execute_and_commit_genesis_block(
             .chain_id(),
         config.block_importer.clone(),
         db.on_chain().clone(),
-        (),
-        (),
+        MockValidator::default(),
+        MockBlockVerifier::default(),
     );
     importer.commit_result(result).await?;
     Ok(())

crates/services/importer/Cargo.toml

@@ -15,19 +15,19 @@ derive_more = { workspace = true }
 fuel-core-metrics = { workspace = true }
 fuel-core-storage = { workspace = true, features = ["std"] }
 fuel-core-types = { workspace = true, features = ["std"] }
-parking_lot = { workspace = true }
+mockall = { workspace = true, optional = true }
 rayon = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true }
 
 [dev-dependencies]
 fuel-core-trace = { path = "./../../trace" }
 fuel-core-types = { path = "./../../types", features = ["test-helpers"] }
-mockall = { workspace = true }
 test-case = { workspace = true }
 
 [features]
 test-helpers = [
+  "dep:mockall",
   "fuel-core-types/test-helpers",
   "fuel-core-storage/test-helpers",
 ]

crates/services/importer/src/error.rs

@@ -0,0 +1,81 @@
+use fuel_core_storage::{
+    Error as StorageError,
+    MerkleRoot,
+};
+use fuel_core_types::{
+    blockchain::primitives::BlockId,
+    fuel_types::BlockHeight,
+    services::executor,
+};
+use tokio::sync::{
+    mpsc,
+    oneshot,
+    TryAcquireError,
+};
+
+#[derive(Debug, derive_more::Display, derive_more::From)]
+pub enum Error {
+    #[display(fmt = "The commit is already in the progress: {_0}.")]
+    Semaphore(TryAcquireError),
+    #[display(
+        fmt = "The wrong state of database during insertion of the genesis block."
+    )]
+    InvalidUnderlyingDatabaseGenesisState,
+    #[display(fmt = "The wrong state of storage after execution of the block.\
+        The actual root is {_1:?}, when the expected root is {_0:?}.")]
+    InvalidDatabaseStateAfterExecution(Option<MerkleRoot>, Option<MerkleRoot>),
+    #[display(fmt = "Got overflow during increasing the height.")]
+    Overflow,
+    #[display(fmt = "The non-generic block can't have zero height.")]
+    ZeroNonGenericHeight,
+    #[display(fmt = "The actual height is {_1}, when the next expected height is {_0}.")]
+    IncorrectBlockHeight(BlockHeight, BlockHeight),
+    #[display(
+        fmt = "Got another block id after validation of the block. Expected {_0} != Actual {_1}"
+    )]
+    BlockIdMismatch(BlockId, BlockId),
+    #[display(fmt = "Some of the block fields are not valid: {_0}.")]
+    FailedVerification(anyhow::Error),
+    #[display(fmt = "The execution of the block failed: {_0}.")]
+    FailedExecution(executor::Error),
+    #[display(fmt = "It is not possible to execute the genesis block.")]
+    ExecuteGenesis,
+    #[display(fmt = "The database already contains the data at the height {_0}.")]
+    NotUnique(BlockHeight),
+    #[display(fmt = "The previous block processing is not finished yet.")]
+    PreviousBlockProcessingNotFinished,
+    #[display(fmt = "The send command to the inner task failed.")]
+    SendCommandToInnerTaskFailed,
+    #[display(fmt = "The inner import task is not running.")]
+    InnerTaskIsNotRunning,
+    #[from]
+    Storage(StorageError),
+    UnsupportedConsensusVariant(String),
+    ActiveBlockResultsSemaphoreClosed(tokio::sync::AcquireError),
+    RayonTaskWasCanceled,
+}
+
+impl From<Error> for anyhow::Error {
+    fn from(error: Error) -> Self {
+        anyhow::Error::msg(error)
+    }
+}
+
+impl From<oneshot::error::RecvError> for Error {
+    fn from(_: oneshot::error::RecvError) -> Self {
+        Error::InnerTaskIsNotRunning
+    }
+}
+
+impl<T> From<mpsc::error::SendError<T>> for Error {
+    fn from(_: mpsc::error::SendError<T>) -> Self {
+        Error::SendCommandToInnerTaskFailed
+    }
+}
+
+#[cfg(test)]
+impl PartialEq for Error {
+    fn eq(&self, other: &Self) -> bool {
+        format!("{self}") == format!("{other}")
+    }
+}