From 764a216b4fb2fbb866b04aeea6c3f6f6fd82c079 Mon Sep 17 00:00:00 2001 From: Boris Oncev Date: Thu, 30 Apr 2026 16:25:22 +0700 Subject: [PATCH 1/3] Reset InMempool txs to Inactive before connecting to the node --- wallet/src/account/mod.rs | 5 ++++ wallet/src/account/output_cache/mod.rs | 5 ++++ wallet/src/account/output_cache/tests.rs | 22 ++++++++++++++ wallet/src/wallet/mod.rs | 9 ++++++ wallet/types/src/wallet_tx.rs | 11 +++++++ wallet/wallet-controller/src/lib.rs | 29 ++++++++++++++----- .../wallet-controller/src/runtime_wallet.rs | 10 +++++++ 7 files changed, 83 insertions(+), 8 deletions(-) diff --git a/wallet/src/account/mod.rs b/wallet/src/account/mod.rs index c733f41b24..c9c7ab8c80 100644 --- a/wallet/src/account/mod.rs +++ b/wallet/src/account/mod.rs @@ -2401,6 +2401,11 @@ impl Account { ) } + /// Reset all transactions that are currently in-mempool to inactive state + pub fn reset_inmempool_txs_to_inactive(&mut self) { + self.output_cache.reset_inmempool_txs_to_inactive(); + } + pub fn update_best_block( &mut self, db_tx: &mut impl WalletStorageWriteLocked, diff --git a/wallet/src/account/output_cache/mod.rs b/wallet/src/account/output_cache/mod.rs index bc4bd44371..d059ba853b 100644 --- a/wallet/src/account/output_cache/mod.rs +++ b/wallet/src/account/output_cache/mod.rs @@ -1431,6 +1431,11 @@ impl OutputCache { Ok(()) } + /// Reset all transactions that are currently in-mempool to inactive state. + pub fn reset_inmempool_txs_to_inactive(&mut self) { + self.txs.values_mut().for_each(|tx| tx.reset_inmempool_to_inactive()); + } + fn is_consumed(&self, utxo_states: UtxoStates, outpoint: &UtxoOutPoint) -> bool { self.consumed .get(outpoint) diff --git a/wallet/src/account/output_cache/tests.rs b/wallet/src/account/output_cache/tests.rs index 979d67ee54..d50547f194 100644 --- a/wallet/src/account/output_cache/tests.rs +++ b/wallet/src/account/output_cache/tests.rs @@ -1608,6 +1608,28 @@ 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(&mut output_cache, &chain_config, &mut rng); + } + + // Reset + output_cache.reset_inmempool_txs_to_inactive(); + + // Check none of the txs are InMempool state + for tx in output_cache.txs.values() { + assert!(!matches!(tx.state(), TxState::InMempool(_))); + } +} + fn add_random_transfer_tx( output_cache: &mut OutputCache, chain_config: &ChainConfig, diff --git a/wallet/src/wallet/mod.rs b/wallet/src/wallet/mod.rs index 2414b82534..d1a9fad65a 100644 --- a/wallet/src/wallet/mod.rs +++ b/wallet/src/wallet/mod.rs @@ -812,6 +812,13 @@ where Ok(()) } + /// Resets all transactions in all accounts from in-mempool state to inactive. + pub fn reset_inmempool_txs_to_inactive(&mut self) { + self.accounts + .values_mut() + .for_each(|account| account.reset_inmempool_txs_to_inactive()); + } + fn reset_wallet_transactions( chain_config: Arc, db_tx: &mut impl WalletStorageWriteLocked, @@ -973,6 +980,8 @@ where .into_iter() .map(|account| (account.account_index(), account)) .collect(); + // reset all inmempool transactions to inactive so we can set them properly from the current node again + accounts.values_mut().for_each(|a| a.reset_inmempool_txs_to_inactive()); let latest_median_time = db_tx.get_median_time()?.unwrap_or(chain_config.genesis_block().timestamp()); diff --git a/wallet/types/src/wallet_tx.rs b/wallet/types/src/wallet_tx.rs index 938ae74480..b9cbe68306 100644 --- a/wallet/types/src/wallet_tx.rs +++ b/wallet/types/src/wallet_tx.rs @@ -164,6 +164,17 @@ impl WalletTx { } } + pub fn reset_inmempool_to_inactive(&mut self) { + match self { + WalletTx::Block(_) => {} // Blocks are never InMempool + WalletTx::Tx(tx) => { + if let TxState::InMempool(counter) = tx.state { + tx.state = TxState::Inactive(counter); + } + } + } + } + pub fn inputs(&self) -> &[TxInput] { match self { WalletTx::Block(block) => block.kernel_inputs(), diff --git a/wallet/wallet-controller/src/lib.rs b/wallet/wallet-controller/src/lib.rs index ddf5362255..94c785c46f 100644 --- a/wallet/wallet-controller/src/lib.rs +++ b/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 { wallet_events: W, mempool_events: MempoolEvents, - finished_initial_sync: SetFlag, + should_resecan_mempool_txs: bool, } impl std::fmt::Debug for Controller { @@ -261,8 +260,19 @@ where wallet: RuntimeWallet, wallet_events: W, ) -> Result> { - let mut controller = - Self::new_unsynced(chain_config, rpc_client, wallet, wallet_events).await?; + let mempool_events = rpc_client + .mempool_subscribe_to_events() + .await + .map_err(ControllerError::NodeCallError)?; + let mut controller = Self { + chain_config, + rpc_client, + wallet, + staking_started: BTreeSet::new(), + wallet_events, + mempool_events, + should_resecan_mempool_txs: true, + }; // In the cold mode, try_sync_once is a no-op, so it doesn't matter whether we call it. // We omit the call to avoid printing the "Syncing the wallet" log line, which looks @@ -299,7 +309,7 @@ where staking_started: BTreeSet::new(), wallet_events, mempool_events, - finished_initial_sync: SetFlag::new(), + should_resecan_mempool_txs: true, }) } @@ -1570,7 +1580,7 @@ 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() { + if self.should_resecan_mempool_txs { let txs = self.rpc_client.mempool_get_transactions().await; match txs { @@ -1580,7 +1590,7 @@ where { log::error!("Error adding mempool transactions: {err}"); } else { - self.finished_initial_sync.set(); + self.should_resecan_mempool_txs = false } } Err(err) => { @@ -1609,8 +1619,11 @@ where Some(e) => e, None => { 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(); + self.should_resecan_mempool_txs = true; + tokio::time::sleep(ERROR_DELAY).await; match self.rpc_client .mempool_subscribe_to_events() .await { diff --git a/wallet/wallet-controller/src/runtime_wallet.rs b/wallet/wallet-controller/src/runtime_wallet.rs index b693edfec7..5b02831af6 100644 --- a/wallet/wallet-controller/src/runtime_wallet.rs +++ b/wallet/wallet-controller/src/runtime_wallet.rs @@ -102,6 +102,16 @@ where } } + pub fn reset_inmempool_txs_to_inactive(&mut self) { + match self { + RuntimeWallet::Software(w) => w.reset_inmempool_txs_to_inactive(), + #[cfg(feature = "trezor")] + RuntimeWallet::Trezor(w) => w.reset_inmempool_txs_to_inactive(), + #[cfg(feature = "ledger")] + RuntimeWallet::Ledger(w) => w.reset_inmempool_txs_to_inactive(), + } + } + pub fn find_account_destination(&self, acc_outpoint: &AccountOutPoint) -> Option { match self { RuntimeWallet::Software(w) => w.find_account_destination(acc_outpoint), From e8db614d69d44fa90e028a7844f539a4a3eddc84 Mon Sep 17 00:00:00 2001 From: Boris Oncev Date: Thu, 28 May 2026 17:22:52 +0200 Subject: [PATCH 2/3] fix comments --- wallet/src/account/output_cache/tests.rs | 19 +++++++++++++++++-- wallet/wallet-controller/src/lib.rs | 2 +- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/wallet/src/account/output_cache/tests.rs b/wallet/src/account/output_cache/tests.rs index d50547f194..f7a74328f1 100644 --- a/wallet/src/account/output_cache/tests.rs +++ b/wallet/src/account/output_cache/tests.rs @@ -1618,7 +1618,12 @@ fn reset_inmempool_txs_to_inactive(#[case] seed: Seed) { // add 10 random txs for _ in 0..10 { - add_random_transfer_tx(&mut output_cache, &chain_config, &mut rng); + add_random_transfer_tx_with_state( + &mut output_cache, + &chain_config, + TxStateTag::InMempool, + &mut rng, + ); } // Reset @@ -1634,6 +1639,16 @@ 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::::random_using(&mut rng); let random_block_id = Id::::random_using(&mut rng); @@ -1649,7 +1664,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)), diff --git a/wallet/wallet-controller/src/lib.rs b/wallet/wallet-controller/src/lib.rs index 94c785c46f..6ad7ec38e8 100644 --- a/wallet/wallet-controller/src/lib.rs +++ b/wallet/wallet-controller/src/lib.rs @@ -1579,7 +1579,7 @@ where match self.wallet_mode { WalletControllerMode::Hot => { - // after the first successful sync to the tip fetch all mempool transactions + // fetch all mempool transactions after a broken connection or after the initial sync if self.should_resecan_mempool_txs { let txs = self.rpc_client.mempool_get_transactions().await; From 1dafd7c68a14691800ca9d96d703717f0421f609 Mon Sep 17 00:00:00 2001 From: Mykhailo Kremniov Date: Mon, 6 Jul 2026 19:29:17 +0300 Subject: [PATCH 3/3] Fix after rebase, minor cleanup --- wallet/wallet-controller/src/lib.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/wallet/wallet-controller/src/lib.rs b/wallet/wallet-controller/src/lib.rs index 6ad7ec38e8..236012f88b 100644 --- a/wallet/wallet-controller/src/lib.rs +++ b/wallet/wallet-controller/src/lib.rs @@ -237,7 +237,7 @@ pub struct Controller { wallet_events: W, mempool_events: MempoolEvents, - should_resecan_mempool_txs: bool, + should_rescan_mempool_txs: bool, } impl std::fmt::Debug for Controller { @@ -260,6 +260,7 @@ where wallet: RuntimeWallet, wallet_events: W, ) -> Result> { + let wallet_mode = rpc_client.is_cold_wallet_node().await; let mempool_events = rpc_client .mempool_subscribe_to_events() .await @@ -268,10 +269,11 @@ where chain_config, rpc_client, wallet, + wallet_mode, staking_started: BTreeSet::new(), wallet_events, mempool_events, - should_resecan_mempool_txs: true, + should_rescan_mempool_txs: true, }; // In the cold mode, try_sync_once is a no-op, so it doesn't matter whether we call it. @@ -309,7 +311,7 @@ where staking_started: BTreeSet::new(), wallet_events, mempool_events, - should_resecan_mempool_txs: true, + should_rescan_mempool_txs: true, }) } @@ -1580,7 +1582,7 @@ where match self.wallet_mode { WalletControllerMode::Hot => { // fetch all mempool transactions after a broken connection or after the initial sync - if self.should_resecan_mempool_txs { + if self.should_rescan_mempool_txs { let txs = self.rpc_client.mempool_get_transactions().await; match txs { @@ -1590,7 +1592,7 @@ where { log::error!("Error adding mempool transactions: {err}"); } else { - self.should_resecan_mempool_txs = false + self.should_rescan_mempool_txs = false } } Err(err) => { @@ -1621,7 +1623,7 @@ where log::error!("Mempool notifications channel is closed"); // reset in-mempool transactions to inactive so we can rescan them when we connect again self.wallet.reset_inmempool_txs_to_inactive(); - self.should_resecan_mempool_txs = true; + self.should_rescan_mempool_txs = true; tokio::time::sleep(ERROR_DELAY).await; match self.rpc_client