fix: protect m_wallet_manager_map with cs_wallet_manager_map

Avoid TSan-reported data race

```
WARNING: ThreadSanitizer: data race (pid=374820)
  Read of size 8 at 0x7b140002ce10 by thread T12:
    #0 _M_ptr /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:154:42 (dashd+0xb58e08) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
    dashpay#1 get /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:361:21 (dashd+0xb58e08)
    dashpay#2 operator-> /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:355:9 (dashd+0xb58e08)
    dashpay#3 CoinJoinWalletManager::DoMaintenance() /src/dash/src/coinjoin/client.cpp:1907:9 (dashd+0xb58e08)
    [...]
  Previous write of size 8 at 0x7b140002ce10 by thread T17 (mutexes: write M0):
    #0 operator new(unsigned long) <null> (dashd+0x162657) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
    dashpay#1 allocate /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/ext/new_allocator.h:114:27 (dashd+0xb772b4) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
    dashpay#2 allocate /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/alloc_traits.h:443:20 (dashd+0xb772b4)
    dashpay#3 _M_get_node /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/stl_tree.h:580:16 (dashd+0xb772b4)
    [...]
    dashpay#8 CoinJoinWalletManager::Add(CWallet&) /src/dash/src/coinjoin/client.cpp:1898:26 (dashd+0xb58c73) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
[...]
SUMMARY: ThreadSanitizer: data race [...]
```

src/coinjoin/client.cpp

@@ -96,7 +96,7 @@ 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 && ranges::any_of(WITH_LOCK(m_walletman.cs_wallet_manager_map, return m_walletman.raw()),
                                          [this, &dmn](const auto &pair) {
                                              return pair.second->TrySubmitDenominate(dmn->pdmnState->addr,
                                                                                      this->connman);
@@ -121,7 +121,7 @@ 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(),
+            ranges::any_of(WITH_LOCK(m_walletman.cs_wallet_manager_map, return m_walletman.raw()),
                            [&dsq](const auto &pair) { return pair.second->MarkAlreadyJoinedQueueAsTried(dsq); });
 
             WITH_LOCK(cs_vecqueue, vecCoinJoinQueue.push_back(dsq));
@@ -159,7 +159,7 @@ CCoinJoinClientSession::CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletMa
                                                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_manager(*Assert(WITH_LOCK(walletman.cs_wallet_manager_map, return walletman.Get(wallet.GetName())))),
     m_dmnman(dmnman),
     m_mn_metaman(mn_metaman),
     m_mn_sync(mn_sync),
@@ -1895,6 +1895,8 @@ void CCoinJoinClientManager::GetJsonInfo(UniValue& obj) const
 }
 
 void CoinJoinWalletManager::Add(CWallet& wallet) {
+    AssertLockHeld(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)
@@ -1903,18 +1905,25 @@ void CoinJoinWalletManager::Add(CWallet& wallet) {
 }
 
 void CoinJoinWalletManager::DoMaintenance() {
+    AssertLockNotHeld(cs_wallet_manager_map);
+    LOCK(cs_wallet_manager_map);
+
     for (auto& [wallet_str, walletman] : m_wallet_manager_map) {
         walletman->DoMaintenance(m_chainstate, m_connman, m_mempool);
     }
 }
 
 void CoinJoinWalletManager::Remove(const std::string& name) {
+    AssertLockHeld(cs_wallet_manager_map);
+
     m_wallet_manager_map.erase(name);
     g_wallet_init_interface.InitCoinJoinSettings(*this);
 }
 
 void CoinJoinWalletManager::Flush(const std::string& name)
 {
+    AssertLockHeld(cs_wallet_manager_map);
+
     auto clientman = Get(name);
     assert(clientman != nullptr);
     clientman->ResetPool();

src/coinjoin/client.h

@@ -72,6 +72,8 @@ class CoinJoinWalletManager {
 public:
     using wallet_name_cjman_map = std::map<const std::string, std::unique_ptr<CCoinJoinClientManager>>;
 
+    Mutex cs_wallet_manager_map;
+
 public:
     CoinJoinWalletManager(CChainState& chainstate, CConnman& connman, CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman, CTxMemPool& mempool,
                           const CMasternodeSync& mn_sync, const std::unique_ptr<CCoinJoinClientQueueManager>& queueman, bool is_masternode)
@@ -85,15 +87,18 @@ class CoinJoinWalletManager {
         }
     }
 
-    void Add(CWallet& wallet);
-    void DoMaintenance();
+    void Add(CWallet& wallet) EXCLUSIVE_LOCKS_REQUIRED(cs_wallet_manager_map);
+    void DoMaintenance() EXCLUSIVE_LOCKS_REQUIRED(!cs_wallet_manager_map);
 
-    void Remove(const std::string& name);
-    void Flush(const std::string& name);
+    void Remove(const std::string& name) EXCLUSIVE_LOCKS_REQUIRED(cs_wallet_manager_map);
+    void Flush(const std::string& name) EXCLUSIVE_LOCKS_REQUIRED(cs_wallet_manager_map);
 
-    CCoinJoinClientManager* Get(const std::string& name) const;
+    CCoinJoinClientManager* Get(const std::string& name) const EXCLUSIVE_LOCKS_REQUIRED(cs_wallet_manager_map);
 
-    const wallet_name_cjman_map& raw() const { return m_wallet_manager_map; }
+    const wallet_name_cjman_map& raw() const EXCLUSIVE_LOCKS_REQUIRED(cs_wallet_manager_map)
+    {
+        return m_wallet_manager_map;
+    }
 
 private:
     CChainState& m_chainstate;
@@ -105,7 +110,7 @@ class CoinJoinWalletManager {
     const std::unique_ptr<CCoinJoinClientQueueManager>& m_queueman;
 
     const bool m_is_masternode;
-    wallet_name_cjman_map m_wallet_manager_map;
+    wallet_name_cjman_map m_wallet_manager_map GUARDED_BY(cs_wallet_manager_map);
 };
 
 class CCoinJoinClientSession : public CCoinJoinBaseSession

src/coinjoin/interfaces.cpp

@@ -69,19 +69,19 @@ class CoinJoinLoaderImpl : public interfaces::CoinJoin::Loader
 
     void AddWallet(CWallet& wallet) override
     {
-        m_walletman.Add(wallet);
+        WITH_LOCK(m_walletman.cs_wallet_manager_map, m_walletman.Add(wallet));
     }
     void RemoveWallet(const std::string& name) override
     {
-        m_walletman.Remove(name);
+        WITH_LOCK(m_walletman.cs_wallet_manager_map, m_walletman.Remove(name));
     }
     void FlushWallet(const std::string& name) override
     {
-        m_walletman.Flush(name);
+        WITH_LOCK(m_walletman.cs_wallet_manager_map, m_walletman.Flush(name));
     }
     std::unique_ptr<interfaces::CoinJoin::Client> GetClient(const std::string& name) override
     {
-        auto clientman = m_walletman.Get(name);
+        auto clientman = WITH_LOCK(m_walletman.cs_wallet_manager_map, return m_walletman.Get(name));
         return clientman ? std::make_unique<CoinJoinClientImpl>(*clientman) : nullptr;
     }
     CoinJoinWalletManager& walletman() override

src/dsnotificationinterface.cpp

@@ -89,7 +89,7 @@ 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()) {
+    for (auto& pair : WITH_LOCK(m_cj_ctx->walletman->cs_wallet_manager_map, return m_cj_ctx->walletman->raw())) {
         pair.second->UpdatedBlockTip(pindexNew);
     }
 #endif // ENABLE_WALLET

src/init.cpp

@@ -2123,7 +2123,10 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
 
 #ifdef ENABLE_WALLET
     node.coinjoin_loader = interfaces::MakeCoinJoinLoader(*node.cj_ctx->walletman);
-    g_wallet_init_interface.InitCoinJoinSettings(*node.cj_ctx->walletman);
+    {
+        LOCK(node.cj_ctx->walletman->cs_wallet_manager_map);
+        g_wallet_init_interface.InitCoinJoinSettings(*node.cj_ctx->walletman);
+    }
 #endif // ENABLE_WALLET
 
     // ********************************************************* Step 7d: Setup other Dash services

src/masternode/utils.cpp

@@ -23,7 +23,7 @@ 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()) {
+    for (auto& pair : WITH_LOCK(cj_ctx.walletman->cs_wallet_manager_map, return cj_ctx.walletman->raw())) {
         pair.second->GetMixingMasternodesInfo(vecDmns);
     }
 #endif // ENABLE_WALLET

src/net_processing.cpp

@@ -5095,7 +5095,7 @@ 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()) {
+        for (auto& pair : WITH_LOCK(m_cj_ctx->walletman->cs_wallet_manager_map, return m_cj_ctx->walletman->raw())) {
             pair.second->ProcessMessage(pfrom, m_chainman.ActiveChainstate(), m_connman, m_mempool, msg_type, vRecv);
         }
 #endif // ENABLE_WALLET

src/rpc/coinjoin.cpp

@@ -84,7 +84,10 @@ static RPCHelpMan coinjoin_reset()
     ValidateCoinJoinArguments();
 
     CHECK_NONFATAL(node.coinjoin_loader);
-    auto cj_clientman = node.coinjoin_loader->walletman().Get(wallet->GetName());
+    auto cj_clientman = [&](){
+        LOCK(node.coinjoin_loader->walletman().cs_wallet_manager_map);
+        return node.coinjoin_loader->walletman().Get(wallet->GetName());
+    }();
 
     CHECK_NONFATAL(cj_clientman);
     cj_clientman->ResetPool();
@@ -127,7 +130,10 @@ static RPCHelpMan coinjoin_start()
     }
 
     CHECK_NONFATAL(node.coinjoin_loader);
-    auto cj_clientman = node.coinjoin_loader->walletman().Get(wallet->GetName());
+    auto cj_clientman = [&](){
+        LOCK(node.coinjoin_loader->walletman().cs_wallet_manager_map);
+        return node.coinjoin_loader->walletman().Get(wallet->GetName());
+    }();
 
     CHECK_NONFATAL(cj_clientman);
     if (!cj_clientman->StartMixing()) {
@@ -169,7 +175,10 @@ static RPCHelpMan coinjoin_stop()
     ValidateCoinJoinArguments();
 
     CHECK_NONFATAL(node.coinjoin_loader);
-    auto cj_clientman = node.coinjoin_loader->walletman().Get(wallet->GetName());
+    auto cj_clientman = [&](){
+        LOCK(node.coinjoin_loader->walletman().cs_wallet_manager_map);
+        return node.coinjoin_loader->walletman().Get(wallet->GetName());
+    }();
 
     CHECK_NONFATAL(cj_clientman);
     if (!cj_clientman->IsMixing()) {
@@ -239,7 +248,10 @@ static RPCHelpMan coinjoinsalt_generate()
 
     const NodeContext& node = EnsureAnyNodeContext(request.context);
     if (node.coinjoin_loader != nullptr) {
-        auto cj_clientman = node.coinjoin_loader->walletman().Get(wallet->GetName());
+        auto cj_clientman = [&](){
+            LOCK(node.coinjoin_loader->walletman().cs_wallet_manager_map);
+            return node.coinjoin_loader->walletman().Get(wallet->GetName());
+        }();
         if (cj_clientman != nullptr && cj_clientman->IsMixing()) {
             throw JSONRPCError(RPC_WALLET_ERROR,
                                strprintf("Wallet \"%s\" is currently mixing, cannot change salt!", str_wallet));
@@ -341,7 +353,10 @@ static RPCHelpMan coinjoinsalt_set()
 
     const NodeContext& node = EnsureAnyNodeContext(request.context);
     if (node.coinjoin_loader != nullptr) {
-        auto cj_clientman = node.coinjoin_loader->walletman().Get(wallet->GetName());
+        auto cj_clientman = [&](){
+            LOCK(node.coinjoin_loader->walletman().cs_wallet_manager_map);
+            return node.coinjoin_loader->walletman().Get(wallet->GetName());
+        }();
         if (cj_clientman != nullptr && cj_clientman->IsMixing()) {
             throw JSONRPCError(RPC_WALLET_ERROR,
                                strprintf("Wallet \"%s\" is currently mixing, cannot change salt!", str_wallet));
@@ -450,7 +465,10 @@ static RPCHelpMan getcoinjoininfo()
         return obj;
     }
 
-    auto manager = node.coinjoin_loader->walletman().Get(wallet->GetName());
+    auto manager = [&](){
+        LOCK(node.coinjoin_loader->walletman().cs_wallet_manager_map);
+        return node.coinjoin_loader->walletman().Get(wallet->GetName());
+    }();
     CHECK_NONFATAL(manager != nullptr);
     manager->GetJsonInfo(obj);
 

src/wallet/init.cpp

@@ -46,7 +46,8 @@ class WalletInit : public WalletInitInterface
 
     // Dash Specific Wallet Init
     void AutoLockMasternodeCollaterals() const override;
-    void InitCoinJoinSettings(const CoinJoinWalletManager& cjwalletman) const override;
+    void InitCoinJoinSettings(const CoinJoinWalletManager& cjwalletman) const override
+        EXCLUSIVE_LOCKS_REQUIRED(cjwalletman.cs_wallet_manager_map);
     bool InitAutoBackup() const override;
 };
 

src/wallet/test/coinjoin_tests.cpp

@@ -209,8 +209,13 @@ class CTransactionBuilderTestSetup : public TestChain100Setup
 
 BOOST_FIXTURE_TEST_CASE(coinjoin_manager_start_stop_tests, CTransactionBuilderTestSetup)
 {
-    BOOST_CHECK_EQUAL(m_node.cj_ctx->walletman->raw().size(), 1);
-    auto& cj_man = m_node.cj_ctx->walletman->raw().begin()->second;
+    CCoinJoinClientManager* cj_man{nullptr};
+    {
+        LOCK(m_node.cj_ctx->walletman->cs_wallet_manager_map);
+        BOOST_CHECK_EQUAL(m_node.cj_ctx->walletman->raw().size(), 1);
+        cj_man = m_node.cj_ctx->walletman->raw().begin()->second.get();
+    }
+    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);