Tests for parallel executor (#2967)

## Linked Issues/PRs
Resolve #2951 

## Description
All tests needed to ensure the correct behavior of the parallel-executor

## Checklist
- [ ] Breaking changes are clearly marked as such in the PR description
and changelog
- [ ] New behavior is reflected in tests
- [ ] [The specification](https://github.com/FuelLabs/fuel-specs/)
matches the implemented behavior (link update PR if changes are needed)

### Before requesting review
- [ ] I have reviewed the code myself
- [ ] I have created follow-up issues caused by this PR and linked them
here

Author: AurelienFT

Cargo.lock

@@ -14,7 +14,15 @@ description = "Fuel Block Parallel Executor"
 wasm-executor = ["fuel-core-upgradable-executor/wasm-executor"]
 
 [dependencies]
+fuel-core-executor = { workspace = true, features = ["std"] }
 fuel-core-storage = { workspace = true, features = ["std"] }
 fuel-core-types = { workspace = true, features = ["std"] }
 fuel-core-upgradable-executor = { workspace = true, features = ["std"] }
-tokio = { workspace = true, features = ["rt-multi-thread"] }
+futures = { workspace = true, features = ["std"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "sync", "macros"] }
+
+[dev-dependencies]
+anyhow = { workspace = true }
+fuel-core-storage = { workspace = true, features = ["test-helpers"] }
+fuel-core-types = { workspace = true, features = ["test-helpers", "serde"] }
+rand = { workspace = true }

crates/services/parallel-executor/Cargo.toml

@@ -1,4 +1,7 @@
-use crate::config::Config;
+use crate::{
+    config::Config,
+    ports::TransactionsSource,
+};
 use fuel_core_storage::transactional::Changes;
 use fuel_core_types::{
     blockchain::block::Block,
@@ -14,10 +17,7 @@ use fuel_core_types::{
         },
     },
 };
-use fuel_core_upgradable_executor::{
-    executor::Executor as UpgradableExecutor,
-    native_executor::ports::TransactionsSource,
-};
+use fuel_core_upgradable_executor::executor::Executor as UpgradableExecutor;
 use std::{
     num::NonZeroUsize,
     sync::{
@@ -77,7 +77,7 @@ impl<S, R> Executor<S, R> {
 
 impl<S, R> Executor<S, R> {
     /// Produces the block and returns the result of the execution without committing the changes.
-    pub fn produce_without_commit_with_source<TxSource>(
+    pub async fn produce_without_commit_with_source<TxSource>(
         &self,
         _components: Components<TxSource>,
     ) -> ExecutorResult<Uncommitted<ExecutionResult, Changes>>

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

@@ -1,2 +1,7 @@
 pub mod config;
 pub mod executor;
+pub mod once_transaction_source;
+pub mod ports;
+
+#[cfg(test)]
+mod tests;

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

@@ -0,0 +1,76 @@
+use std::sync::Mutex;
+
+use fuel_core_executor::ports::{
+    MaybeCheckedTransaction,
+    TransactionsSource as ExecutorTransactionsSource,
+};
+use fuel_core_types::fuel_vm::checked_transaction::CheckedTransaction;
+
+use crate::ports::{
+    TransactionFiltered,
+    TransactionSourceExecutableTransactions,
+    TransactionsSource,
+};
+
+pub struct OnceTransactionsSource {
+    transactions: Mutex<Vec<CheckedTransaction>>,
+    consensus_parameters_version: u32,
+}
+
+impl OnceTransactionsSource {
+    pub fn new(
+        transactions: Vec<CheckedTransaction>,
+        consensus_parameters_version: u32,
+    ) -> Self {
+        Self {
+            transactions: Mutex::new(transactions),
+            consensus_parameters_version,
+        }
+    }
+}
+
+impl ExecutorTransactionsSource for OnceTransactionsSource {
+    fn next(
+        &self,
+        _gas_limit: u64,
+        transactions_limit: u16,
+        _block_transaction_size_limit: u32,
+    ) -> Vec<fuel_core_executor::ports::MaybeCheckedTransaction> {
+        let mut transactions = self.transactions.lock().expect("Mutex poisoned");
+        // Avoid panicking if we request more transactions than there are in the vector
+        let transactions_limit = (transactions_limit as usize).min(transactions.len());
+        transactions
+            .drain(..transactions_limit)
+            .map(|tx| {
+                MaybeCheckedTransaction::CheckedTransaction(
+                    tx,
+                    self.consensus_parameters_version,
+                )
+            })
+            .collect()
+    }
+}
+
+impl TransactionsSource for OnceTransactionsSource {
+    fn get_executable_transactions(
+        &mut self,
+        _gas_limit: u64,
+        tx_count_limit: u16,
+        _block_transaction_size_limit: u32,
+        filter: crate::ports::Filter,
+    ) -> TransactionSourceExecutableTransactions {
+        let mut transactions = self.transactions.lock().expect("Mutex poisoned");
+        // Avoid panicking if we request more transactions than there are in the vector
+        let transactions_limit = (tx_count_limit as usize).min(transactions.len());
+        let txs = transactions.drain(..transactions_limit).collect();
+        TransactionSourceExecutableTransactions {
+            transactions: txs,
+            filtered: TransactionFiltered::NotFiltered,
+            filter,
+        }
+    }
+    fn get_new_transactions_notifier(&mut self) -> tokio::sync::Notify {
+        // This is a one-time source, so we don't need to notify about new transactions
+        tokio::sync::Notify::new()
+    }
+}

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

@@ -0,0 +1,76 @@
+use std::collections::HashSet;
+
+use fuel_core_storage::Result as StorageResult;
+use fuel_core_types::{
+    blockchain::primitives::DaBlockHeight,
+    entities::coins::coin::CompressedCoin,
+    fuel_tx::{
+        ConsensusParameters,
+        ContractId,
+        UtxoId,
+    },
+    fuel_types::BlockHeight,
+    fuel_vm::checked_transaction::CheckedTransaction,
+};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TransactionFiltered {
+    /// Some transactions were filtered out and so could be fetched in the future
+    Filtered,
+    /// No transactions were filtered out
+    NotFiltered,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Filter {
+    /// The set of contract IDs to filter out
+    pub excluded_contract_ids: HashSet<ContractId>,
+}
+
+impl Filter {
+    pub fn new(excluded_contract_ids: HashSet<ContractId>) -> Self {
+        Self {
+            excluded_contract_ids,
+        }
+    }
+}
+
+pub struct TransactionSourceExecutableTransactions {
+    /// The transactions that can be executed
+    pub transactions: Vec<CheckedTransaction>,
+    /// Indicates whether some transactions were filtered out based on the filter
+    pub filtered: TransactionFiltered,
+    /// The filter used to fetch these transactions
+    pub filter: Filter,
+}
+
+pub trait TransactionsSource {
+    /// Returns the a batch of transactions to satisfy the given parameters
+    fn get_executable_transactions(
+        &mut self,
+        gas_limit: u64,
+        tx_count_limit: u16,
+        block_transaction_size_limit: u32,
+        filter: Filter,
+    ) -> TransactionSourceExecutableTransactions;
+
+    /// Returns a notification receiver for new transactions
+    fn get_new_transactions_notifier(&mut self) -> tokio::sync::Notify;
+}
+
+pub trait Storage {
+    /// Get a coin by a UTXO
+    fn get_coin(&self, utxo: &UtxoId) -> StorageResult<Option<CompressedCoin>>;
+
+    /// Get the DA block height based on provided height
+    fn get_da_height_by_l2_height(
+        &self,
+        block_height: &BlockHeight,
+    ) -> StorageResult<Option<DaBlockHeight>>;
+
+    /// Get consensus parameters based on a version
+    fn get_consensus_parameters(
+        &self,
+        consensus_parameters_version: u32,
+    ) -> StorageResult<ConsensusParameters>;
+}