Reset InMempool txs to Inactive before connecting to the node

CHANGELOG.md

@@ -62,6 +62,10 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/
     - Fixed an issue where the mempool wouldn't track non-UTXO dependencies between transactions, which could
       prevent e.g. several token management transactions from being included in the same block.
 
+  - Wallet:
+    - Fixed an issue where wallet transactions would remain marked as in-mempool after reopening the wallet when they are
+      actually no longer in the mempool.
+
 ## [1.3.1] - 2026-06-03
 
 ### Fixed

wallet/src/account/mod.rs

@@ -2401,6 +2401,26 @@ impl<K: AccountKeyChains> Account<K> {
         )
     }
 
+    /// Reset all transactions that are currently in-mempool to inactive state
+    pub fn reset_inmempool_txs_to_inactive<B: storage::Backend>(
+        &mut self,
+        db_tx: &mut StoreTxRw<B>,
+        wallet_events: Option<&impl WalletEvents>,
+    ) -> WalletResult<()> {
+        let acc_id = self.get_account_id();
+        let account_index = self.account_index();
+
+        for tx in self.output_cache.reset_inmempool_txs_to_inactive() {
+            db_tx.set_transaction(&AccountWalletTxId::new(acc_id.clone(), tx.id()), tx)?;
+
+            if let Some(wallet_events) = wallet_events {
+                wallet_events.set_transaction(account_index, tx);
+            }
+        }
+
+        Ok(())
+    }
+
     pub fn update_best_block(
         &mut self,
         db_tx: &mut impl WalletStorageWriteLocked,

wallet/src/account/output_cache/mod.rs

@@ -600,6 +600,7 @@ pub struct OutputCache {
     //  - no confirmed txs are allowed;
     //  - confirmed tx cannot have unconfirmed parent.
     unconfirmed_descendants: BTreeMap<OutPointSourceId, BTreeSet<OutPointSourceId>>,
+    // Map of consumed utxos; the value is the state of the tx that consumed the corresponding utxo.
     consumed: BTreeMap<UtxoOutPoint, TxState>,
 
     pools: BTreeMap<PoolId, PoolData>,
@@ -1431,6 +1432,19 @@ impl OutputCache {
         Ok(())
     }
 
+    /// Reset all transactions that are currently in-mempool to inactive state.
+    ///
+    /// Return an iterator of the transactions that were reset.
+    pub fn reset_inmempool_txs_to_inactive(&mut self) -> impl Iterator<Item = &WalletTx> {
+        self.consumed.values_mut().for_each(|tx_state| {
+            tx_state.reset_inmempool_to_inactive();
+        });
+
+        self.txs
+            .values_mut()
+            .filter_map(|tx| tx.reset_inmempool_to_inactive().then_some(&*tx))
+    }
+
     fn is_consumed(&self, utxo_states: UtxoStates, outpoint: &UtxoOutPoint) -> bool {
         self.consumed
             .get(outpoint)

wallet/src/account/output_cache/tests.rs

@@ -1608,10 +1608,66 @@ fn update_conflicting_txs_order_v1_freeze(#[case] seed: Seed) {
     assert!(!output_cache.orders[&order_id].is_concluded);
 }
 
+#[rstest]
+#[trace]
+#[case(Seed::from_entropy())]
+fn reset_inmempool_txs_to_inactive(#[case] seed: Seed) {
+    let mut rng = make_seedable_rng(seed);
+    let chain_config = create_unit_test_config();
+    let mut output_cache = OutputCache::empty();
+
+    // add 10 random txs
+    for _ in 0..10 {
+        add_random_transfer_tx_with_state(
+            &mut output_cache,
+            &chain_config,
+            TxStateTag::InMempool,
+            &mut rng,
+        );
+    }
+
+    // Sanity check - the `consumed` collection is not empty and all the tx states are InMempool.
+    assert_eq!(output_cache.consumed.len(), 10);
+    for tx_state in output_cache.consumed.values() {
+        assert!(matches!(tx_state, TxState::InMempool(_)));
+    }
+
+    let all_tx_ids = output_cache.txs.values().map(|tx| tx.id()).collect::<BTreeSet<_>>();
+
+    // Reset
+    let reset_tx_ids = output_cache
+        .reset_inmempool_txs_to_inactive()
+        .map(|tx| tx.id())
+        .collect::<BTreeSet<_>>();
+
+    assert_eq!(reset_tx_ids, all_tx_ids);
+
+    // Check that all txs are now Inactive
+    for tx in output_cache.txs.values() {
+        assert!(matches!(tx.state(), TxState::Inactive(_)));
+    }
+
+    // Check that inside `consumed` the state of the consuming tx has been updated as well.
+    assert_eq!(output_cache.consumed.len(), 10);
+    for tx_state in output_cache.consumed.values() {
+        assert!(matches!(tx_state, TxState::Inactive(_)));
+    }
+}
+
 fn add_random_transfer_tx(
     output_cache: &mut OutputCache,
     chain_config: &ChainConfig,
     mut rng: impl Rng,
+) {
+    let state_tag = TxStateTag::iter().choose(&mut rng).unwrap();
+    add_random_transfer_tx_with_state(output_cache, chain_config, state_tag, rng);
+}
+
+fn add_random_transfer_tx_with_state(
+    output_cache: &mut OutputCache,
+    chain_config: &ChainConfig,
+    state_tag: TxStateTag,
+    mut rng: impl Rng,
 ) {
     let random_tx_id = Id::<Transaction>::random_using(&mut rng);
     let random_block_id = Id::<GenBlock>::random_using(&mut rng);
@@ -1627,7 +1683,7 @@ fn add_random_transfer_tx(
         .build();
     let tx_id = tx.transaction().get_id();
 
-    let tx_state = match TxStateTag::iter().choose(&mut rng).unwrap() {
+    let tx_state = match state_tag {
         TxStateTag::Confirmed => TxState::Confirmed(
             BlockHeight::new(rng.random_range(0..100)),
             BlockTimestamp::from_int_seconds(rng.random_range(0..100)),

wallet/src/wallet/mod.rs

@@ -812,6 +812,22 @@ where
         Ok(())
     }
 
+    /// Resets all transactions in all accounts from in-mempool state to inactive.
+    pub fn reset_inmempool_txs_to_inactive(
+        &mut self,
+        wallet_events: Option<&impl WalletEvents>,
+    ) -> WalletResult<()> {
+        let mut db_tx = self.db.transaction_rw(None)?;
+
+        for account in self.accounts.values_mut() {
+            account.reset_inmempool_txs_to_inactive(&mut db_tx, wallet_events)?;
+        }
+
+        db_tx.commit()?;
+
+        Ok(())
+    }
+
     fn reset_wallet_transactions(
         chain_config: Arc<ChainConfig>,
         db_tx: &mut impl WalletStorageWriteLocked,
@@ -979,6 +995,21 @@ where
 
         db_tx.close();
 
+        {
+            // Reset all in-mempool transactions to inactive so we can set them properly from the current node again.
+
+            let mut db_tx = db.transaction_rw(None)?;
+
+            for account in accounts.values_mut() {
+                account.reset_inmempool_txs_to_inactive(
+                    &mut db_tx,
+                    Option::<&WalletEventsNoOp>::None,
+                )?;
+            }
+
+            db_tx.commit()?;
+        }
+
         let next_unused_account = accounts.pop_last().ok_or(WalletError::WalletNotInitialized)?;
 
         Ok(WalletCreation::Wallet(Wallet {

wallet/types/src/wallet_tx.rs

@@ -91,6 +91,20 @@ impl TxState {
             TxState::Abandoned => "Abandoned",
         }
     }
+
+    pub fn reset_inmempool_to_inactive(&mut self) -> bool {
+        match self {
+            TxState::InMempool(unconfirmed_tx_counter) => {
+                *self = TxState::Inactive(*unconfirmed_tx_counter);
+                true
+            }
+
+            TxState::Confirmed(_, _, _)
+            | TxState::Conflicted(_)
+            | TxState::Inactive(_)
+            | TxState::Abandoned => false,
+        }
+    }
 }
 
 impl Display for TxState {
@@ -164,6 +178,13 @@ impl WalletTx {
         }
     }
 
+    pub fn reset_inmempool_to_inactive(&mut self) -> bool {
+        match self {
+            WalletTx::Block(_) => false, // Blocks are never InMempool
+            WalletTx::Tx(tx) => tx.state.reset_inmempool_to_inactive(),
+        }
+    }
+
     pub fn inputs(&self) -> &[TxInput] {
         match self {
             WalletTx::Block(block) => block.kernel_inputs(),

wallet/wallet-controller/src/lib.rs

@@ -54,7 +54,6 @@ use types::{
     SeedWithPassPhrase, SignatureStats, TransactionToInspect, ValidatedSignatures, WalletInfo,
     WalletTypeArgsComputed,
 };
-use utils::set_flag::SetFlag;
 use wallet_storage::DefaultBackend;
 
 use read::ReadOnlyController;
@@ -238,7 +237,7 @@ pub struct Controller<T, W, B: storage::Backend + 'static> {
     wallet_events: W,
 
     mempool_events: MempoolEvents,
-    finished_initial_sync: SetFlag,
+    should_rescan_mempool_txs: bool,
 }
 
 impl<T, WalletEvents, B: storage::Backend> std::fmt::Debug for Controller<T, WalletEvents, B> {
@@ -299,7 +298,7 @@ where
             staking_started: BTreeSet::new(),
             wallet_events,
             mempool_events,
-            finished_initial_sync: SetFlag::new(),
+            should_rescan_mempool_txs: true,
         })
     }
 
@@ -1569,8 +1568,8 @@ where
 
             match self.wallet_mode {
                 WalletControllerMode::Hot => {
-                    // after the first successful sync to the tip fetch all mempool transactions
-                    if !self.finished_initial_sync.test() {
+                    // fetch all mempool transactions after a broken connection or after the initial sync
+                    if self.should_rescan_mempool_txs {
                         let txs = self.rpc_client.mempool_get_transactions().await;
 
                         match txs {
@@ -1580,7 +1579,7 @@ where
                                 {
                                     log::error!("Error adding mempool transactions: {err}");
                                 } else {
-                                    self.finished_initial_sync.set();
+                                    self.should_rescan_mempool_txs = false
                                 }
                             }
                             Err(err) => {
@@ -1608,9 +1607,17 @@ where
                         let event = match maybe_event {
                             Some(e) => e,
                             None => {
+                                // Note: currently the wallet is unable to automatically reconnect to the node when
+                                // the connection is dropped, so for now this branch mostly handles a hypothetical
+                                // situation when the connection is still up, but the stream itself somehow got closed.
+
                                 log::error!("Mempool notifications channel is closed");
-                                tokio::time::sleep(ERROR_DELAY).await;
 
+                                // Reset in-mempool transactions to inactive so we can rescan them when we connect again.
+                                self.wallet.reset_inmempool_txs_to_inactive(Some(&self.wallet_events))?;
+                                self.should_rescan_mempool_txs = true;
+
+                                tokio::time::sleep(ERROR_DELAY).await;
                                 match self.rpc_client
                                     .mempool_subscribe_to_events()
                                     .await {
@@ -1626,6 +1633,21 @@ where
                         };
 
                         match event {
+                            // TODO: mempool can evict transactions - there is a size limit for the entire mempool
+                            // (MAX_MEMPOOL_SIZE_BYTES by default, which is 300Mb) and an expiration time for each tx
+                            // (DEFAULT_MEMPOOL_EXPIRY, which is currently 2 weeks). The wallet must be aware of this
+                            // and mark all its in-mempool txs that have been evicted as inactive.
+                            // At this moment eviction happens when a new tx is being added to the mempool, but note
+                            // that a `NewTransaction` event is not guaranteed in this case (e.g. if the newly added
+                            // tx gets evicted too). Additionally, txs may be evicted after the mempool has been explicitly
+                            // trimmed via `set_max_size`.
+                            // So, the simpler (but not 100% reliable) solution would be to check all of the wallet's
+                            // in-mempool txs for whether they're still in mempool (preferably via a dedicated rpc method,
+                            // to avoid overhead) whenever a `NewTransaction` event arrives.
+                            // A better solution is to introduce a separate mempool event, `TransactionsRemoved`,
+                            // that would contain ids of all txs that have been evicted or removed due to other reasons
+                            // (plus maybe the reason for the eviction/removal).
+
                             MempoolEvent::NewTransaction { tx_id } => {
                                 let transaction = self.rpc_client
                                     .mempool_get_transaction(tx_id)

wallet/wallet-controller/src/runtime_wallet.rs

@@ -102,6 +102,19 @@ where
         }
     }
 
+    pub fn reset_inmempool_txs_to_inactive(
+        &mut self,
+        wallet_events: Option<&impl WalletEvents>,
+    ) -> WalletResult<()> {
+        match self {
+            RuntimeWallet::Software(w) => w.reset_inmempool_txs_to_inactive(wallet_events),
+            #[cfg(feature = "trezor")]
+            RuntimeWallet::Trezor(w) => w.reset_inmempool_txs_to_inactive(wallet_events),
+            #[cfg(feature = "ledger")]
+            RuntimeWallet::Ledger(w) => w.reset_inmempool_txs_to_inactive(wallet_events),
+        }
+    }
+
     pub fn find_account_destination(&self, acc_outpoint: &AccountOutPoint) -> Option<Destination> {
         match self {
             RuntimeWallet::Software(w) => w.find_account_destination(acc_outpoint),