Txpool v2 insertion (#2193)

REVIEWER NOTE : Review #2162
first for context
## Linked Issues/PRs
Closes #2186 

## Description

This PR allows you to insert `Transaction` in the `TxPool v2` service
and so perform all the necessary verifications and the insertion.

### Changes in code logic from `TxPool v1`

- The verifications are performed in a new order specified in #2186. The
goal is to avoid making the computation heavy work if the simple checks
aren't valid. In this new version we also ensure that verifications are
done in order by having wrapper type around each step to allow only one
verification path.

- The insertion is performed in a separate thread pool, the goal is to
not block the pool on any verifications/insertions and to manage the
ressources we allocate to these works

- The insertion rules and conditions has change to the following : 
- A transaction with dependencies can collide only with one other
transaction
- A transaction without dependencies can collide with multiple
transaction
  - Rules to free up space for new transaction
- If a transaction is colliding with another verify if deleting the
colliding transaction and dependents subtree is enough otherwise refuses
the tx
- If a transaction is dependent and not enough space, don't accept
transaction
- If a transaction is executable, try to free has much space used by
less profitable transactions as possible in the pool to include it

- New limits on the size of the pool : `max_pool_bytes_size and
max_pool_gas`

### Changes from the base branch `txpool-v2`

All the tests has been refactored to be way more clearer to understand
and easier to read. They also now all follow the naming and the GWT
convention.

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

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

---------

Co-authored-by: Green Baneling <[email protected]>

Author: AurelienFT

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
 

Cargo.lock

@@ -18,8 +18,18 @@ 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/Cargo.toml

@@ -31,15 +31,14 @@ use fuel_core_types::{
 use num_rational::Ratio;
 
 use crate::{
-    error::Error,
+    error::{
+        CollisionReason,
+        Error,
+    },
     storage::StorageData,
 };
 
-use super::{
-    CollisionManager,
-    CollisionReason,
-    Collisions,
-};
+use super::CollisionManager;
 
 pub trait BasicCollisionManagerStorage {
     type StorageIndex: Copy + Debug + Hash + PartialEq + Eq;
@@ -76,104 +75,113 @@ impl<S: BasicCollisionManagerStorage> Default for BasicCollisionManager<S> {
 }
 
 impl<S: BasicCollisionManagerStorage> BasicCollisionManager<S> {
-    fn gather_colliding_txs(
+    fn is_better_than_collision(
         &self,
         tx: &PoolTransaction,
-    ) -> Result<Collisions<S::StorageIndex>, Error> {
-        let mut collisions = Collisions::new();
-        if let PoolTransaction::Blob(checked_tx, _) = tx {
+        collision: S::StorageIndex,
+        storage: &S,
+    ) -> bool {
+        let new_tx_ratio = Ratio::new(tx.tip(), tx.max_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,
+        );
+        new_tx_ratio > colliding_tx_ratio
+    }
+}
+
+impl<S: BasicCollisionManagerStorage> CollisionManager for BasicCollisionManager<S> {
+    type Storage = S;
+    type StorageIndex = S::StorageIndex;
+
+    fn collect_colliding_transactions(
+        &self,
+        transaction: &PoolTransaction,
+    ) -> Result<HashMap<Self::StorageIndex, Vec<CollisionReason>>, Error> {
+        let mut collisions = HashMap::new();
+        if let PoolTransaction::Blob(checked_tx, _) = transaction {
             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.insert(*state);
+                collisions.insert(*state, vec![CollisionReason::Blob(*blob_id)]);
             }
         }
-        for input in tx.inputs() {
+        for input in transaction.inputs() {
             match input {
                 Input::CoinSigned(CoinSigned { utxo_id, .. })
                 | 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.insert(*tx_id);
+                    if let Some(storage_id) = self.coins_spenders.get(utxo_id) {
+                        let entry = collisions.entry(*storage_id).or_default();
+                        entry.push(CollisionReason::Utxo(*utxo_id));
                     }
                 }
                 Input::MessageCoinSigned(MessageCoinSigned { nonce, .. })
                 | Input::MessageCoinPredicate(MessageCoinPredicate { nonce, .. })
                 | Input::MessageDataSigned(MessageDataSigned { nonce, .. })
                 | 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.insert(*tx_id);
+                    if let Some(storage_id) = self.messages_spenders.get(nonce) {
+                        let entry = collisions.entry(*storage_id).or_default();
+                        entry.push(CollisionReason::Message(*nonce));
                     }
                 }
                 // No collision for contract inputs
                 _ => {}
             }
         }
 
-        for output in tx.outputs() {
+        for output in transaction.outputs() {
             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.insert(*tx_id);
+                if let Some(storage_id) = self.contracts_creators.get(contract_id) {
+                    let entry = collisions.entry(*storage_id).or_default();
+                    entry.push(CollisionReason::ContractCreation(*contract_id));
                 }
             }
         }
         Ok(collisions)
     }
 
-    fn is_better_than_collisions(
-        &self,
-        tx: &PoolTransaction,
-        collisions: &Collisions<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 collision_tx_ratio = Ratio::new(total_tip, total_gas);
-
-        new_tx_ratio > collision_tx_ratio
-    }
-}
-
-impl<S: BasicCollisionManagerStorage> CollisionManager for BasicCollisionManager<S> {
-    type Storage = S;
-    type StorageIndex = S::StorageIndex;
-
-    fn collect_colliding_transactions(
+    /// Rules:
+    // A transaction with dependencies can collide only with one other transaction if it is less worth it
+    // A transaction without dependencies can collide with multiple transaction if they are less worth it
+    fn can_store_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)
+        has_dependencies: bool,
+        colliding_transactions: &HashMap<Self::StorageIndex, Vec<CollisionReason>>,
+        storage: &Self::Storage,
+    ) -> Result<(), CollisionReason> {
+        if colliding_transactions.is_empty() {
+            return Ok(());
+        }
+        if has_dependencies {
+            if colliding_transactions.len() > 1 {
+                return Err(CollisionReason::MultipleCollisions);
+            }
+            let (collision, reason) = colliding_transactions.iter().next().unwrap();
+            if !self.is_better_than_collision(transaction, *collision, storage) {
+                if let Some(reason) = reason.first() {
+                    return Err(reason.clone());
+                } else {
+                    return Err(CollisionReason::Unknown);
+                }
+            }
         } else {
-            Err(Error::Collided(format!(
-                "Transaction collides with other transactions: {:?}",
-                collisions.colliding_txs
-            )))
+            for (collision, reason) in colliding_transactions.iter() {
+                if !self.is_better_than_collision(transaction, *collision, storage) {
+                    if let Some(reason) = reason.first() {
+                        return Err(reason.clone());
+                    } else {
+                        return Err(CollisionReason::Unknown);
+                    }
+                }
+            }
         }
+        Ok(())
     }
 
     fn on_stored_transaction(

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

@@ -1,5 +1,8 @@
 use std::{
-    collections::HashSet,
+    collections::{
+        HashMap,
+        HashSet,
+    },
     fmt::Debug,
 };
 
@@ -13,61 +16,35 @@ use fuel_core_types::{
     services::txpool::PoolTransaction,
 };
 
-use crate::error::Error;
+use crate::error::{
+    CollisionReason,
+    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: HashSet<Idx>,
-}
-
-impl<Idx> Collisions<Idx> {
-    /// Create a new empty collision information.
-    pub fn new() -> Self {
-        Self {
-            reasons: HashSet::default(),
-            colliding_txs: HashSet::default(),
-        }
-    }
-
-    /// 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) -> &HashSet<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.
-    /// It returns an error if the transaction is less worthy than the colliding transactions.
-    /// It returns the information about the collisions.
+    /// Collect the transaction that collide with the given transaction.
+    /// Returns a list of storage indexes of the colliding transactions and the reason of the collision.
     fn collect_colliding_transactions(
         &self,
         transaction: &PoolTransaction,
+    ) -> Result<HashMap<Self::StorageIndex, Vec<CollisionReason>>, Error>;
+
+    /// Determine if the collisions allow the transaction to be stored.
+    /// Returns the reason of the collision if the transaction cannot be stored.
+    fn can_store_transaction(
+        &self,
+        transaction: &PoolTransaction,
+        has_dependencies: bool,
+        colliding_transactions: &HashMap<Self::StorageIndex, Vec<CollisionReason>>,
         storage: &Self::Storage,
-    ) -> Result<Collisions<Self::StorageIndex>, Error>;
+    ) -> Result<(), CollisionReason>;
 
     /// Inform the collision manager that a transaction was stored.
     fn on_stored_transaction(

crates/services/txpool_v2/src/collision_manager/mod.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_to_verify_transactions: usize,
+    /// Maximum of tasks in the heavy async processing queue.
+    pub size_of_verification_queue: 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_to_verify_transactions: 4,
+                size_of_verification_queue: 100,
+            },
         }
     }
 }