Increase usage of new Fuel_txpool (#238)

Adds the new mempool to find_tx_by_hash and submit_tx. Future PR will complete the issue adding find_dependent and fixing find_tx_by_hash to use tx status rather than mempool existence

fuel-core/src/schema/tx.rs

@@ -41,8 +41,17 @@ impl TxQuery {
     ) -> async_graphql::Result<Option<Transaction>> {
         let db = ctx.data_unchecked::<Database>();
         let key = id.0;
-        Ok(Storage::<fuel_types::Bytes32, FuelTx>::get(db, &key)?
-            .map(|tx| Transaction(tx.into_owned())))
+
+        let tx_pool = ctx.data::<Arc<TxPool>>().unwrap();
+
+        let found_tx = tx_pool.pool().find(&[key]).await;
+
+        if let Some(Some(transaction)) = found_tx.get(0) {
+            Ok(Some(Transaction((transaction.deref()).clone())))
+        } else {
+            Ok(Storage::<fuel_types::Bytes32, FuelTx>::get(db, &key)?
+                .map(|tx| Transaction(tx.into_owned())))
+        }
     }
 
     async fn transactions(

fuel-core/src/tx_pool.rs

@@ -37,6 +37,8 @@ pub enum Error {
     Database(Box<dyn StdError>),
     #[error("unexpected block execution error {0:?}")]
     Execution(crate::executor::Error),
+    #[error("Tx is invalid, insertion failed {0:?}")]
+    Other(#[from] anyhow::Error),
 }
 
 impl From<KvStoreError> for Error {
@@ -83,9 +85,43 @@ impl TxPool {
 
     pub async fn submit_tx(&self, tx: Transaction) -> Result<Bytes32, Error> {
         let db = self.db.clone();
-        let tx_id = tx.id();
+
+        let mut tx_to_exec = tx.clone();
+
+        let includable_txs: Vec<Transaction>;
+
+        if self.executor.config.utxo_validation {
+            if tx_to_exec.metadata().is_none() {
+                tx_to_exec.precompute_metadata();
+            }
+
+            self.fuel_txpool
+                .insert(vec![Arc::new(tx_to_exec.clone())])
+                .await
+                .into_iter()
+                .collect::<Result<Vec<_>, _>>()?;
+
+            let includable_arc_txs = self.fuel_txpool.includable().await;
+
+            includable_txs = includable_arc_txs
+                .iter()
+                .map(|arc| Transaction::clone(&*arc))
+                .collect();
+
+            for included_tx in includable_arc_txs {
+                self.fuel_txpool.remove(&[included_tx.id()]).await;
+            }
+        } else {
+            includable_txs = vec![tx];
+        }
+
+        let tx_id = tx_to_exec.id();
+
         // set status to submitted
-        db.update_tx_status(&tx_id, TransactionStatus::Submitted { time: Utc::now() })?;
+        db.update_tx_status(
+            &tx_id.clone(),
+            TransactionStatus::Submitted { time: Utc::now() },
+        )?;
 
         // setup and execute block
         let current_height = db.get_block_height()?.unwrap_or_default();
@@ -103,7 +139,7 @@ impl TxPool {
                 // TODO: compute the current merkle root of all blocks
                 prev_root: Default::default(),
             },
-            transactions: vec![tx],
+            transactions: includable_txs,
         };
         // immediately execute block
         self.executor

fuel-tests/tests/tx.rs

@@ -8,6 +8,7 @@ use fuel_core::{
 };
 use fuel_gql_client::client::types::TransactionStatus;
 use fuel_gql_client::client::{FuelClient, PageDirection, PaginationRequest};
+use fuel_vm::util::test_helpers::TestBuilder as TxBuilder;
 use fuel_vm::{consts::*, prelude::*};
 use itertools::Itertools;
 use rand::{rngs::StdRng, Rng, SeedableRng};
@@ -121,6 +122,50 @@ async fn submit() {
     assert_eq!(tx.id(), ret_tx.id());
 }
 
+#[tokio::test]
+async fn submit_utxo_verified_tx() {
+    let config = Config {
+        utxo_validation: true,
+        ..Config::local_node()
+    };
+
+    let srv = FuelService::new_node(config).await.unwrap();
+    let client = FuelClient::from(srv.bound_address);
+
+    let transactions = (1..10 + 1)
+        .into_iter()
+        .map(|i| TxBuilder::new(2322u64).gas_limit(i * 1000).build())
+        .collect_vec();
+
+    for tx in transactions {
+        let id = client.submit(&tx).await.unwrap();
+        // verify that the tx returned from the api matches the submitted tx
+        let ret_tx = client
+            .transaction(&id.0.to_string())
+            .await
+            .unwrap()
+            .unwrap()
+            .transaction;
+
+        let transaction_result = client
+            .transaction_status(&ret_tx.id().to_string())
+            .await
+            .ok()
+            .unwrap();
+
+        if let TransactionStatus::Success { block_id, .. } = transaction_result.clone() {
+            let block_exists = client.block(&block_id).await.unwrap();
+
+            assert!(block_exists.is_some());
+        }
+
+        assert!(matches!(
+            transaction_result,
+            TransactionStatus::Success { .. }
+        ));
+    }
+}
+
 #[tokio::test]
 async fn receipts() {
     let transaction = fuel_tx::Transaction::default();