diff --git a/src/coinjoin/client.cpp b/src/coinjoin/client.cpp index 1a16f48acadd..13be545cd2d8 100644 --- a/src/coinjoin/client.cpp +++ b/src/coinjoin/client.cpp @@ -1229,8 +1229,9 @@ bool CCoinJoinClientSession::JoinExistingQueue(CAmount nBalanceNeedsAnonymized, SetState(POOL_STATE_QUEUE); nTimeLastSuccessfulStep = GetTime(); WalletCJLogPrint(m_wallet, /* Continued */ - "CCoinJoinClientSession::JoinExistingQueue -- pending connection, masternode=%s, nSessionDenom=%d (%s)\n", - dmn->proTxHash.ToString(), nSessionDenom, CoinJoin::DenominationToString(nSessionDenom)); + "CCoinJoinClientSession::JoinExistingQueue -- pending connection, masternode=%s, " + "nSessionDenom=%d (%s)\n", + dmn->proTxHash.ToString(), nSessionDenom.load(), CoinJoin::DenominationToString(nSessionDenom)); strAutoDenomResult = _("Trying to connect…"); return true; } @@ -1310,9 +1311,11 @@ bool CCoinJoinClientSession::StartNewQueue(CAmount nBalanceNeedsAnonymized, CCon pendingDsaRequest = CPendingDsaRequest(dmn->proTxHash, CCoinJoinAccept(nSessionDenom, txMyCollateral)); SetState(POOL_STATE_QUEUE); nTimeLastSuccessfulStep = GetTime(); - WalletCJLogPrint( /* Continued */ - m_wallet, "CCoinJoinClientSession::StartNewQueue -- pending connection, masternode=%s, nSessionDenom=%d (%s)\n", - dmn->proTxHash.ToString(), nSessionDenom, CoinJoin::DenominationToString(nSessionDenom)); + WalletCJLogPrint(/* Continued */ + m_wallet, + "CCoinJoinClientSession::StartNewQueue -- pending connection, masternode=%s, nSessionDenom=%d " + "(%s)\n", + dmn->proTxHash.ToString(), nSessionDenom.load(), CoinJoin::DenominationToString(nSessionDenom)); strAutoDenomResult = _("Trying to connect…"); return true; } @@ -1424,7 +1427,7 @@ bool CCoinJoinClientSession::SubmitDenominate(CConnman& connman) return a.second > b.second || (a.second == b.second && a.first < b.first); }); - WalletCJLogPrint(m_wallet, "vecInputsByRounds for denom %d\n", nSessionDenom); + WalletCJLogPrint(m_wallet, "vecInputsByRounds for denom %d\n", nSessionDenom.load()); for (const auto& pair : vecInputsByRounds) { WalletCJLogPrint(m_wallet, "vecInputsByRounds: rounds: %d, inputs: %d\n", pair.first, pair.second); } diff --git a/src/coinjoin/coinjoin.h b/src/coinjoin/coinjoin.h index 20662233091e..684bb10b901f 100644 --- a/src/coinjoin/coinjoin.h +++ b/src/coinjoin/coinjoin.h @@ -342,7 +342,12 @@ class CCoinJoinBaseSession PoolMessage& nMessageIDRet, bool* fConsumeCollateralRet) const; public: - int nSessionDenom{0}; // Users must submit a denom matching this + // Users must submit a denom matching this. Atomic like its sibling session fields: it is + // written by the message-handling thread when a session opens and by either thread in + // SetNull(), and read unlocked by both plus RPC threads via GetJsonInfo(). + // Note when logging: LogPrint() takes its arguments by const reference so a bare read works, + // but WalletCJLogPrint() takes them by value and needs an explicit .load(). + std::atomic nSessionDenom{0}; CCoinJoinBaseSession() = default; virtual ~CCoinJoinBaseSession() = default; diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index 6d9a3ba16578..bade4b3e4f25 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -41,7 +41,6 @@ CCoinJoinServer::CCoinJoinServer(PeerManagerInternal* peer_manager, ChainstateMa m_mn_activeman{mn_activeman}, m_mn_sync{mn_sync}, m_isman{isman}, - vecSessionCollaterals{}, fUnitTest{false} { } @@ -67,7 +66,7 @@ void CCoinJoinServer::ProcessDSACCEPT(CNode& peer, CDataStream& vRecv) { assert(m_mn_metaman.IsValid()); - if (IsSessionReady()) { + if (WITH_LOCK(cs_coinjoin, return IsSessionReady())) { // too many users in this session already, reject new ones LogPrint(BCLog::COINJOIN, "DSACCEPT -- queue is already full!\n"); PushStatus(peer, STATUS_REJECTED, ERR_QUEUE_FULL); @@ -86,7 +85,7 @@ void CCoinJoinServer::ProcessDSACCEPT(CNode& peer, CDataStream& vRecv) return; } - if (vecSessionCollaterals.empty()) { + if (WITH_LOCK(cs_coinjoin, return m_session_collaterals.empty())) { { const auto hasQueue = m_queueman.TryHasQueueFromMasternode(m_mn_activeman.GetOutPoint()); if (!hasQueue.has_value()) return; @@ -200,7 +199,7 @@ void CCoinJoinServer::ProcessDSQUEUE(NodeId from, CDataStream& vRecv) void CCoinJoinServer::ProcessDSVIN(CNode& peer, CDataStream& vRecv) { //do we have enough users in the current session? - if (!IsSessionReady()) { + if (!WITH_LOCK(cs_coinjoin, return IsSessionReady())) { LogPrint(BCLog::COINJOIN, "DSVIN -- session not complete!\n"); PushStatus(peer, STATUS_REJECTED, ERR_SESSION); return; @@ -282,8 +281,7 @@ void CCoinJoinServer::SetNull() { AssertLockHeld(cs_coinjoin); // MN side - vecSessionCollaterals.clear(); - setSessionCollateralPrevouts.clear(); + m_session_collaterals.Clear(); CCoinJoinBaseSession::SetNull(); m_queueman.SetNull(); @@ -294,42 +292,79 @@ void CCoinJoinServer::SetNull() // void CCoinJoinServer::CheckPool() { - if (int entries = GetEntriesCount(); entries != 0) - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckPool -- entries count %lu\n", entries); + AssertLockNotHeld(cs_coinjoin); - // If we have an entry for each collateral, then create final tx - if (nState == POOL_STATE_ACCEPTING_ENTRIES && size_t(GetEntriesCount()) == vecSessionCollaterals.size()) { - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckPool -- FINALIZE TRANSACTIONS\n"); - CreateFinalTransaction(); - return; + // Both the scheduler thread and the message-handling thread get here. Skip the round if the + // other one is already in it rather than blocking msghand behind its mempool work: whichever + // thread holds the lock is performing the same check we would. + TRY_LOCK(cs_check_pool, lock_check_pool); + if (!lock_check_pool) return; + + // Decide what to do from a single consistent snapshot. Sampling nState, the entry count and + // the collateral count under separate lock acquisitions let a concurrent SetNull() land + // between them, so an already-reset session could be read as "0 entries == 0 collaterals" + // and finalized: an empty final transaction, a dead session put back into SIGNING, and no + // new session accepted until that timed out. + enum class Action { + None, + Finalize, + ChargeAndFinalize, + Commit + }; + Action action{Action::None}; + int session_id{0}; + { + LOCK(cs_coinjoin); + session_id = nSessionID; + const int entries{GetEntriesCountLocked()}; + if (entries != 0) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckPool -- entries count %lu\n", entries); + } + if (nState == POOL_STATE_ACCEPTING_ENTRIES) { + if (size_t(entries) == m_session_collaterals.size()) { + // We have an entry for each collateral + action = Action::Finalize; + } else if (CCoinJoinServer::HasTimedOut() && entries >= CoinJoin::GetMinPoolParticipants()) { + // We timed out while accepting entries but still have more than the minimum, so + // punish the misbehaving participants and complete the session without them + action = Action::ChargeAndFinalize; + } + } else if (nState == POOL_STATE_SIGNING && IsSignaturesComplete()) { + action = Action::Commit; + } } - // Check for Time Out - // If we timed out while accepting entries, then if we have more than minimum, create final tx - if (nState == POOL_STATE_ACCEPTING_ENTRIES && CCoinJoinServer::HasTimedOut() && - GetEntriesCount() >= CoinJoin::GetMinPoolParticipants()) { - // Punish misbehaving participants + switch (action) { + case Action::None: + return; + case Action::ChargeAndFinalize: ChargeFees(); - // Try to complete this session ignoring the misbehaving ones - CreateFinalTransaction(); + [[fallthrough]]; + case Action::Finalize: + CreateFinalTransaction(session_id); return; - } - - // If we have all the signatures, try to compile the transaction - if (nState == POOL_STATE_SIGNING && IsSignaturesComplete()) { + case Action::Commit: LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckPool -- SIGNING\n"); - CommitFinalTransaction(); + CommitFinalTransaction(session_id); return; } } -void CCoinJoinServer::CreateFinalTransaction() +void CCoinJoinServer::CreateFinalTransaction(int session_id) { AssertLockNotHeld(cs_coinjoin); LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateFinalTransaction -- FINALIZE TRANSACTIONS\n"); LOCK(cs_coinjoin); + // Finalizing a session that is already gone would put it back into SIGNING and reject every + // new one until that timed out. + if (nSessionID != session_id) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateFinalTransaction -- session %d is gone, not finalizing\n", + session_id); + return; + } + CMutableTransaction txNew; // make our new transaction @@ -354,11 +389,22 @@ void CCoinJoinServer::CreateFinalTransaction() RelayFinalTransaction(CTransaction(finalMutableTransaction)); } -void CCoinJoinServer::CommitFinalTransaction() +void CCoinJoinServer::CommitFinalTransaction(int session_id) { AssertLockNotHeld(cs_coinjoin); - CTransactionRef finalTransaction = WITH_LOCK(cs_coinjoin, return MakeTransactionRef(finalMutableTransaction)); + CTransactionRef finalTransaction; + { + LOCK(cs_coinjoin); + // Committing a session that is already gone would push a cleared finalMutableTransaction + // through ATMP and notify the participants of a failure that never happened. + if (nSessionID != session_id) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CommitFinalTransaction -- session %d is gone, not committing\n", + session_id); + return; + } + finalTransaction = MakeTransactionRef(finalMutableTransaction); + } uint256 hashTx = finalTransaction->GetHash(); LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CommitFinalTransaction -- finalTransaction=%s", /* Continued */ @@ -424,33 +470,42 @@ void CCoinJoinServer::ChargeFees() const if (GetRand(/*nMax=*/100) > 33) return; std::vector vecOffendersCollaterals; + size_t nSessionCollaterals{0}; + PoolState state{POOL_STATE_IDLE}; - if (nState == POOL_STATE_ACCEPTING_ENTRIES) { - LOCK(cs_coinjoin); - for (const auto& txCollateral : vecSessionCollaterals) { - bool fFound = std::ranges::any_of(vecEntries, [&txCollateral](const auto& entry) { - return *entry.txCollateral == *txCollateral; - }); - - // This queue entry didn't send us the promised transaction - if (!fFound) { - LogPrint(BCLog::COINJOIN, /* Continued */ - "CCoinJoinServer::ChargeFees -- found uncooperative node (didn't send transaction), found " - "offence\n"); - vecOffendersCollaterals.push_back(txCollateral); - } - } - } - - if (nState == POOL_STATE_SIGNING) { - // who didn't sign? + { LOCK(cs_coinjoin); - for (const auto& entry : vecEntries) { - for (const auto& txdsin : entry.vecTxDSIn) { - if (!txdsin.fHasSig) { + // Sample the state under the lock, together with the data it describes. Reading nState + // separately per branch let a concurrent transition select the "didn't send" offenders + // and then charge and log them as "didn't sign", or pick offenders from a session the + // state no longer describes. + state = nState; + nSessionCollaterals = m_session_collaterals.size(); + + if (state == POOL_STATE_ACCEPTING_ENTRIES) { + for (const auto& txCollateral : m_session_collaterals.txs()) { + bool fFound = std::ranges::any_of(vecEntries, [&txCollateral](const auto& entry) { + return *entry.txCollateral == *txCollateral; + }); + + // This queue entry didn't send us the promised transaction + if (!fFound) { LogPrint(BCLog::COINJOIN, /* Continued */ - "CCoinJoinServer::ChargeFees -- found uncooperative node (didn't sign), found offence\n"); - vecOffendersCollaterals.push_back(entry.txCollateral); + "CCoinJoinServer::ChargeFees -- found uncooperative node (didn't send transaction), found " + "offence\n"); + vecOffendersCollaterals.push_back(txCollateral); + } + } + } else if (state == POOL_STATE_SIGNING) { + // who didn't sign? + for (const auto& entry : vecEntries) { + for (const auto& txdsin : entry.vecTxDSIn) { + if (!txdsin.fHasSig) { + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::ChargeFees -- found uncooperative node (didn't sign), found " + "offence\n"); + vecOffendersCollaterals.push_back(entry.txCollateral); + } } } } @@ -460,20 +515,18 @@ void CCoinJoinServer::ChargeFees() const if (vecOffendersCollaterals.empty()) return; //mostly offending? Charge sometimes - if (vecOffendersCollaterals.size() >= vecSessionCollaterals.size() - 1 && GetRand(/*nMax=*/100) > 33) return; + if (vecOffendersCollaterals.size() >= nSessionCollaterals - 1 && GetRand(/*nMax=*/100) > 33) return; //everyone is an offender? That's not right - if (vecOffendersCollaterals.size() >= vecSessionCollaterals.size()) return; + if (vecOffendersCollaterals.size() >= nSessionCollaterals) return; //charge one of the offenders randomly Shuffle(vecOffendersCollaterals.begin(), vecOffendersCollaterals.end(), FastRandomContext()); - if (nState == POOL_STATE_ACCEPTING_ENTRIES || nState == POOL_STATE_SIGNING) { - LogPrint(BCLog::COINJOIN, /* Continued */ - "CCoinJoinServer::ChargeFees -- found uncooperative node (didn't %s transaction), charging fees: %s", - (nState == POOL_STATE_SIGNING) ? "sign" : "send", vecOffendersCollaterals[0]->ToString()); - ConsumeCollateral(vecOffendersCollaterals[0]); - } + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::ChargeFees -- found uncooperative node (didn't %s transaction), charging fees: %s", + (state == POOL_STATE_SIGNING) ? "sign" : "send", vecOffendersCollaterals[0]->ToString()); + ConsumeCollateral(vecOffendersCollaterals[0]); } /* @@ -490,7 +543,14 @@ void CCoinJoinServer::ChargeFees() const */ void CCoinJoinServer::ChargeRandomFees() const { - for (const auto& txCollateral : vecSessionCollaterals) { + AssertLockNotHeld(cs_coinjoin); + + // Copy the collaterals out before consuming any: ConsumeCollateral() takes cs_main, which + // must not be held under cs_coinjoin, and iterating the member directly meant a concurrent + // SetNull() destroyed the transactions this loop was walking. + const std::vector collaterals{WITH_LOCK(cs_coinjoin, return m_session_collaterals.CopyTxs())}; + + for (const auto& txCollateral : collaterals) { if (GetRand(/*nMax=*/100) > 10) return; LogPrint(BCLog::COINJOIN, /* Continued */ "CCoinJoinServer::ChargeRandomFees -- charging random fees, txCollateral=%s", txCollateral->ToString()); @@ -541,17 +601,34 @@ void CCoinJoinServer::CheckTimeout() */ void CCoinJoinServer::CheckForCompleteQueue() { - if (nState == POOL_STATE_QUEUE && IsSessionReady()) { - SetState(POOL_STATE_ACCEPTING_ENTRIES); + AssertLockNotHeld(cs_coinjoin); - CCoinJoinQueue dsq(nSessionDenom, m_mn_activeman.GetOutPoint(), m_mn_activeman.GetProTxHash(), - GetAdjustedTime(), true); - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckForCompleteQueue -- queue is ready, signing and relaying (%s) " /* Continued */ - "with %d participants\n", dsq.ToString(), vecSessionCollaterals.size()); - dsq.vchSig = m_mn_activeman.SignBasic(dsq.GetSignatureHash()); - m_peer_manager->PeerRelayDSQ(dsq); - m_queueman.AddQueue(std::move(dsq)); + int nDenom{0}; + size_t nParticipants{0}; + { + // Test the readiness condition and perform the transition under one lock, so that a + // message-handling thread revalidating POOL_STATE_QUEUE cannot have the state flipped + // out from under it and commit a collateral into a session that has already announced + // itself ready. The denom and participant count are captured for the log and dsq below, + // which run after the lock is released. + LOCK(cs_coinjoin); + if (nState != POOL_STATE_QUEUE || !IsSessionReady()) return; + + SetState(POOL_STATE_ACCEPTING_ENTRIES); + nDenom = nSessionDenom; + nParticipants = m_session_collaterals.size(); } + + // Signing and relaying happen with cs_coinjoin released: BLS signing and network sends have + // no business holding the session lock. + CCoinJoinQueue dsq(nDenom, m_mn_activeman.GetOutPoint(), m_mn_activeman.GetProTxHash(), GetAdjustedTime(), true); + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::CheckForCompleteQueue -- queue is ready, signing and relaying (%s) " /* Continued */ + "with %d participants\n", + dsq.ToString(), nParticipants); + dsq.vchSig = m_mn_activeman.SignBasic(dsq.GetSignatureHash()); + m_peer_manager->PeerRelayDSQ(dsq); + m_queueman.AddQueue(std::move(dsq)); } // Check to make sure a given input matches an input in the pool and its scriptSig is valid @@ -606,7 +683,12 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag { AssertLockNotHeld(cs_coinjoin); - if (size_t(GetEntriesCount()) >= vecSessionCollaterals.size()) { + // Remember which session we are admitting this entry to. The validation below releases + // cs_coinjoin and takes cs_main, so the session can be reset underneath us before we commit. + const int session_id{nSessionID}; + + // Cheap gate before the cs_main work below; the authoritative check is at commit time. + if (WITH_LOCK(cs_coinjoin, return size_t(GetEntriesCountLocked()) >= m_session_collaterals.size())) { LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: entries is full!\n", __func__); nMessageIDRet = ERR_ENTRIES_FULL; return false; @@ -621,10 +703,11 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag CTransactionRef txCollateralToConsume; { LOCK(cs_coinjoin); - const auto it = std::ranges::find_if(vecSessionCollaterals, [&entry](const auto& txCollateral) { + const auto& txs = m_session_collaterals.txs(); + const auto it = std::ranges::find_if(txs, [&entry](const auto& txCollateral) { return *entry.txCollateral == *txCollateral; }); - if (it != vecSessionCollaterals.end()) { + if (it != txs.end()) { txCollateralToConsume = *it; } } @@ -641,21 +724,24 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag } std::vector vin; - for (const auto& txin : entry.vecTxDSIn) { - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- txin=%s\n", __func__, txin.ToString()); + { LOCK(cs_coinjoin); - for (const auto& inner_entry : vecEntries) { - if (std::ranges::any_of(inner_entry.vecTxDSIn, - [&txin](const auto& txdsin) { return txdsin.prevout == txin.prevout; })) { - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: already have this txin in entries\n", __func__); - nMessageIDRet = ERR_ALREADY_HAVE; - // Two peers sent the same input? Can't really say who is the malicious one here, - // could be that someone is picking someone else's inputs randomly trying to force - // collateral consumption. Do not punish. - return false; + for (const auto& txin : entry.vecTxDSIn) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- txin=%s\n", __func__, txin.ToString()); + for (const auto& inner_entry : vecEntries) { + if (std::ranges::any_of(inner_entry.vecTxDSIn, + [&txin](const auto& txdsin) { return txdsin.prevout == txin.prevout; })) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: already have this txin in entries\n", + __func__); + nMessageIDRet = ERR_ALREADY_HAVE; + // Two peers sent the same input? Can't really say who is the malicious one here, + // could be that someone is picking someone else's inputs randomly trying to force + // collateral consumption. Do not punish. + return false; + } } + vin.emplace_back(txin); } - vin.emplace_back(txin); } bool fConsumeCollateral{false}; @@ -668,9 +754,32 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag return false; } - WITH_LOCK(cs_coinjoin, vecEntries.push_back(entry)); + int nEntries{0}; + { + LOCK(cs_coinjoin); + + // IsCollateralValid() and IsValidInOuts() above take cs_main and can block for a long + // time behind block validation, so a scheduler-thread timeout can reset the session in + // that window. Committing then would leave an entry of a dead session in vecEntries: the + // next session inherits it, counts it towards its own participants, and finalizes a + // transaction containing an input nobody present is going to sign. + if (nSessionID != session_id) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: session %d is gone!\n", __func__, session_id); + nMessageIDRet = ERR_SESSION; + return false; + } + if (size_t(GetEntriesCountLocked()) >= m_session_collaterals.size()) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: entries is full!\n", __func__); + nMessageIDRet = ERR_ENTRIES_FULL; + return false; + } + + vecEntries.push_back(entry); + nEntries = GetEntriesCountLocked(); + } - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- adding entry %d of %d required\n", __func__, GetEntriesCount(), CoinJoin::GetMaxPoolParticipants()); + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- adding entry %d of %d required\n", __func__, nEntries, + CoinJoin::GetMaxPoolParticipants()); nMessageIDRet = MSG_ENTRIES_ADDED; return true; @@ -717,8 +826,7 @@ bool CCoinJoinServer::AddScriptSig(const CTxIn& txinNew) // Check to make sure everything is signed bool CCoinJoinServer::IsSignaturesComplete() const { - AssertLockNotHeld(cs_coinjoin); - LOCK(cs_coinjoin); + AssertLockHeld(cs_coinjoin); return std::ranges::all_of(vecEntries, [](const auto& entry) { return std::ranges::all_of(entry.vecTxDSIn, [](const auto& txdsin) { return txdsin.fHasSig; }); @@ -744,15 +852,6 @@ bool CCoinJoinServer::IsAcceptableDSA(const CCoinJoinAccept& dsa, PoolMessage& n return true; } -void CCoinJoinServer::CommitSessionCollateral(const CMutableTransaction& txCollateral) -{ - AssertLockHeld(cs_coinjoin); - vecSessionCollaterals.push_back(MakeTransactionRef(txCollateral)); - for (const auto& txin : txCollateral.vin) { - setSessionCollateralPrevouts.insert(txin.prevout); - } -} - bool CCoinJoinServer::CreateNewSession(const CCoinJoinAccept& dsa, PoolMessage& nMessageIDRet) { if (nSessionID != 0) return false; @@ -768,6 +867,8 @@ bool CCoinJoinServer::CreateNewSession(const CCoinJoinAccept& dsa, PoolMessage& return false; } + int nDenom{0}; + size_t nParticipants{0}; { LOCK(cs_coinjoin); @@ -786,28 +887,34 @@ bool CCoinJoinServer::CreateNewSession(const CCoinJoinAccept& dsa, PoolMessage& SetState(POOL_STATE_QUEUE); - CommitSessionCollateral(dsa.txCollateral); + m_session_collaterals.Add(dsa.txCollateral); + nDenom = nSessionDenom; + nParticipants = m_session_collaterals.size(); } if (!fUnitTest) { //broadcast that I'm accepting entries, only if it's the first entry through - CCoinJoinQueue dsq(nSessionDenom, m_mn_activeman.GetOutPoint(), m_mn_activeman.GetProTxHash(), - GetAdjustedTime(), false); + CCoinJoinQueue dsq(nDenom, m_mn_activeman.GetOutPoint(), m_mn_activeman.GetProTxHash(), GetAdjustedTime(), false); LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateNewSession -- signing and relaying new queue: %s\n", dsq.ToString()); dsq.vchSig = m_mn_activeman.SignBasic(dsq.GetSignatureHash()); m_peer_manager->PeerRelayDSQ(dsq); m_queueman.AddQueue(std::move(dsq)); } - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateNewSession -- new session created, nSessionID: %d nSessionDenom: %d (%s) vecSessionCollaterals.size(): %d CoinJoin::GetMaxPoolParticipants(): %d\n", - nSessionID, nSessionDenom, CoinJoin::DenominationToString(nSessionDenom), vecSessionCollaterals.size(), CoinJoin::GetMaxPoolParticipants()); + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::CreateNewSession -- new session created, nSessionID: %d nSessionDenom: %d (%s) " + "participants: %d CoinJoin::GetMaxPoolParticipants(): %d\n", + nSessionID, nSessionDenom, CoinJoin::DenominationToString(nSessionDenom), nParticipants, + CoinJoin::GetMaxPoolParticipants()); return true; } bool CCoinJoinServer::AddUserToExistingSession(const CCoinJoinAccept& dsa, PoolMessage& nMessageIDRet) { - if (nSessionID == 0 || IsSessionReady()) return false; + // Cheap gates first: IsAcceptableDSA() below runs a mempool test-accept, which a full or + // absent session must not pay for. + if (nSessionID == 0 || WITH_LOCK(cs_coinjoin, return IsSessionReady())) return false; if (!IsAcceptableDSA(dsa, nMessageIDRet)) { return false; @@ -832,30 +939,31 @@ bool CCoinJoinServer::AddUserToExistingSession(const CCoinJoinAccept& dsa, PoolM // A scheduler-thread timeout can reset the session via SetNull() between the checks above // and taking cs_coinjoin, so revalidate: a collateral must never be committed to a session // that no longer exists. - if (nSessionID == 0 || nState != POOL_STATE_QUEUE) { + if (nSessionID == 0 || nState != POOL_STATE_QUEUE || IsSessionReady()) { nMessageIDRet = ERR_MODE; return false; } - // Session collaterals are only ever test-accepted, never added to the mempool, so nothing - // pins their identity: the same UTXO can be re-signed into arbitrarily many distinct txids. - // Match on input prevouts so a resent or replayed dsa cannot be counted as a new participant. - for (const auto& txin : dsa.txCollateral.vin) { - if (setSessionCollateralPrevouts.contains(txin.prevout)) { - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::AddUserToExistingSession -- collateral %s spends prevout %s already committed to this session\n", - dsa.txCollateral.GetHash().ToString(), txin.prevout.ToStringShort()); - nMessageIDRet = ERR_ALREADY_HAVE; - return false; - } + // A resent or replayed dsa must not be counted as a new participant; see SessionCollaterals. + if (const auto prevout = m_session_collaterals.FindCommittedPrevout(dsa.txCollateral)) { + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::AddUserToExistingSession -- collateral %s spends prevout %s already committed to " + "this session\n", + dsa.txCollateral.GetHash().ToString(), prevout->ToStringShort()); + nMessageIDRet = ERR_ALREADY_HAVE; + return false; } // count new user as accepted to an existing session nMessageIDRet = MSG_NOERR; - CommitSessionCollateral(dsa.txCollateral); + m_session_collaterals.Add(dsa.txCollateral); - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::AddUserToExistingSession -- new user accepted, nSessionID: %d nSessionDenom: %d (%s) vecSessionCollaterals.size(): %d CoinJoin::GetMaxPoolParticipants(): %d\n", - nSessionID, nSessionDenom, CoinJoin::DenominationToString(nSessionDenom), vecSessionCollaterals.size(), CoinJoin::GetMaxPoolParticipants()); + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::AddUserToExistingSession -- new user accepted, nSessionID: %d nSessionDenom: %d (%s) " + "participants: %d CoinJoin::GetMaxPoolParticipants(): %d\n", + nSessionID, nSessionDenom, CoinJoin::DenominationToString(nSessionDenom), m_session_collaterals.size(), + CoinJoin::GetMaxPoolParticipants()); return true; } @@ -863,11 +971,13 @@ bool CCoinJoinServer::AddUserToExistingSession(const CCoinJoinAccept& dsa, PoolM // Returns true if either max size has been reached or if the mix timed out and min size was reached bool CCoinJoinServer::IsSessionReady() const { + AssertLockHeld(cs_coinjoin); + if (nState == POOL_STATE_QUEUE) { - if ((int)vecSessionCollaterals.size() >= CoinJoin::GetMaxPoolParticipants()) { + if ((int)m_session_collaterals.size() >= CoinJoin::GetMaxPoolParticipants()) { return true; } - if (CCoinJoinServer::HasTimedOut() && (int)vecSessionCollaterals.size() >= CoinJoin::GetMinPoolParticipants()) { + if (CCoinJoinServer::HasTimedOut() && (int)m_session_collaterals.size() >= CoinJoin::GetMinPoolParticipants()) { return true; } } @@ -965,6 +1075,8 @@ void CCoinJoinServer::RelayCompletedTransaction(PoolMessage nMessageID) void CCoinJoinServer::SetState(PoolState nStateNew) { + AssertLockHeld(cs_coinjoin); + if (nStateNew == POOL_STATE_ERROR) { LogPrint(BCLog::COINJOIN, "CCoinJoinServer::SetState -- Can't set state to ERROR as a Masternode. \n"); return; diff --git a/src/coinjoin/server.h b/src/coinjoin/server.h index 6e148871b9d0..52a4811b1172 100644 --- a/src/coinjoin/server.h +++ b/src/coinjoin/server.h @@ -12,6 +12,7 @@ #include #include +#include #include class CActiveMasternodeManager; @@ -43,15 +44,61 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler const CMasternodeSync& m_mn_sync; const llmq::CInstantSendManager& m_isman; - // Mixing uses collateral transactions to trust parties entering the pool - // to behave honestly. If they don't it takes their money. - std::vector vecSessionCollaterals; - // Input prevouts of every transaction in vecSessionCollaterals, so a dsa whose collateral - // reuses one of them can be rejected without rescanning them all. - std::unordered_set setSessionCollateralPrevouts GUARDED_BY(cs_coinjoin); + /// The collateral transactions of every peer admitted to the current session. + /// + /// Mixing uses collateral transactions to trust parties entering the pool to behave + /// honestly. If they don't it takes their money. + /// + /// Session collaterals are only ever test-accepted, never added to the mempool, so nothing + /// pins their identity: the same UTXO can be re-signed into arbitrarily many distinct txids. + /// Matching on input prevouts is what makes a resent or replayed dsa recognisable as the same + /// participant, which is why the prevout index lives here rather than beside it. + class SessionCollaterals + { + public: + void Add(const CMutableTransaction& txCollateral) + { + m_txs.push_back(MakeTransactionRef(txCollateral)); + for (const auto& txin : txCollateral.vin) { + m_prevouts.insert(txin.prevout); + } + } + void Clear() + { + m_txs.clear(); + m_prevouts.clear(); + } + //! The first input of txCollateral that an already admitted collateral also spends, if any. + std::optional FindCommittedPrevout(const CMutableTransaction& txCollateral) const + { + for (const auto& txin : txCollateral.vin) { + if (m_prevouts.contains(txin.prevout)) return txin.prevout; + } + return std::nullopt; + } + const std::vector& txs() const { return m_txs; } + //! For callers that need the collaterals after releasing cs_coinjoin. Returning by value + //! keeps the copy inside the lock scope; WITH_LOCK deduces decltype(auto), so returning + //! txs() through it would hand back a reference and copy after the lock was released. + std::vector CopyTxs() const { return m_txs; } + size_t size() const { return m_txs.size(); } + bool empty() const { return m_txs.empty(); } + + private: + std::vector m_txs; + std::unordered_set m_prevouts; + }; + SessionCollaterals m_session_collaterals GUARDED_BY(cs_coinjoin); bool fUnitTest; + /// Serializes CheckPool() against itself. CheckPool() runs both on the scheduler thread and + /// on the message-handling thread, and its finalize and commit steps have to be single-shot: + /// relaying DSFINALTX twice makes every client sign twice, and the duplicate signatures then + /// abort the session for all of them. Always acquired with TRY_LOCK and never taken by any + /// other code path, so a contended caller skips the round rather than blocking msghand. + Mutex cs_check_pool; + /// Add a clients entry to the pool bool AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessageIDRet) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Add signature to a txin @@ -60,32 +107,33 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler /// Charge fees to bad actors (Charge clients a fee if they're abusive) void ChargeFees() const EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Rarely charge fees to pay miners - void ChargeRandomFees() const; + void ChargeRandomFees() const EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Consume collateral in cases when peer misbehaved void ConsumeCollateral(const CTransactionRef& txref) const; /// Check for process - void CheckPool(); + void CheckPool() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin, !cs_check_pool); - void CreateFinalTransaction() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); - void CommitFinalTransaction() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void CreateFinalTransaction(int session_id) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void CommitFinalTransaction(int session_id) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Is this nDenom and txCollateral acceptable? bool IsAcceptableDSA(const CCoinJoinAccept& dsa, PoolMessage& nMessageIDRet) const; - /// Record an accepted collateral and index its input prevouts - void CommitSessionCollateral(const CMutableTransaction& txCollateral) EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); bool CreateNewSession(const CCoinJoinAccept& dsa, PoolMessage& nMessageIDRet) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); bool AddUserToExistingSession(const CCoinJoinAccept& dsa, PoolMessage& nMessageIDRet) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Do we have enough users to take entries? - bool IsSessionReady() const; + bool IsSessionReady() const EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); /// Check that all inputs are signed. (Are all inputs signed?) - bool IsSignaturesComplete() const EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + bool IsSignaturesComplete() const EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); /// Check to make sure a given input matches an input in the pool and its scriptSig is valid bool IsInputScriptSigValid(const CTxIn& txin) const EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); - // Set the 'state' value, with some logging and capturing when the state changed - void SetState(PoolState nStateNew); + // Set the 'state' value, with some logging and capturing when the state changed. + // Requires cs_coinjoin so that a transition and the session data it describes are always + // observed together: code that revalidates nState under the lock must not have it changed + // out from under it by a concurrent transition. + void SetState(PoolState nStateNew) EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); /// Relay mixing Messages void RelayFinalTransaction(const CTransaction& txFinal) EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); @@ -95,8 +143,8 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler void ProcessDSACCEPT(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); void ProcessDSQUEUE(NodeId from, CDataStream& vRecv); - void ProcessDSVIN(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); - void ProcessDSSIGNFINALTX(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void ProcessDSVIN(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin, !cs_check_pool); + void ProcessDSSIGNFINALTX(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin, !cs_check_pool); void SetNull() override EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); @@ -110,14 +158,15 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler const CMasternodeSync& mn_sync, const llmq::CInstantSendManager& isman); ~CCoinJoinServer(); - void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv) override; + void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv) override + EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin, !cs_check_pool); bool ProcessGetData(CNode& pfrom, const CInv& inv, const CNetMsgMaker& msgMaker) override; bool AlreadyHave(const CInv& inv) override; void Schedule(CScheduler& scheduler) override; bool HasTimedOut() const; - void CheckTimeout(); - void CheckForCompleteQueue(); + void CheckTimeout() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void CheckForCompleteQueue() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); void GetJsonInfo(UniValue& obj) const; };