Txpool v2 insertion

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2182](https://github.com/FuelLabs/fuel-core/pull/2151): Limit number of transactions that can be fetched via TxSource::next
 - [2189](https://github.com/FuelLabs/fuel-core/pull/2151): Select next DA height to never include more than u16::MAX -1 transactions from L1.
 - [2162](https://github.com/FuelLabs/fuel-core/pull/2162): Pool structure with dependencies, etc.. for the next transaction pool module.
+- [2193](https://github.com/FuelLabs/fuel-core/pull/2193): Insertion in PoolV2 and tests refactoring
+
 
 ### Changed
 

crates/services/txpool_v2/Cargo.toml

@@ -18,8 +18,19 @@ derive_more = { workspace = true }
 fuel-core-services = { workspace = true }
 fuel-core-storage = { workspace = true, features = ["std"] }
 fuel-core-types = { workspace = true, features = ["test-helpers"] }
+futures = { workspace = true }
+mockall = { workspace = true, optional = true }
 num-rational = { workspace = true }
+parking_lot = { workspace = true }
 petgraph = "0.6.5"
+rayon = { workspace = true }
 tokio = { workspace = true, default-features = false, features = ["sync"] }
 tokio-rayon = { workspace = true }
 tracing = { workspace = true }
+
+[features]
+test-helpers = [
+  "dep:mockall",
+  "fuel-core-types/test-helpers",
+  "fuel-core-storage/test-helpers",
+]

crates/services/txpool_v2/src/collision_manager/basic.rs

@@ -34,14 +34,10 @@ use crate::{
     storage::StorageData,
 };
 
-use super::{
-    CollisionManager,
-    CollisionReason,
-    Collisions,
-};
+use super::CollisionManager;
 
 pub trait BasicCollisionManagerStorage {
-    type StorageIndex: Copy + Debug;
+    type StorageIndex: Copy + Debug + PartialEq + Eq;
 
     fn get(&self, index: &Self::StorageIndex) -> Result<&StorageData, Error>;
 }
@@ -75,16 +71,23 @@ impl<S: BasicCollisionManagerStorage> Default for BasicCollisionManager<S> {
 }
 
 impl<S: BasicCollisionManagerStorage> BasicCollisionManager<S> {
-    fn gather_colliding_txs(
+    fn gather_colliding_tx(
         &self,
         tx: &PoolTransaction,
-    ) -> Result<Collisions<S::StorageIndex>, Error> {
-        let mut collisions = Collisions::new();
+    ) -> Result<Option<S::StorageIndex>, Error> {
+        let mut collision: Option<S::StorageIndex> = None;
         if let PoolTransaction::Blob(checked_tx, _) = tx {
             let blob_id = checked_tx.transaction().blob_id();
             if let Some(state) = self.blobs_users.get(blob_id) {
-                collisions.reasons.insert(CollisionReason::Blob(*blob_id));
-                collisions.colliding_txs.push(*state);
+                if let Some(collision) = collision {
+                    if state != &collision {
+                        return Err(Error::Collided(format!(
+                            "Transaction collides with other transactions: {:?}",
+                            collision
+                        )));
+                    }
+                }
+                collision = Some(*state);
             }
         }
         for input in tx.inputs() {
@@ -93,8 +96,15 @@ impl<S: BasicCollisionManagerStorage> BasicCollisionManager<S> {
                 | Input::CoinPredicate(CoinPredicate { utxo_id, .. }) => {
                     // Check if the utxo is already spent by another transaction in the pool
                     if let Some(tx_id) = self.coins_spenders.get(utxo_id) {
-                        collisions.reasons.insert(CollisionReason::Coin(*utxo_id));
-                        collisions.colliding_txs.push(*tx_id);
+                        if let Some(collision) = collision {
+                            if tx_id != &collision {
+                                return Err(Error::Collided(format!(
+                                    "Transaction collides with other transactions: {:?}",
+                                    collision
+                                )));
+                            }
+                        }
+                        collision = Some(*tx_id);
                     }
                 }
                 Input::MessageCoinSigned(MessageCoinSigned { nonce, .. })
@@ -103,8 +113,15 @@ impl<S: BasicCollisionManagerStorage> BasicCollisionManager<S> {
                 | Input::MessageDataPredicate(MessageDataPredicate { nonce, .. }) => {
                     // Check if the message is already spent by another transaction in the pool
                     if let Some(tx_id) = self.messages_spenders.get(nonce) {
-                        collisions.reasons.insert(CollisionReason::Message(*nonce));
-                        collisions.colliding_txs.push(*tx_id);
+                        if let Some(collision) = collision {
+                            if tx_id != &collision {
+                                return Err(Error::Collided(format!(
+                                    "Transaction collides with other transactions: {:?}",
+                                    collision
+                                )));
+                            }
+                        }
+                        collision = Some(*tx_id);
                     }
                 }
                 // No collision for contract inputs
@@ -116,62 +133,60 @@ impl<S: BasicCollisionManagerStorage> BasicCollisionManager<S> {
             if let Output::ContractCreated { contract_id, .. } = output {
                 // Check if the contract is already created by another transaction in the pool
                 if let Some(tx_id) = self.contracts_creators.get(contract_id) {
-                    collisions
-                        .reasons
-                        .insert(CollisionReason::ContractCreation(*contract_id));
-                    collisions.colliding_txs.push(*tx_id);
+                    if let Some(collision) = collision {
+                        if tx_id != &collision {
+                            return Err(Error::Collided(format!(
+                                "Transaction collides with other transactions: {:?}",
+                                collision
+                            )));
+                        }
+                    }
+                    collision = Some(*tx_id);
                 }
             }
         }
-        Ok(collisions)
+        Ok(collision)
     }
 
-    fn is_better_than_collisions(
+    fn is_better_than_collision(
         &self,
         tx: &PoolTransaction,
-        collisions: &Collisions<S::StorageIndex>,
+        collision: S::StorageIndex,
         storage: &S,
     ) -> bool {
         let new_tx_ratio = Ratio::new(tx.tip(), tx.max_gas());
-        let (total_tip, total_gas) = collisions.colliding_txs.iter().fold(
-            (0u64, 0u64),
-            |(total_tip, total_gas), node_id| {
-                let dependent_tx = storage
-                    .get(node_id)
-                    .expect("Transaction always should exist in storage");
-                let total_tip =
-                    total_tip.saturating_add(dependent_tx.dependents_cumulative_tip);
-                let total_gas =
-                    total_gas.saturating_add(dependent_tx.dependents_cumulative_gas);
-                (total_tip, total_gas)
-            },
+        let colliding_tx = storage
+            .get(&collision)
+            .expect("Transaction always should exist in storage");
+        let colliding_tx_ratio = Ratio::new(
+            colliding_tx.dependents_cumulative_tip,
+            colliding_tx.dependents_cumulative_gas,
         );
-
-        let collision_tx_ratio = Ratio::new(total_tip, total_gas);
-
-        new_tx_ratio > collision_tx_ratio
+        new_tx_ratio > colliding_tx_ratio
     }
 }
 
 impl<S: BasicCollisionManagerStorage> CollisionManager for BasicCollisionManager<S> {
     type Storage = S;
     type StorageIndex = S::StorageIndex;
 
-    fn collect_colliding_transactions(
+    fn collect_colliding_transaction(
         &self,
         transaction: &PoolTransaction,
         storage: &S,
-    ) -> Result<Collisions<S::StorageIndex>, Error> {
-        let collisions = self.gather_colliding_txs(transaction)?;
-        if collisions.colliding_txs.is_empty() {
-            Ok(Collisions::new())
-        } else if self.is_better_than_collisions(transaction, &collisions, storage) {
-            Ok(collisions)
+    ) -> Result<Option<S::StorageIndex>, Error> {
+        let collision = self.gather_colliding_tx(transaction)?;
+        if let Some(collision) = collision {
+            if self.is_better_than_collision(transaction, collision, storage) {
+                Ok(Some(collision))
+            } else {
+                Err(Error::Collided(format!(
+                    "Transaction collides with other transactions: {:?}",
+                    collision
+                )))
+            }
         } else {
-            Err(Error::Collided(format!(
-                "Transaction collides with other transactions: {:?}",
-                collisions.colliding_txs
-            )))
+            Ok(None)
         }
     }
 

crates/services/txpool_v2/src/collision_manager/mod.rs

@@ -17,57 +17,21 @@ use crate::error::Error;
 
 pub mod basic;
 
-/// The reason why a transaction collides with another.
-/// It also contains additional information about the collision.
-#[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)]
-pub enum CollisionReason {
-    Coin(UtxoId),
-    Blob(BlobId),
-    Message(Nonce),
-    ContractCreation(ContractId),
-}
-
-/// Contains all the information about the collisions of a transaction.
-#[derive(Default, Debug)]
-pub struct Collisions<Idx> {
-    reasons: HashSet<CollisionReason>,
-    colliding_txs: Vec<Idx>,
-}
-
-impl<Idx> Collisions<Idx> {
-    /// Create a new empty collision information.
-    pub fn new() -> Self {
-        Self {
-            reasons: HashSet::default(),
-            colliding_txs: vec![],
-        }
-    }
-
-    /// Get the reasons of the collisions.
-    pub fn reasons(&self) -> &HashSet<CollisionReason> {
-        &self.reasons
-    }
-
-    /// Get the transactions that collide with the transaction.
-    pub fn colliding_txs(&self) -> &[Idx] {
-        &self.colliding_txs
-    }
-}
-
 pub trait CollisionManager {
     /// Storage type of the collision manager.
     type Storage;
     /// Index that identifies a transaction in the storage.
     type StorageIndex;
 
-    /// Collect all the transactions that collide with the given transaction.
+    /// Collect the transaction that collide with the given transaction.
     /// It returns an error if the transaction is less worthy than the colliding transactions.
-    /// It returns the information about the collisions.
-    fn collect_colliding_transactions(
+    /// It returns an error if the transaction collide with two transactions.
+    /// It returns the information about the collision if exists, none otherwise.
+    fn collect_colliding_transaction(
         &self,
         transaction: &PoolTransaction,
         storage: &Self::Storage,
-    ) -> Result<Collisions<Self::StorageIndex>, Error>;
+    ) -> Result<Option<Self::StorageIndex>, Error>;
 
     /// Inform the collision manager that a transaction was stored.
     fn on_stored_transaction(

crates/services/txpool_v2/src/config.rs

@@ -115,6 +115,7 @@ impl BlackList {
     }
 }
 
+#[derive(Clone)]
 pub struct Config {
     /// Enable UTXO validation (will check if UTXO exists in the database and has correct data).
     pub utxo_validation: bool,
@@ -124,14 +125,34 @@ pub struct Config {
     pub max_block_gas: u64,
     /// Maximum transactions per dependencies chain.
     pub max_txs_chain_count: usize,
-    /// Maximum transactions in the pool.
-    pub max_txs: u64,
+    /// Pool limits
+    pub pool_limits: PoolLimits,
     /// Maximum transaction time to live.
     pub max_txs_ttl: Duration,
+    /// Heavy async processing configuration.
+    pub heavy_work: HeavyWorkConfig,
     /// Blacklist. Transactions with blacklisted inputs will not be accepted.
     pub black_list: BlackList,
 }
 
+#[derive(Clone)]
+pub struct PoolLimits {
+    /// Maximum number of transactions in the pool.
+    pub max_txs: usize,
+    /// Maximum number of gas in the pool.
+    pub max_gas: u64,
+    /// Maximum number of bytes in the pool.
+    pub max_bytes_size: usize,
+}
+
+#[derive(Clone)]
+pub struct HeavyWorkConfig {
+    /// Maximum of threads for managing verifications/insertions.
+    pub number_threads_verif_insert_transactions: usize,
+    /// Maximum of tasks in the heavy async processing queue.
+    pub number_pending_tasks_threads_verif_insert_transactions: usize,
+}
+
 #[cfg(test)]
 impl Default for Config {
     fn default() -> Self {
@@ -140,9 +161,17 @@ impl Default for Config {
             max_block_gas: 100000000,
             max_block_size: 1000000000,
             max_txs_chain_count: 1000,
-            max_txs: 10000,
+            pool_limits: PoolLimits {
+                max_txs: 10000,
+                max_gas: 100_000_000_000,
+                max_bytes_size: 10_000_000_000,
+            },
             max_txs_ttl: Duration::from_secs(60 * 10),
             black_list: BlackList::default(),
+            heavy_work: HeavyWorkConfig {
+                number_threads_verif_insert_transactions: 4,
+                number_pending_tasks_threads_verif_insert_transactions: 100,
+            },
         }
     }
 }

crates/services/txpool_v2/src/error.rs

@@ -9,6 +9,8 @@ use fuel_core_types::{
     fuel_vm::checked_transaction::CheckError,
 };
 
+use crate::ports::WasmValidityError;
+
 #[derive(Debug, derive_more::Display)]
 pub enum Error {
     #[display(fmt = "Gas price not found for block height {_0}")]
@@ -30,6 +32,8 @@ pub enum Error {
     // TODO: Make more specific errors: https://github.com/FuelLabs/fuel-core/issues/2185
     #[display(fmt = "Transaction collided: {_0}")]
     Collided(String),
+    #[display(fmt = "Transaction is not inserted. Collision is also a dependency")]
+    NotInsertedCollisionIsDependency,
     #[display(fmt = "Utxo not found: {_0}")]
     UtxoNotFound(UtxoId),
     #[display(fmt = "The UTXO `{_0}` is blacklisted")]
@@ -82,6 +86,10 @@ pub enum Error {
     NotInsertedLimitHit,
     #[display(fmt = "Storage error: {_0}")]
     Storage(String),
+    #[display(fmt = "Error with Wasm validity: {:?}", _0)]
+    WasmValidity(WasmValidityError),
+    #[display(fmt = "Transaction is not inserted. Mint transaction is not allowed")]
+    MintIsDisallowed,
 }
 
 impl From<CheckError> for Error {