diff --git a/wallet/cli/src/core/console.rs b/wallet/cli/src/core/console.rs index 942be88..74b3d76 100644 --- a/wallet/cli/src/core/console.rs +++ b/wallet/cli/src/core/console.rs @@ -134,21 +134,30 @@ 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: {}\n", total_balance, - utxo_count, spendable_balance, total_balance.checked_sub(&spendable_balance).unwrap(), unconfirmed_balance, + outgoing_balance, + utxo_count, ); } Command::Address(key_type) => { 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/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/cli/src/main.rs b/wallet/cli/src/main.rs index e88260d..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")] @@ -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/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 7c57686..1a20d98 100644 --- a/wallet/sdk/src/scanners/mempool.rs +++ b/wallet/sdk/src/scanners/mempool.rs @@ -1,15 +1,94 @@ -use nyks_standards::wallet::keys::viewing_key::ViewingKey; +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 crate::state::utxos::UtxoKey; +use crate::state::utxos::index::UtxoIndex; + +/// Scans mempool transactions against the current UTXO pool. pub struct MempoolScanner { - keys: Vec, + index: UtxoIndex, + // 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 } impl MempoolScanner { - pub fn new(keys: Vec) -> Self { - MempoolScanner { keys } + pub fn new(index: UtxoIndex) -> Self { + MempoolScanner { + index, + checked_ids: HashSet::new(), + pending_spends: HashMap::new(), + } } - pub fn add_key(&mut self, key: ViewingKey) { - self.keys.push(key); + /// 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| !self.checked_ids.contains(id)) + .cloned() + .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 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 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); + spent_inputs.push(utxo_key); + } + } + + 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.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. + 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/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 4739bac..fbf556b 100644 --- a/wallet/sdk/src/state/utxos/mod.rs +++ b/wallet/sdk/src/state/utxos/mod.rs @@ -1,2 +1,198 @@ +pub mod index; 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..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; @@ -10,10 +11,11 @@ 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; +use crate::state::utxos::index::UtxoIndex; /// Max index sets per `restore_membership_proof` call. const RESTORE_BATCH_LIMIT: usize = 128; @@ -25,6 +27,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 +50,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 +90,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)); } @@ -89,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`. @@ -96,6 +114,7 @@ impl UtxoPool { &mut self, amount: NativeCurrencyAmount, timestamp: Timestamp, + exclude: Option>, ) -> UtxosSelection { let mut invalidated_utxos = Vec::new(); @@ -110,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; } @@ -171,8 +194,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)); } } @@ -211,6 +235,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/state/utxos/utxo.rs b/wallet/sdk/src/state/utxos/utxo.rs deleted file mode 100644 index 12af714..0000000 --- a/wallet/sdk/src/state/utxos/utxo.rs +++ /dev/null @@ -1,186 +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(), - ) - } -} - -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 8e5c05b..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; @@ -13,7 +14,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,10 +23,11 @@ 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::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; @@ -39,12 +40,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. @@ -54,6 +50,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 { @@ -64,6 +68,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)] @@ -71,6 +79,7 @@ pub struct Wallet { rpc: HttpClient, addresses: Arc>, scanner: Arc>, + mempool_scanner: Arc>, utxos: Arc>, pub network: Network, @@ -88,15 +97,18 @@ impl Wallet { network: Network, ) -> Self { let addresses = AddressBook::new(entropy); + let utxos = UtxoPool::new(rpc.clone()); + let view_keys = addresses.view_keys().to_vec(); 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(RwLock::new(MempoolScanner::new(utxos.index()))), + utxos: Arc::new(RwLock::new(utxos)), network, pending_events: Arc::new(RwLock::new(Vec::new())), } @@ -109,7 +121,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); } } @@ -140,18 +152,36 @@ 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() } + 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 @@ -167,12 +197,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; + + events.extend(self.sync_mempool().await); + + if let Some(chain_events) = self.sync_chain(network_height).await? { + events.extend(chain_events); + } + + 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 +251,47 @@ 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) -> Vec { + let mempool_txs = self.rpc.transactions().await.unwrap().transactions; + 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((id, kernel)); + } + } + + let mut mempool_scanner = self.mempool_scanner.write().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`) @@ -226,12 +313,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; - 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;