Moved insertion of the blocks into the BlockImporter instead of the…

… executor (#1577) Related to #1549 We don't need to insert the block in the executor because it is the block importer's responsibility. It verifies the block's validity and decides whether we want to insert a new block or not. Plus, storing blocks and transactions is not part of the state transition. This change also adds the ability to produce the block with a defined order of the transactions. It may be useful in the tests.
FuelLabs · Jan 5, 2024 · 3cab654 · 3cab654
1 parent 5ce757f
commit 3cab654
Show file tree

Hide file tree

Showing 32 changed files with 450 additions and 345 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 Description of the upcoming release here.
 
+
+### Changed
+
+- [#1577](https://github.com/FuelLabs/fuel-core/pull/1577): Moved insertion of sealed blocks into the `BlockImporter` instead of the executor.
+
 ## [Version 0.22.0]
 
 ### Added

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bin/fuel-core/src/cli/run.rs b/bin/fuel-core/src/cli/run.rs
@@ -300,6 +300,9 @@ impl Command {
             max_wait_time: max_wait_time.into(),
         };
 
+        let block_importer =
+            fuel_core::service::config::fuel_core_importer::Config::new(&chain_conf);
+
         let config = Config {
             addr,
             api_request_timeout: api_request_timeout.into(),
@@ -328,8 +331,7 @@ impl Command {
                 coinbase_recipient,
                 metrics,
             },
-            block_executor: Default::default(),
-            block_importer: Default::default(),
+            block_importer,
             #[cfg(feature = "relayer")]
             relayer: relayer_cfg,
             #[cfg(feature = "p2p")]

diff --git a/crates/fuel-core/src/database.rs b/crates/fuel-core/src/database.rs
@@ -150,6 +150,8 @@ pub enum Column {
     ContractsStateMerkleData = 23,
     /// See [`ContractsStateMerkleMetadata`](storage::ContractsStateMerkleMetadata)
     ContractsStateMerkleMetadata = 24,
+    /// See [`ProcessedTransactions`](storage::ProcessedTransactions)
+    ProcessedTransactions = 25,
 }
 
 impl Column {

diff --git a/crates/fuel-core/src/database/storage.rs b/crates/fuel-core/src/database/storage.rs
@@ -3,6 +3,7 @@ use crate::database::{
     Database,
 };
 use fuel_core_storage::{
+    tables::ProcessedTransactions,
     Error as StorageError,
     Mappable,
     MerkleRoot,
@@ -160,6 +161,12 @@ impl DatabaseColumn for FuelBlockSecondaryKeyBlockHeights {
     }
 }
 
+impl DatabaseColumn for ProcessedTransactions {
+    fn column() -> Column {
+        Column::ProcessedTransactions
+    }
+}
+
 impl DatabaseColumn for FuelBlockMerkleData {
     fn column() -> Column {
         Column::FuelBlockMerkleData

diff --git a/crates/fuel-core/src/executor.rs b/crates/fuel-core/src/executor.rs
@@ -20,7 +20,6 @@ mod tests {
             ContractsRawCode,
             Messages,
             Receipts,
-            Transactions,
         },
         StorageAsMut,
     };
@@ -1571,7 +1570,7 @@ mod tests {
             .into();
         let db = &mut Database::default();
 
-        let mut executor = create_executor(
+        let executor = create_executor(
             db.clone(),
             Config {
                 utxo_validation_default: false,
@@ -1607,16 +1606,6 @@ mod tests {
         assert_eq!(executed_tx.inputs()[0].balance_root(), Some(&empty_state));
         assert_eq!(executed_tx.outputs()[0].state_root(), Some(&empty_state));
         assert_eq!(executed_tx.outputs()[0].balance_root(), Some(&empty_state));
-
-        let expected_tx = block.transactions()[1].clone();
-        let storage_tx = executor
-            .database
-            .storage::<Transactions>()
-            .get(&executed_tx.id(&ChainId::default()))
-            .unwrap()
-            .unwrap()
-            .into_owned();
-        assert_eq!(storage_tx, expected_tx);
     }
 
     #[test]
@@ -1638,7 +1627,7 @@ mod tests {
             .into();
         let db = &mut Database::default();
 
-        let mut executor = create_executor(
+        let executor = create_executor(
             db.clone(),
             Config {
                 utxo_validation_default: false,
@@ -1680,16 +1669,6 @@ mod tests {
         );
         assert_eq!(executed_tx.inputs()[0].state_root(), Some(&empty_state));
         assert_eq!(executed_tx.inputs()[0].balance_root(), Some(&empty_state));
-
-        let expected_tx = block.transactions()[1].clone();
-        let storage_tx = executor
-            .database
-            .storage::<Transactions>()
-            .get(&expected_tx.id(&ChainId::default()))
-            .unwrap()
-            .unwrap()
-            .into_owned();
-        assert_eq!(storage_tx, expected_tx);
     }
 
     #[test]
@@ -1751,7 +1730,7 @@ mod tests {
             .clone();
         let db = &mut Database::default();
 
-        let mut executor = create_executor(
+        let executor = create_executor(
             db.clone(),
             Config {
                 utxo_validation_default: false,
@@ -1793,16 +1772,6 @@ mod tests {
             executed_tx.inputs()[0].balance_root(),
             executed_tx.outputs()[0].balance_root()
         );
-
-        let expected_tx = block.transactions()[1].clone();
-        let storage_tx = executor
-            .database
-            .storage::<Transactions>()
-            .get(&expected_tx.id(&ChainId::default()))
-            .unwrap()
-            .unwrap()
-            .into_owned();
-        assert_eq!(storage_tx, expected_tx);
     }
 
     #[test]

diff --git a/crates/fuel-core/src/service.rs b/crates/fuel-core/src/service.rs
@@ -19,6 +19,7 @@ pub use config::{
 };
 pub use fuel_core_services::Service as ServiceTrait;
 
+use crate::service::adapters::PoAAdapter;
 pub use fuel_core_consensus_module::RelayerVerifierConfig;
 
 use self::adapters::BlockImporterAdapter;
@@ -32,6 +33,8 @@ pub mod sub_services;
 
 #[derive(Clone)]
 pub struct SharedState {
+    /// The PoA adaptor around the shared state of the consensus module.
+    pub poa_adapter: PoAAdapter,
     /// The transaction pool shared state.
     pub txpool: fuel_core_txpool::service::SharedState<P2PAdapter, Database>,
     /// The P2P network shared state.

diff --git a/crates/fuel-core/src/service/adapters/block_importer.rs b/crates/fuel-core/src/service/adapters/block_importer.rs
@@ -18,7 +18,11 @@ use fuel_core_importer::{
 };
 use fuel_core_poa::ports::RelayerPort;
 use fuel_core_storage::{
-    tables::SealedBlockConsensus,
+    tables::{
+        FuelBlocks,
+        SealedBlockConsensus,
+        Transactions,
+    },
     transactional::StorageTransaction,
     Result as StorageResult,
     StorageAsMut,
@@ -27,13 +31,14 @@ use fuel_core_types::{
     blockchain::{
         block::Block,
         consensus::Consensus,
-        primitives::{
-            BlockId,
-            DaBlockHeight,
-        },
+        primitives::DaBlockHeight,
         SealedBlock,
     },
-    fuel_types::BlockHeight,
+    fuel_tx::UniqueIdentifier,
+    fuel_types::{
+        BlockHeight,
+        ChainId,
+    },
     services::executor::{
         ExecutionTypes,
         Result as ExecutorResult,
@@ -42,7 +47,10 @@ use fuel_core_types::{
 };
 use std::sync::Arc;
 
-use super::MaybeRelayerAdapter;
+use super::{
+    MaybeRelayerAdapter,
+    TransactionsSource,
+};
 
 impl BlockImporterAdapter {
     pub fn new(
@@ -112,8 +120,8 @@ impl RelayerPort for MaybeRelayerAdapter {
 }
 
 impl ImporterDatabase for Database {
-    fn latest_block_height(&self) -> StorageResult<BlockHeight> {
-        self.latest_height()
+    fn latest_block_height(&self) -> StorageResult<Option<BlockHeight>> {
+        Ok(self.ids_of_latest_block()?.map(|(height, _)| height))
     }
 
     fn increase_tx_count(&self, new_txs_count: u64) -> StorageResult<u64> {
@@ -122,14 +130,29 @@ impl ImporterDatabase for Database {
 }
 
 impl ExecutorDatabase for Database {
-    fn seal_block(
+    fn store_new_block(
         &mut self,
-        block_id: &BlockId,
-        consensus: &Consensus,
-    ) -> StorageResult<Option<Consensus>> {
-        self.storage::<SealedBlockConsensus>()
-            .insert(block_id, consensus)
-            .map_err(Into::into)
+        chain_id: &ChainId,
+        block: &SealedBlock,
+    ) -> StorageResult<bool> {
+        let block_id = block.entity.id();
+        let mut found = self
+            .storage::<FuelBlocks>()
+            .insert(&block_id, &block.entity.compress(chain_id))?
+            .is_some();
+        found |= self
+            .storage::<SealedBlockConsensus>()
+            .insert(&block_id, &block.consensus)?
+            .is_some();
+
+        // TODO: Use `batch_insert` from https://github.com/FuelLabs/fuel-core/pull/1576
+        for tx in block.entity.transactions() {
+            found |= self
+                .storage::<Transactions>()
+                .insert(&tx.id(chain_id), tx)?
+                .is_some();
+        }
+        Ok(!found)
     }
 }
 
@@ -141,6 +164,8 @@ impl Executor for ExecutorAdapter {
         block: Block,
     ) -> ExecutorResult<UncommittedExecutionResult<StorageTransaction<Self::Database>>>
     {
-        self._execute_without_commit(ExecutionTypes::Validation(block))
+        self._execute_without_commit::<TransactionsSource>(ExecutionTypes::Validation(
+            block,
+        ))
     }
 }
diff --git a/crates/fuel-core/src/service/adapters/consensus_module/poa.rs b/crates/fuel-core/src/service/adapters/consensus_module/poa.rs
@@ -17,8 +17,12 @@ use fuel_core_poa::{
         BlockImporter,
         P2pPort,
         TransactionPool,
+        TransactionsSource,
+    },
+    service::{
+        Mode,
+        SharedState,
     },
-    service::SharedState,
 };
 use fuel_core_services::stream::BoxStream;
 use fuel_core_storage::transactional::StorageTransaction;
@@ -45,6 +49,18 @@ impl PoAAdapter {
     pub fn new(shared_state: Option<SharedState>) -> Self {
         Self { shared_state }
     }
+
+    pub async fn manually_produce_blocks(
+        &self,
+        start_time: Option<Tai64>,
+        mode: Mode,
+    ) -> anyhow::Result<()> {
+        self.shared_state
+            .as_ref()
+            .ok_or(anyhow!("The block production is disabled"))?
+            .manually_produce_block(start_time, mode)
+            .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -54,10 +70,7 @@ impl ConsensusModulePort for PoAAdapter {
         start_time: Option<Tai64>,
         number_of_blocks: u32,
     ) -> anyhow::Result<()> {
-        self.shared_state
-            .as_ref()
-            .ok_or(anyhow!("The block production is disabled"))?
-            .manually_produce_block(start_time, number_of_blocks)
+        self.manually_produce_blocks(start_time, Mode::Blocks { number_of_blocks })
             .await
     }
 }
@@ -91,11 +104,23 @@ impl fuel_core_poa::ports::BlockProducer for BlockProducerAdapter {
         &self,
         height: BlockHeight,
         block_time: Tai64,
+        source: TransactionsSource,
         max_gas: Word,
     ) -> anyhow::Result<UncommittedResult<StorageTransaction<Database>>> {
-        self.block_producer
-            .produce_and_execute_block(height, block_time, max_gas)
-            .await
+        match source {
+            TransactionsSource::TxPool => {
+                self.block_producer
+                    .produce_and_execute_block_txpool(height, block_time, max_gas)
+                    .await
+            }
+            TransactionsSource::SpecificTransactions(txs) => {
+                self.block_producer
+                    .produce_and_execute_block_transactions(
+                        height, block_time, txs, max_gas,
+                    )
+                    .await
+            }
+        }
     }
 }
 

diff --git a/crates/fuel-core/src/service/adapters/executor.rs b/crates/fuel-core/src/service/adapters/executor.rs
@@ -50,10 +50,13 @@ impl fuel_core_executor::ports::TransactionsSource for TransactionsSource {
 }
 
 impl ExecutorAdapter {
-    pub(crate) fn _execute_without_commit(
+    pub(crate) fn _execute_without_commit<TxSource>(
         &self,
-        block: ExecutionBlockWithSource<TransactionsSource>,
-    ) -> ExecutorResult<UncommittedResult<StorageTransaction<Database>>> {
+        block: ExecutionBlockWithSource<TxSource>,
+    ) -> ExecutorResult<UncommittedResult<StorageTransaction<Database>>>
+    where
+        TxSource: fuel_core_executor::ports::TransactionsSource,
+    {
         let executor = Executor {
             database: self.relayer.database.clone(),
             relayer: self.relayer.clone(),