TxPool v2 General architecture

CHANGELOG.md

@@ -11,6 +11,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.
+- [2193](https://github.com/FuelLabs/fuel-core/pull/2193): Insertion in PoolV2 and tests refactoring
 
 
 ### Changed

Cargo.toml

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

crates/services/txpool/src/containers/dependency.rs

@@ -294,12 +294,15 @@ impl Dependency {
                                 }
                             } else {
                                 // tx output is in pool
-                                let output_tx = txs.get(utxo_id.tx_id()).unwrap();
-                                let output =
-                                    &output_tx.outputs()[utxo_id.output_index() as usize];
-                                Self::check_if_coin_input_can_spend_output(
-                                    output, input, false,
-                                )?;
+                                let output_tx = txs.get(utxo_id.tx_id());
+                                let output = output_tx.and_then(|tx| {
+                                    tx.outputs().get(utxo_id.output_index() as usize)
+                                });
+                                if let Some(output) = output {
+                                    Self::check_if_coin_input_can_spend_output(
+                                        output, input, false,
+                                    )?;
+                                }
                             }
 
                             collided.push(*spend_by);

crates/services/txpool/src/transaction_selector.rs

@@ -72,7 +72,10 @@ mod tests {
             checked_transaction::builder::TransactionBuilderExt,
             SecretKey,
         },
-        services::txpool::PoolTransaction,
+        services::txpool::{
+            Metadata,
+            PoolTransaction,
+        },
     };
     use itertools::Itertools;
     use std::sync::Arc;
@@ -157,7 +160,7 @@ mod tests {
                 // so it doesn't need to compute valid sigs for tests
                 PoolTransaction::Script(
                     script.finalize_checked_basic(Default::default()),
-                    ConsensusParametersVersion::MIN,
+                    Metadata::new(ConsensusParametersVersion::MIN),
                 )
             })
             .map(Arc::new)

crates/services/txpool/src/txpool.rs

@@ -64,14 +64,19 @@ use fuel_core_types::{
         checked_transaction::CheckPredicateParams,
         interpreter::Memory,
     },
-    services::executor::TransactionExecutionStatus,
+    services::{
+        executor::TransactionExecutionStatus,
+        txpool::Metadata,
+    },
 };
 use std::{
     collections::HashMap,
     ops::Deref,
     sync::Arc,
 };
 
+#[cfg(test)]
+mod stability_test;
 #[cfg(test)]
 mod test_helpers;
 #[cfg(test)]
@@ -343,26 +348,35 @@ where
         tx: Checked<Transaction>,
     ) -> Result<InsertionResult, Error> {
         let view = self.database.latest_view().unwrap();
-        self.insert_inner(tx, ConsensusParametersVersion::MIN, &view)
+        self.insert_inner(tx, Metadata::new(ConsensusParametersVersion::MIN), &view)
+    }
+
+    #[cfg(test)]
+    fn insert_single_with_metadata(
+        &mut self,
+        tx: Checked<Transaction>,
+        metadata: Metadata,
+    ) -> Result<InsertionResult, Error> {
+        let view = self.database.latest_view().unwrap();
+        self.insert_inner(tx, metadata, &view)
     }
 
-    #[tracing::instrument(level = "debug", skip_all, fields(tx_id = %tx.id()), ret, err)]
     // this is atomic operation. Return removed(pushed out/replaced) transactions
     fn insert_inner(
         &mut self,
         tx: Checked<Transaction>,
-        version: ConsensusParametersVersion,
+        metadata: Metadata,
         view: &View,
     ) -> Result<InsertionResult, Error> {
         let tx: CheckedTransaction = tx.into();
 
         let tx = Arc::new(match tx {
-            CheckedTransaction::Script(tx) => PoolTransaction::Script(tx, version),
-            CheckedTransaction::Create(tx) => PoolTransaction::Create(tx, version),
+            CheckedTransaction::Script(tx) => PoolTransaction::Script(tx, metadata),
+            CheckedTransaction::Create(tx) => PoolTransaction::Create(tx, metadata),
             CheckedTransaction::Mint(_) => return Err(Error::MintIsDisallowed),
-            CheckedTransaction::Upgrade(tx) => PoolTransaction::Upgrade(tx, version),
-            CheckedTransaction::Upload(tx) => PoolTransaction::Upload(tx, version),
-            CheckedTransaction::Blob(tx) => PoolTransaction::Blob(tx, version),
+            CheckedTransaction::Upgrade(tx) => PoolTransaction::Upgrade(tx, metadata),
+            CheckedTransaction::Upload(tx) => PoolTransaction::Upload(tx, metadata),
+            CheckedTransaction::Blob(tx) => PoolTransaction::Blob(tx, metadata),
         });
 
         self.check_blacklisting(tx.as_ref())?;
@@ -450,7 +464,7 @@ where
         };
 
         for tx in txs.into_iter() {
-            res.push(self.insert_inner(tx, version, &view));
+            res.push(self.insert_inner(tx, Metadata::new(version), &view));
         }
 
         // announce to subscribers

crates/services/txpool/src/txpool/stability_test.rs

@@ -0,0 +1,150 @@
+#![allow(clippy::cast_possible_truncation)]
+#![allow(clippy::arithmetic_side_effects)]
+
+use crate::{
+    test_helpers::TextContext,
+    types::TxId,
+    Config,
+};
+use fuel_core_storage::rand::{
+    prelude::SliceRandom,
+    rngs::StdRng,
+    thread_rng,
+    Rng,
+    RngCore,
+    SeedableRng,
+};
+use fuel_core_types::{
+    fuel_tx::{
+        field::Tip,
+        ConsensusParameters,
+        Finalizable,
+        GasCosts,
+        Input,
+        Output,
+        Transaction,
+        TransactionBuilder,
+        UtxoId,
+    },
+    fuel_types::AssetId,
+    fuel_vm::checked_transaction::{
+        Checked,
+        IntoChecked,
+    },
+    services::txpool::Metadata,
+};
+
+#[derive(Debug, Clone, Copy)]
+struct Limits {
+    max_inputs: usize,
+    min_outputs: u16,
+    max_outputs: u16,
+    utxo_id_range: u8,
+    gas_limit_range: u64,
+}
+
+fn some_transaction(
+    limits: Limits,
+    tip: u64,
+    rng: &mut StdRng,
+) -> (Checked<Transaction>, Metadata) {
+    const AMOUNT: u64 = 10000;
+
+    let mut consensus_parameters = ConsensusParameters::standard();
+    consensus_parameters.set_gas_costs(GasCosts::free());
+
+    let mut builder = TransactionBuilder::script(vec![], vec![]);
+    builder.with_params(consensus_parameters.clone());
+
+    let mut owner = [0u8; 32];
+    owner[0] = 123;
+    owner[1] = 222;
+    let owner = owner.into();
+
+    let mut random_ids = (0..limits.utxo_id_range)
+        .map(|_| rng.gen_range(0..limits.utxo_id_range))
+        .collect::<Vec<_>>();
+    random_ids.sort();
+    random_ids.dedup();
+    random_ids.shuffle(rng);
+
+    let tx_id_byte = random_ids.pop().expect("No random ids");
+    let tx_id: TxId = [tx_id_byte; 32].into();
+    let max_gas = rng.gen_range(0..limits.gas_limit_range) + 1;
+
+    let inputs_count = limits.max_inputs.min(random_ids.len());
+    let inputs_count = rng.gen_range(1..inputs_count);
+    let inputs = random_ids.into_iter().take(inputs_count);
+
+    for input_utxo_id in inputs {
+        let output_index = rng.gen_range(0..limits.max_outputs);
+        let utxo_id = UtxoId::new([input_utxo_id; 32].into(), output_index);
+
+        builder.add_input(Input::coin_signed(
+            utxo_id,
+            owner,
+            AMOUNT,
+            AssetId::BASE,
+            Default::default(),
+            Default::default(),
+        ));
+    }
+
+    // We can't have more outputs than inputs.
+    let outputs = rng
+        .gen_range(limits.min_outputs..limits.max_outputs)
+        .min(inputs_count as u16);
+
+    for _ in 0..outputs {
+        let output = Output::coin(owner, AMOUNT, AssetId::BASE);
+        builder.add_output(output);
+    }
+
+    builder.add_witness(Default::default());
+
+    let mut tx = builder.finalize();
+    tx.set_tip(tip);
+    let tx = Transaction::from(tx);
+
+    let checked = tx
+        .into_checked_basic(0u32.into(), &consensus_parameters)
+        .unwrap();
+    let metadata = Metadata::new_test(0, Some(max_gas), Some(tx_id));
+
+    (checked, metadata)
+}
+
+#[test]
+fn stability_test() {
+    let config = Config {
+        utxo_validation: false,
+        ..Default::default()
+    };
+
+    let limit = Limits {
+        max_inputs: 4,
+        min_outputs: 1,
+        max_outputs: 4,
+        utxo_id_range: 12,
+        gas_limit_range: 1000,
+    };
+
+    for _ in 0..100 {
+        let seed = thread_rng().next_u64();
+        let mut rng = StdRng::seed_from_u64(seed);
+        tracing::info!("Running stability test with seed: {}", seed);
+
+        const ROUNDS: usize = 1000;
+        let mut errors = 0;
+
+        let mut txpool = TextContext::default().config(config.clone()).build();
+        for tip in 0..ROUNDS {
+            let (checked, metadata) = some_transaction(limit, tip as u64, &mut rng);
+            let result = txpool.insert_single_with_metadata(checked, metadata);
+            errors += result.is_err() as usize;
+        }
+
+        assert_ne!(ROUNDS, errors);
+        tracing::info!("Error rate: {:.02}", errors as f64 / ROUNDS as f64);
+    }
+}

crates/services/txpool_v2/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "fuel-core-txpool-v2"
+version = { workspace = true }
+authors = { workspace = true }
+categories = ["cryptography::cryptocurrencies"]
+edition = { workspace = true }
+homepage = { workspace = true }
+keywords = ["blockchain", "cryptocurrencies", "fuel-vm", "vm"]
+license = { workspace = true }
+repository = { workspace = true }
+description = "Transaction pool v2"
+publish = false
+
+[dependencies]
+anyhow = { workspace = true }
+async-trait = { workspace = true }
+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"] }
+tracing = { workspace = true }
+
+[dev-dependencies]
+fuel-core-trace = { workspace = true }
+rand = { workspace = true }
+
+[features]
+test-helpers = [
+  "dep:mockall",
+  "fuel-core-types/test-helpers",
+  "fuel-core-storage/test-helpers",
+]