Pruning of the transaction by TTL

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

@@ -45,7 +45,6 @@ use fuel_core_types::{
         txpool::{
             InsertionResult,
             TransactionStatus,
-            TxInfo,
         },
     },
     tai64::Tai64,
@@ -148,7 +147,9 @@ pub trait DatabaseChain {
 }
 
 pub trait TxPoolPort: Send + Sync {
-    fn find_one(&self, id: TxId) -> Option<TxInfo>;
+    fn transaction(&self, id: TxId) -> Option<Transaction>;
+
+    fn submission_time(&self, id: TxId) -> Option<Tai64>;
 
     fn insert(&self, txs: Vec<Arc<Transaction>>) -> Vec<anyhow::Result<InsertionResult>>;
 

crates/fuel-core/src/schema/tx.rs

@@ -48,7 +48,6 @@ use futures::{
 use itertools::Itertools;
 use std::{
     iter,
-    ops::Deref,
     sync::Arc,
 };
 use types::Transaction;
@@ -74,8 +73,8 @@ impl TxQuery {
         let id = id.0;
         let txpool = ctx.data_unchecked::<TxPool>();
 
-        if let Some(transaction) = txpool.find_one(id) {
-            Ok(Some(Transaction(transaction.tx().clone().deref().into())))
+        if let Some(transaction) = txpool.transaction(id) {
+            Ok(Some(Transaction(transaction)))
         } else {
             query.transaction(&id).into_api_result()
         }

crates/fuel-core/src/schema/tx/types.rs

@@ -500,11 +500,10 @@ pub(super) async fn get_tx_status(
         .into_api_result::<txpool::TransactionStatus, StorageError>()?
     {
         Some(status) => Ok(Some(status.into())),
-        None => match txpool.find_one(id) {
-            Some(transaction_in_pool) => {
-                let time = transaction_in_pool.submitted_time();
-                Ok(Some(TransactionStatus::Submitted(SubmittedStatus(time))))
-            }
+        None => match txpool.submission_time(id) {
+            Some(submitted_time) => Ok(Some(TransactionStatus::Submitted(
+                SubmittedStatus(submitted_time),
+            ))),
             _ => Ok(None),
         },
     }

crates/fuel-core/src/service/adapters/graphql_api.rs

@@ -59,8 +59,12 @@ use fuel_core_types::{
             TransactionStatus,
         },
     },
+    tai64::Tai64,
+};
+use std::{
+    ops::Deref,
+    sync::Arc,
 };
-use std::sync::Arc;
 use tokio_stream::wrappers::{
     errors::BroadcastStreamRecvError,
     BroadcastStream,
@@ -202,6 +206,18 @@ impl DatabaseChain for Database {
 impl DatabasePort for Database {}
 
 impl TxPoolPort for TxPoolAdapter {
+    fn transaction(&self, id: TxId) -> Option<Transaction> {
+        self.service
+            .find_one(id)
+            .map(|info| info.tx().clone().deref().into())
+    }
+
+    fn submission_time(&self, id: TxId) -> Option<Tai64> {
+        self.service
+            .find_one(id)
+            .map(|info| Tai64::from_unix(info.submitted_time().as_secs() as i64))
+    }
+
     fn insert(&self, txs: Vec<Arc<Transaction>>) -> Vec<anyhow::Result<InsertionResult>> {
         self.service.insert(txs)
     }
@@ -211,10 +227,6 @@ impl TxPoolPort for TxPoolAdapter {
     ) -> BoxStream<Result<TxUpdate, BroadcastStreamRecvError>> {
         Box::pin(BroadcastStream::new(self.service.tx_update_subscribe()))
     }
-
-    fn find_one(&self, id: TxId) -> Option<fuel_core_types::services::txpool::TxInfo> {
-        self.service.find_one(id)
-    }
 }
 
 #[async_trait]

crates/services/txpool/Cargo.toml

@@ -29,6 +29,10 @@ fuel-core-txpool = { path = "", features = ["test-helpers"] }
 itertools = { workspace = true }
 mockall = { workspace = true }
 rstest = "0.15"
+tokio = { workspace = true, features = [
+    "sync",
+    "test-util",
+] }
 
 [features]
 test-helpers = [

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

@@ -2,6 +2,7 @@ use crate::{
     ports::TxPoolDb,
     types::*,
     Error,
+    TxInfo,
 };
 use anyhow::anyhow;
 use fuel_core_types::{
@@ -15,10 +16,7 @@ use fuel_core_types::{
         UtxoId,
     },
     fuel_types::MessageId,
-    services::txpool::{
-        ArcPoolTx,
-        TxInfo,
-    },
+    services::txpool::ArcPoolTx,
 };
 use std::collections::{
     HashMap,

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

@@ -4,8 +4,8 @@ use crate::{
         SortableKey,
     },
     types::*,
+    TxInfo,
 };
-use fuel_core_types::services::txpool::TxInfo;
 use std::cmp;
 
 /// all transactions sorted by min/max price

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

@@ -1,8 +1,8 @@
-use crate::types::*;
-use fuel_core_types::services::txpool::{
-    ArcPoolTx,
+use crate::{
+    types::*,
     TxInfo,
 };
+use fuel_core_types::services::txpool::ArcPoolTx;
 use std::collections::BTreeMap;
 
 #[derive(Debug, Clone)]

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

@@ -4,28 +4,36 @@ use crate::{
         SortableKey,
     },
     types::*,
+    TxInfo,
 };
-use fuel_core_types::{
-    services::txpool::TxInfo,
-    tai64::Tai64,
+use core::{
+    cmp,
+    time::Duration,
 };
-use std::cmp;
 
 /// all transactions sorted by min/max time
 pub type TimeSort = Sort<TimeSortKey>;
 
 #[derive(Clone, Debug)]
 pub struct TimeSortKey {
-    time: Tai64,
+    time: Duration,
+    created: tokio::time::Instant,
     tx_id: TxId,
 }
 
+impl TimeSortKey {
+    pub fn created(&self) -> &tokio::time::Instant {
+        &self.created
+    }
+}
+
 impl SortableKey for TimeSortKey {
-    type Value = Tai64;
+    type Value = Duration;
 
     fn new(info: &TxInfo) -> Self {
         Self {
             time: info.submitted_time(),
+            created: info.created(),
             tx_id: info.tx().id(),
         }
     }

crates/services/txpool/src/lib.rs

@@ -1,5 +1,11 @@
 #![deny(unused_crate_dependencies)]
 
+use fuel_core_types::services::txpool::ArcPoolTx;
+use std::{
+    ops::Deref,
+    time::Duration,
+};
+
 pub mod config;
 mod containers;
 pub mod ports;
@@ -26,3 +32,45 @@ pub(crate) mod test_helpers;
 
 #[cfg(test)]
 fuel_core_trace::enable_tracing!();
+
+/// Information of a transaction fetched from the txpool.
+#[derive(Debug, Clone)]
+pub struct TxInfo {
+    tx: ArcPoolTx,
+    submitted_time: Duration,
+    creation_instant: tokio::time::Instant,
+}
+
+#[allow(missing_docs)]
+impl TxInfo {
+    pub fn new(tx: ArcPoolTx) -> Self {
+        let since_epoch = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .expect("Now is bellow of the `UNIX_EPOCH`");
+
+        Self {
+            tx,
+            submitted_time: since_epoch,
+            creation_instant: tokio::time::Instant::now(),
+        }
+    }
+
+    pub fn tx(&self) -> &ArcPoolTx {
+        &self.tx
+    }
+
+    pub fn submitted_time(&self) -> Duration {
+        self.submitted_time
+    }
+
+    pub fn created(&self) -> tokio::time::Instant {
+        self.creation_instant
+    }
+}
+
+impl Deref for TxInfo {
+    type Target = ArcPoolTx;
+    fn deref(&self) -> &Self::Target {
+        &self.tx
+    }
+}

crates/services/txpool/src/service.rs

@@ -7,6 +7,7 @@ use crate::{
     transaction_selector::select_transactions,
     Config,
     Error as TxPoolError,
+    TxInfo,
     TxPool,
 };
 use fuel_core_services::{
@@ -35,7 +36,6 @@ use fuel_core_types::{
             ArcPoolTx,
             Error,
             InsertionResult,
-            TxInfo,
             TxStatus,
         },
     },
@@ -129,7 +129,8 @@ where
         self.shared.clone()
     }
 
-    async fn into_task(self, _: &StateWatcher) -> anyhow::Result<Self::Task> {
+    async fn into_task(mut self, _: &StateWatcher) -> anyhow::Result<Self::Task> {
+        self.ttl_timer.reset();
         Ok(self)
     }
 }
@@ -156,8 +157,6 @@ where
                     self.shared.tx_status_sender.send_squeezed_out(tx.id(), Error::TTLReason);
                 }
 
-                self.ttl_timer.reset();
-
                 should_continue = true
             }
 
@@ -352,7 +351,7 @@ where
     let gossiped_tx_stream = p2p.gossiped_transaction_events();
     let committed_block_stream = importer.block_events();
     let mut ttl_timer = tokio::time::interval(config.transaction_ttl);
-    ttl_timer.set_missed_tick_behavior(MissedTickBehavior::Burst);
+    ttl_timer.set_missed_tick_behavior(MissedTickBehavior::Skip);
     let txpool = Arc::new(ParkingMutex::new(TxPool::new(config, db)));
     let task = Task {
         gossiped_tx_stream,

crates/services/txpool/src/service/tests.rs

@@ -47,9 +47,9 @@ async fn test_find() {
     service.stop_and_await().await.unwrap();
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[tokio::test(start_paused = true)]
 async fn test_prune_transactions() {
-    const TIMEOUT: u64 = 1;
+    const TIMEOUT: u64 = 10;
 
     let config = Config {
         transaction_ttl: Duration::from_secs(TIMEOUT),
@@ -76,14 +76,14 @@ async fn test_prune_transactions() {
     assert!(out[1].is_ok(), "Tx2 should be OK, got err:{out:?}");
     assert!(out[2].is_ok(), "Tx3 should be OK, got err:{out:?}");
 
-    tokio::time::sleep(Duration::from_secs(TIMEOUT)).await;
+    tokio::time::sleep(Duration::from_secs(TIMEOUT / 2)).await;
     let out = service.shared.find(vec![tx1.id(), tx2.id(), tx3.id()]);
     assert_eq!(out.len(), 3, "Should be len 3:{out:?}");
     assert!(out[0].is_some(), "Tx1 should exist");
     assert!(out[1].is_some(), "Tx2 should exist");
     assert!(out[2].is_some(), "Tx3 should exist");
 
-    tokio::time::sleep(Duration::from_secs(2 * TIMEOUT)).await;
+    tokio::time::sleep(Duration::from_secs(TIMEOUT / 2 + 1)).await;
     let out = service.shared.find(vec![tx1.id(), tx2.id(), tx3.id()]);
     assert_eq!(out.len(), 3, "Should be len 3:{out:?}");
     assert!(out[0].is_none(), "Tx1 should be pruned");
@@ -93,6 +93,56 @@ async fn test_prune_transactions() {
     service.stop_and_await().await.unwrap();
 }
 
+#[tokio::test(start_paused = true)]
+async fn test_prune_transactions_the_oldest() {
+    const TIMEOUT: u64 = 10;
+    const DELAY: u64 = 2;
+
+    let config = Config {
+        transaction_ttl: Duration::from_secs(TIMEOUT),
+        ..Default::default()
+    };
+    let ctx = TestContextBuilder::new()
+        .with_config(config)
+        .build_and_start()
+        .await;
+
+    let tx1 = Arc::new(ctx.setup_script_tx(10));
+    let tx2 = Arc::new(ctx.setup_script_tx(20));
+    let tx3 = Arc::new(ctx.setup_script_tx(30));
+
+    let service = ctx.service();
+
+    let out = service.shared.insert(vec![tx1.clone()]);
+    assert!(out[0].is_ok(), "Tx1 should be OK, got err:{out:?}");
+
+    tokio::time::sleep(Duration::from_secs(TIMEOUT - DELAY)).await;
+    let out = service.shared.insert(vec![tx2.clone()]);
+    assert!(out[0].is_ok(), "Tx2 should be OK, got err:{out:?}");
+
+    let out = service.shared.find(vec![tx1.id(), tx2.id(), tx3.id()]);
+    assert!(out[0].is_some(), "Tx1 should exist");
+    assert!(out[1].is_some(), "Tx2 should exist");
+
+    tokio::time::sleep(Duration::from_secs(TIMEOUT)).await;
+    let out = service.shared.insert(vec![tx3.clone()]);
+    assert!(out[0].is_ok(), "Tx3 should be OK, got err:{out:?}");
+
+    let out = service.shared.find(vec![tx1.id(), tx2.id(), tx3.id()]);
+    assert!(out[0].is_none(), "Tx1 should pruned");
+    assert!(out[1].is_some(), "Tx2 should exist");
+    assert!(out[2].is_some(), "Tx3 should exist");
+
+    tokio::time::sleep(Duration::from_secs(TIMEOUT)).await;
+
+    let out = service.shared.find(vec![tx1.id(), tx2.id(), tx3.id()]);
+    assert!(out[0].is_none(), "Tx1 should pruned");
+    assert!(out[1].is_none(), "Tx2 should pruned");
+    assert!(out[2].is_some(), "Tx3 should exist");
+
+    service.stop_and_await().await.unwrap();
+}
+
 #[tokio::test]
 async fn simple_insert_removal_subscription() {
     let ctx = TestContextBuilder::new().build_and_start().await;