fix(wallet): keep mixed CoinJoin inputs locked until the finalized tx is observed

doc/release-notes-7480.md

@@ -0,0 +1,28 @@
+Updated RPCs
+------------
+
+* `getcoinjoininfo` gained a `pending_inputs` field, reporting how many
+  successfully mixed inputs are currently kept locked while waiting for the
+  finalized mixing transaction spending them to be observed.
+
+CoinJoin
+--------
+
+* After a mixing session completes successfully, the inputs a client
+  contributed to it are no longer released as soon as the `DSCOMPLETE` message
+  is processed. They stay locked until the wallet observes a transaction
+  spending them, or for at most an hour if that transaction never propagates.
+  Previously the completion message routinely overtook the trickle-relayed
+  mixing transaction, leaving a window in which another session (with
+  `-coinjoinmultisession=1`, where the one-block cooldown does not apply) could
+  select and sign the very same inputs and produce a second, conflicting
+  CoinJoin transaction.
+
+  These locks are persisted, so they survive a restart. They are tracked in a
+  new wallet database record, `cj_pending_obs`, which is written only when a
+  mixing session completes successfully. Older clients loading such a wallet
+  ignore the record, counting it as an unknown record; the accompanying
+  `lockedutxo` entries are understood by them regardless, so the inputs stay
+  locked either way. Locks the user set themselves via `lockunspent` are never
+  adopted or released by this mechanism, and unlocking a pending input manually
+  purges it from the pending set.

src/coinjoin/client.cpp

@@ -14,10 +14,12 @@
 
 #include <chain.h>
 #include <chainparams.h>
+#include <coins.h>
 #include <core_io.h>
 #include <net.h>
 #include <netmessagemaker.h>
 #include <shutdown.h>
+#include <txmempool.h>
 #include <util/check.h>
 #include <util/fs_helpers.h>
 #include <util/moneystr.h>
@@ -28,8 +30,10 @@
 #include <wallet/coinselection.h>
 #include <wallet/receive.h>
 #include <wallet/spend.h>
+#include <wallet/walletdb.h>
 
 #include <memory>
+#include <optional>
 #include <ranges>
 
 #include <univalue.h>
@@ -577,6 +581,33 @@ void CCoinJoinClientSession::CompletedTransaction(PoolMessage nMessageID)
     if (nMessageID == MSG_SUCCESS) {
         m_clientman.UpdatedSuccessBlock();
         keyHolderStorage.KeepAll();
+        // Our inputs are now committed to the finalized mixing transaction but the
+        // DSCOMPLETE message may arrive well before the (trickle-relayed) DSTX does.
+        // Do not release those inputs yet: keep them locked until the wallet actually
+        // observes a transaction spending them, otherwise another session could pick
+        // them up in the meantime (esp. with -coinjoinmultisession) and double-spend
+        // them. Anything else we locked (e.g. collateral inputs which may never be
+        // spent) is unlocked immediately as before.
+        std::set<COutPoint> setMixingInputs;
+        {
+            LOCK(cs_coinjoin);
+            for (const auto& entry : vecEntries) {
+                for (const auto& txdsin : entry.vecTxDSIn) {
+                    setMixingInputs.emplace(txdsin.prevout);
+                }
+            }
+        }
+        std::vector<COutPoint> vecPending;
+        std::vector<COutPoint> vecUnlockNow;
+        for (const auto& outpoint : vecOutPointLocked) {
+            if (setMixingInputs.count(outpoint)) {
+                vecPending.push_back(outpoint);
+            } else {
+                vecUnlockNow.push_back(outpoint);
+            }
+        }
+        vecOutPointLocked = std::move(vecUnlockNow);
+        m_clientman.AddPendingObservation(vecPending);
         WalletCJLogPrint(m_wallet, "CompletedTransaction -- success\n");
     } else {
         WITH_LOCK(m_wallet->cs_wallet, keyHolderStorage.ReturnAll());
@@ -592,6 +623,165 @@ void CCoinJoinClientManager::UpdatedSuccessBlock()
     nCachedLastSuccessBlock = nCachedBlockHeight;
 }
 
+void CCoinJoinClientManager::AddPendingObservation(const std::vector<COutPoint>& outpoints)
+{
+    AssertLockNotHeld(cs_pending_obs);
+    if (outpoints.empty()) return;
+
+    LOCK(m_wallet->cs_wallet);
+    LOCK(cs_pending_obs);
+    wallet::WalletBatch batch(m_wallet->GetDatabase());
+    LoadPendingObservations(batch);
+
+    const int64_t nNow{GetTime()};
+    for (const auto& outpoint : outpoints) {
+        m_pending_obs.emplace(outpoint, nNow);
+    }
+
+    // NOTE: the record which owns these locks has to be written BEFORE the locks
+    // themselves. These are two separate writes and a crash in between must not leave
+    // persistently locked coins behind with nothing tracking them: nothing would ever
+    // release them again. The opposite order is self-healing, CheckPendingObservations()
+    // drops a pending entry as soon as it sees its coin is not locked.
+    bool fPersisted{batch.WriteCoinJoinPendingObs(m_pending_obs)};
+    for (const auto& outpoint : outpoints) {
+        // The coins are locked in memory already (see PrepareDenominate), this only
+        // persists the lock so that a restart before the finalized transaction is
+        // observed cannot make the input available for selection again. Skip persisting
+        // it if the record above could not be written though, a persistent lock with
+        // nothing left to release it strands the input.
+        if (!m_wallet->LockCoin(outpoint, fPersisted ? &batch : nullptr)) fPersisted = false;
+        WalletCJLogPrint(m_wallet, "CCoinJoinClientManager::%s -- %s is locked until the finalized mixing transaction is observed\n",
+                         __func__, outpoint.ToStringShort());
+    }
+    if (!fPersisted) {
+        // The in-memory lock still protects these inputs for as long as this process
+        // runs, but a restart would make them selectable again while a valid mixing
+        // transaction spending them may already be in flight. Nothing we can do about
+        // it here beyond making the failure loud - the wallet database is broken.
+        LogPrintf("CCoinJoinClientManager::%s -- ERROR: failed to persist locks for %d successfully mixed input(s), " /* Continued */
+                  "they will not survive a restart\n",
+                  __func__, outpoints.size());
+    }
+}
+
+void CCoinJoinClientManager::LoadPendingObservations(wallet::WalletBatch& batch)
+{
+    AssertLockHeld(cs_pending_obs);
+
+    if (m_pending_obs_loaded) return;
+    m_pending_obs_loaded = true;
+    // Missing record simply means nothing was pending when we last shut down
+    if (!batch.ReadCoinJoinPendingObs(m_pending_obs)) return;
+    WalletCJLogPrint(m_wallet, "CCoinJoinClientManager::%s -- loaded %d pending observation(s)\n", __func__,
+                     m_pending_obs.size());
+}
+
+void CCoinJoinClientManager::CheckPendingObservations(const CTxMemPool& mempool)
+{
+    AssertLockNotHeld(cs_pending_obs);
+
+    // nothing to do in the common case, don't touch the wallet at all
+    if (WITH_LOCK(cs_pending_obs, return m_pending_obs_loaded && m_pending_obs.empty())) return;
+
+    LOCK(m_wallet->cs_wallet);
+    LOCK(cs_pending_obs);
+
+    if (!m_pending_obs_loaded) {
+        // Read-only, no need to checkpoint the database on the way out
+        wallet::WalletBatch batch_load(m_wallet->GetDatabase(), /*_fFlushOnClose=*/false);
+        LoadPendingObservations(batch_load);
+    }
+
+    // Constructed on the first write only: this keeps running for as long as anything is
+    // pending and every WalletBatch checkpoints the wallet database when it goes out of
+    // scope, which is far too expensive to pay for a pass which changes nothing.
+    std::optional<wallet::WalletBatch> batch;
+    const auto get_batch = [&]() -> wallet::WalletBatch& {
+        if (!batch.has_value()) batch.emplace(m_wallet->GetDatabase());
+        return batch.value();
+    };
+
+    const int64_t nNow{GetTime()};
+    const bool fSynced{m_mn_sync.IsBlockchainSynced()};
+    bool fChanged{false};
+    for (auto it = m_pending_obs.begin(); it != m_pending_obs.end();) {
+        const COutPoint& outpoint = it->first;
+        if (!m_wallet->IsLockedCoin(outpoint)) {
+            // The user released the lock manually (e.g. via lockunspent), respect that
+            it = m_pending_obs.erase(it);
+            fChanged = true;
+            continue;
+        }
+        if (m_wallet->IsSpent(outpoint)) {
+            // The wallet observed a transaction spending this input, it is no longer
+            // selectable anyway, drop the protective lock
+            WalletCJLogPrint(m_wallet, "CCoinJoinClientManager::%s -- observed spend of %s, releasing lock\n", __func__,
+                             outpoint.ToStringShort());
+            m_wallet->UnlockCoin(outpoint, &get_batch());
+            it = m_pending_obs.erase(it);
+            fChanged = true;
+            continue;
+        }
+        // Only ever consult chain/mempool spentness on a synced chain: while catching up
+        // the spending transaction may sit in a block we have not downloaded yet, and
+        // releasing the input on the strength of that would defeat the whole point
+        if (fSynced && nNow - it->second >= COINJOIN_PENDING_OBSERVATION_TIMEOUT) {
+            // NOTE: findCoins() reports outputs which exist in the chain UTXO set or are
+            // created by a mempool transaction; it does NOT know whether a mempool
+            // transaction spends them (CCoinsViewMemPool::GetCoin never looks at
+            // mapNextTx). The finalized mixing transaction sitting unprocessed in our own
+            // mempool is exactly the case we must not misread as "never propagated", so
+            // ask the mempool about spentness explicitly as well.
+            std::map<COutPoint, Coin> coins{{outpoint, Coin{}}};
+            m_wallet->chain().findCoins(coins);
+            if (!coins.at(outpoint).IsSpent() && !mempool.isSpent(outpoint)) {
+                // Still unspent in chain and mempool long after the session completed -
+                // the finalized transaction most likely never propagated, release the
+                // input so the wallet does not lose it forever
+                LogPrintf("CCoinJoinClientManager::%s -- WARNING: never observed finalized mixing transaction for %s, " /* Continued */
+                          "releasing lock after %d seconds\n",
+                          __func__, outpoint.ToStringShort(), COINJOIN_PENDING_OBSERVATION_TIMEOUT);
+                m_wallet->UnlockCoin(outpoint, &get_batch());
+                it = m_pending_obs.erase(it);
+                fChanged = true;
+                continue;
+            }
+            // Spent according to chain/mempool but the wallet has not recorded the
+            // spending transaction yet, keep waiting for it. Note that this can be
+            // terminal: if the wallet never learns about the spend (restored from an old
+            // backup and never rescanned, say) the entry stays for good. Releasing it
+            // would be wrong - such a coin still looks unspent to the wallet and could be
+            // selected for another session - so only refresh the timer to keep the check
+            // above from running on every pass. The refresh is deliberately not persisted,
+            // the worst a restart can do is re-run the check once.
+            WalletCJLogPrint(m_wallet, "CCoinJoinClientManager::%s -- %s is spent in chain/mempool but not according to " /* Continued */
+                                       "the wallet, keeping it locked\n",
+                             __func__, outpoint.ToStringShort());
+            it->second = nNow;
+        }
+        ++it;
+    }
+    if (fChanged && !get_batch().WriteCoinJoinPendingObs(m_pending_obs)) {
+        LogPrintf("CCoinJoinClientManager::%s -- ERROR: failed to persist %d pending observation(s)\n", __func__,
+                  m_pending_obs.size());
+    }
+}
+
+bool CCoinJoinClientManager::IsPendingObservation(const COutPoint& outpoint) const
+{
+    AssertLockNotHeld(cs_pending_obs);
+    LOCK(cs_pending_obs);
+    return m_pending_obs.count(outpoint) > 0;
+}
+
+size_t CCoinJoinClientManager::GetPendingObservationCount() const
+{
+    AssertLockNotHeld(cs_pending_obs);
+    LOCK(cs_pending_obs);
+    return m_pending_obs.size();
+}
+
 bool CCoinJoinClientManager::WaitForAnotherBlock() const
 {
     if (!m_mn_sync.IsBlockchainSynced()) return true;
@@ -1751,6 +1941,7 @@ UniValue CCoinJoinClientManager::getJsonInfo() const
 {
     UniValue obj(UniValue::VOBJ);
     obj.pushKV("running", isMixing());
+    obj.pushKV("pending_inputs", static_cast<int64_t>(GetPendingObservationCount()));
 
     UniValue arrSessions(UniValue::VARR);
     AssertLockNotHeld(cs_deqsessions);

src/coinjoin/client.h

@@ -12,10 +12,15 @@
 #include <util/translation.h>
 
 #include <deque>
+#include <map>
 #include <memory>
 #include <ranges>
 #include <utility>
 
+namespace wallet {
+class WalletBatch;
+} // namespace wallet
+
 class CCoinJoinClientManager;
 class CConnman;
 class CDeterministicMNManager;
@@ -174,6 +179,21 @@ class CCoinJoinClientManager : public interfaces::CoinJoin::Client
     // how many blocks to wait for after one successful mixing tx in non-multisession mode
     static constexpr int nMinBlocksToWait{1};
 
+    mutable Mutex cs_pending_obs;
+    //! Inputs of successfully completed mixing sessions which must stay locked until
+    //! the wallet observes the finalized mixing transaction spending them. Without this
+    //! a client which processes DSCOMPLETE(MSG_SUCCESS) before the (trickle-relayed)
+    //! finalized DSTX arrives could select the very same inputs for another session
+    //! (esp. with -coinjoinmultisession) and double-spend them.
+    //! Maps outpoint to the time it was added. Mirrored to the wallet database so that
+    //! both the locks and their grace period survive a restart, and so that these are
+    //! never confused with locks the user set themselves via `lockunspent`.
+    std::map<COutPoint, int64_t> m_pending_obs GUARDED_BY(cs_pending_obs);
+    bool m_pending_obs_loaded GUARDED_BY(cs_pending_obs){false};
+
+    /// Populate m_pending_obs from the wallet database, once per run
+    void LoadPendingObservations(wallet::WalletBatch& batch) EXCLUSIVE_LOCKS_REQUIRED(cs_pending_obs);
+
     // Keep track of current block height
     int nCachedBlockHeight{0};
 
@@ -211,6 +231,14 @@ class CCoinJoinClientManager : public interfaces::CoinJoin::Client
 
     void UpdatedSuccessBlock();
 
+    /// Keep the given successfully mixed inputs locked (persistently) until the wallet
+    /// observes a transaction spending them
+    void AddPendingObservation(const std::vector<COutPoint>& outpoints) EXCLUSIVE_LOCKS_REQUIRED(!cs_pending_obs);
+    /// Release pending inputs whose spend has been observed by the wallet
+    void CheckPendingObservations(const CTxMemPool& mempool) EXCLUSIVE_LOCKS_REQUIRED(!cs_pending_obs);
+    bool IsPendingObservation(const COutPoint& outpoint) const EXCLUSIVE_LOCKS_REQUIRED(!cs_pending_obs);
+    size_t GetPendingObservationCount() const EXCLUSIVE_LOCKS_REQUIRED(!cs_pending_obs);
+
     void UpdatedBlockTip(const CBlockIndex* pindex);
 
     void DoMaintenance(ChainstateManager& chainman, CConnman& connman, const CTxMemPool& mempool)
@@ -219,7 +247,7 @@ class CCoinJoinClientManager : public interfaces::CoinJoin::Client
     // interfaces::CoinJoin::Client overrides
     void disableAutobackups() override { fCreateAutoBackups = false; }
     void resetPool() override EXCLUSIVE_LOCKS_REQUIRED(!cs_deqsessions);
-    UniValue getJsonInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!cs_deqsessions);
+    UniValue getJsonInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!cs_deqsessions, !cs_pending_obs);
     std::vector<std::string> getSessionStatuses() const override EXCLUSIVE_LOCKS_REQUIRED(!cs_deqsessions);
     std::string getSessionDenoms() const override EXCLUSIVE_LOCKS_REQUIRED(!cs_deqsessions);
     bool isMixing() const override;

src/coinjoin/coinjoin.h

@@ -47,6 +47,10 @@ static constexpr int COINJOIN_AUTO_TIMEOUT_MIN = 5;
 static constexpr int COINJOIN_AUTO_TIMEOUT_MAX = 15;
 static constexpr int COINJOIN_QUEUE_TIMEOUT = 30;
 static constexpr int COINJOIN_SIGNING_TIMEOUT = 15;
+//! How long a successfully mixed input is kept locked while waiting for the finalized
+//! mixing transaction before chain/mempool spentness is double-checked and the input is
+//! potentially released, in seconds
+static constexpr int64_t COINJOIN_PENDING_OBSERVATION_TIMEOUT = 60 * 60;
 
 static constexpr size_t COINJOIN_ENTRY_MAX_SIZE = 9;
 

src/coinjoin/walletman.cpp

@@ -73,6 +73,7 @@ class CJWalletManagerImpl final : public CJWalletManager
     std::map<const std::string, std::unique_ptr<CCoinJoinClientManager>> m_wallet_manager_map GUARDED_BY(cs_wallet_manager_map);
 
     void DoMaintenance(CConnman& connman) EXCLUSIVE_LOCKS_REQUIRED(!cs_wallet_manager_map);
+    void CheckPendingObservations() EXCLUSIVE_LOCKS_REQUIRED(!cs_wallet_manager_map);
 
     [[nodiscard]] MessageProcessingResult ProcessDSQueue(NodeId from, CConnman& connman, std::string_view msg_type,
                                                          CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_ProcessDSQueue, !cs_wallet_manager_map);
@@ -119,7 +120,19 @@ CJWalletManagerImpl::~CJWalletManagerImpl()
 
 void CJWalletManagerImpl::Schedule(CConnman& connman, CScheduler& scheduler)
 {
+    // Inputs of a successfully completed session stay locked until the finalized mixing
+    // transaction is observed and those locks are persisted, so they are restored with
+    // the wallet. Releasing them has to keep running independently of mixing itself:
+    // even with CoinJoin disabled, mixing stopped or transaction relay unavailable
+    // (block-only mode, where nothing below is scheduled at all) nothing else would ever
+    // unlock them. NOTE: no CJWalletManager exists on a masternode (see init.cpp), a
+    // wallet with pending observations opened there keeps its inputs locked until it is
+    // opened on a regular node again or the user unlocks them via `lockunspent`.
+    scheduler.scheduleEvery(std::bind(&CJWalletManagerImpl::CheckPendingObservations, this), std::chrono::minutes{1});
+
+    // Mixing needs transaction relay
     if (!m_relay_txes) return;
+
     scheduler.scheduleEvery(std::bind(&CJWalletManagerImpl::DoMaintenance, this, std::ref(connman)),
                             std::chrono::seconds{1});
 }
@@ -205,6 +218,15 @@ void CJWalletManagerImpl::DoMaintenance(CConnman& connman)
     }
 }
 
+void CJWalletManagerImpl::CheckPendingObservations()
+{
+    if (ShutdownRequested()) return;
+    LOCK(cs_wallet_manager_map);
+    for (auto& [_, clientman] : m_wallet_manager_map) {
+        clientman->CheckPendingObservations(m_mempool);
+    }
+}
+
 MessageProcessingResult CJWalletManagerImpl::processMessage(CNode& pfrom, Chainstate& chainstate, CConnman& connman,
                                                             CTxMemPool& mempool, std::string_view msg_type,
                                                             CDataStream& vRecv)

src/rpc/coinjoin.cpp

@@ -429,6 +429,7 @@ static RPCHelpMan getcoinjoininfo()
                             {RPCResult::Type::NUM, "denoms_hardcap", "Maximum limit of how many inputs of each denominated amount to create"},
                             {RPCResult::Type::NUM, "queue_size", "How many queues there are currently on the network"},
                             {RPCResult::Type::BOOL, "running", "Whether mixing is currently running"},
+                            {RPCResult::Type::NUM, "pending_inputs", "The number of successfully mixed inputs kept locked until the transaction spending them is observed"},
                             {RPCResult::Type::ARR, "sessions", "",
                             {
                                 {RPCResult::Type::OBJ, "", "",