From 0d082239274d8dd8d14cea0b8143bf37357a334d Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Wed, 5 Aug 2026 04:53:44 +0300 Subject: [PATCH 01/11] feat(wallet-sdk): Add mempool scanners baseline --- wallet/sdk/src/scanners/mempool.rs | 44 +++++++++++++++++++--- wallet/sdk/src/state/utxos/utxo.rs | 9 +++++ wallet/sdk/src/wallet.rs | 59 +++++++++++++++++++++++++----- 3 files changed, 97 insertions(+), 15 deletions(-) diff --git a/wallet/sdk/src/scanners/mempool.rs b/wallet/sdk/src/scanners/mempool.rs index 7c57686..83220e7 100644 --- a/wallet/sdk/src/scanners/mempool.rs +++ b/wallet/sdk/src/scanners/mempool.rs @@ -1,15 +1,47 @@ -use nyks_standards::wallet::keys::viewing_key::ViewingKey; +use std::{collections::HashMap, sync::Arc}; +use nyks_consensus::mutator_set::removal_record::absolute_index_set::AbsoluteIndexSet; +use nyks_rpc_client::block::transaction_kernel::RpcTransactionKernel; +use tokio::sync::RwLock; +use tracing::info; + +use crate::state::utxos::{pool::UtxoPool, utxo::UtxoKey}; + +/// Scans invidual transactions batch independant of chain +/// interacting with current utxo pool pub struct MempoolScanner { - keys: Vec, + utxos: Arc>, } impl MempoolScanner { - pub fn new(keys: Vec) -> Self { - MempoolScanner { keys } + pub fn new(utxos: Arc>) -> Self { + MempoolScanner { utxos } + } + + async fn indices(&self) -> HashMap { + self.utxos + .read() + .await + .utxos + .iter() + .map(|(key, utxo)| (utxo.indices(), *key)) + .collect() } - pub fn add_key(&mut self, key: ViewingKey) { - self.keys.push(key); + pub async fn scan(&self, transactions: Vec) { + let current_indices = self.indices().await; + + for transaction in &transactions { + info!("{} inputs", transaction.inputs.len()); + for input in &transaction.inputs { + let indices = input.absolute_indices; + + if current_indices.contains_key(&indices) { + let utxo_key = current_indices.get(&indices).unwrap(); + + info!("{} is being spent on mempool", utxo_key.aocl_index); + } + } + } } } diff --git a/wallet/sdk/src/state/utxos/utxo.rs b/wallet/sdk/src/state/utxos/utxo.rs index 12af714..0ee50d1 100644 --- a/wallet/sdk/src/state/utxos/utxo.rs +++ b/wallet/sdk/src/state/utxos/utxo.rs @@ -149,6 +149,15 @@ impl MonitoredUtxo { self.membership_proof.receiver_preimage.hash(), ) } + + /// Returns the mutator set indices associated with this UTXO. + /// + /// These indices correspond to the removal record required to spend this UTXO + /// and can be used to match on-chain inputs to this UTXO. + pub fn indices(&self) -> AbsoluteIndexSet { + self.membership_proof + .compute_indices(self.mutator_set_item()) + } } impl Deref for MonitoredUtxo { diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index 8e5c05b..859be8c 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -13,7 +13,6 @@ use nyks_standards::wallet::keys::address::Address; use nyks_standards::wallet::keys::address::Recipient; use nyks_standards::wallet::keys::key::KeyType; use nyks_standards::wallet::keys::key::Spender; - use nyks_wallet_core::entropy::wallet_entropy::WalletEntropy; use nyks_wallet_core::transaction::builder::TransactionBuilder; use nyks_wallet_core::transaction::builder::output::TxOutput; @@ -23,6 +22,7 @@ use tokio::sync::RwLock; use crate::scanners::chain::AdvanceError; use crate::scanners::chain::ChainScanner; +use crate::scanners::mempool::MempoolScanner; use crate::state::address_book::AddressBook; use crate::state::utxos::pool::UtxoPool; use crate::state::utxos::utxo::MonitoredUtxo; @@ -71,6 +71,7 @@ pub struct Wallet { rpc: HttpClient, addresses: Arc>, scanner: Arc>, + mempool_scanner: Arc, utxos: Arc>, pub network: Network, @@ -89,14 +90,16 @@ impl Wallet { ) -> Self { let addresses = AddressBook::new(entropy); let view_keys = addresses.view_keys().to_vec(); + let utxos = Arc::new(RwLock::new(UtxoPool::new(rpc.clone()))); Wallet { - rpc: rpc.clone(), + rpc, addresses: Arc::new(RwLock::new(addresses)), scanner: Arc::new(RwLock::new(ChainScanner::new( height, None, view_keys, network, ))), - utxos: Arc::new(RwLock::new(UtxoPool::new(rpc))), + mempool_scanner: Arc::new(MempoolScanner::new(utxos.clone())), + utxos, network, pending_events: Arc::new(RwLock::new(Vec::new())), } @@ -167,12 +170,35 @@ impl Wallet { pub async fn sync(&self) -> Result, SyncError> { let network_height = self.rpc.height().await.unwrap().height; + let mut events = self.drain_pending_events().await; + + if let Some(chain_events) = self.sync_chain(network_height).await? { + events.extend(chain_events); + } + + self.sync_mempool().await; + + Ok(events) + } + + /// Advances the chain scanner by at most `BATCH_SIZE` blocks (up to + /// `network_height`) and updates the UTXO pool with any newly confirmed + /// UTXOs. + /// + /// Returns `None` if the scanner is already caught up to + /// `network_height`, in which case there is nothing to do. Otherwise + /// returns the `UtxoReceived` events discovered in this batch (which may + /// be empty). + async fn sync_chain( + &self, + network_height: BlockHeight, + ) -> Result>, SyncError> { // Use scan_tip (including unconfirmed blocks) for both the check and the start height. let scan_tip = { let scanner = self.scanner.read().await; let tip = scanner.tip_height(); if tip >= network_height { - return Ok(self.drain_pending_events().await); + return Ok(None); } tip }; @@ -198,13 +224,28 @@ impl Wallet { utxos.add_utxos(confirmed).await }; - let mut events = self.drain_pending_events().await; - events.extend( + Ok(Some( keys.into_iter() - .map(|(key, utxo)| WalletEvent::utxo_received(key, utxo)), - ); + .map(|(key, utxo)| WalletEvent::utxo_received(key, utxo)) + .collect(), + )) + } - Ok(events) + /// Fetches the current mempool's transactions and feeds their kernels + /// to the mempool scanner. + async fn sync_mempool(&self) { + let mempool_txs = self.rpc.transactions().await.unwrap().transactions; + + let mut kernels = Vec::with_capacity(mempool_txs.len()); + for id in mempool_txs { + let kernel = self.rpc.get_transaction_kernel(id).await.unwrap().kernel; + + if let Some(kernel) = kernel { + kernels.push(kernel); + } + } + + self.mempool_scanner.scan(kernels).await; } /// Takes and returns all events queued by other operations (e.g. `send`) From 0d0f477fedbcf7c5fd0c447b662b02cf29820581 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Wed, 5 Aug 2026 05:04:10 +0300 Subject: [PATCH 02/11] feat(wallet-sdk): Move UTXOs to module file --- wallet/sdk/src/scanners/chain.rs | 2 +- wallet/sdk/src/scanners/mempool.rs | 3 +- wallet/sdk/src/state/utxos/mod.rs | 197 ++++++++++++++++++++++++++++- wallet/sdk/src/state/utxos/pool.rs | 8 +- wallet/sdk/src/state/utxos/utxo.rs | 195 ---------------------------- wallet/sdk/src/wallet.rs | 4 +- 6 files changed, 205 insertions(+), 204 deletions(-) delete mode 100644 wallet/sdk/src/state/utxos/utxo.rs diff --git a/wallet/sdk/src/scanners/chain.rs b/wallet/sdk/src/scanners/chain.rs index 60cd7a0..438a969 100644 --- a/wallet/sdk/src/scanners/chain.rs +++ b/wallet/sdk/src/scanners/chain.rs @@ -19,7 +19,7 @@ use nyks_standards::wallet::keys::viewing_key::Decryptor; use nyks_standards::wallet::keys::viewing_key::ViewingKey; use thiserror::Error; -use crate::state::utxos::utxo::IncomingUtxo; +use crate::state::utxos::IncomingUtxo; #[derive(Debug, Copy, Clone, Error)] pub enum AdvanceError { diff --git a/wallet/sdk/src/scanners/mempool.rs b/wallet/sdk/src/scanners/mempool.rs index 83220e7..0f72670 100644 --- a/wallet/sdk/src/scanners/mempool.rs +++ b/wallet/sdk/src/scanners/mempool.rs @@ -5,7 +5,8 @@ use nyks_rpc_client::block::transaction_kernel::RpcTransactionKernel; use tokio::sync::RwLock; use tracing::info; -use crate::state::utxos::{pool::UtxoPool, utxo::UtxoKey}; +use crate::state::utxos::UtxoKey; +use crate::state::utxos::pool::UtxoPool; /// Scans invidual transactions batch independant of chain /// interacting with current utxo pool diff --git a/wallet/sdk/src/state/utxos/mod.rs b/wallet/sdk/src/state/utxos/mod.rs index 4739bac..aabed51 100644 --- a/wallet/sdk/src/state/utxos/mod.rs +++ b/wallet/sdk/src/state/utxos/mod.rs @@ -1,2 +1,197 @@ pub mod pool; -pub mod utxo; + +use std::ops::Deref; + +use nyks_consensus::block::block_height::BlockHeight; +use nyks_consensus::mutator_set::addition_record::AdditionRecord; +use nyks_consensus::mutator_set::commit; +use nyks_consensus::mutator_set::ms_membership_proof::MsMembershipProof; +use nyks_consensus::mutator_set::removal_record::absolute_index_set::AbsoluteIndexSet; +use nyks_consensus::transaction::utxo::Utxo; +use nyks_consensus::twenty_first::tip5::Digest; +use nyks_consensus::twenty_first::tip5::Tip5; +use serde::Deserialize; +use serde::Serialize; + +/// A received UTXO that has not yet been anchored with a membership proof. +/// +/// Contains the data required to monitor the mutator set and later recover +/// or reconstruct the membership proof. +#[derive(Debug, Clone)] +pub struct IncomingUtxo { + pub utxo: Utxo, + pub sender_randomness: Digest, + pub receiver_preimage: Digest, + pub aocl_leaf_index: u64, + + // A metadata so scanner and wallet can handle reorgs easier. + pub inclusion_block: Option<(Digest, BlockHeight)>, +} + +impl IncomingUtxo { + pub fn new( + utxo: Utxo, + sender_randomness: Digest, + receiver_preimage: Digest, + aocl_leaf_index: u64, + inclusion_block: Option<(Digest, BlockHeight)>, + ) -> Self { + IncomingUtxo { + utxo, + sender_randomness, + receiver_preimage, + aocl_leaf_index, + inclusion_block, + } + } + + /// Returns the mutator set indices associated with this UTXO. + /// + /// These indices can be used to recover the membership proof from + /// archival nodes. + pub fn indices(&self) -> AbsoluteIndexSet { + AbsoluteIndexSet::compute( + Tip5::hash(&self.utxo), + self.sender_randomness, + self.receiver_preimage, + self.aocl_leaf_index, + ) + } + + /// Finalize this UTXO with a membership proof. + /// + /// Panics if the proof does not match the expected data for this UTXO. + pub fn finalize(self, membership_proof: MsMembershipProof) -> MonitoredUtxo { + assert!( + self.sender_randomness == membership_proof.sender_randomness, + "Sender randomness doesnt match" + ); + assert!( + self.receiver_preimage == membership_proof.receiver_preimage, + "Receiver digest doesnt match" + ); + assert!( + self.aocl_leaf_index == membership_proof.aocl_leaf_index, + "AOCL leaf index doesnt match" + ); + + MonitoredUtxo { + utxo: self.utxo, + membership_proof, + status: MonitoredUtxoStatus::Unspent, + inclusion_block: self.inclusion_block, + } + } +} + +impl Deref for IncomingUtxo { + type Target = Utxo; + + fn deref(&self) -> &Self::Target { + &self.utxo + } +} + +#[derive(Clone, Debug, Hash, Serialize, Deserialize)] +pub struct ExpectedUtxo { + pub utxo: Utxo, + pub sender_randomness: Digest, + pub receiver_preimage: Digest, + pub inclusion_block: Option<(Digest, BlockHeight)>, +} + +/// Enumerates the possible spent spend-statuses of a monitored UTXO. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub enum MonitoredUtxoStatus { + /// No spend of this UTXO was ever recorded. + Unspent, + + /// UTXO is spent but the node does not know in which block it was spent. + /// To correctly handle reorganizations of the block in which the UTXO was + /// spent, a check against the archival mutator set must be performed each + /// time its status is queried. + SpentInUnknownBlock, + + /// UTXO is spent and the block in which it was spent is known. + SpentIn { + block_hash: Digest, + block_height: BlockHeight, + }, +} + +/// A mined [`Utxo`] managed by the wallet. +/// +/// The UTXO must, at one point, have been mined, although the block in which +/// it was mined might have been abandoned. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct MonitoredUtxo { + pub utxo: Utxo, + + /// Current membership proof of the UTXO + pub membership_proof: MsMembershipProof, + + /// Hash and other metadata of the block, if any, in which a spend of this + /// UTXO was observed. + pub status: MonitoredUtxoStatus, + + /// Hash and other metadata of the block in which this UTXO was confirmed. + pub inclusion_block: Option<(Digest, BlockHeight)>, +} + +impl MonitoredUtxo { + /// Return the `item` from the perspective of the mutator set + pub fn mutator_set_item(&self) -> Digest { + Tip5::hash(&self.utxo) + } + + pub fn addition_record(&self) -> AdditionRecord { + commit( + self.mutator_set_item(), + self.membership_proof.sender_randomness, + self.membership_proof.receiver_preimage.hash(), + ) + } + + /// Returns the mutator set indices associated with this UTXO. + /// + /// These indices correspond to the removal record required to spend this UTXO + /// and can be used to match on-chain inputs to this UTXO. + pub fn indices(&self) -> AbsoluteIndexSet { + self.membership_proof + .compute_indices(self.mutator_set_item()) + } +} + +impl Deref for MonitoredUtxo { + type Target = Utxo; + + fn deref(&self) -> &Self::Target { + &self.utxo + } +} + +/// An ID for UTXOs that defines uniqueness of a UTXO even in the case of +/// reorganizations. In the case of reorganizations both the AOCL leaf index and +/// the addition record is required to identify a UTXO across multiple forks. We +/// do not use the block digest in which the UTXO was mined in this definition +// since a reorganization that repeats some UTXOs at the same location in the +/// AOCL is not considered to introduce new UTXOs, rather the same UTXOs are +/// present, but they were just mined in different blocks. +/// +/// From the perspective of the mutator set, two UTXOs with the same +/// [`UtxoKey`] will always have the same lockscript and mutator set +/// membership proofs. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct UtxoKey { + pub addition_record: AdditionRecord, + pub aocl_index: u64, +} + +impl UtxoKey { + pub fn new(utxo: &MonitoredUtxo) -> Self { + UtxoKey { + addition_record: utxo.addition_record(), + aocl_index: utxo.membership_proof.aocl_leaf_index, + } + } +} diff --git a/wallet/sdk/src/state/utxos/pool.rs b/wallet/sdk/src/state/utxos/pool.rs index 495a754..385d505 100644 --- a/wallet/sdk/src/state/utxos/pool.rs +++ b/wallet/sdk/src/state/utxos/pool.rs @@ -10,10 +10,10 @@ use nyks_rpc_client::RpcApi; use nyks_rpc_client::http::HttpClient; use nyks_rpc_client::wallet::mutator_set::RpcMsMembershipProofPrivacyPreserving; -use crate::state::utxos::utxo::IncomingUtxo; -use crate::state::utxos::utxo::MonitoredUtxo; -use crate::state::utxos::utxo::MonitoredUtxoStatus; -use crate::state::utxos::utxo::UtxoKey; +use crate::state::utxos::IncomingUtxo; +use crate::state::utxos::MonitoredUtxo; +use crate::state::utxos::MonitoredUtxoStatus; +use crate::state::utxos::UtxoKey; /// Max index sets per `restore_membership_proof` call. const RESTORE_BATCH_LIMIT: usize = 128; diff --git a/wallet/sdk/src/state/utxos/utxo.rs b/wallet/sdk/src/state/utxos/utxo.rs deleted file mode 100644 index 0ee50d1..0000000 --- a/wallet/sdk/src/state/utxos/utxo.rs +++ /dev/null @@ -1,195 +0,0 @@ -use std::ops::Deref; - -use nyks_consensus::block::block_height::BlockHeight; -use nyks_consensus::mutator_set::addition_record::AdditionRecord; -use nyks_consensus::mutator_set::commit; -use nyks_consensus::mutator_set::ms_membership_proof::MsMembershipProof; -use nyks_consensus::mutator_set::removal_record::absolute_index_set::AbsoluteIndexSet; -use nyks_consensus::transaction::utxo::Utxo; -use nyks_consensus::twenty_first::tip5::Digest; -use nyks_consensus::twenty_first::tip5::Tip5; -use serde::Deserialize; -use serde::Serialize; - -/// A received UTXO that has not yet been anchored with a membership proof. -/// -/// Contains the data required to monitor the mutator set and later recover -/// or reconstruct the membership proof. -#[derive(Debug, Clone)] -pub struct IncomingUtxo { - pub utxo: Utxo, - pub sender_randomness: Digest, - pub receiver_preimage: Digest, - pub aocl_leaf_index: u64, - - // A metadata so scanner and wallet can handle reorgs easier. - pub inclusion_block: Option<(Digest, BlockHeight)>, -} - -impl IncomingUtxo { - pub fn new( - utxo: Utxo, - sender_randomness: Digest, - receiver_preimage: Digest, - aocl_leaf_index: u64, - inclusion_block: Option<(Digest, BlockHeight)>, - ) -> Self { - IncomingUtxo { - utxo, - sender_randomness, - receiver_preimage, - aocl_leaf_index, - inclusion_block, - } - } - - /// Returns the mutator set indices associated with this UTXO. - /// - /// These indices can be used to recover the membership proof from - /// archival nodes. - pub fn indices(&self) -> AbsoluteIndexSet { - AbsoluteIndexSet::compute( - Tip5::hash(&self.utxo), - self.sender_randomness, - self.receiver_preimage, - self.aocl_leaf_index, - ) - } - - /// Finalize this UTXO with a membership proof. - /// - /// Panics if the proof does not match the expected data for this UTXO. - pub fn finalize(self, membership_proof: MsMembershipProof) -> MonitoredUtxo { - assert!( - self.sender_randomness == membership_proof.sender_randomness, - "Sender randomness doesnt match" - ); - assert!( - self.receiver_preimage == membership_proof.receiver_preimage, - "Receiver digest doesnt match" - ); - assert!( - self.aocl_leaf_index == membership_proof.aocl_leaf_index, - "AOCL leaf index doesnt match" - ); - - MonitoredUtxo { - utxo: self.utxo, - membership_proof, - status: MonitoredUtxoStatus::Unspent, - inclusion_block: self.inclusion_block, - } - } -} - -impl Deref for IncomingUtxo { - type Target = Utxo; - - fn deref(&self) -> &Self::Target { - &self.utxo - } -} - -#[derive(Clone, Debug, Hash, Serialize, Deserialize)] -pub struct ExpectedUtxo { - pub utxo: Utxo, - pub sender_randomness: Digest, - pub receiver_preimage: Digest, - pub inclusion_block: Option<(Digest, BlockHeight)>, -} - -/// Enumerates the possible spent spend-statuses of a monitored UTXO. -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] -pub enum MonitoredUtxoStatus { - /// No spend of this UTXO was ever recorded. - Unspent, - - /// UTXO is spent but the node does not know in which block it was spent. - /// To correctly handle reorganizations of the block in which the UTXO was - /// spent, a check against the archival mutator set must be performed each - /// time its status is queried. - SpentInUnknownBlock, - - /// UTXO is spent and the block in which it was spent is known. - SpentIn { - block_hash: Digest, - block_height: BlockHeight, - }, -} - -/// A mined [`Utxo`] managed by the wallet. -/// -/// The UTXO must, at one point, have been mined, although the block in which -/// it was mined might have been abandoned. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct MonitoredUtxo { - pub utxo: Utxo, - - /// Current membership proof of the UTXO - pub membership_proof: MsMembershipProof, - - /// Hash and other metadata of the block, if any, in which a spend of this - /// UTXO was observed. - pub status: MonitoredUtxoStatus, - - /// Hash and other metadata of the block in which this UTXO was confirmed. - pub inclusion_block: Option<(Digest, BlockHeight)>, -} - -impl MonitoredUtxo { - /// Return the `item` from the perspective of the mutator set - pub fn mutator_set_item(&self) -> Digest { - Tip5::hash(&self.utxo) - } - - pub fn addition_record(&self) -> AdditionRecord { - commit( - self.mutator_set_item(), - self.membership_proof.sender_randomness, - self.membership_proof.receiver_preimage.hash(), - ) - } - - /// Returns the mutator set indices associated with this UTXO. - /// - /// These indices correspond to the removal record required to spend this UTXO - /// and can be used to match on-chain inputs to this UTXO. - pub fn indices(&self) -> AbsoluteIndexSet { - self.membership_proof - .compute_indices(self.mutator_set_item()) - } -} - -impl Deref for MonitoredUtxo { - type Target = Utxo; - - fn deref(&self) -> &Self::Target { - &self.utxo - } -} - -/// An ID for UTXOs that defines uniqueness of a UTXO even in the case of -/// reorganizations. In the case of reorganizations both the AOCL leaf index and -/// the addition record is required to identify a UTXO across multiple forks. We -/// do not use the block digest in which the UTXO was mined in this definition -// since a reorganization that repeats some UTXOs at the same location in the -/// AOCL is not considered to introduce new UTXOs, rather the same UTXOs are -/// present, but they were just mined in different blocks. -/// -/// From the perspective of the mutator set, two UTXOs with the same -/// [`UtxoKey`] will always have the same lockscript and mutator set -/// membership proofs. -#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] -pub struct UtxoKey { - pub addition_record: AdditionRecord, - pub aocl_index: u64, -} - -impl UtxoKey { - pub fn new(utxo: &MonitoredUtxo) -> Self { - UtxoKey { - addition_record: utxo.addition_record(), - aocl_index: utxo.membership_proof.aocl_leaf_index, - } - } -} diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index 859be8c..a6aaa67 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -24,9 +24,9 @@ use crate::scanners::chain::AdvanceError; use crate::scanners::chain::ChainScanner; use crate::scanners::mempool::MempoolScanner; use crate::state::address_book::AddressBook; +use crate::state::utxos::MonitoredUtxo; +use crate::state::utxos::UtxoKey; use crate::state::utxos::pool::UtxoPool; -use crate::state::utxos::utxo::MonitoredUtxo; -use crate::state::utxos::utxo::UtxoKey; const BATCH_SIZE: usize = 100; From 5a2db3ccea9ddf914437b89824d90edc8eef5613 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Wed, 5 Aug 2026 05:32:16 +0300 Subject: [PATCH 03/11] feat(wallet-sdk): Index indices => utxo id on UtxoPool --- wallet/sdk/src/state/utxos/index.rs | 30 +++++++++++++++++++++++++++++ wallet/sdk/src/state/utxos/mod.rs | 1 + wallet/sdk/src/state/utxos/pool.rs | 23 ++++++++++++++++++---- wallet/sdk/src/wallet.rs | 9 ++------- 4 files changed, 52 insertions(+), 11 deletions(-) create mode 100644 wallet/sdk/src/state/utxos/index.rs diff --git a/wallet/sdk/src/state/utxos/index.rs b/wallet/sdk/src/state/utxos/index.rs new file mode 100644 index 0000000..b9b3c09 --- /dev/null +++ b/wallet/sdk/src/state/utxos/index.rs @@ -0,0 +1,30 @@ +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::RwLock; + +use nyks_consensus::mutator_set::removal_record::absolute_index_set::AbsoluteIndexSet; + +use crate::state::utxos::UtxoKey; + +/// Cheap, independently-lockable index from a UTXO's absolute index set to +/// its pool key. Kept in sync with `UtxoPool` at every insert/evict. +#[derive(Clone, Default)] +pub struct UtxoIndex(Arc>>); + +impl UtxoIndex { + pub fn new() -> Self { + UtxoIndex(Arc::new(RwLock::new(HashMap::new()))) + } + + pub async fn get(&self, idx: &AbsoluteIndexSet) -> Option { + self.0.read().await.get(idx).copied() + } + + pub(crate) async fn insert(&self, idx: AbsoluteIndexSet, key: UtxoKey) { + self.0.write().await.insert(idx, key); + } + + pub(crate) async fn remove(&self, idx: &AbsoluteIndexSet) { + self.0.write().await.remove(idx); + } +} diff --git a/wallet/sdk/src/state/utxos/mod.rs b/wallet/sdk/src/state/utxos/mod.rs index aabed51..fbf556b 100644 --- a/wallet/sdk/src/state/utxos/mod.rs +++ b/wallet/sdk/src/state/utxos/mod.rs @@ -1,3 +1,4 @@ +pub mod index; pub mod pool; use std::ops::Deref; diff --git a/wallet/sdk/src/state/utxos/pool.rs b/wallet/sdk/src/state/utxos/pool.rs index 385d505..7a35893 100644 --- a/wallet/sdk/src/state/utxos/pool.rs +++ b/wallet/sdk/src/state/utxos/pool.rs @@ -14,6 +14,7 @@ use crate::state::utxos::IncomingUtxo; use crate::state::utxos::MonitoredUtxo; use crate::state::utxos::MonitoredUtxoStatus; use crate::state::utxos::UtxoKey; +use crate::state::utxos::index::UtxoIndex; /// Max index sets per `restore_membership_proof` call. const RESTORE_BATCH_LIMIT: usize = 128; @@ -25,6 +26,7 @@ const RESTORE_BATCH_LIMIT: usize = 128; pub struct UtxoPool { pub rpc: HttpClient, pub utxos: HashMap, + pub index: UtxoIndex, } /// Result of [`UtxoPool::select_utxos`]. @@ -47,12 +49,22 @@ impl UtxoPool { UtxoPool { rpc, utxos: HashMap::new(), + index: UtxoIndex::new(), } } + /// Shared handle to this pool's index, for callers (e.g. `MempoolScanner`) + /// that only need index lookups, not full pool access. + pub fn index(&self) -> UtxoIndex { + self.index.clone() + } + /// Returns true if it was a new UTXO - pub fn import_utxo(&mut self, utxo: MonitoredUtxo) -> bool { - self.utxos.insert(UtxoKey::new(&utxo), utxo).is_none() + pub async fn import_utxo(&mut self, utxo: MonitoredUtxo) -> bool { + let key = UtxoKey::new(&utxo); + + self.index.insert(utxo.indices(), key).await; + self.utxos.insert(key, utxo).is_none() } /// Ingests UTXOs, restoring proofs only for the new ones. @@ -77,9 +89,11 @@ impl UtxoPool { ) .unwrap(); let utxo = utxo.finalize(utxo_msmp); + let utxo_key = UtxoKey::new(&utxo); + self.index.insert(utxo.indices(), utxo_key).await; + self.utxos.insert(utxo_key, utxo.clone()); - self.utxos.insert(utxo_key.clone(), utxo.clone()); added_utxos.push((utxo_key, utxo)); } @@ -171,8 +185,9 @@ impl UtxoPool { // Drop spent UTXOs and retry against what remains. for key in spent_utxos { let mut utxo = self.utxos.remove(&key).expect("key was just selected"); - utxo.status = MonitoredUtxoStatus::SpentInUnknownBlock; + self.index.remove(&utxo.indices()).await; + utxo.status = MonitoredUtxoStatus::SpentInUnknownBlock; invalidated_utxos.push((key, utxo)); } } diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index a6aaa67..6b49c18 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -39,12 +39,7 @@ pub enum SyncError { Advance(#[from] AdvanceError), } -/// Events emitted by the wallet as a result of chain-sync activity. -/// -/// This is intentionally an enum (rather than just returning the raw utxo -/// pairs) so that future sync-driven activity (e.g. spent-utxo detection, -/// reorgs, balance-threshold crossings, etc.) can be represented without -/// changing the signature of `Wallet::sync`. +/// Events emitted by the wallet as a result of sync and scan activity. #[derive(Debug, Clone)] pub enum WalletEvent { /// A new UTXO was discovered and added to the wallet's UTXO pool. @@ -112,7 +107,7 @@ impl Wallet { let mut utxos_pool = self.utxos.write().await; for utxo in utxos { - let is_unique = utxos_pool.import_utxo(utxo); + let is_unique = utxos_pool.import_utxo(utxo).await; assert!(is_unique); } } From 5c01671994e16327829f07b8f45760ba28a0f41e Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Wed, 5 Aug 2026 05:35:18 +0300 Subject: [PATCH 04/11] feat(wallet-sdk): Wire mempool scanner into index --- wallet/cli/src/core/storage.rs | 4 ++-- wallet/sdk/src/scanners/mempool.rs | 31 +++++------------------------- wallet/sdk/src/wallet.rs | 7 ++++--- 3 files changed, 11 insertions(+), 31 deletions(-) diff --git a/wallet/cli/src/core/storage.rs b/wallet/cli/src/core/storage.rs index 0f5b3bf..8e9b1dd 100644 --- a/wallet/cli/src/core/storage.rs +++ b/wallet/cli/src/core/storage.rs @@ -5,8 +5,8 @@ use std::path::PathBuf; use nyks_consensus::block::block_height::BlockHeight; use nyks_standards::wallet::keys::key::KeyType; -use nyks_wallet_sdk::state::utxos::utxo::MonitoredUtxo; -use nyks_wallet_sdk::state::utxos::utxo::UtxoKey; +use nyks_wallet_sdk::state::utxos::MonitoredUtxo; +use nyks_wallet_sdk::state::utxos::UtxoKey; use serde::Deserialize; use serde::Serialize; diff --git a/wallet/sdk/src/scanners/mempool.rs b/wallet/sdk/src/scanners/mempool.rs index 0f72670..a2b51c1 100644 --- a/wallet/sdk/src/scanners/mempool.rs +++ b/wallet/sdk/src/scanners/mempool.rs @@ -1,45 +1,24 @@ -use std::{collections::HashMap, sync::Arc}; - -use nyks_consensus::mutator_set::removal_record::absolute_index_set::AbsoluteIndexSet; use nyks_rpc_client::block::transaction_kernel::RpcTransactionKernel; -use tokio::sync::RwLock; use tracing::info; -use crate::state::utxos::UtxoKey; -use crate::state::utxos::pool::UtxoPool; +use crate::state::utxos::index::UtxoIndex; /// Scans invidual transactions batch independant of chain /// interacting with current utxo pool pub struct MempoolScanner { - utxos: Arc>, + index: UtxoIndex, } impl MempoolScanner { - pub fn new(utxos: Arc>) -> Self { - MempoolScanner { utxos } - } - - async fn indices(&self) -> HashMap { - self.utxos - .read() - .await - .utxos - .iter() - .map(|(key, utxo)| (utxo.indices(), *key)) - .collect() + pub fn new(index: UtxoIndex) -> Self { + MempoolScanner { index } } pub async fn scan(&self, transactions: Vec) { - let current_indices = self.indices().await; - for transaction in &transactions { info!("{} inputs", transaction.inputs.len()); for input in &transaction.inputs { - let indices = input.absolute_indices; - - if current_indices.contains_key(&indices) { - let utxo_key = current_indices.get(&indices).unwrap(); - + if let Some(utxo_key) = self.index.get(&input.absolute_indices).await { info!("{} is being spent on mempool", utxo_key.aocl_index); } } diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index 6b49c18..8e44e0b 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -84,8 +84,9 @@ impl Wallet { network: Network, ) -> Self { let addresses = AddressBook::new(entropy); + let utxos = UtxoPool::new(rpc.clone()); + let view_keys = addresses.view_keys().to_vec(); - let utxos = Arc::new(RwLock::new(UtxoPool::new(rpc.clone()))); Wallet { rpc, @@ -93,8 +94,8 @@ impl Wallet { scanner: Arc::new(RwLock::new(ChainScanner::new( height, None, view_keys, network, ))), - mempool_scanner: Arc::new(MempoolScanner::new(utxos.clone())), - utxos, + mempool_scanner: Arc::new(MempoolScanner::new(utxos.index())), + utxos: Arc::new(RwLock::new(utxos)), network, pending_events: Arc::new(RwLock::new(Vec::new())), } From de5a03061b4f245a875a19d8c73567928cce1f70 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Wed, 5 Aug 2026 18:04:36 +0300 Subject: [PATCH 05/11] feat(wallet-sdk): Only scan relevant mempool transactions --- wallet/sdk/src/scanners/mempool.rs | 61 +++++++++++++++++++++++++++--- wallet/sdk/src/wallet.rs | 32 +++++++++++----- 2 files changed, 78 insertions(+), 15 deletions(-) diff --git a/wallet/sdk/src/scanners/mempool.rs b/wallet/sdk/src/scanners/mempool.rs index a2b51c1..76cc8b8 100644 --- a/wallet/sdk/src/scanners/mempool.rs +++ b/wallet/sdk/src/scanners/mempool.rs @@ -1,27 +1,76 @@ +use std::collections::HashMap; +use std::collections::HashSet; + +use nyks_consensus::transaction::transaction_kernel_id::TransactionKernelId; use nyks_rpc_client::block::transaction_kernel::RpcTransactionKernel; use tracing::info; use crate::state::utxos::index::UtxoIndex; -/// Scans invidual transactions batch independant of chain -/// interacting with current utxo pool +#[derive(Clone, Copy, PartialEq, Eq)] +enum TxStatus { + Relevant, + Unrelated, +} + +/// Scans mempool transactions against the current UTXO pool. pub struct MempoolScanner { index: UtxoIndex, + // IDs we've already checked. + cache: HashMap, } impl MempoolScanner { pub fn new(index: UtxoIndex) -> Self { - MempoolScanner { index } + MempoolScanner { + index, + cache: HashMap::new(), + } + } + + /// Returns IDs that need to be fetched. + /// Unrelated transactions are skipped if already cached. + pub async fn ids_to_fetch( + &self, + mempool_ids: &[TransactionKernelId], + ) -> Vec { + mempool_ids + .iter() + .filter(|id| match self.cache.get(id) { + Some(TxStatus::Unrelated) => false, + Some(TxStatus::Relevant) => true, + None => true, + }) + .cloned() + .collect() } - pub async fn scan(&self, transactions: Vec) { - for transaction in &transactions { - info!("{} inputs", transaction.inputs.len()); + /// Checks the transactions and updates their status in the cache. + pub async fn scan(&mut self, transactions: Vec<(TransactionKernelId, RpcTransactionKernel)>) { + for (id, transaction) in &transactions { + let mut relevant = false; + for input in &transaction.inputs { if let Some(utxo_key) = self.index.get(&input.absolute_indices).await { info!("{} is being spent on mempool", utxo_key.aocl_index); + relevant = true; } } + + self.cache.insert( + *id, + if relevant { + TxStatus::Relevant + } else { + TxStatus::Unrelated + }, + ); } } + + /// Removes transactions that are no longer in the mempool. + pub async fn evict_stale(&mut self, current_mempool_ids: Vec) { + let current_mempool_ids: HashSet<_> = current_mempool_ids.into_iter().collect(); + self.cache.retain(|id, _| current_mempool_ids.contains(id)); + } } diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index 8e44e0b..0ac3e37 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -66,7 +66,7 @@ pub struct Wallet { rpc: HttpClient, addresses: Arc>, scanner: Arc>, - mempool_scanner: Arc, + mempool_scanner: Arc>, utxos: Arc>, pub network: Network, @@ -94,7 +94,7 @@ impl Wallet { scanner: Arc::new(RwLock::new(ChainScanner::new( height, None, view_keys, network, ))), - mempool_scanner: Arc::new(MempoolScanner::new(utxos.index())), + mempool_scanner: Arc::new(RwLock::new(MempoolScanner::new(utxos.index()))), utxos: Arc::new(RwLock::new(utxos)), network, pending_events: Arc::new(RwLock::new(Vec::new())), @@ -231,17 +231,31 @@ impl Wallet { /// to the mempool scanner. async fn sync_mempool(&self) { let mempool_txs = self.rpc.transactions().await.unwrap().transactions; - - let mut kernels = Vec::with_capacity(mempool_txs.len()); - for id in mempool_txs { - let kernel = self.rpc.get_transaction_kernel(id).await.unwrap().kernel; - + let ids_to_fetch = self + .mempool_scanner + .read() + .await + .ids_to_fetch(&mempool_txs) + .await; + + let mut kernels = Vec::with_capacity(ids_to_fetch.len()); + for id in ids_to_fetch { + let kernel = self + .rpc + .get_transaction_kernel(id.clone()) + .await + .unwrap() + .kernel; if let Some(kernel) = kernel { - kernels.push(kernel); + kernels.push((id, kernel)); } } - self.mempool_scanner.scan(kernels).await; + let mut mempool_scanner = self.mempool_scanner.write().await; + mempool_scanner.scan(kernels).await; + + // keep cache from growing unbounded as txs leave the mempool + mempool_scanner.evict_stale(mempool_txs).await; } /// Takes and returns all events queued by other operations (e.g. `send`) From 37f5a7e44524589f4684c8baf3daf95fb377e9f5 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Wed, 5 Aug 2026 18:24:56 +0300 Subject: [PATCH 06/11] feat(wallet-sdk): Add `outgoing_balance` by mempool scanning --- wallet/sdk/src/scanners/mempool.rs | 27 +++++++++++++++++++++++++-- wallet/sdk/src/state/utxos/pool.rs | 11 +++++++++++ wallet/sdk/src/wallet.rs | 13 ++++++++++++- 3 files changed, 48 insertions(+), 3 deletions(-) diff --git a/wallet/sdk/src/scanners/mempool.rs b/wallet/sdk/src/scanners/mempool.rs index 76cc8b8..fa0dabe 100644 --- a/wallet/sdk/src/scanners/mempool.rs +++ b/wallet/sdk/src/scanners/mempool.rs @@ -3,8 +3,8 @@ use std::collections::HashSet; use nyks_consensus::transaction::transaction_kernel_id::TransactionKernelId; use nyks_rpc_client::block::transaction_kernel::RpcTransactionKernel; -use tracing::info; +use crate::state::utxos::UtxoKey; use crate::state::utxos::index::UtxoIndex; #[derive(Clone, Copy, PartialEq, Eq)] @@ -18,6 +18,8 @@ pub struct MempoolScanner { index: UtxoIndex, // IDs we've already checked. cache: HashMap, + // UTXOs currently observed as spent by relevant mempool transactions. + pub pending_spends: HashMap>, // Kept supporting multiple txs just in case } impl MempoolScanner { @@ -25,6 +27,7 @@ impl MempoolScanner { MempoolScanner { index, cache: HashMap::new(), + pending_spends: HashMap::new(), } } @@ -45,6 +48,18 @@ impl MempoolScanner { .collect() } + /// Returns true if the given UTXO is currently being spent by a + /// transaction sitting in the mempool. + pub fn is_pending_spend(&self, utxo_key: &UtxoKey) -> bool { + self.pending_spends.contains_key(utxo_key) + } + + /// Returns the UTXO keys currently observed as spent by a transaction + /// sitting in the mempool. + pub fn pending_spend_utxos(&self) -> impl Iterator + '_ { + self.pending_spends.keys() + } + /// Checks the transactions and updates their status in the cache. pub async fn scan(&mut self, transactions: Vec<(TransactionKernelId, RpcTransactionKernel)>) { for (id, transaction) in &transactions { @@ -52,7 +67,7 @@ impl MempoolScanner { for input in &transaction.inputs { if let Some(utxo_key) = self.index.get(&input.absolute_indices).await { - info!("{} is being spent on mempool", utxo_key.aocl_index); + self.pending_spends.entry(utxo_key).or_default().insert(*id); relevant = true; } } @@ -71,6 +86,14 @@ impl MempoolScanner { /// Removes transactions that are no longer in the mempool. pub async fn evict_stale(&mut self, current_mempool_ids: Vec) { let current_mempool_ids: HashSet<_> = current_mempool_ids.into_iter().collect(); + self.cache.retain(|id, _| current_mempool_ids.contains(id)); + + // Drop stale tx ids from each UTXO's spender set, and drop the + // UTXO entry entirely once no live tx is spending it anymore. + self.pending_spends.retain(|_, spender_ids| { + spender_ids.retain(|id| current_mempool_ids.contains(id)); + !spender_ids.is_empty() + }); } } diff --git a/wallet/sdk/src/state/utxos/pool.rs b/wallet/sdk/src/state/utxos/pool.rs index 7a35893..eebeb06 100644 --- a/wallet/sdk/src/state/utxos/pool.rs +++ b/wallet/sdk/src/state/utxos/pool.rs @@ -226,6 +226,17 @@ impl UtxoPool { total_amount } + /// Sums the amounts of UTXOs matching the given utxo keys + pub async fn total_balance_from_utxos<'a>( + &self, + utxos: impl Iterator, + ) -> NativeCurrencyAmount { + utxos + .filter_map(|key| self.utxos.get(key)) + .map(|utxo| utxo.get_native_currency_amount()) + .fold(NativeCurrencyAmount::zero(), |acc, amt| acc + amt) + } + /// Calls `restore_membership_proof` in chunks of at most /// [`RESTORE_BATCH_LIMIT`], retrying if the tip changes mid-flight. /// diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index 0ac3e37..da44bb3 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -151,6 +151,17 @@ impl Wallet { self.scanner.read().await.unconfirmed_balance() } + pub async fn outgoing_balance(&self) -> NativeCurrencyAmount { + let mempool_scanner = self.mempool_scanner.read().await; + let utxos = mempool_scanner.pending_spend_utxos(); + + self.utxos + .read() + .await + .total_balance_from_utxos(utxos) + .await + } + /// Sync wallet forward by at most `BATCH_SIZE` blocks. /// /// Does not necessarily reach the current chain tip in one call, call @@ -281,7 +292,7 @@ impl Wallet { // Generate "spendable" UTXOs and prepare them for spending. let timestamp = Timestamp::now(); let mut utxos = self.utxos.write().await; - let selection = utxos.select_utxos(amount + fee, timestamp).await; + let selection = utxos.select_utxos(amount + fee, timestamp).await; // TODO: Allow an external filter callback drop(utxos); if !selection.invalidated_utxos.is_empty() { From bf848342e60e862f2716e5d979c15876d1173bd3 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Wed, 5 Aug 2026 18:35:01 +0300 Subject: [PATCH 07/11] feat(wallet-cli): Show outgoing UTXO amount --- wallet/cli/src/core/console.rs | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/wallet/cli/src/core/console.rs b/wallet/cli/src/core/console.rs index 942be88..e36124a 100644 --- a/wallet/cli/src/core/console.rs +++ b/wallet/cli/src/core/console.rs @@ -134,14 +134,23 @@ pub async fn start_console(wallet: Wallet) { let spendable_balance = wallet.spendable_balance().await; let total_balance = wallet.total_balance().await; let unconfirmed_balance = wallet.unconfirmed_balance().await; - - info!( - "Balance: {} NYKS ({} UTXOs; {} spendable, {} timelocked, {} unconfirmed).", + let outgoing_balance = wallet.outgoing_balance().await; + + println!( + "\n\ + Balance\n\ + ├─ Total: {} NYKS\n\ + ├─ Spendable: {} NYKS\n\ + ├─ Timelocked: {} NYKS\n\ + ├─ Unconfirmed: {} NYKS\n\ + ├─ Outgoing: {} NYKS\n\ + └─ UTXOs: {}", total_balance, - utxo_count, spendable_balance, total_balance.checked_sub(&spendable_balance).unwrap(), unconfirmed_balance, + outgoing_balance, + utxo_count, ); } Command::Address(key_type) => { From e6b81caa19475fcda0da8604c424fdf1f61ebd0d Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Thu, 6 Aug 2026 04:13:50 +0300 Subject: [PATCH 08/11] feat(wallet-sdk): Exclude UTXOs that are on mempool --- wallet/sdk/src/state/utxos/pool.rs | 9 +++++++++ wallet/sdk/src/wallet.rs | 14 +++++++++++--- 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/wallet/sdk/src/state/utxos/pool.rs b/wallet/sdk/src/state/utxos/pool.rs index eebeb06..8d79ed8 100644 --- a/wallet/sdk/src/state/utxos/pool.rs +++ b/wallet/sdk/src/state/utxos/pool.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::collections::HashSet; use num_traits::CheckedSub; use num_traits::Zero; @@ -103,6 +104,9 @@ impl UtxoPool { /// Greedily selects UTXOs covering `amount`, syncing only the selected /// ones against current chain state. /// + /// UTXOs whose keys appear in `exclude` are skipped entirely, e.g. ones + /// already committed to another in-flight transaction. + /// /// Spent UTXOs are evicted and selection retries against the rest; all /// evictions are collected and returned. Every returned UTXO is valid /// against the returned `msa`. @@ -110,6 +114,7 @@ impl UtxoPool { &mut self, amount: NativeCurrencyAmount, timestamp: Timestamp, + exclude: Option>, ) -> UtxosSelection { let mut invalidated_utxos = Vec::new(); @@ -124,6 +129,10 @@ impl UtxoPool { break; } + if exclude.as_ref().is_some_and(|e| e.contains(key)) { + continue; + } + if !utxo.can_spend_at(timestamp) { continue; } diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index da44bb3..fb3ba67 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -288,12 +288,20 @@ impl Wallet { fee: NativeCurrencyAmount, ) -> Result { let height = self.tip_height().await; + let timestamp = Timestamp::now(); // Generate "spendable" UTXOs and prepare them for spending. - let timestamp = Timestamp::now(); let mut utxos = self.utxos.write().await; - let selection = utxos.select_utxos(amount + fee, timestamp).await; // TODO: Allow an external filter callback - drop(utxos); + let excluded_utxos = self + .mempool_scanner + .read() + .await + .pending_spend_utxos() + .copied() + .collect(); + let selection = utxos + .select_utxos(amount + fee, timestamp, Some(excluded_utxos)) + .await; if !selection.invalidated_utxos.is_empty() { let mut pending = self.pending_events.write().await; From f961f3a0ff71e9ba4796c11807689f9ec0675009 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Thu, 6 Aug 2026 05:18:32 +0300 Subject: [PATCH 09/11] feat(wallet): UtxosOutgoing event and log on CLI --- wallet/cli/src/main.rs | 3 ++ wallet/sdk/src/scanners/mempool.rs | 53 ++++++++++++++---------------- wallet/sdk/src/wallet.rs | 25 +++++++++++--- 3 files changed, 48 insertions(+), 33 deletions(-) diff --git a/wallet/cli/src/main.rs b/wallet/cli/src/main.rs index e88260d..5c6fee0 100644 --- a/wallet/cli/src/main.rs +++ b/wallet/cli/src/main.rs @@ -96,6 +96,9 @@ async fn main() -> Result<()> { storage.utxos.put(&key, &utxo); } + WalletEvent::UtxosOutgoing { id, utxos } => { + info!("{} UTXOs being spent on transaction {}.", utxos.len(), id); + } } } diff --git a/wallet/sdk/src/scanners/mempool.rs b/wallet/sdk/src/scanners/mempool.rs index fa0dabe..1a20d98 100644 --- a/wallet/sdk/src/scanners/mempool.rs +++ b/wallet/sdk/src/scanners/mempool.rs @@ -7,17 +7,11 @@ use nyks_rpc_client::block::transaction_kernel::RpcTransactionKernel; use crate::state::utxos::UtxoKey; use crate::state::utxos::index::UtxoIndex; -#[derive(Clone, Copy, PartialEq, Eq)] -enum TxStatus { - Relevant, - Unrelated, -} - /// Scans mempool transactions against the current UTXO pool. pub struct MempoolScanner { index: UtxoIndex, - // IDs we've already checked. - cache: HashMap, + // IDs we've already checked, relevant or not - never rescanned. + checked_ids: HashSet, // UTXOs currently observed as spent by relevant mempool transactions. pub pending_spends: HashMap>, // Kept supporting multiple txs just in case } @@ -26,24 +20,19 @@ impl MempoolScanner { pub fn new(index: UtxoIndex) -> Self { MempoolScanner { index, - cache: HashMap::new(), + checked_ids: HashSet::new(), pending_spends: HashMap::new(), } } - /// Returns IDs that need to be fetched. - /// Unrelated transactions are skipped if already cached. + /// Returns IDs that need to be fetched, i.e. ones we haven't checked yet. pub async fn ids_to_fetch( &self, mempool_ids: &[TransactionKernelId], ) -> Vec { mempool_ids .iter() - .filter(|id| match self.cache.get(id) { - Some(TxStatus::Unrelated) => false, - Some(TxStatus::Relevant) => true, - None => true, - }) + .filter(|id| !self.checked_ids.contains(id)) .cloned() .collect() } @@ -60,34 +49,40 @@ impl MempoolScanner { self.pending_spends.keys() } - /// Checks the transactions and updates their status in the cache. - pub async fn scan(&mut self, transactions: Vec<(TransactionKernelId, RpcTransactionKernel)>) { + /// Checks the transactions and marks them as checked. Each transaction + /// is scanned at most once ever; relevant ones are returned only here. + pub async fn scan( + &mut self, + transactions: Vec<(TransactionKernelId, RpcTransactionKernel)>, + ) -> Vec<(TransactionKernelId, Vec)> { + let mut relevant_transactions = Vec::new(); + for (id, transaction) in &transactions { - let mut relevant = false; + let mut spent_inputs = Vec::new(); for input in &transaction.inputs { if let Some(utxo_key) = self.index.get(&input.absolute_indices).await { self.pending_spends.entry(utxo_key).or_default().insert(*id); - relevant = true; + spent_inputs.push(utxo_key); } } - self.cache.insert( - *id, - if relevant { - TxStatus::Relevant - } else { - TxStatus::Unrelated - }, - ); + self.checked_ids.insert(*id); + + if !spent_inputs.is_empty() { + relevant_transactions.push((*id, spent_inputs)); + } } + + relevant_transactions } /// Removes transactions that are no longer in the mempool. pub async fn evict_stale(&mut self, current_mempool_ids: Vec) { let current_mempool_ids: HashSet<_> = current_mempool_ids.into_iter().collect(); - self.cache.retain(|id, _| current_mempool_ids.contains(id)); + self.checked_ids + .retain(|id| current_mempool_ids.contains(id)); // Drop stale tx ids from each UTXO's spender set, and drop the // UTXO entry entirely once no live tx is spending it anymore. diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index fb3ba67..ffda709 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -49,6 +49,14 @@ pub enum WalletEvent { /// invalid) while syncing membership proofs, and was evicted from the /// pool. UtxoInvalidated { key: UtxoKey, utxo: MonitoredUtxo }, + + /// A mempool transaction was found to spend one or more of the + /// wallet's UTXOs. Emitted once per transaction, the first time it's + /// observed as relevant. + UtxosOutgoing { + id: TransactionKernelId, + utxos: Vec, + }, } impl WalletEvent { @@ -59,6 +67,10 @@ impl WalletEvent { pub fn utxo_invalidated(key: UtxoKey, utxo: MonitoredUtxo) -> Self { WalletEvent::UtxoInvalidated { key, utxo } } + + pub fn utxos_outgoing(id: TransactionKernelId, utxos: Vec) -> Self { + WalletEvent::UtxosOutgoing { id, utxos } + } } #[derive(Clone)] @@ -179,12 +191,12 @@ impl Wallet { let mut events = self.drain_pending_events().await; + events.extend(self.sync_mempool().await); + if let Some(chain_events) = self.sync_chain(network_height).await? { events.extend(chain_events); } - self.sync_mempool().await; - Ok(events) } @@ -240,7 +252,7 @@ impl Wallet { /// Fetches the current mempool's transactions and feeds their kernels /// to the mempool scanner. - async fn sync_mempool(&self) { + async fn sync_mempool(&self) -> Vec { let mempool_txs = self.rpc.transactions().await.unwrap().transactions; let ids_to_fetch = self .mempool_scanner @@ -263,10 +275,15 @@ impl Wallet { } let mut mempool_scanner = self.mempool_scanner.write().await; - mempool_scanner.scan(kernels).await; + let outgoing_utxos = mempool_scanner.scan(kernels).await; // keep cache from growing unbounded as txs leave the mempool mempool_scanner.evict_stale(mempool_txs).await; + + outgoing_utxos + .into_iter() + .map(|(id, utxos)| WalletEvent::utxos_outgoing(id, utxos)) + .collect() } /// Takes and returns all events queued by other operations (e.g. `send`) From ec3838328e61a0c30c8ef1060ff7bce5b999bb95 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Thu, 6 Aug 2026 05:23:53 +0300 Subject: [PATCH 10/11] feat(wallet-cli): Add more spacing to println logs --- wallet/cli/src/core/console.rs | 4 ++-- wallet/cli/src/main.rs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/wallet/cli/src/core/console.rs b/wallet/cli/src/core/console.rs index e36124a..74b3d76 100644 --- a/wallet/cli/src/core/console.rs +++ b/wallet/cli/src/core/console.rs @@ -144,7 +144,7 @@ pub async fn start_console(wallet: Wallet) { ├─ Timelocked: {} NYKS\n\ ├─ Unconfirmed: {} NYKS\n\ ├─ Outgoing: {} NYKS\n\ - └─ UTXOs: {}", + └─ UTXOs: {}\n", total_balance, spendable_balance, total_balance.checked_sub(&spendable_balance).unwrap(), @@ -157,7 +157,7 @@ pub async fn start_console(wallet: Wallet) { let key_type = key_type.unwrap_or(KeyType::Generation); let address = wallet.address(key_type).await; - println!("\n{}", address.to_bech32m(wallet.network)); + println!("\n{}\n", address.to_bech32m(wallet.network)); } Command::Send { recipient, diff --git a/wallet/cli/src/main.rs b/wallet/cli/src/main.rs index 5c6fee0..e943ef0 100644 --- a/wallet/cli/src/main.rs +++ b/wallet/cli/src/main.rs @@ -1,3 +1,5 @@ +pub mod core; + use std::panic; use std::time::Duration; @@ -14,8 +16,6 @@ use tracing_subscriber::EnvFilter; use crate::core::storage::Storage; -pub mod core; - #[derive(Parser)] #[command(name = "nyks-wallet")] #[command(about = "A nyks daemon wallet")] From e3d6ef4e070b71257e3440a1aa931b2a89978a45 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Fri, 7 Aug 2026 03:06:25 +0300 Subject: [PATCH 11/11] feat(wallet-sdk): Exclude outgoing balance from spendable --- wallet/sdk/src/wallet.rs | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/wallet/sdk/src/wallet.rs b/wallet/sdk/src/wallet.rs index ffda709..d2a5c84 100644 --- a/wallet/sdk/src/wallet.rs +++ b/wallet/sdk/src/wallet.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use num_traits::CheckedSub; use nyks_consensus::block::block_height::BlockHeight; use nyks_consensus::network::Network; use nyks_consensus::proof_abstractions::timestamp::Timestamp; @@ -151,14 +152,21 @@ impl Wallet { self.utxos.read().await.utxo_count() } - pub async fn spendable_balance(&self) -> NativeCurrencyAmount { - self.utxos.read().await.spendable_balance() - } - pub async fn total_balance(&self) -> NativeCurrencyAmount { self.utxos.read().await.total_balance() } + pub async fn spendable_balance(&self) -> NativeCurrencyAmount { + let mempool_scanner = self.mempool_scanner.read().await; + let pending_spend_utxos = mempool_scanner.pending_spend_utxos(); + + let utxos = self.utxos.read().await; + let spendable = utxos.spendable_balance(); + let outgoing = utxos.total_balance_from_utxos(pending_spend_utxos).await; + + spendable.checked_sub(&outgoing).unwrap() + } + pub async fn unconfirmed_balance(&self) -> NativeCurrencyAmount { self.scanner.read().await.unconfirmed_balance() }