fix(key-wallet): reserve selected UTXOs to prevent double-spend

dash-spv/tests/dashd_sync/tests_transaction.rs

@@ -1,5 +1,6 @@
 use dash_spv::sync::ProgressPercentage;
 use dashcore::{Address, Amount, Network};
+use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
@@ -345,6 +346,72 @@ async fn test_spend_change_balance() {
     client_handle.stop().await;
 }
 
+/// Two transactions built from the same wallet before either is broadcast must
+/// not select the same UTXO. The first build reserves its inputs so the second
+/// build's coin selection skips them, preventing a double-spend the network
+/// would reject. Without the reservation both builds pick the same UTXO and the
+/// `tx_a`/`tx_b` inputs overlap, so the disjoint-input assertion below fails.
+///
+/// The two builds run sequentially: this guards that the first build's
+/// reservation outlives its wallet lock and is observed by the second build,
+/// not that the two builds are scheduled concurrently.
+#[tokio::test]
+async fn test_concurrent_builds_do_not_double_spend() {
+    let Some(ctx) = TestContext::new(TestChain::Minimal).await else {
+        return;
+    };
+    if !ctx.dashd.supports_mining {
+        eprintln!("Skipping test (dashd RPC miner not available)");
+        return;
+    }
+
+    let (wallet, wallet_id) = create_test_wallet(EMPTY_MNEMONIC, Network::Regtest);
+    let mut client_handle = create_and_start_client(&ctx.client_config, Arc::clone(&wallet)).await;
+    wait_for_sync(&mut client_handle.progress_receiver, ctx.dashd.initial_height).await;
+
+    // Fund the wallet with two separate, equal UTXOs at the same receive
+    // address. Either one alone covers a single send, so naive coin selection
+    // would pick the same UTXO for both builds.
+    let receive_address = reserve_first_address(EMPTY_MNEMONIC).await;
+    let utxo_amount = Amount::from_sat(200_000_000);
+    ctx.dashd.node.send_to_address(&receive_address, utxo_amount);
+    ctx.dashd.node.send_to_address(&receive_address, utxo_amount);
+
+    let miner_address = ctx.dashd.node.get_new_address_from_wallet("default");
+    ctx.dashd.node.generate_blocks(1, &miner_address);
+    let funded_height = ctx.dashd.initial_height + 1;
+    wait_for_sync(&mut client_handle.progress_receiver, funded_height).await;
+    wait_for_wallet_synced(&wallet, &wallet_id, funded_height).await;
+
+    // Build both transactions before broadcasting either, so neither spend has
+    // been processed back into the wallet when the second build selects inputs.
+    let dest_a = Address::dummy(Network::Regtest, 1);
+    let dest_b = Address::dummy(Network::Regtest, 2);
+    let (tx_a, _) =
+        build_and_sign(&wallet, &wallet_id, &dest_a, 50_000_000).await.expect("build tx_a");
+    let (tx_b, _) =
+        build_and_sign(&wallet, &wallet_id, &dest_b, 50_000_000).await.expect("build tx_b");
+
+    let inputs_a: HashSet<_> = tx_a.input.iter().map(|i| i.previous_output).collect();
+    let overlap = tx_b.input.iter().any(|i| inputs_a.contains(&i.previous_output));
+    assert!(!overlap, "tx_a and tx_b must not share an input (double-spend)");
+
+    // Both broadcasts succeed and both transactions reach the mempool: a
+    // double-spend would have the second rejected by the network.
+    client_handle.client.broadcast_transaction(&tx_a).await.expect("broadcast tx_a");
+    let detected_a = wait_for_mempool_tx(&mut client_handle.wallet_event_receiver, MEMPOOL_TIMEOUT)
+        .await
+        .expect("detect tx_a");
+    assert_eq!(detected_a, tx_a.txid());
+    client_handle.client.broadcast_transaction(&tx_b).await.expect("broadcast tx_b");
+    let detected_b = wait_for_mempool_tx(&mut client_handle.wallet_event_receiver, MEMPOOL_TIMEOUT)
+        .await
+        .expect("detect tx_b");
+    assert_eq!(detected_b, tx_b.txid());
+
+    client_handle.stop().await;
+}
+
 /// Spend an incoming mempool UTXO (we own the output, none of the inputs)
 /// before it confirms.
 #[tokio::test]

key-wallet/src/managed_account/managed_core_funds_account.rs

@@ -19,6 +19,7 @@ use crate::managed_account::address_pool;
 use crate::managed_account::managed_account_trait::ManagedAccountTrait;
 use crate::managed_account::managed_account_type::ManagedAccountType;
 use crate::managed_account::managed_core_keys_account::ManagedCoreKeysAccount;
+use crate::managed_account::reservation::ReservationSet;
 use crate::managed_account::transaction_record::{
     InputDetail, OutputDetail, OutputRole, TransactionDirection,
 };
@@ -61,6 +62,12 @@ pub struct ManagedCoreFundsAccount {
     /// Rebuilt from `transactions` during deserialization.
     #[cfg_attr(feature = "serde", serde(skip_serializing))]
     spent_outpoints: HashSet<OutPoint>,
+    /// Outpoints reserved by in-flight transaction builds so concurrent builds
+    /// do not select the same UTXO before the first build's transaction is
+    /// processed. Empty after a restart, where chain and mempool sync
+    /// re-establish which coins are spent.
+    #[cfg_attr(feature = "serde", serde(skip))]
+    reservations: ReservationSet,
 }
 
 impl ManagedCoreFundsAccount {
@@ -71,6 +78,7 @@ impl ManagedCoreFundsAccount {
             balance: WalletCoreBalance::default(),
             utxos: BTreeMap::new(),
             spent_outpoints: HashSet::new(),
+            reservations: ReservationSet::default(),
         }
     }
 
@@ -97,9 +105,31 @@ impl ManagedCoreFundsAccount {
             balance: WalletCoreBalance::default(),
             utxos: BTreeMap::new(),
             spent_outpoints: HashSet::new(),
+            reservations: ReservationSet::default(),
         }
     }
 
+    /// Reservation set tracking outpoints chosen by in-flight transaction
+    /// builds, consulted by coin selection so concurrent builds do not pick the
+    /// same UTXO.
+    ///
+    /// The reservation is only effective when each build runs `set_funding`
+    /// through `assemble_unsigned` under a single uninterrupted hold of the
+    /// wallet lock. If two builds interleave between observing the UTXO set and
+    /// reserving their inputs, both can see the same UTXO as free and select it.
+    pub(crate) fn reservations(&self) -> &ReservationSet {
+        &self.reservations
+    }
+
+    /// Release the reservations held for `tx`'s inputs. Call this when a built
+    /// transaction will not be broadcast, e.g. the user cancelled, so its inputs
+    /// become selectable again immediately instead of waiting out the TTL
+    /// backstop. Broadcast transactions release on their own once the spend is
+    /// processed back into the wallet, so this is only for abandoned builds.
+    pub fn release_reservation(&self, tx: &Transaction) {
+        self.reservations.release(tx.input.iter().map(|input| &input.previous_output));
+    }
+
     /// Get a reference to the inner keys-account state.
     pub fn keys(&self) -> &ManagedCoreKeysAccount {
         &self.keys
@@ -208,7 +238,11 @@ impl ManagedCoreFundsAccount {
                     }
                 }
 
-                // Remove UTXOs spent by this transaction and track spent outpoints
+                // Remove UTXOs spent by this transaction and track spent outpoints.
+                // Processing the spend also hands the inputs off from the
+                // ephemeral build reservation to the durable spent set, so the
+                // reservation taken when this transaction was built is released.
+                self.reservations.release(tx.input.iter().map(|input| &input.previous_output));
                 for input in &tx.input {
                     self.spent_outpoints.insert(input.previous_output);
 
@@ -726,6 +760,7 @@ impl<'de> Deserialize<'de> for ManagedCoreFundsAccount {
             balance: helper.balance,
             utxos: helper.utxos,
             spent_outpoints,
+            reservations: ReservationSet::default(),
         })
     }
 }

key-wallet/src/managed_account/mod.rs

@@ -21,6 +21,7 @@ pub mod managed_core_funds_account;
 pub mod managed_core_keys_account;
 pub mod managed_platform_account;
 pub mod platform_address;
+pub(crate) mod reservation;
 pub mod transaction_record;
 
 pub use managed_account_ref::{ManagedAccountRef, ManagedAccountRefMut, OwnedManagedCoreAccount};

key-wallet/src/managed_account/reservation.rs

@@ -0,0 +1,178 @@
+//! Ephemeral reservation of UTXOs selected by in-flight transaction builds.
+//!
+//! Coin selection reads the UTXO set, but a freshly built transaction is not
+//! reflected in that set until the broadcast transaction is processed back into
+//! the wallet (its inputs leave `utxos`). In the window between selecting inputs
+//! and that processing, a second build sees the same UTXOs and can pick them
+//! again, producing a double-spend the network rejects.
+//!
+//! A [`ReservationSet`] bridges exactly that window: selected inputs are
+//! reserved at build time and skipped by subsequent coin selection. The set is
+//! shared via `Arc<Mutex<_>>` so a reservation taken while the wallet lock is
+//! held survives the lock being released and is observed by the next build. It
+//! is never persisted: after a restart it is empty, where chain and mempool
+//! sync are the source of truth for which coins are already spent.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::{Arc, Mutex, MutexGuard};
+
+use dashcore::blockdata::transaction::OutPoint;
+
+/// Number of blocks after which a stale reservation is reclaimed, about one
+/// hour at the mainnet 2.5-minute block target.
+///
+/// A reservation is normally released the moment its broadcast transaction is
+/// processed (the input leaves the UTXO set), and a restart drops all
+/// reservations, so this backstop only matters for a reservation whose
+/// transaction was built but never processed in a long-running process, e.g. a
+/// silently failed broadcast. The value is kept comfortably above the real
+/// selection-to-processing latency (seconds): reclaiming a still-in-flight
+/// reservation too early would let coin selection re-pick the input and rebuild
+/// the very double-spend this guards against.
+const RESERVATION_TTL_BLOCKS: u32 = 24;
+
+/// Ephemeral, in-memory set of reserved outpoints. Cloning shares the
+/// underlying state, which is what lets a build's reservation outlive the
+/// wallet lock and be seen by the next build.
+#[derive(Debug, Clone, Default)]
+pub(crate) struct ReservationSet {
+    inner: Arc<Mutex<HashMap<OutPoint, u32>>>,
+}
+
+impl ReservationSet {
+    /// Recovers from a poisoned mutex rather than panicking: the guarded data is
+    /// a plain `HashMap` with no invariant a partial write could break, and
+    /// panicking here would strand all later coin selection in a long-running
+    /// node.
+    fn lock(&self) -> MutexGuard<'_, HashMap<OutPoint, u32>> {
+        self.inner.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    fn sweep(reserved: &mut HashMap<OutPoint, u32>, current_height: u32) {
+        // Height 0 means the wallet has no processed height yet, so the elapsed
+        // span is unknown and no entry can be reliably judged stale. An entry
+        // stamped at height 0 therefore relies on a later non-zero height being
+        // presented before it can ever be reclaimed by the TTL backstop.
+        if current_height == 0 {
+            return;
+        }
+        reserved.retain(|_, reserved_at| {
+            current_height.saturating_sub(*reserved_at) < RESERVATION_TTL_BLOCKS
+        });
+    }
+
+    /// Reserve `outpoints` as of `current_height`, dropping expired entries
+    /// first. Re-reserving an outpoint refreshes its height.
+    pub(crate) fn reserve(&self, outpoints: &[OutPoint], current_height: u32) {
+        let mut reserved = self.lock();
+        Self::sweep(&mut reserved, current_height);
+        for outpoint in outpoints {
+            reserved.insert(*outpoint, current_height);
+        }
+    }
+
+    /// Return the currently reserved outpoints, dropping expired entries first.
+    pub(crate) fn reserved(&self, current_height: u32) -> HashSet<OutPoint> {
+        let mut reserved = self.lock();
+        Self::sweep(&mut reserved, current_height);
+        reserved.keys().copied().collect()
+    }
+
+    /// Release the given outpoints. Idempotent: releasing an outpoint that is
+    /// not reserved does nothing.
+    pub(crate) fn release<'a>(&self, outpoints: impl IntoIterator<Item = &'a OutPoint>) {
+        let mut reserved = self.lock();
+        for outpoint in outpoints {
+            reserved.remove(outpoint);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use dashcore::Txid;
+
+    fn outpoint(byte: u8, vout: u32) -> OutPoint {
+        OutPoint::new(Txid::from([byte; 32]), vout)
+    }
+
+    #[test]
+    fn reserve_then_skip_then_release() {
+        let set = ReservationSet::default();
+        let a = outpoint(0x01, 0);
+        let b = outpoint(0x02, 1);
+
+        set.reserve(&[a], 100);
+        assert!(set.reserved(100).contains(&a));
+        assert!(!set.reserved(100).contains(&b));
+        assert_eq!(set.reserved(100), HashSet::from([a]));
+
+        set.release([&a]);
+        assert!(set.reserved(100).is_empty());
+    }
+
+    #[test]
+    fn release_is_idempotent() {
+        let set = ReservationSet::default();
+        let a = outpoint(0x03, 0);
+        // Releasing an unreserved outpoint is a no-op.
+        set.release([&a]);
+        set.reserve(&[a], 10);
+        set.release([&a]);
+        set.release([&a]);
+        assert!(!set.reserved(10).contains(&a));
+    }
+
+    #[test]
+    fn ttl_reclaims_stale_reservation_by_height() {
+        let set = ReservationSet::default();
+        let a = outpoint(0x04, 0);
+        set.reserve(&[a], 100);
+
+        // The boundary is exclusive, so the entry survives until exactly
+        // `reserved_at + RESERVATION_TTL_BLOCKS`.
+        assert_eq!(set.reserved(100 + RESERVATION_TTL_BLOCKS - 1), HashSet::from([a]));
+        // At that height the sweep's strict less-than drops the entry.
+        assert!(set.reserved(100 + RESERVATION_TTL_BLOCKS).is_empty());
+    }
+
+    #[test]
+    fn zero_height_disables_sweep() {
+        let set = ReservationSet::default();
+        let a = outpoint(0x07, 0);
+        set.reserve(&[a], 0);
+
+        // Height 0 means the wallet has no processed height yet, so the elapsed
+        // span is unknown and the sweep is suppressed: the reservation stands.
+        assert!(set.reserved(0).contains(&a));
+
+        // Once a real height is known the TTL backstop applies again.
+        set.reserve(&[a], 1);
+        assert!(set.reserved(1 + RESERVATION_TTL_BLOCKS).is_empty());
+    }
+
+    #[test]
+    fn re_reserving_refreshes_the_ttl_height() {
+        let set = ReservationSet::default();
+        let a = outpoint(0x06, 0);
+        set.reserve(&[a], 100);
+        // A later reserve replaces the stored height, so the TTL is measured
+        // from the most recent reservation, not the first.
+        set.reserve(&[a], 150);
+
+        assert_eq!(set.reserved(150 + RESERVATION_TTL_BLOCKS - 1), HashSet::from([a]));
+        assert!(set.reserved(150 + RESERVATION_TTL_BLOCKS).is_empty());
+    }
+
+    #[test]
+    fn clone_shares_state() {
+        let set = ReservationSet::default();
+        let clone = set.clone();
+        let a = outpoint(0x05, 0);
+        // A reservation taken on one handle is visible through the other, which
+        // is what lets a build's reservation outlive the wallet lock.
+        set.reserve(&[a], 1);
+        assert!(clone.reserved(1).contains(&a));
+    }
+}