Linking TxPoolV2 in place of V1 (#2263)

Closes #2020
Closes #1961
Closes #868
Closes #2185
Closes #2186
Closes #2255
Closes #2160
Closes #1967
Closes #2187

## Description
Remove code of the previous txpool and connect all modules to this one

TODO: Improve description @AurelienFT 

## 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 <XgreenX9999@gmail.com>

Author: AurelienFT

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### Added
 - [2309](https://github.com/FuelLabs/fuel-core/pull/2309): Limit number of concurrent queries to the graphql service.
 - [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)
+- [2263](https://github.com/FuelLabs/fuel-core/pull/2263): Use the Txpool v2 in the whole codebase
 
 ### Removed
 - [2306](https://github.com/FuelLabs/fuel-core/pull/2306): Removed hack for genesis asset contract from the code.
@@ -27,8 +28,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2131](https://github.com/FuelLabs/fuel-core/pull/2131): Add flow in TxPool in order to ask to newly connected peers to share their transaction pool
 - [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. Also adds insertion/verification process in PoolV2 and tests refactoring
 - [2265](https://github.com/FuelLabs/fuel-core/pull/2265): Integrate Block Committer API for DA Block Costs.
+- [2162](https://github.com/FuelLabs/fuel-core/pull/2162): Pool structure with dependencies, etc.. for the next transaction pool module. Also adds insertion/verification process in PoolV2 and tests refactoring
 - [2280](https://github.com/FuelLabs/fuel-core/pull/2280): Allow comma separated relayer addresses in cli
 - [2299](https://github.com/FuelLabs/fuel-core/pull/2299): Support blobs in the predicates.
 - [2300](https://github.com/FuelLabs/fuel-core/pull/2300): Added new function to `fuel-core-client` for checking whether a blob exists.

Cargo.lock

@@ -25,7 +25,6 @@ members = [
   "crates/services/producer",
   "crates/services/relayer",
   "crates/services/sync",
-  "crates/services/txpool",
   "crates/services/txpool_v2",
   "crates/services/upgradable-executor",
   "crates/services/upgradable-executor/wasm-executor",
@@ -77,8 +76,7 @@ fuel-core-p2p = { version = "0.37.1", path = "./crates/services/p2p" }
 fuel-core-producer = { version = "0.37.1", path = "./crates/services/producer" }
 fuel-core-relayer = { version = "0.37.1", path = "./crates/services/relayer" }
 fuel-core-sync = { version = "0.37.1", path = "./crates/services/sync" }
-fuel-core-txpool = { version = "0.37.1", path = "./crates/services/txpool" }
-fuel-core-txpool-v2 = { version = "0.37.1", path = "./crates/services/txpool_v2" }
+fuel-core-txpool = { version = "0.37.1", path = "./crates/services/txpool_v2" }
 fuel-core-storage = { version = "0.37.1", path = "./crates/storage", default-features = false }
 fuel-core-trace = { version = "0.37.1", path = "./crates/trace" }
 fuel-core-types = { version = "0.37.1", path = "./crates/types", default-features = false }

Cargo.toml

@@ -27,7 +27,6 @@ use fuel_core::{
         Config,
         FuelService,
     },
-    txpool::types::Word,
 };
 use fuel_core_benches::{
     default_gas_costs::default_gas_costs,
@@ -57,6 +56,7 @@ use fuel_core_types::{
         GTFArgs,
         Instruction,
         RegId,
+        Word,
     },
     fuel_crypto::{
         secp256r1,
@@ -428,11 +428,8 @@ fn run_with_service_with_extra_inputs(
                 let mut sub = shared.block_importer.block_importer.subscribe();
                 shared
                     .txpool_shared_state
-                    .insert(vec![std::sync::Arc::new(tx)])
+                    .insert(tx)
                     .await
-                    .into_iter()
-                    .next()
-                    .expect("Should be at least 1 element")
                     .expect("Should include transaction successfully");
                 let res = sub.recv().await.expect("Should produce a block");
                 assert_eq!(res.tx_status.len(), 2, "res.tx_status: {:?}", res.tx_status);

benches/benches/block_target_gas.rs

@@ -104,20 +104,15 @@ where
                     for _ in 0..iters {
                         let mut test_builder = test_builder.clone();
                         let sealed_block = {
-                            let transactions = transactions
-                                .iter()
-                                .map(|tx| Arc::new(tx.clone()))
-                                .collect();
+                            let transactions: Vec<Transaction> =
+                                transactions.iter().cloned().collect();
                             // start the producer node
                             let TestContext { srv, client, .. } =
                                 test_builder.finalize().await;
 
                             // insert all transactions
-                            let results =
-                                srv.shared.txpool_shared_state.insert(transactions).await;
-                            for result in results {
-                                let result = result.expect("Should insert transaction");
-                                assert_eq!(result.removed.len(), 0);
+                            for tx in transactions {
+                                srv.shared.txpool_shared_state.insert(tx).await.unwrap();
                             }
                             let _ = client.produce_blocks(1, None).await;
 

benches/benches/transaction_throughput.rs

@@ -4,12 +4,12 @@ use fuel_core::service::{
 };
 
 // Add methods on commands
-use fuel_core::txpool::types::ContractId;
 use fuel_core_chain_config::{
     SnapshotMetadata,
     SnapshotReader,
 };
 use fuel_core_e2e_client::config::SuiteConfig;
+use fuel_core_types::fuel_tx::ContractId;
 use std::{
     fs,
     str::FromStr,

bin/e2e-test-client/tests/integration_tests.rs

@@ -33,9 +33,12 @@ use fuel_core::{
         RelayerConsensusConfig,
         VMConfig,
     },
-    txpool::{
-        config::BlackList,
+    txpool::config::{
+        BlackList,
         Config as TxPoolConfig,
+        HeavyWorkConfig,
+        PoolLimits,
+        ServiceChannelLimits,
     },
     types::{
         fuel_tx::ContractId,
@@ -438,22 +441,50 @@ impl Command {
 
         let TxPoolArgs {
             tx_pool_ttl,
+            tx_ttl_check_interval,
             tx_max_number,
-            tx_max_depth,
+            tx_max_total_bytes,
+            tx_max_total_gas,
+            tx_max_chain_count,
             tx_number_active_subscriptions,
             tx_blacklist_addresses,
             tx_blacklist_coins,
             tx_blacklist_messages,
             tx_blacklist_contracts,
+            tx_number_threads_to_verify_transactions,
+            tx_size_of_verification_queue,
+            tx_number_threads_p2p_sync,
+            tx_size_of_p2p_sync_queue,
+            tx_max_pending_read_requests,
+            tx_max_pending_write_requests,
         } = tx_pool;
 
-        let blacklist = BlackList::new(
+        let black_list = BlackList::new(
             tx_blacklist_addresses,
             tx_blacklist_coins,
             tx_blacklist_messages,
             tx_blacklist_contracts,
         );
 
+        let pool_limits = PoolLimits {
+            max_txs: tx_max_number,
+            max_gas: tx_max_total_gas,
+            max_bytes_size: tx_max_total_bytes,
+        };
+
+        let pool_heavy_work_config = HeavyWorkConfig {
+            number_threads_to_verify_transactions:
+                tx_number_threads_to_verify_transactions,
+            size_of_verification_queue: tx_size_of_verification_queue,
+            number_threads_p2p_sync: tx_number_threads_p2p_sync,
+            size_of_p2p_sync_queue: tx_size_of_p2p_sync_queue,
+        };
+
+        let service_channel_limits = ServiceChannelLimits {
+            max_pending_read_pool_requests: tx_max_pending_read_requests,
+            max_pending_write_pool_requests: tx_max_pending_write_requests,
+        };
+
         let config = Config {
             graphql_config: GraphQLConfig {
                 addr,
@@ -477,15 +508,17 @@ impl Command {
             vm: VMConfig {
                 backtrace: vm_backtrace,
             },
-            txpool: TxPoolConfig::new(
-                tx_max_number,
-                tx_max_depth,
+            txpool: TxPoolConfig {
+                max_txs_chain_count: tx_max_chain_count,
+                max_txs_ttl: tx_pool_ttl.into(),
+                ttl_check_interval: tx_ttl_check_interval.into(),
                 utxo_validation,
-                metrics,
-                tx_pool_ttl.into(),
-                tx_number_active_subscriptions,
-                blacklist,
-            ),
+                max_tx_update_subscriptions: tx_number_active_subscriptions,
+                black_list,
+                pool_limits,
+                heavy_work: pool_heavy_work_config,
+                service_channel_limits,
+            },
             block_producer: ProducerConfig {
                 coinbase_recipient,
                 metrics,

bin/fuel-core/src/cli/run.rs

@@ -189,6 +189,14 @@ pub struct P2PArgs {
     /// For peer reputations, the maximum time since last heartbeat before penalty
     #[clap(long = "heartbeat-max-time-since-last", default_value = "40", env)]
     pub heartbeat_max_time_since_last: u64,
+
+    /// Number of threads to read from the database.
+    #[clap(long = "p2p-database-read-threads", default_value = "2", env)]
+    pub database_read_threads: usize,
+
+    /// Number of threads to read from the tx pool.
+    #[clap(long = "p2p-txpool-threads", default_value = "0", env)]
+    pub tx_pool_threads: usize,
 }
 
 #[derive(Debug, Clone, Args)]
@@ -335,6 +343,8 @@ impl P2PArgs {
             info_interval: Some(Duration::from_secs(self.info_interval)),
             identify_interval: Some(Duration::from_secs(self.identify_interval)),
             metrics,
+            database_read_threads: self.database_read_threads,
+            tx_pool_threads: self.tx_pool_threads,
             state: NotInitialized,
         };
         Ok(Some(config))

bin/fuel-core/src/cli/run/p2p.rs

@@ -1,9 +1,9 @@
 //! Clap configuration related to TxPool service.
 
-use fuel_core::txpool::types::ContractId;
 use fuel_core_types::{
     fuel_tx::{
         Address,
+        ContractId,
         UtxoId,
     },
     fuel_types::Nonce,
@@ -15,13 +15,25 @@ pub struct TxPoolArgs {
     #[clap(long = "tx-pool-ttl", default_value = "5m", env)]
     pub tx_pool_ttl: humantime::Duration,
 
+    /// The interval for checking the time to live of transactions.
+    #[clap(long = "tx-ttl-check-interval", default_value = "1m", env)]
+    pub tx_ttl_check_interval: humantime::Duration,
+
     /// The max number of transactions that the `TxPool` can simultaneously store.
     #[clap(long = "tx-max-number", default_value = "4064", env)]
     pub tx_max_number: usize,
 
-    /// The max depth of the dependent transactions that supported by the `TxPool`.
-    #[clap(long = "tx-max-depth", default_value = "10", env)]
-    pub tx_max_depth: usize,
+    /// The max number of gas the `TxPool` can simultaneously store.
+    #[clap(long = "tx-max-total-gas", default_value = "30000000000", env)]
+    pub tx_max_total_gas: u64,
+
+    /// The max number of bytes that the `TxPool` can simultaneously store.
+    #[clap(long = "tx-max-total-bytes", default_value = "131072000", env)]
+    pub tx_max_total_bytes: usize,
+
+    /// The max number of tx in a chain of dependent transactions that supported by the `TxPool`.
+    #[clap(long = "tx-max-depth", default_value = "32", env)]
+    pub tx_max_chain_count: usize,
 
     /// The maximum number of active subscriptions that supported by the `TxPool`.
     #[clap(long = "tx-number-active-subscriptions", default_value = "4064", env)]
@@ -42,6 +54,34 @@ pub struct TxPoolArgs {
     /// The list of banned contracts ignored by the `TxPool`.
     #[clap(long = "tx-blacklist-contracts", value_delimiter = ',', env)]
     pub tx_blacklist_contracts: Vec<ContractId>,
+
+    /// Number of threads for managing verifications/insertions.
+    #[clap(
+        long = "tx-number-threads-to-verify-transactions",
+        default_value = "4",
+        env
+    )]
+    pub tx_number_threads_to_verify_transactions: usize,
+
+    /// Maximum number of tasks in the verifications/insertions queue.
+    #[clap(long = "tx-size-of-verification-queue", default_value = "2000", env)]
+    pub tx_size_of_verification_queue: usize,
+
+    /// Number of threads for managing the p2p synchronisation.
+    #[clap(long = "tx-number-threads-p2p-sync", default_value = "2", env)]
+    pub tx_number_threads_p2p_sync: usize,
+
+    /// Maximum number of tasks in the p2p synchronisation queue.
+    #[clap(long = "tx-size-of-p2p-sync-queue", default_value = "20", env)]
+    pub tx_size_of_p2p_sync_queue: usize,
+
+    /// Maximum number of pending write requests in the service.
+    #[clap(long = "tx-max-pending-write-requests", default_value = "500", env)]
+    pub tx_max_pending_write_requests: usize,
+
+    /// Maximum number of pending read requests in the service.
+    #[clap(long = "tx-max-pending-read-requests", default_value = "1000", env)]
+    pub tx_max_pending_read_requests: usize,
 }
 
 #[cfg(test)]

bin/fuel-core/src/cli/run/tx_pool.rs

@@ -18,11 +18,11 @@ use fuel_core_storage::{
     Result as StorageResult,
     StorageAsRef,
 };
-use fuel_core_txpool::types::TxId;
 use fuel_core_types::{
     entities::coins::coin::CompressedCoin,
     fuel_tx::{
         Address,
+        TxId,
         UtxoId,
     },
 };