Txpool v2 features for futures connections with other modules

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Added
+- [2216](https://github.com/FuelLabs/fuel-core/pull/2216): Add more function to the state and task of TxPoolV2 to handle the future interactions with others modules (PoA, BlockProducer, BlockImporter and P2P)
+
 ## [Version 0.37.1]
 
 ### Fixed

crates/services/txpool/src/txpool.rs

@@ -452,7 +452,8 @@ where
         };
 
         for tx in txs.into_iter() {
-            res.push(self.insert_inner(tx, Metadata::new(version), &view));
+            // Not used in this module
+            res.push(self.insert_inner(tx, Metadata::new(version, 0, 0), &view));
         }
 
         // announce to subscribers

crates/services/txpool_v2/Cargo.toml

@@ -19,21 +19,26 @@ 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-stream = { workspace = true }
 tracing = { workspace = true }
 
 [dev-dependencies]
+fuel-core-storage = { workspace = true, features = ["std", "test-helpers"] }
 fuel-core-trace = { workspace = true }
+mockall = { workspace = true }
+proptest = { workspace = true }
 rand = { workspace = true }
+test-strategy = { workspace = true }
+tokio = { workspace = true, features = ["sync", "test-util"] }
+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

@@ -1,5 +1,8 @@
 use std::{
-    collections::HashMap,
+    collections::{
+        BTreeMap,
+        HashMap,
+    },
     fmt::Debug,
     hash::Hash,
 };
@@ -23,6 +26,7 @@ use fuel_core_types::{
         ContractId,
         Input,
         Output,
+        TxId,
         UtxoId,
     },
     fuel_types::Nonce,
@@ -33,6 +37,7 @@ use crate::{
     error::{
         CollisionReason,
         Error,
+        InputValidationError,
     },
     storage::StorageData,
 };
@@ -46,7 +51,7 @@ pub struct BasicCollisionManager<StorageIndex> {
     /// Message -> Transaction that currently use the Message
     messages_spenders: HashMap<Nonce, StorageIndex>,
     /// Coins -> Transaction that currently use the UTXO
-    coins_spenders: HashMap<UtxoId, StorageIndex>,
+    coins_spenders: BTreeMap<UtxoId, StorageIndex>,
     /// Contract -> Transaction that currently create the contract
     contracts_creators: HashMap<ContractId, StorageIndex>,
     /// Blob -> Transaction that currently create the blob
@@ -57,7 +62,7 @@ impl<StorageIndex> BasicCollisionManager<StorageIndex> {
     pub fn new() -> Self {
         Self {
             messages_spenders: HashMap::new(),
-            coins_spenders: HashMap::new(),
+            coins_spenders: BTreeMap::new(),
             contracts_creators: HashMap::new(),
             blobs_users: HashMap::new(),
         }
@@ -84,6 +89,13 @@ where
 {
     type StorageIndex = StorageIndex;
 
+    fn get_coins_spenders(&self, tx_creator_id: &TxId) -> Vec<Self::StorageIndex> {
+        self.coins_spenders
+            .range(UtxoId::new(*tx_creator_id, 0)..UtxoId::new(*tx_creator_id, u16::MAX))
+            .map(|(_, storage_id)| *storage_id)
+            .collect()
+    }
+
     fn find_collisions(
         &self,
         transaction: &PoolTransaction,
@@ -137,7 +149,9 @@ where
                     );
 
                     if self.coins_spenders.contains_key(&utxo_id) {
-                        return Err(Error::DuplicateTxId(transaction.id()));
+                        return Err(Error::InputValidation(
+                            InputValidationError::DuplicateTxId(transaction.id()),
+                        ));
                     }
                 }
                 Output::Contract(_) => {}
@@ -152,7 +166,7 @@ where
         storage_id: StorageIndex,
         store_entry: &StorageData,
     ) {
-        if let PoolTransaction::Blob(checked_tx, _) = &store_entry.transaction {
+        if let PoolTransaction::Blob(checked_tx, _) = store_entry.transaction.as_ref() {
             let blob_id = checked_tx.transaction().blob_id();
             self.blobs_users.insert(*blob_id, storage_id);
         }

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

@@ -2,7 +2,10 @@ use crate::error::{
     CollisionReason,
     Error,
 };
-use fuel_core_types::services::txpool::PoolTransaction;
+use fuel_core_types::{
+    fuel_tx::TxId,
+    services::txpool::PoolTransaction,
+};
 use std::collections::HashMap;
 
 use crate::storage::StorageData;
@@ -21,6 +24,9 @@ pub trait CollisionManager {
         transaction: &PoolTransaction,
     ) -> Result<Collisions<Self::StorageIndex>, Error>;
 
+    /// Get spenders of coins UTXO created by a transaction ID.
+    fn get_coins_spenders(&self, tx_creator_id: &TxId) -> Vec<Self::StorageIndex>;
+
     /// Inform the collision manager that a transaction was stored.
     fn on_stored_transaction(
         &mut self,

crates/services/txpool_v2/src/config.rs

@@ -26,7 +26,7 @@ use fuel_core_types::{
     services::txpool::PoolTransaction,
 };
 
-use crate::error::Error;
+use crate::error::BlacklistedError;
 
 #[derive(Default, Debug, Clone, PartialEq, Eq)]
 pub struct BlackList {
@@ -57,21 +57,26 @@ impl BlackList {
     }
 
     /// Check if the transaction has blacklisted inputs.
-    pub fn check_blacklisting(&self, tx: &PoolTransaction) -> Result<(), Error> {
+    pub fn check_blacklisting(
+        &self,
+        tx: &PoolTransaction,
+    ) -> Result<(), BlacklistedError> {
         for input in tx.inputs() {
             match input {
                 Input::CoinSigned(CoinSigned { utxo_id, owner, .. })
                 | Input::CoinPredicate(CoinPredicate { utxo_id, owner, .. }) => {
                     if self.coins.contains(utxo_id) {
-                        return Err(Error::BlacklistedUTXO(*utxo_id));
+                        return Err(BlacklistedError::BlacklistedUTXO(*utxo_id));
                     }
                     if self.owners.contains(owner) {
-                        return Err(Error::BlacklistedOwner(*owner));
+                        return Err(BlacklistedError::BlacklistedOwner(*owner));
                     }
                 }
                 Input::Contract(contract) => {
                     if self.contracts.contains(&contract.contract_id) {
-                        return Err(Error::BlacklistedContract(contract.contract_id));
+                        return Err(BlacklistedError::BlacklistedContract(
+                            contract.contract_id,
+                        ));
                     }
                 }
                 Input::MessageCoinSigned(MessageCoinSigned {
@@ -99,13 +104,13 @@ impl BlackList {
                     ..
                 }) => {
                     if self.messages.contains(nonce) {
-                        return Err(Error::BlacklistedMessage(*nonce));
+                        return Err(BlacklistedError::BlacklistedMessage(*nonce));
                     }
                     if self.owners.contains(sender) {
-                        return Err(Error::BlacklistedOwner(*sender));
+                        return Err(BlacklistedError::BlacklistedOwner(*sender));
                     }
                     if self.owners.contains(recipient) {
-                        return Err(Error::BlacklistedOwner(*recipient));
+                        return Err(BlacklistedError::BlacklistedOwner(*recipient));
                     }
                 }
             }
@@ -119,14 +124,16 @@ impl BlackList {
 pub struct Config {
     /// Enable UTXO validation (will check if UTXO exists in the database and has correct data).
     pub utxo_validation: bool,
-    /// Maximum block size in bytes.
-    pub max_block_size: u64,
-    /// Maximum gas per block.
-    pub max_block_gas: u64,
+    /// Maximum of subscriptions to listen to updates of a transaction.
+    pub max_tx_update_subscriptions: usize,
     /// Maximum transactions per dependencies chain.
     pub max_txs_chain_count: usize,
     /// Pool limits
     pub pool_limits: PoolLimits,
+    /// Service channel limits
+    pub service_channel_limits: ServiceChannelLimits,
+    /// Interval for checking the time to live of transactions.
+    pub ttl_check_interval: Duration,
     /// Maximum transaction time to live.
     pub max_txs_ttl: Duration,
     /// Heavy async processing configuration.
@@ -145,32 +152,55 @@ pub struct PoolLimits {
     pub max_bytes_size: usize,
 }
 
+#[derive(Clone)]
+pub struct ServiceChannelLimits {
+    /// Maximum number of pending requests waiting in the write pool channel.
+    pub max_pending_write_pool_requests: usize,
+    /// Maximum number of pending requests waiting in the read pool channel.
+    pub max_pending_read_pool_requests: usize,
+    /// Maximum number of pending requests waiting in the select transactions channel.
+    pub max_pending_select_transactions_requests: 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,
+    /// Maximum number of threads for managing the p2p synchronisation
+    pub number_threads_p2p_sync: usize,
+    /// Maximum number of tasks in the p2p sync queue
+    pub size_of_p2p_sync_queue: usize,
 }
 
 #[cfg(test)]
 impl Default for Config {
     fn default() -> Self {
         Self {
             utxo_validation: true,
-            max_block_gas: 100000000,
-            max_block_size: 1000000000,
-            max_txs_chain_count: 1000,
+            max_tx_update_subscriptions: 1000,
+            max_txs_chain_count: 50,
+            ttl_check_interval: Duration::from_secs(60),
+            max_txs_ttl: Duration::from_secs(60 * 10),
+            black_list: BlackList::default(),
             pool_limits: PoolLimits {
                 max_txs: 10000,
                 max_gas: 100_000_000_000,
-                max_bytes_size: 10_000_000_000,
+                max_bytes_size: 1_000_000_000,
             },
-            max_txs_ttl: Duration::from_secs(60 * 10),
-            black_list: BlackList::default(),
             heavy_work: HeavyWorkConfig {
-                number_threads_to_verify_transactions: 4,
+                // It is important for tests to have only one thread for verification
+                // because some of them rely on the ordering of insertion.
+                number_threads_to_verify_transactions: 1,
                 size_of_verification_queue: 100,
+                number_threads_p2p_sync: 1,
+                size_of_p2p_sync_queue: 100,
+            },
+            service_channel_limits: ServiceChannelLimits {
+                max_pending_write_pool_requests: 1000,
+                max_pending_read_pool_requests: 1000,
+                max_pending_select_transactions_requests: 1000,
             },
         }
     }