Merge #6192: fix: protect CoinJoinWalletManager::m_wallet_manager_map with Mutex, avoid data race, partial bitcoin#22868

87d775d27c partial bitcoin#22868: Call load handlers without cs_wallet locked (Kittywhiskers Van Gogh)
c602ca15e1 coinjoin: protect `m_wallet_manager_map` with `cs_wallet_manager_map` (Kittywhiskers Van Gogh)

Pull request description:

  ## Additional Information

  This pull request aims to deal with regressions ([build](https://gitlab.com/dashpay/dash/-/jobs/7550859495)) spotted in `develop` after the merger of [dash#6143](https://github.com/dashpay/dash/pull/6143), namely, a failure to build the multiprocess variant and two data races, one involving `ChainstateManager::m_best_header` and another involving `CoinJoinWalletManager::m_wallet_manager_map`.

  * Fixes for the build failure and the first data race (b136742f98c6792735232cc170123f849fd13bad and 9e7c6850139e69cb64cfe89813e6f0e4b1e626e1), have been spun off into [dash#6199](https://github.com/dashpay/dash/pull/6199) and merged

  * The second data race is being avoided by protected `m_wallet_manager_map` with a new `RecursiveMutex`, `cs_wallet_manager_map` (the contents of this PR). ~~A version of these changes are available using a regular `Mutex` but prove far more cumbersome ([source](b89457dc9a)) when taking practicalities into account ([comment](https://github.com/dashpay/dash/pull/6192#discussion_r1713207664)).~~ A `Mutex` version is now available courtesy of UdjinM6 (see [patch](ecf5c1ba76)), thanks!

  ## Checklist:

  - [x] I have performed a self-review of my own code
  - [x] I have commented my code, particularly in hard-to-understand areas **(note: N/A)**
  - [x] I have added or updated relevant unit/integration/functional/e2e tests
  - [x] I have made corresponding changes to the documentation **(note: N/A)**
  - [x] I have assigned this pull request to a milestone _(for repository code-owners and collaborators only)_

ACKs for top commit:
  UdjinM6:
    light-ACK 87d775d27c

Tree-SHA512: 52545a1547357a45fb6bf4d3948fc75a5e88f1d86e9809b1060007cd74dd383249691d8d9777f647c7534d458fe126ae818fa2b47f0e5b0cee78a469af1ed0c9

src/coinjoin/client.cpp

@@ -96,11 +96,9 @@ PeerMsgRet CCoinJoinClientQueueManager::ProcessDSQueue(const CNode& peer, CDataS
         }
 
         // if the queue is ready, submit if we can
-        if (dsq.fReady && ranges::any_of(m_walletman.raw(),
+        if (dsq.fReady && m_walletman.ForAnyCJClientMan([this, &dmn](std::unique_ptr<CCoinJoinClientManager>& clientman) {
-                                         [this, &dmn](const auto &pair) {
+                return clientman->TrySubmitDenominate(dmn->pdmnState->addr, this->connman);
-                                             return pair.second->TrySubmitDenominate(dmn->pdmnState->addr,
+            })) {
-                                                                                     this->connman);
-                                         })) {
             LogPrint(BCLog::COINJOIN, "DSQUEUE -- CoinJoin queue (%s) is ready on masternode %s\n", dsq.ToString(),
                      dmn->pdmnState->addr.ToString());
             return {};
@@ -121,8 +119,9 @@ PeerMsgRet CCoinJoinClientQueueManager::ProcessDSQueue(const CNode& peer, CDataS
             LogPrint(BCLog::COINJOIN, "DSQUEUE -- new CoinJoin queue (%s) from masternode %s\n", dsq.ToString(),
                      dmn->pdmnState->addr.ToString());
 
-            ranges::any_of(m_walletman.raw(),
+            m_walletman.ForAnyCJClientMan([&dsq](const std::unique_ptr<CCoinJoinClientManager>& clientman) {
-                           [&dsq](const auto &pair) { return pair.second->MarkAlreadyJoinedQueueAsTried(dsq); });
+                return clientman->MarkAlreadyJoinedQueueAsTried(dsq);
+            });
 
             WITH_LOCK(cs_vecqueue, vecCoinJoinQueue.push_back(dsq));
         }
@@ -155,11 +154,14 @@ void CCoinJoinClientManager::ProcessMessage(CNode& peer, CChainState& active_cha
     }
 }
 
-CCoinJoinClientSession::CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman, CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman,
+CCoinJoinClientSession::CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman,
-                                               const CMasternodeSync& mn_sync, const std::unique_ptr<CCoinJoinClientQueueManager>& queueman, bool is_masternode) :
+                                               CCoinJoinClientManager& clientman, CDeterministicMNManager& dmnman,
+                                               CMasternodeMetaMan& mn_metaman, const CMasternodeSync& mn_sync,
+                                               const std::unique_ptr<CCoinJoinClientQueueManager>& queueman,
+                                               bool is_masternode) :
     m_wallet(wallet),
     m_walletman(walletman),
-    m_manager(*Assert(walletman.Get(wallet.GetName()))),
+    m_clientman(clientman),
     m_dmnman(dmnman),
     m_mn_metaman(mn_metaman),
     m_mn_sync(mn_sync),
@@ -684,7 +686,7 @@ void CCoinJoinClientSession::CompletedTransaction(PoolMessage nMessageID)
     if (m_is_masternode) return;
 
     if (nMessageID == MSG_SUCCESS) {
-        m_manager.UpdatedSuccessBlock();
+        m_clientman.UpdatedSuccessBlock();
         keyHolderStorage.KeepAll();
         WalletCJLogPrint(m_wallet, "CompletedTransaction -- success\n");
     } else {
@@ -995,7 +997,8 @@ bool CCoinJoinClientManager::DoAutomaticDenominating(CChainState& active_chainst
     AssertLockNotHeld(cs_deqsessions);
     LOCK(cs_deqsessions);
     if (int(deqSessions.size()) < CCoinJoinClientOptions::GetSessions()) {
-        deqSessions.emplace_back(m_wallet, m_walletman, m_dmnman, m_mn_metaman, m_mn_sync, m_queueman, m_is_masternode);
+        deqSessions.emplace_back(m_wallet, m_walletman, *this, m_dmnman, m_mn_metaman, m_mn_sync, m_queueman,
+                                 m_is_masternode);
     }
     for (auto& session : deqSessions) {
         if (!CheckAutomaticBackup()) return false;
@@ -1100,7 +1103,7 @@ bool CCoinJoinClientSession::JoinExistingQueue(CAmount nBalanceNeedsAnonymized,
             continue;
         }
 
-        m_manager.AddUsedMasternode(dsq.masternodeOutpoint);
+        m_clientman.AddUsedMasternode(dsq.masternodeOutpoint);
 
         if (connman.IsMasternodeOrDisconnectRequested(dmn->pdmnState->addr)) {
             WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::JoinExistingQueue -- skipping masternode connection, addr=%s\n", dmn->pdmnState->addr.ToString());
@@ -1145,14 +1148,14 @@ bool CCoinJoinClientSession::StartNewQueue(CAmount nBalanceNeedsAnonymized, CCon
 
     // otherwise, try one randomly
     while (nTries < 10) {
-        auto dmn = m_manager.GetRandomNotUsedMasternode();
+        auto dmn = m_clientman.GetRandomNotUsedMasternode();
         if (!dmn) {
             strAutoDenomResult = _("Can't find random Masternode.");
             WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::StartNewQueue -- %s\n", strAutoDenomResult.original);
             return false;
         }
 
-        m_manager.AddUsedMasternode(dmn->collateralOutpoint);
+        m_clientman.AddUsedMasternode(dmn->collateralOutpoint);
 
         // skip next mn payments winners
         if (dmn->pdmnState->nLastPaidHeight + nWeightedMnCount < mnList.GetHeight() + WinnersToSkip()) {
@@ -1526,7 +1529,7 @@ bool CCoinJoinClientSession::MakeCollateralAmounts(const CompactTallyItem& tally
         return false;
     }
 
-    m_manager.UpdatedSuccessBlock();
+    m_clientman.UpdatedSuccessBlock();
 
     WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::%s -- txid: %s\n", __func__, strResult.original);
 
@@ -1803,7 +1806,7 @@ bool CCoinJoinClientSession::CreateDenominated(CAmount nBalanceToDenominate, con
     }
 
     // use the same nCachedLastSuccessBlock as for DS mixing to prevent race
-    m_manager.UpdatedSuccessBlock();
+    m_clientman.UpdatedSuccessBlock();
 
     WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::%s -- txid: %s\n", __func__, strResult.original);
 
@@ -1894,35 +1897,43 @@ void CCoinJoinClientManager::GetJsonInfo(UniValue& obj) const
     obj.pushKV("sessions",  arrSessions);
 }
 
-void CoinJoinWalletManager::Add(CWallet& wallet) {
+void CoinJoinWalletManager::Add(CWallet& wallet)
-    m_wallet_manager_map.try_emplace(
+{
-        wallet.GetName(),
+    {
-        std::make_unique<CCoinJoinClientManager>(wallet, *this, m_dmnman, m_mn_metaman, m_mn_sync, m_queueman, m_is_masternode)
+        LOCK(cs_wallet_manager_map);
-    );
+        m_wallet_manager_map.try_emplace(wallet.GetName(),
+                                         std::make_unique<CCoinJoinClientManager>(wallet, *this, m_dmnman, m_mn_metaman,
+                                                                                  m_mn_sync, m_queueman, m_is_masternode));
+    }
     g_wallet_init_interface.InitCoinJoinSettings(*this);
 }
 
-void CoinJoinWalletManager::DoMaintenance() {
+void CoinJoinWalletManager::DoMaintenance()
-    for (auto& [wallet_str, walletman] : m_wallet_manager_map) {
+{
-        walletman->DoMaintenance(m_chainstate, m_connman, m_mempool);
+    LOCK(cs_wallet_manager_map);
+    for (auto& [_, clientman] : m_wallet_manager_map) {
+        clientman->DoMaintenance(m_chainstate, m_connman, m_mempool);
     }
 }
 
 void CoinJoinWalletManager::Remove(const std::string& name) {
-    m_wallet_manager_map.erase(name);
+    {
+        LOCK(cs_wallet_manager_map);
+        m_wallet_manager_map.erase(name);
+    }
     g_wallet_init_interface.InitCoinJoinSettings(*this);
 }
 
 void CoinJoinWalletManager::Flush(const std::string& name)
 {
-    auto clientman = Get(name);
+    auto clientman = Assert(Get(name));
-    assert(clientman != nullptr);
     clientman->ResetPool();
     clientman->StopMixing();
 }
 
 CCoinJoinClientManager* CoinJoinWalletManager::Get(const std::string& name) const
 {
+    LOCK(cs_wallet_manager_map);
     auto it = m_wallet_manager_map.find(name);
     return (it != m_wallet_manager_map.end()) ? it->second.get() : nullptr;
 }

src/coinjoin/client.h

@@ -9,6 +9,7 @@
 #include <coinjoin/coinjoin.h>
 
 #include <net_types.h>
+#include <util/ranges.h>
 #include <util/translation.h>
 
 #include <atomic>
@@ -80,6 +81,7 @@ public:
     {}
 
     ~CoinJoinWalletManager() {
+        LOCK(cs_wallet_manager_map);
         for (auto& [wallet_name, cj_man] : m_wallet_manager_map) {
             cj_man.reset();
         }
@@ -93,7 +95,21 @@ public:
 
     CCoinJoinClientManager* Get(const std::string& name) const;
 
-    const wallet_name_cjman_map& raw() const { return m_wallet_manager_map; }
+    template <typename Callable>
+    void ForEachCJClientMan(Callable&& func)
+    {
+        LOCK(cs_wallet_manager_map);
+        for (auto&& [_, clientman] : m_wallet_manager_map) {
+            func(clientman);
+        }
+    };
+
+    template <typename Callable>
+    bool ForAnyCJClientMan(Callable&& func)
+    {
+        LOCK(cs_wallet_manager_map);
+        return ranges::any_of(m_wallet_manager_map, [&](auto& pair) { return func(pair.second); });
+    };
 
 private:
     CChainState& m_chainstate;
@@ -105,7 +121,9 @@ private:
     const std::unique_ptr<CCoinJoinClientQueueManager>& m_queueman;
 
     const bool m_is_masternode;
-    wallet_name_cjman_map m_wallet_manager_map;
+
+    mutable Mutex cs_wallet_manager_map;
+    wallet_name_cjman_map m_wallet_manager_map GUARDED_BY(cs_wallet_manager_map);
 };
 
 class CCoinJoinClientSession : public CCoinJoinBaseSession
@@ -113,7 +131,7 @@ class CCoinJoinClientSession : public CCoinJoinBaseSession
 private:
     CWallet& m_wallet;
     CoinJoinWalletManager& m_walletman;
-    CCoinJoinClientManager& m_manager;
+    CCoinJoinClientManager& m_clientman;
     CDeterministicMNManager& m_dmnman;
     CMasternodeMetaMan& m_mn_metaman;
     const CMasternodeSync& m_mn_sync;
@@ -168,8 +186,10 @@ private:
     void SetNull() override EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin);
 
 public:
-    explicit CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman, CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman,
+    explicit CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman,
-                                    const CMasternodeSync& mn_sync, const std::unique_ptr<CCoinJoinClientQueueManager>& queueman, bool is_masternode);
+                                    CCoinJoinClientManager& clientman, CDeterministicMNManager& dmnman,
+                                    CMasternodeMetaMan& mn_metaman, const CMasternodeSync& mn_sync,
+                                    const std::unique_ptr<CCoinJoinClientQueueManager>& queueman, bool is_masternode);
 
     void ProcessMessage(CNode& peer, CChainState& active_chainstate, CConnman& connman, const CTxMemPool& mempool, std::string_view msg_type, CDataStream& vRecv);
 

src/dsnotificationinterface.cpp

@@ -86,9 +86,8 @@ void CDSNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, con
 
     m_cj_ctx->dstxman->UpdatedBlockTip(pindexNew, *m_llmq_ctx->clhandler, m_mn_sync);
 #ifdef ENABLE_WALLET
-    for (auto& pair : m_cj_ctx->walletman->raw()) {
+    m_cj_ctx->walletman->ForEachCJClientMan(
-        pair.second->UpdatedBlockTip(pindexNew);
+        [&pindexNew](std::unique_ptr<CCoinJoinClientManager>& clientman) { clientman->UpdatedBlockTip(pindexNew); });
-    }
 #endif // ENABLE_WALLET
 
     m_llmq_ctx->isman->UpdatedBlockTip(pindexNew);

src/masternode/utils.cpp

@@ -23,9 +23,9 @@ void CMasternodeUtils::DoMaintenance(CConnman& connman, CDeterministicMNManager&
 
     std::vector<CDeterministicMNCPtr> vecDmns; // will be empty when no wallet
 #ifdef ENABLE_WALLET
-    for (auto& pair : cj_ctx.walletman->raw()) {
+    cj_ctx.walletman->ForEachCJClientMan([&vecDmns](const std::unique_ptr<CCoinJoinClientManager>& clientman) {
-        pair.second->GetMixingMasternodesInfo(vecDmns);
+        clientman->GetMixingMasternodesInfo(vecDmns);
-    }
+    });
 #endif // ENABLE_WALLET
 
     // Don't disconnect masternode connections when we have less then the desired amount of outbound nodes

src/net_processing.cpp

@@ -5095,9 +5095,9 @@ void PeerManagerImpl::ProcessMessage(
         //probably one the extensions
 #ifdef ENABLE_WALLET
         ProcessPeerMsgRet(m_cj_ctx->queueman->ProcessMessage(pfrom, msg_type, vRecv), pfrom);
-        for (auto& pair : m_cj_ctx->walletman->raw()) {
+        m_cj_ctx->walletman->ForEachCJClientMan([this, &pfrom, &msg_type, &vRecv](std::unique_ptr<CCoinJoinClientManager>& clientman) {
-            pair.second->ProcessMessage(pfrom, m_chainman.ActiveChainstate(), m_connman, m_mempool, msg_type, vRecv);
+            clientman->ProcessMessage(pfrom, m_chainman.ActiveChainstate(), m_connman, m_mempool, msg_type, vRecv);
-        }
+        });
 #endif // ENABLE_WALLET
         ProcessPeerMsgRet(m_cj_ctx->server->ProcessMessage(pfrom, msg_type, vRecv), pfrom);
         ProcessPeerMsgRet(m_sporkman.ProcessMessage(pfrom, m_connman, *this, msg_type, vRecv), pfrom);

src/wallet/test/coinjoin_tests.cpp

@@ -208,8 +208,8 @@ public:
 
 BOOST_FIXTURE_TEST_CASE(coinjoin_manager_start_stop_tests, CTransactionBuilderTestSetup)
 {
-    BOOST_CHECK_EQUAL(m_node.cj_ctx->walletman->raw().size(), 1);
+    CCoinJoinClientManager* cj_man = m_node.cj_ctx->walletman->Get("");
-    auto& cj_man = m_node.cj_ctx->walletman->raw().begin()->second;
+    BOOST_ASSERT(cj_man != nullptr);
     BOOST_CHECK_EQUAL(cj_man->IsMixing(), false);
     BOOST_CHECK_EQUAL(cj_man->StartMixing(), true);
     BOOST_CHECK_EQUAL(cj_man->IsMixing(), true);

src/wallet/test/wallet_tests.cpp

@@ -1298,18 +1298,14 @@ BOOST_FIXTURE_TEST_CASE(CreateWallet, TestChain100Setup)
     // deadlock during the sync and simulates a new block notification happening
     // as soon as possible.
     addtx_count = 0;
-    auto handler = HandleLoadWallet([&](std::unique_ptr<interfaces::Wallet> wallet) EXCLUSIVE_LOCKS_REQUIRED(wallet->wallet()->cs_wallet, cs_wallets) {
+    auto handler = HandleLoadWallet([&](std::unique_ptr<interfaces::Wallet> wallet) {
             BOOST_CHECK(rescan_completed);
             m_coinbase_txns.push_back(CreateAndProcessBlock({}, GetScriptForRawPubKey(coinbaseKey.GetPubKey())).vtx[0]);
             block_tx = TestSimpleSpend(*m_coinbase_txns[2], 0, coinbaseKey, GetScriptForRawPubKey(key.GetPubKey()));
             m_coinbase_txns.push_back(CreateAndProcessBlock({block_tx}, GetScriptForRawPubKey(coinbaseKey.GetPubKey())).vtx[0]);
             mempool_tx = TestSimpleSpend(*m_coinbase_txns[3], 0, coinbaseKey, GetScriptForRawPubKey(key.GetPubKey()));
             BOOST_CHECK(m_node.chain->broadcastTransaction(MakeTransactionRef(mempool_tx), DEFAULT_TRANSACTION_MAXFEE, false, error));
-            LEAVE_CRITICAL_SECTION(cs_wallets);
-            LEAVE_CRITICAL_SECTION(wallet->wallet()->cs_wallet);
             SyncWithValidationInterfaceQueue();
-            ENTER_CRITICAL_SECTION(wallet->wallet()->cs_wallet);
-            ENTER_CRITICAL_SECTION(cs_wallets);
         });
     wallet = TestLoadWallet(m_node.chain.get(), m_node.coinjoin_loader.get());
     BOOST_CHECK_EQUAL(addtx_count, 4);

src/wallet/wallet.cpp

@@ -4949,8 +4949,6 @@ std::shared_ptr<CWallet> CWallet::Create(interfaces::Chain* chain, interfaces::C
     // Try to top up keypool. No-op if the wallet is locked.
     walletInstance->TopUpKeyPool();
 
-    LOCK(walletInstance->cs_wallet);
-
     if (chain && !AttachChain(walletInstance, *chain, error, warnings)) {
         return nullptr;
     }
@@ -4966,9 +4964,10 @@ std::shared_ptr<CWallet> CWallet::Create(interfaces::Chain* chain, interfaces::C
         }
     }
 
-    walletInstance->SetBroadcastTransactions(gArgs.GetBoolArg("-walletbroadcast", DEFAULT_WALLETBROADCAST));
 
     {
+        LOCK(walletInstance->cs_wallet);
+        walletInstance->SetBroadcastTransactions(gArgs.GetBoolArg("-walletbroadcast", DEFAULT_WALLETBROADCAST));
         walletInstance->WalletLogPrintf("setExternalKeyPool.size() = %u\n",   walletInstance->KeypoolCountExternalKeys());
         walletInstance->WalletLogPrintf("GetKeyPoolSize() = %u\n",   walletInstance->GetKeyPoolSize());
         walletInstance->WalletLogPrintf("mapWallet.size() = %u\n",            walletInstance->mapWallet.size());