feat(mempool): evict expired transactions after configurable ttl (#3210)

nodes/node/binary/src/config/deployment/settings.yaml

@@ -72,3 +72,4 @@ time:
   slot_duration: '1.000000000'
 mempool:
   pubsub_topic: /logos-blockchain/mempool/X.Y.Z
+  tx_ttl: '86400.000000000'

nodes/node/binary/src/config/mempool/deployment.rs

@@ -1,6 +1,22 @@
-use serde::{Deserialize, Serialize};
+use core::time::Duration;
 
+use lb_tx_service::backend::DEFAULT_TX_TTL;
+use lb_utils::bounded_duration::{MinimalBoundedDuration, SECOND};
+use serde::{Deserialize, Serialize};
+use serde_with::serde_as;
+
+#[serde_as]
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Settings {
     pub pubsub_topic: String,
+    /// How long a pending transaction may stay in the mempool before it is
+    /// evicted. `None` disables expiry-based eviction.
+    #[serde_as(as = "Option<MinimalBoundedDuration<1, SECOND>>")]
+    #[serde(default = "default_tx_ttl")]
+    pub tx_ttl: Option<Duration>,
+}
+
+#[must_use]
+pub const fn default_tx_ttl() -> Option<Duration> {
+    Some(DEFAULT_TX_TTL)
 }

nodes/node/binary/src/config/mempool/mod.rs

@@ -5,7 +5,8 @@ use lb_core::mantle::{
 };
 use lb_services_utils::overwatch::RecoveryData;
 use lb_tx_service::{
-    TxMempoolSettings, network::adapters::libp2p::Settings as Libp2pNetworkAdapterSettings,
+    TxMempoolSettings, backend::MempoolSettings,
+    network::adapters::libp2p::Settings as Libp2pNetworkAdapterSettings,
 };
 
 use crate::config::mempool::deployment::Settings as DeploymentSettings;
@@ -21,14 +22,18 @@ impl ServiceConfig {
     pub fn into_mempool_service_settings(
         self,
         recovery_data: RecoveryData,
-    ) -> TxMempoolSettings<(), Libp2pNetworkAdapterSettings<TxHash, SignedMantleTx<Preverified>>>
-    {
+    ) -> TxMempoolSettings<
+        MempoolSettings,
+        Libp2pNetworkAdapterSettings<TxHash, SignedMantleTx<Preverified>>,
+    > {
         TxMempoolSettings {
             network_adapter: Libp2pNetworkAdapterSettings {
                 id: SignedMantleTx::<Preverified>::hash,
                 topic: self.deployment.pubsub_topic,
             },
-            pool: (),
+            pool: MempoolSettings {
+                tx_ttl: self.deployment.tx_ttl,
+            },
             recovery_data,
         }
     }

services/tx-service/src/backend/evictor.rs

@@ -0,0 +1,52 @@
+use std::hash::Hash;
+
+use indexmap::IndexMap;
+
+use crate::backend::{policy::TtlPolicy, pool::MempoolSettings};
+
+/// Decides which valid transactions the pool stops carrying.
+///
+/// Composes eviction policies and drives their lifecycle; it never removes
+/// anything itself — the pool executes the key lists it returns.
+pub struct Evictor<Key> {
+    ttl: TtlPolicy<Key>,
+}
+
+impl<Key> Evictor<Key>
+where
+    Key: Hash + Eq + Clone,
+{
+    #[must_use]
+    pub fn new(settings: &MempoolSettings) -> Self {
+        Self {
+            ttl: TtlPolicy::new(settings.tx_ttl),
+        }
+    }
+
+    pub fn on_add(&mut self, key: Key, now: u64) {
+        self.ttl.on_add(key, now);
+    }
+
+    pub fn on_remove(&mut self, key: &Key) {
+        self.ttl.on_remove(key);
+    }
+
+    /// Keys the pool should stop carrying, according to the enabled
+    /// policies. The cause stays internal to the policies.
+    #[must_use]
+    pub fn select_evictions(&self, now: u64) -> Vec<Key> {
+        self.ttl.expired(now)
+    }
+
+    #[must_use]
+    pub fn save(&self) -> IndexMap<Key, u64> {
+        self.ttl.save()
+    }
+
+    #[must_use]
+    pub const fn recover(settings: &MempoolSettings, state: IndexMap<Key, u64>) -> Self {
+        Self {
+            ttl: TtlPolicy::recover(settings.tx_ttl, state),
+        }
+    }
+}

services/tx-service/src/backend/mod.rs

@@ -1,3 +1,5 @@
+pub mod evictor;
+pub mod policy;
 pub mod pool;
 
 use core::hash::Hash;
@@ -5,7 +7,7 @@ use std::pin::Pin;
 
 use futures::Stream;
 use lb_core::mantle::transactions::hash::PrefixedKey;
-pub use pool::{Mempool, PoolRecoveryState};
+pub use pool::{DEFAULT_TX_TTL, Mempool, MempoolSettings, PoolRecoveryState};
 use serde::{Deserialize, Serialize};
 
 #[derive(thiserror::Error, Debug)]

services/tx-service/src/backend/policy/mod.rs

@@ -0,0 +1,3 @@
+pub mod ttl;
+
+pub use ttl::TtlPolicy;

services/tx-service/src/backend/policy/ttl.rs

@@ -0,0 +1,67 @@
+use std::{hash::Hash, time::Duration};
+
+use indexmap::IndexMap;
+
+/// Evicts transactions that stayed in the pool longer than the configured
+/// TTL, regardless of pool fill.
+///
+/// Owns its own state: the insertion timestamp of every pending transaction.
+/// Timestamps are tracked even when the TTL is disabled so that transaction
+/// age survives recovery and a later configuration change.
+pub struct TtlPolicy<Key> {
+    tx_ttl: Option<Duration>,
+    inserted: IndexMap<Key, u64>,
+}
+
+impl<Key> TtlPolicy<Key>
+where
+    Key: Hash + Eq + Clone,
+{
+    #[must_use]
+    pub fn new(tx_ttl: Option<Duration>) -> Self {
+        Self {
+            tx_ttl,
+            inserted: IndexMap::new(),
+        }
+    }
+
+    pub fn on_add(&mut self, key: Key, now: u64) {
+        self.inserted.insert(key, now);
+    }
+
+    pub fn on_remove(&mut self, key: &Key) {
+        self.inserted.shift_remove(key);
+    }
+
+    /// Keys whose age exceeds the TTL.
+    ///
+    /// `inserted` is insertion-ordered with non-decreasing timestamps, so
+    /// expired keys are always at the front and the scan stops at the first
+    /// non-expired entry.
+    #[must_use]
+    pub fn expired(&self, now: u64) -> Vec<Key> {
+        let Some(tx_ttl) = self.tx_ttl else {
+            return Vec::new();
+        };
+        let ttl_millis = tx_ttl.as_millis() as u64;
+
+        self.inserted
+            .iter()
+            .take_while(|(_, inserted_at)| now.saturating_sub(**inserted_at) >= ttl_millis)
+            .map(|(key, _)| key.clone())
+            .collect()
+    }
+
+    #[must_use]
+    pub fn save(&self) -> IndexMap<Key, u64> {
+        self.inserted.clone()
+    }
+
+    #[must_use]
+    pub const fn recover(tx_ttl: Option<Duration>, state: IndexMap<Key, u64>) -> Self {
+        Self {
+            tx_ttl,
+            inserted: state,
+        }
+    }
+}

services/tx-service/src/backend/pool.rs

@@ -8,14 +8,14 @@ use std::{
 
 use async_trait::async_trait;
 use futures::Stream;
-use indexmap::IndexSet;
+use indexmap::{IndexMap, IndexSet};
 use lb_core::mantle::transactions::hash::PrefixedKey;
 use lb_log_targets::mempool;
 use serde::{Deserialize, Serialize};
 
 use super::Status;
 use crate::{
-    backend::{MemPool, MempoolError, RecoverableMempool},
+    backend::{MemPool, MempoolError, RecoverableMempool, evictor::Evictor},
     metrics,
     storage::MempoolStorageAdapter,
 };
@@ -23,12 +23,41 @@ use crate::{
 const REMOVED_ITEM_GRACE_PERIOD: Duration = Duration::from_mins(10);
 const LOG_TARGET: &str = mempool::POOL;
 
+/// Default time a pending transaction may stay in the pool before eviction.
+pub const DEFAULT_TX_TTL: Duration = Duration::from_hours(24);
+
+/// Settings for the [`Mempool`].
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct MempoolSettings {
+    /// How long a pending transaction is allowed to stay in the pool before
+    /// it is evicted. `None` disables expiry-based eviction.
+    #[serde(default = "default_tx_ttl")]
+    pub tx_ttl: Option<Duration>,
+}
+
+impl Default for MempoolSettings {
+    fn default() -> Self {
+        Self {
+            tx_ttl: default_tx_ttl(),
+        }
+    }
+}
+
+#[expect(
+    clippy::unnecessary_wraps,
+    reason = "Serde default for an `Option` field."
+)]
+const fn default_tx_ttl() -> Option<Duration> {
+    Some(DEFAULT_TX_TTL)
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct PoolRecoveryState<Key>
 where
     Key: Hash + Eq + Ord,
 {
-    pub pending_items: IndexSet<Key>,
+    /// Pending item keys mapped to their insertion timestamp in milliseconds.
+    pub pending_items: IndexMap<Key, u64>,
     pub removed_items: BTreeMap<Key, u64>,
     pub last_item_timestamp: u64,
 }
@@ -48,6 +77,7 @@ where
     by_prefix: HashMap<Key::Prefix, Vec<Key>>,
     removed_items: BTreeMap<Key, u64>,
     last_item_timestamp: u64,
+    evictor: Evictor<Key>,
     storage_adapter: Storage,
     _phantom: std::marker::PhantomData<(BlockId, Item, RuntimeServiceId)>,
 }
@@ -66,7 +96,7 @@ where
             .field("removed_items", &self.removed_items)
             .field("last_item_timestamp", &self.last_item_timestamp)
             .field("storage_adapter", &"<StorageAdapter>")
-            .finish()
+            .finish_non_exhaustive()
     }
 }
 
@@ -89,18 +119,19 @@ where
     Storage::Error: Debug,
     RuntimeServiceId: Send + Sync,
 {
-    type Settings = ();
+    type Settings = MempoolSettings;
     type Item = Item;
     type Key = Key;
     type BlockId = BlockId;
     type Storage = Storage;
 
-    fn new(_settings: Self::Settings, storage: Self::Storage) -> Self {
+    fn new(settings: Self::Settings, storage: Self::Storage) -> Self {
         Self {
             pending_items: IndexSet::new(),
             by_prefix: HashMap::new(),
             removed_items: BTreeMap::new(),
             last_item_timestamp: 0,
+            evictor: Evictor::new(&settings),
             storage_adapter: storage,
             _phantom: std::marker::PhantomData,
         }
@@ -126,6 +157,7 @@ where
 
         self.removed_items.remove(&key);
         self.index_by_prefix(&key);
+        self.evictor.on_add(key.clone(), timestamp);
         self.pending_items.insert(key);
         self.last_item_timestamp = timestamp;
         tracing::debug!(
@@ -175,17 +207,19 @@ where
     async fn remove(&mut self, keys: &[Self::Key]) {
         self.prune_removed_items().await;
 
-        let removed_count = keys.len();
         let removed_at = current_timestamp_millis();
 
-        for key in keys {
-            self.pending_items.shift_remove(key);
-            self.unindex_by_prefix(key);
-            self.removed_items.insert(key.clone(), removed_at);
-        }
-        log_removed_items(removed_count, self.pending_items.len());
+        self.retire(keys.iter().cloned(), removed_at, "removed");
+
+        // `remove` is called once per applied canonical block, so it doubles
+        // as the periodic trigger for eviction — without it the pool has no
+        // clock.
+        self.retire(
+            self.evictor.select_evictions(removed_at),
+            removed_at,
+            "evicted",
+        );
 
-        metrics::mempool_transactions_removed(removed_count);
         metrics::mempool_transactions_pending(self.pending_items.len());
     }
 
@@ -235,21 +269,21 @@ where
 
     fn save(&self) -> Self::RecoveryState {
         PoolRecoveryState {
-            pending_items: self.pending_items.clone(),
+            pending_items: self.evictor.save(),
             removed_items: self.removed_items.clone(),
             last_item_timestamp: self.last_item_timestamp,
         }
     }
 
     fn recover(
-        _settings: <Self as MemPool>::Settings,
+        settings: <Self as MemPool>::Settings,
         state: Self::RecoveryState,
         storage: <Self as MemPool>::Storage,
     ) -> Self {
         // `by_prefix` is derived, so it is rebuilt rather than restored.
         let mut by_prefix: HashMap<Key::Prefix, Vec<Key>> =
             HashMap::with_capacity(state.pending_items.len());
-        for key in &state.pending_items {
+        for key in state.pending_items.keys() {
             by_prefix
                 .entry(key.key_prefix())
                 .or_default()
@@ -257,10 +291,11 @@ where
         }
 
         Self {
-            pending_items: state.pending_items,
+            pending_items: state.pending_items.keys().cloned().collect(),
             by_prefix,
             removed_items: state.removed_items,
             last_item_timestamp: state.last_item_timestamp,
+            evictor: Evictor::recover(&settings, state.pending_items),
             storage_adapter: storage,
             _phantom: std::marker::PhantomData,
         }
@@ -306,6 +341,22 @@ where
         }
     }
 
+    /// The single exit for pending items: every removal — included in a
+    /// block or evicted — goes through here so the evictor stays in sync.
+    fn retire(&mut self, keys: impl IntoIterator<Item = Key>, at: u64, cause: &str) {
+        let mut count = 0usize;
+        for key in keys {
+            self.pending_items.shift_remove(&key);
+            self.unindex_by_prefix(&key);
+            self.evictor.on_remove(&key);
+            self.removed_items.insert(key, at);
+            count += 1;
+        }
+
+        log_retired_items(count, cause, self.pending_items.len());
+        metrics::mempool_transactions_removed(count);
+    }
+
     async fn prune_removed_items(&mut self) {
         let now = current_timestamp_millis();
         let grace_period_millis = REMOVED_ITEM_GRACE_PERIOD.as_millis() as u64;
@@ -339,16 +390,16 @@ fn current_timestamp_millis() -> u64 {
         .as_millis() as u64
 }
 
-fn log_removed_items(removed_count: usize, pending_items: usize) {
-    if removed_count == 0 {
+fn log_retired_items(count: usize, cause: &str, pending_items: usize) {
+    if count == 0 {
         tracing::trace!(
             target: LOG_TARGET,
-            "Removed {removed_count} items from mempool; pending_items={pending_items}"
+            "{cause}: 0 items from mempool; pending_items={pending_items}"
         );
     } else {
         tracing::debug!(
             target: LOG_TARGET,
-            "Removed {removed_count} items from mempool; pending_items={pending_items}"
+            "{cause}: {count} items from mempool; pending_items={pending_items}"
         );
     }
 }

services/tx-service/tests/mock.rs

@@ -3,11 +3,12 @@ use std::{
     convert::Infallible,
     pin::Pin,
     sync::{Arc, Mutex, atomic::AtomicBool},
+    time::Duration,
 };
 
 use async_trait::async_trait;
 use futures::{Stream, StreamExt as _, stream};
-use indexmap::IndexSet;
+use indexmap::IndexMap;
 use lb_core::{
     block::MAX_BLOCK_TRANSACTIONS_SIZE,
     codec::{DeserializeOp as _, SerializeOp as _},
@@ -36,7 +37,8 @@ use lb_utils::noop_service::NoService;
 use logos_blockchain_tx_service::{
     MempoolMsg, TxMempoolSettings,
     backend::{
-        MemPool as _, Mempool, MempoolError, PoolRecoveryState, RecoverableMempool as _, Status,
+        MemPool as _, Mempool, MempoolError, MempoolSettings, PoolRecoveryState,
+        RecoverableMempool as _, Status,
     },
     network::adapters::mock::{MOCK_TX_CONTENT_TOPIC, MockAdapter},
     storage::{MempoolStorageAdapter, adapters::rocksdb::RocksStorageAdapter},
@@ -50,8 +52,8 @@ use overwatch_derive::*;
 use tempfile::TempDir;
 
 type MockRecoveryBackend = StorageRecoveryBackend<
-    TxMempoolState<PoolRecoveryState<MockTxId>, (), ()>,
-    TxMempoolSettings<(), ()>,
+    TxMempoolState<PoolRecoveryState<MockTxId>, MempoolSettings, ()>,
+    TxMempoolSettings<MempoolSettings, ()>,
     RocksBackend,
     RuntimeServiceId,
 >;
@@ -99,7 +101,7 @@ fn mock_pool_node_settings(
             network: NetworkConfig {
                 backend: MockConfig {
                     predefined_messages,
-                    duration: tokio::time::Duration::from_millis(100),
+                    duration: Duration::from_millis(100),
                     seed: 0,
                     version: 1,
                     weights: None,
@@ -111,7 +113,7 @@ fn mock_pool_node_settings(
                 column_family: None,
             },
             mockpool: TxMempoolSettings {
-                pool: (),
+                pool: MempoolSettings::default(),
                 network_adapter: (),
                 recovery_data: RecoveryData::default(),
             },
@@ -282,7 +284,7 @@ impl MempoolStorageAdapter<RuntimeServiceId> for FailingStorageAdapter {
 #[test]
 fn test_mock_pool_recovery_state() {
     let recovery_state = PoolRecoveryState::<MockTxId> {
-        pending_items: IndexSet::new(),
+        pending_items: IndexMap::new(),
         removed_items: BTreeMap::new(),
         last_item_timestamp: 1_234_567_890,
     };
@@ -300,6 +302,136 @@ fn test_mock_pool_recovery_state() {
     );
 }
 
+const fn ttl_settings(tx_ttl: Duration) -> MempoolSettings {
+    MempoolSettings {
+        tx_ttl: Some(tx_ttl),
+    }
+}
+
+const TEST_TX_TTL: Duration = Duration::from_hours(1);
+
+/// Build a pool holding one pending tx whose insertion timestamp is backdated
+/// to the unix epoch, so it is expired for any reasonable TTL.
+async fn pool_with_backdated_tx(
+    settings: MempoolSettings,
+) -> (
+    Mempool<
+        HeaderId,
+        MockTransaction<MockMessage>,
+        MockTxId,
+        InMemoryStorageAdapter,
+        RuntimeServiceId,
+    >,
+    MockTransaction<MockMessage>,
+) {
+    let storage = InMemoryStorageAdapter::default();
+    let mut pool = Mempool::<
+        HeaderId,
+        MockTransaction<MockMessage>,
+        MockTxId,
+        InMemoryStorageAdapter,
+        RuntimeServiceId,
+    >::new(settings, storage.clone());
+
+    let tx = sample_removed_tx();
+    let tx_id = tx.id();
+
+    pool.add_item(tx_id, tx.clone())
+        .await
+        .expect("tx should be added");
+
+    let mut saved_state = pool.save();
+    saved_state.pending_items.insert(tx_id, 0);
+
+    (Mempool::recover(settings, saved_state, storage), tx)
+}
+
+#[tokio::test]
+async fn expired_tx_is_evicted_on_next_sweep() {
+    let storage = InMemoryStorageAdapter::default();
+    let mut pool = Mempool::<
+        HeaderId,
+        MockTransaction<MockMessage>,
+        MockTxId,
+        InMemoryStorageAdapter,
+        RuntimeServiceId,
+    >::new(ttl_settings(TEST_TX_TTL), storage.clone());
+
+    let old_tx = sample_removed_tx();
+    let old_tx_id = old_tx.id();
+    let old_tx_prefix = old_tx_id.key_prefix();
+
+    let fresh_tx = MockTransaction::new(MockMessage {
+        payload: "fresh".to_owned(),
+        content_topic: MOCK_TX_CONTENT_TOPIC,
+        version: 0,
+        timestamp: 1,
+    });
+    let fresh_tx_id = fresh_tx.id();
+
+    pool.add_item(old_tx_id, old_tx.clone())
+        .await
+        .expect("old tx should be added");
+    pool.add_item(fresh_tx_id, fresh_tx.clone())
+        .await
+        .expect("fresh tx should be added");
+
+    let mut saved_state = pool.save();
+    saved_state.pending_items.insert(old_tx_id, 0);
+    let mut pool = Mempool::<
+        HeaderId,
+        MockTransaction<MockMessage>,
+        MockTxId,
+        InMemoryStorageAdapter,
+        RuntimeServiceId,
+    >::recover(ttl_settings(TEST_TX_TTL), saved_state, storage);
+
+    assert_eq!(pool.pending_item_count(), 2);
+    assert_eq!(pool.status(&[old_tx_id]), vec![Status::Pending]);
+    assert_eq!(
+        pool.keys_by_prefix(&old_tx_prefix)
+            .copied()
+            .collect::<Vec<_>>(),
+        vec![old_tx_id]
+    );
+
+    pool.remove(&[]).await;
+
+    assert_eq!(pool.pending_item_count(), 1);
+    assert_eq!(pool.status(&[old_tx_id]), vec![Status::Unknown]);
+    assert_eq!(pool.status(&[fresh_tx_id]), vec![Status::Pending]);
+    assert!(pool.keys_by_prefix(&old_tx_prefix).next().is_none());
+
+    let pending_after_eviction = pool
+        .view([0; 32].into())
+        .await
+        .expect("pending view should load")
+        .collect::<Vec<_>>()
+        .await;
+    assert_eq!(pending_after_eviction, vec![fresh_tx]);
+
+    let fetched_after_eviction = pool
+        .get_items_by_keys([old_tx_id])
+        .await
+        .expect("evicted tx should still be fetchable during grace period")
+        .collect::<Vec<_>>()
+        .await;
+    assert_eq!(fetched_after_eviction, vec![old_tx]);
+}
+
+#[tokio::test]
+async fn expired_tx_is_not_evicted_when_ttl_is_disabled() {
+    let (mut pool, tx) = pool_with_backdated_tx(MempoolSettings { tx_ttl: None }).await;
+    let tx_id = tx.id();
+
+    for _ in 0..10 {
+        pool.remove(&[]).await;
+    }
+
+    assert_eq!(pool.pending_item_count(), 1);
+    assert_eq!(pool.status(&[tx_id]), vec![Status::Pending]);
+}
+
 #[tokio::test]
 async fn storage_failure_does_not_mark_tx_pending() {
     let mut pool = Mempool::<
@@ -308,7 +440,7 @@ async fn storage_failure_does_not_mark_tx_pending() {
         MockTxId,
         FailingStorageAdapter,
         RuntimeServiceId,
-    >::new((), FailingStorageAdapter);
+    >::new(MempoolSettings::default(), FailingStorageAdapter);
 
     let tx = sample_removed_tx();
     let tx_id = tx.id();
@@ -333,7 +465,7 @@ async fn removed_items_are_not_pending_but_still_fetchable() {
         MockTxId,
         InMemoryStorageAdapter,
         RuntimeServiceId,
-    >::new((), storage.clone());
+    >::new(MempoolSettings::default(), storage.clone());
 
     let tx = sample_removed_tx();
     let tx_id = tx.id();
@@ -385,7 +517,7 @@ async fn removed_items_remain_fetchable_after_recovery() {
         MockTxId,
         InMemoryStorageAdapter,
         RuntimeServiceId,
-    >::new((), storage.clone());
+    >::new(MempoolSettings::default(), storage.clone());
 
     let tx = sample_removed_tx();
     let tx_id = tx.id();
@@ -405,7 +537,7 @@ async fn removed_items_remain_fetchable_after_recovery() {
         MockTxId,
         InMemoryStorageAdapter,
         RuntimeServiceId,
-    >::recover((), saved_state, storage);
+    >::recover(MempoolSettings::default(), saved_state, storage);
 
     assert_eq!(recovered_pool.pending_item_count(), 0);
     assert_eq!(recovered_pool.status(&[tx_id]), vec![Status::Unknown]);
@@ -579,7 +711,7 @@ fn test_mock_mempool() {
 
             // try to wait all ops to be stored in mempool
             loop {
-                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+                tokio::time::sleep(Duration::from_secs(1)).await;
                 let (mtx, mrx) = tokio::sync::oneshot::channel();
                 mempool_outbound
                     .send(MempoolMsg::View {
@@ -605,7 +737,7 @@ fn test_mock_mempool() {
         });
 
         while !exist2.load(std::sync::atomic::Ordering::SeqCst) {
-            std::thread::sleep(std::time::Duration::from_millis(200));
+            std::thread::sleep(Duration::from_millis(200));
         }
 
         drop(app.runtime().handle().block_on(app.handle().shutdown()));
@@ -618,7 +750,7 @@ fn test_mock_mempool() {
         })
         .expect("Should load recovery data from storage.");
         let recovery_settings = TxMempoolSettings {
-            pool: (),
+            pool: MempoolSettings::default(),
             network_adapter: (),
             recovery_data,
         };

tests/src/cucumber/steps/manual_mempool/actions.rs

@@ -241,7 +241,7 @@ fn read_recovered_mempool_pending_hashes(
 
     Ok(recovery_state
         .pool()
-        .map(|pool| pool.pending_items.iter().copied().collect()))
+        .map(|pool| pool.pending_items.keys().copied().collect()))
 }
 
 fn wallet_transaction_error(error: &WalletTransactionError) -> StepError {

tools/config/src/deployment.rs

@@ -19,7 +19,7 @@ use lb_node::config::{
         EpochConfig, ServiceParameters, Settings as CryptarchiaDeploymentSettings,
     },
     deployment::DeploymentSettings,
-    mempool::deployment::Settings as MempoolDeploymentSettings,
+    mempool::deployment::{Settings as MempoolDeploymentSettings, default_tx_ttl},
     network::deployment::Settings as NetworkDeploymentSettings,
     time::deployment::Settings as TimeDeploymentSettings,
 };
@@ -149,6 +149,7 @@ pub fn e2e_deployment_settings_with_genesis_block(
         },
         mempool: MempoolDeploymentSettings {
             pubsub_topic: MEMPOOL_TOPIC.to_owned(),
+            tx_ttl: default_tx_ttl(),
         },
     }
 }