Create validation in parallel-executor

crates/services/executor/src/executor.rs

@@ -896,7 +896,7 @@ where
     }
 
     #[allow(clippy::too_many_arguments)]
-    fn execute_transaction_and_commit<'a, W>(
+    pub fn execute_transaction_and_commit<'a, W>(
         &'a self,
         block: &'a mut PartialFuelBlock,
         storage_tx: &mut BlockStorageTransaction<W>,

crates/services/parallel-executor/src/scheduler/coin.rs → crates/services/parallel-executor/src/coin.rs

@@ -24,8 +24,7 @@ use fuel_core_types::{
 };
 use fxhash::FxHashMap;
 
-use super::SchedulerError;
-
+use crate::scheduler::SchedulerError;
 #[derive(Debug, Eq)]
 pub(crate) struct CoinInBatch {
     /// The utxo id

crates/services/parallel-executor/src/executor.rs

@@ -7,6 +7,11 @@ use crate::{
         SchedulerExecutionResult,
     },
     tx_waiter::NoWaitTxs,
+    txs_ext::TxCoinbaseExt,
+    validator::{
+        self,
+        Validator,
+    },
 };
 use fuel_core_executor::{
     executor::{
@@ -37,9 +42,12 @@ use fuel_core_storage::{
     },
 };
 use fuel_core_types::{
-    blockchain::block::{
-        Block,
-        PartialFuelBlock,
+    blockchain::{
+        block::{
+            Block,
+            PartialFuelBlock,
+        },
+        header::PartialBlockHeader,
     },
     fuel_tx::{
         Bytes32,
@@ -75,7 +83,10 @@ pub struct Executor<S, R, P> {
     memory_pool: Option<Vec<MemoryInstance>>,
 }
 
-impl<S, R, P> Executor<S, R, P> {
+impl<S, R, P> Executor<S, R, P>
+where
+    R: Clone,
+{
     pub fn new(
         storage_view_provider: S,
         relayer: R,
@@ -197,20 +208,28 @@ where
             return Ok(Default::default());
         };
 
-        let prev_block = structured_storage
+        let prev_da = structured_storage
             .storage::<FuelBlocks>()
             .get(&prev_height)?
             .ok_or_else(|| {
                 SchedulerError::InternalError("Previous block not found".to_string())
-            })?;
+            })?
+            .header()
+            .da_height();
 
-        if prev_block.header().da_height() != components.header_to_produce.da_height {
-            let (storage_tx, event_inbox_root) = self.process_l1_txs(
+        let mut storage_tx = StorageTransaction::transaction(
+            structured_storage.into_storage(),
+            ConflictPolicy::Fail,
+            Default::default(),
+        );
+
+        if prev_da != components.header_to_produce.da_height {
+            let event_inbox_root = self.process_l1_txs(
                 partial_block,
                 components.coinbase_recipient,
                 execution_data,
                 memory,
-                structured_storage.into_storage(),
+                &mut storage_tx,
                 executor,
             )?;
             Ok((storage_tx.into_changes(), event_inbox_root))
@@ -226,24 +245,18 @@ where
         coinbase_contract_id: ContractId,
         execution_data: &mut ExecutionData,
         memory: &mut MemoryInstance,
-        view: View,
+        storage_tx: &mut StorageTransaction<View>,
         executor: &mut BlockExecutor<R, NoWaitTxs, P>,
-    ) -> Result<(StorageTransaction<View>, Bytes32), SchedulerError> {
-        let mut storage_tx = StorageTransaction::transaction(
-            view,
-            ConflictPolicy::Fail,
-            Default::default(),
-        );
-
+    ) -> Result<Bytes32, SchedulerError> {
         executor.process_l1_txs(
             partial_block,
             coinbase_contract_id,
-            &mut storage_tx,
+            storage_tx,
             execution_data,
             memory,
         )?;
 
-        Ok((storage_tx, execution_data.event_inbox_root))
+        Ok(execution_data.event_inbox_root)
     }
 
     /// Run the parallel executor for L2 transactions
@@ -418,3 +431,76 @@ where
         unimplemented!("Dry run not implemented yet");
     }
 }
+
+impl<S, R, View> Executor<S, R, validator::NoPreconfirmationSender>
+where
+    R: RelayerPort + Clone + Send + 'static,
+    S: AtomicView<LatestView = View> + Clone + Send + 'static,
+    View: KeyValueInspect<Column = Column> + Send + Sync + 'static,
+{
+    async fn validate_block(
+        mut self,
+        block: &Block,
+        mut block_storage_tx: StorageTransaction<View>,
+    ) -> Result<validator::ValidationResult, SchedulerError> {
+        let mut data = ExecutionData::new();
+
+        let partial_header = PartialBlockHeader::from(block.header());
+        let mut partial_block = PartialFuelBlock::new(partial_header, vec![]);
+        let transactions = block.transactions();
+        let mut memory = MemoryInstance::new();
+        let consensus_parameters = {
+            block_storage_tx
+                .storage::<ConsensusParametersVersions>()
+                .get(&block.header().consensus_parameters_version())?
+                .ok_or_else(|| {
+                    SchedulerError::InternalError(
+                        "Consensus parameters not found".to_string(),
+                    )
+                })?
+                .into_owned()
+        };
+        // Initialize block executor
+        let mut executor = BlockExecutor::new(
+            self.relayer.clone(),
+            ExecutionOptions {
+                forbid_unauthorized_inputs: true,
+                forbid_fake_utxo: false,
+                backtrace: false,
+            },
+            consensus_parameters.clone(),
+            NoWaitTxs,
+            validator::NoPreconfirmationSender,
+            true, // dry run
+        )
+        .map_err(|e| {
+            SchedulerError::InternalError(format!("Failed to create executor: {e}"))
+        })?;
+        let validator = Validator::new(self.config.clone(), executor.clone());
+
+        let (gas_price, coinbase_contract_id) = transactions.coinbase()?;
+
+        let _event_root = self.process_l1_txs(
+            &mut partial_block,
+            coinbase_contract_id,
+            &mut data,
+            &mut memory,
+            &mut block_storage_tx,
+            &mut executor,
+        )?;
+        let processed_l1_tx_count = partial_block.transactions.len();
+
+        let components = Components {
+            header_to_produce: partial_block.header,
+            transactions_source: transactions.iter().cloned().skip(processed_l1_tx_count),
+            coinbase_recipient: coinbase_contract_id,
+            gas_price,
+        };
+
+        let executed_block_result = validator
+            .validate_block(components, block_storage_tx, block)
+            .await?;
+
+        Ok(executed_block_result)
+    }
+}

crates/services/parallel-executor/src/lib.rs

@@ -4,9 +4,12 @@ pub mod config;
 pub mod executor;
 pub(crate) mod l1_execution_data;
 pub mod ports;
+pub(crate) mod txs_ext;
 
+mod coin;
 mod once_transaction_source;
 mod tx_waiter;
+mod validator;
 
 #[cfg(test)]
 mod tests;

crates/services/parallel-executor/src/scheduler.rs

@@ -18,7 +18,6 @@
 //! This can be done because we assume that the transaction pool is sending us transactions that are alTransactionsReadyForPickup correctly verified.
 //! If we have a transaction that end up being skipped (only possible cause if consensus parameters changes) then we will have to
 //! fallback a sequential execution of the transaction that used the skipped one as a dependency.
-mod coin;
 mod contracts_changes;
 
 use std::{
@@ -98,6 +97,10 @@ use tokio::runtime::Runtime;
 
 use crate::{
     checked_transaction_ext::CheckedTransactionExt,
+    coin::{
+        CoinDependencyChainVerifier,
+        CoinInBatch,
+    },
     column_adapter::ContractColumnsIterator,
     config::Config,
     l1_execution_data::L1ExecutionData,
@@ -109,10 +112,6 @@ use crate::{
     },
     tx_waiter::NoWaitTxs,
 };
-use coin::{
-    CoinDependencyChainVerifier,
-    CoinInBatch,
-};
 
 pub struct Scheduler<R, S, PreconfirmationSender> {
     /// Config
@@ -230,6 +229,14 @@ pub enum SchedulerError {
     StorageError(StorageError),
     /// Internal error
     InternalError(String),
+    /// Mint missing error
+    MintMissing,
+    /// Skipped transaction error
+    SkippedTransaction(ExecutorError),
+    /// Block mismatch
+    BlockMismatch,
+    /// Consensus parameters not found
+    ConsensusParametersNotFound(u32),
 }
 
 impl From<StorageError> for SchedulerError {

crates/services/parallel-executor/src/txs_ext.rs

@@ -0,0 +1,25 @@
+use fuel_core_types::fuel_tx::{
+    ContractId,
+    Transaction,
+    field::{
+        InputContract,
+        MintGasPrice,
+    },
+};
+
+use crate::scheduler::SchedulerError;
+
+/// Returns the coinbase transaction details if it exists.
+pub trait TxCoinbaseExt {
+    fn coinbase(&self) -> Result<(u64, ContractId), SchedulerError>;
+}
+
+impl TxCoinbaseExt for &[Transaction] {
+    fn coinbase(&self) -> Result<(u64, ContractId), SchedulerError> {
+        if let Some(Transaction::Mint(mint)) = self.last() {
+            Ok((*mint.gas_price(), mint.input_contract().contract_id))
+        } else {
+            Err(SchedulerError::MintMissing)
+        }
+    }
+}