Make transaction status stream work

Cargo.toml

@@ -99,6 +99,8 @@ serde_json = "1.0"
 reqwest = { version = "0.11.16", default-features = false, features = ["rustls-tls", "cookies"] }
 mockall = "0.11"
 test-case = "2.2"
+test-strategy = "0.3"
+proptest = "1.1"
 axum = "0.5"
 lazy_static = "1.4"
 libp2p-prom-client = { package = "prometheus-client", version = "0.18" }

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

@@ -163,11 +163,25 @@ pub struct Command {
 
     #[clap(long = "verify_max_da_lag", default_value = "10", env)]
     pub max_da_lag: u64,
+
     #[clap(long = "verify_max_relayer_wait", default_value = "30s", env)]
     pub max_wait_time: humantime::Duration,
 
+    /// The max time to live of the transaction inside of the `TxPool`.
     #[clap(long = "tx-pool-ttl", default_value = "5m", env)]
     pub tx_pool_ttl: 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 maximum number of active subscriptions that supported by the `TxPool`.
+    #[clap(long = "tx-number-active-subscriptions", default_value = "4064", env)]
+    pub tx_number_active_subscriptions: usize,
 }
 
 impl Command {
@@ -198,6 +212,9 @@ impl Command {
             max_da_lag,
             max_wait_time,
             tx_pool_ttl,
+            tx_max_number,
+            tx_max_depth,
+            tx_number_active_subscriptions,
         } = self;
 
         let addr = net::SocketAddr::new(ip, port);
@@ -265,11 +282,14 @@ impl Command {
                 backtrace: vm_backtrace,
             },
             txpool: TxPoolConfig::new(
+                tx_max_number,
+                tx_max_depth,
                 chain_conf,
                 min_gas_price,
                 utxo_validation,
                 metrics,
                 tx_pool_ttl.into(),
+                tx_number_active_subscriptions,
             ),
             block_producer: ProducerConfig {
                 utxo_validation,

crates/fuel-core/Cargo.toml

@@ -67,7 +67,9 @@ fuel-core-storage = { path = "./../storage", features = ["test-helpers"] }
 fuel-core-trace = { path = "./../trace" }
 fuel-core-types = { path = "./../types", features = ["test-helpers"] }
 mockall = { workspace = true }
+proptest = { workspace = true }
 test-case = { workspace = true }
+test-strategy = { workspace = true }
 
 [features]
 dap = ["dep:uuid"]

crates/fuel-core/src/graphql_api/ports.rs

@@ -1,5 +1,8 @@
 use async_trait::async_trait;
-use fuel_core_services::stream::BoxStream;
+use fuel_core_services::stream::{
+    BoxFuture,
+    BoxStream,
+};
 use fuel_core_storage::{
     iter::{
         BoxedIter,
@@ -21,7 +24,7 @@ use fuel_core_storage::{
     Result as StorageResult,
     StorageInspect,
 };
-use fuel_core_txpool::service::TxUpdate;
+use fuel_core_txpool::service::TxStatusMessage;
 use fuel_core_types::{
     blockchain::primitives::{
         BlockId,
@@ -55,7 +58,6 @@ use fuel_core_types::{
     tai64::Tai64,
 };
 use std::sync::Arc;
-use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
 
 /// The database port expected by GraphQL API service.
 pub trait DatabasePort:
@@ -161,7 +163,8 @@ pub trait TxPoolPort: Send + Sync {
 
     fn tx_update_subscribe(
         &self,
-    ) -> BoxStream<anyhow::Result<TxUpdate, BroadcastStreamRecvError>>;
+        tx_id: TxId,
+    ) -> BoxFuture<'_, BoxStream<TxStatusMessage>>;
 }
 
 #[async_trait]

crates/fuel-core/src/p2p_test_helpers.rs

@@ -21,6 +21,7 @@ use fuel_core_storage::{
     tables::Transactions,
     StorageAsRef,
 };
+use fuel_core_txpool::ports::BlockImporter;
 use fuel_core_types::{
     fuel_asm::{
         op,
@@ -349,10 +350,10 @@ impl Node {
     /// Wait for the node to reach consistency with the given transactions.
     pub async fn consistency(&mut self, txs: &HashMap<Bytes32, Transaction>) {
         let Self { db, .. } = self;
-        let mut tx_status = self.node.shared.txpool.tx_status_subscribe();
+        let mut blocks = self.node.shared.block_importer.block_events();
         while !not_found_txs(db, txs).is_empty() {
             tokio::select! {
-                result = tx_status.recv() => {
+                result = blocks.next() => {
                     result.unwrap();
                 }
                 _ = self.node.await_stop() => {