From 32e344833eeb51b97258fad23855f2aea4a368b2 Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 3 Aug 2026 13:06:09 -0500 Subject: [PATCH 1/8] fix: make CoinJoin nSessionDenom atomic nSessionDenom was the one CCoinJoinBaseSession field that was neither atomic nor guarded, while its siblings nState, nSessionID and nTimeLastSuccessfulStep are all std::atomic. On the server it is written by the message-handling thread in CreateNewSession() and by the scheduler thread in SetNull(), and read without any lock by CheckForCompleteQueue(), AddUserToExistingSession(), IsValidInOuts(), the relay logging, and by RPC threads via GetJsonInfo(). Concurrent unsynchronized access to a plain int is a data race: benign on the hardware we support, but formally UB and reportable by TSan. --- src/coinjoin/client.cpp | 15 +++++++++------ src/coinjoin/coinjoin.h | 4 +++- 2 files changed, 12 insertions(+), 7 deletions(-) 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 0cc205bdcf66..863774b84dfb 100644 --- a/src/coinjoin/coinjoin.h +++ b/src/coinjoin/coinjoin.h @@ -342,7 +342,9 @@ class CCoinJoinBaseSession PoolMessage& nMessageIDRet, bool* fConsumeCollateralRet) const; public: - int nSessionDenom{0}; // Users must submit a denom matching this + // Atomic because the message-handling and scheduler threads write it while those threads and + // RPC callers also read it without holding cs_coinjoin. + std::atomic nSessionDenom{0}; CCoinJoinBaseSession() = default; virtual ~CCoinJoinBaseSession() = default; From 943fff67ba2708fb94557b6b09ba02487fb69bdc Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 3 Aug 2026 13:11:11 -0500 Subject: [PATCH 2/8] fix: decide CoinJoin server state transitions under cs_coinjoin CheckPool() read nState, the entry count and the collateral count under separate lock acquisitions (or none at all) and then acted on the result: it sampled nState, then took and released cs_coinjoin for GetEntriesCount(), then read vecSessionCollaterals.size() unlocked. A scheduler-thread SetNull() landing between the samples made an already-reset session read as '0 entries == 0 collaterals' and get finalized, putting a dead session back into POOL_STATE_SIGNING and rejecting every new dsa until the 15s signing timeout expired. It now decides from one snapshot and acts afterwards, and CreateFinalTransaction()/CommitFinalTransaction() revalidate nSessionID because the decision is made with the lock released. CheckPool() also runs on both the scheduler thread and the message-handling thread, so two concurrent calls could both finalize: clients would receive DSFINALTX twice, sign twice, and the duplicate signatures make AddScriptSig() fail and abort the session for everyone. A TRY_LOCK-only cs_check_pool makes it single-shot without ever blocking msghand. SetState() and IsSessionReady() now require cs_coinjoin, so a transition and the session data it describes can only be observed together; this is what makes the admission revalidation dash#7596 introduced in AddUserToExistingSession() and CheckForCompleteQueue() effective, and ProcessDSVIN()'s readiness gate now takes the lock as well. ChargeFees() samples nState once instead of three times, which previously let it select 'didn't send' offenders and then charge and log them as 'didn't sign'. --- src/coinjoin/server.cpp | 174 +++++++++++++++++++++++++++------------- src/coinjoin/server.h | 35 +++++--- 2 files changed, 140 insertions(+), 69 deletions(-) diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index e380df099c4d..34bb25585dd1 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -200,7 +200,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; @@ -294,42 +294,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 : uint8_t { + 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 (static_cast(entries) == vecSessionCollaterals.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 +391,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 +472,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 = vecSessionCollaterals.size(); + + if (state == POOL_STATE_ACCEPTING_ENTRIES) { + 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 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 +517,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]); } /* @@ -549,6 +604,8 @@ void CCoinJoinServer::CheckTimeout() */ void CCoinJoinServer::CheckForCompleteQueue() { + AssertLockNotHeld(cs_coinjoin); + int session_denom; size_t participants; { @@ -731,8 +788,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; }); @@ -886,6 +942,8 @@ 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()) { return true; @@ -988,6 +1046,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 49cb7a3002d7..8b90e73ab05b 100644 --- a/src/coinjoin/server.h +++ b/src/coinjoin/server.h @@ -52,6 +52,13 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler 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 @@ -65,10 +72,10 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler 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; @@ -77,15 +84,18 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler 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 +105,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 +120,15 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler const CMasternodeSync& mn_sync, const llmq::CInstantSendManager& isman); ~CCoinJoinServer() override; - 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; }; From 0532dea416f3d44b923b647c418159650dfa4f93 Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 3 Aug 2026 13:13:40 -0500 Subject: [PATCH 3/8] fix: guard the CoinJoin session collaterals with cs_coinjoin vecSessionCollaterals had no GUARDED_BY and was reached from both threads with no lock at all: the message-handling thread read it in ProcessDSACCEPT(), IsSessionReady() and AddEntry(), while the scheduler thread read it in CheckPool(), CheckForCompleteQueue(), ChargeFees() and ChargeRandomFees(). The only synchronized accesses were the clear() in SetNull() and the copies dash#7596 and dash#7598 recently put under the lock. Committing a collateral still raced every remaining read. The transactions and their prevout index are now a single SessionCollaterals member so they cannot drift apart, and GUARDED_BY on that member makes every access - including the calls on it - checked by -Wthread-safety. Reintroducing an unlocked read is now a compile error rather than a review finding. --- src/coinjoin/server.cpp | 79 ++++++++++++++++++++--------------------- src/coinjoin/server.h | 50 +++++++++++++++++++++----- 2 files changed, 80 insertions(+), 49 deletions(-) diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index 34bb25585dd1..09fe78c63e16 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} { } @@ -86,7 +85,7 @@ void CCoinJoinServer::ProcessDSACCEPT(CNode& peer, CDataStream& vRecv) return; } - if (WITH_LOCK(cs_coinjoin, return 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; @@ -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(); @@ -323,7 +321,7 @@ void CCoinJoinServer::CheckPool() LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckPool -- entries count %lu\n", entries); } if (nState == POOL_STATE_ACCEPTING_ENTRIES) { - if (static_cast(entries) == vecSessionCollaterals.size()) { + if (static_cast(entries) == m_session_collaterals.size()) { // We have an entry for each collateral action = Action::Finalize; } else if (CCoinJoinServer::HasTimedOut() && entries >= CoinJoin::GetMinPoolParticipants()) { @@ -482,10 +480,10 @@ void CCoinJoinServer::ChargeFees() const // and then charge and log them as "didn't sign", or pick offenders from a session the // state no longer describes. state = nState; - nSessionCollaterals = vecSessionCollaterals.size(); + nSessionCollaterals = m_session_collaterals.size(); if (state == POOL_STATE_ACCEPTING_ENTRIES) { - for (const auto& txCollateral : vecSessionCollaterals) { + for (const auto& txCollateral : m_session_collaterals.txs()) { bool fFound = std::ranges::any_of(vecEntries, [&txCollateral](const auto& entry) { return *entry.txCollateral == *txCollateral; }); @@ -550,7 +548,7 @@ void CCoinJoinServer::ChargeRandomFees() const std::vector session_collaterals; { LOCK(cs_coinjoin); - session_collaterals = vecSessionCollaterals; + session_collaterals = m_session_collaterals.txs(); } for (const auto& txCollateral : session_collaterals) { @@ -614,7 +612,7 @@ void CCoinJoinServer::CheckForCompleteQueue() SetState(POOL_STATE_ACCEPTING_ENTRIES); session_denom = nSessionDenom; - participants = vecSessionCollaterals.size(); + participants = m_session_collaterals.size(); } CCoinJoinQueue dsq(session_denom, m_mn_activeman.GetOutPoint(), m_mn_activeman.GetProTxHash(), GetAdjustedTime(), true); @@ -677,7 +675,7 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag { AssertLockNotHeld(cs_coinjoin); - if (size_t(GetEntriesCount()) >= vecSessionCollaterals.size()) { + 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; @@ -692,10 +690,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; } } @@ -814,15 +813,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; @@ -838,6 +828,8 @@ bool CCoinJoinServer::CreateNewSession(const CCoinJoinAccept& dsa, PoolMessage& return false; } + int nDenom{0}; + size_t nParticipants{0}; { LOCK(cs_coinjoin); @@ -856,21 +848,25 @@ 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; } @@ -916,25 +912,26 @@ bool CCoinJoinServer::AddUserToExistingSession(const CCoinJoinAccept& dsa, PoolM 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; } @@ -945,10 +942,10 @@ 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; } } diff --git a/src/coinjoin/server.h b/src/coinjoin/server.h index 8b90e73ab05b..8782affc1a35 100644 --- a/src/coinjoin/server.h +++ b/src/coinjoin/server.h @@ -12,6 +12,7 @@ #include #include +#include #include class CActiveMasternodeManager; @@ -43,12 +44,47 @@ 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. + 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; } + 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; @@ -79,8 +115,6 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler /// 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? From a873f1dc770b615b30155cc970e2ec0d34f0e1af Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 3 Aug 2026 13:14:22 -0500 Subject: [PATCH 4/8] fix: revalidate the CoinJoin session before committing an entry AddEntry() checked its bound, then ran IsCollateralValid() and IsValidInOuts() - both of which take cs_main and can block behind block validation - and only then took cs_coinjoin again to push_back. A scheduler-thread CheckTimeout() in that window calls SetNull(), so the entry was committed to a session that no longer existed. The consequence outlives the window: vecEntries keeps the orphaned entry while vecSessionCollaterals is empty, so the next session starts one entry ahead of its own participant count. CheckPool()'s entries == collaterals test then fires early and finalizes a transaction containing an input from the dead session, which nobody present will sign, stalling the new session to its signing timeout and charging its honest participants in ChargeFees(). The bound check and the push_back now share one lock scope, and the session identity captured before validation is rechecked inside it, so an entry can only ever be committed to the session it was validated for. --- src/coinjoin/server.cpp | 61 +++++++++++++++++++++++++++++++---------- 1 file changed, 46 insertions(+), 15 deletions(-) diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index 09fe78c63e16..9e51f37da144 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -675,7 +675,12 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag { AssertLockNotHeld(cs_coinjoin); - if (WITH_LOCK(cs_coinjoin, return size_t(GetEntriesCountLocked()) >= m_session_collaterals.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 static_cast(GetEntriesCountLocked()) >= m_session_collaterals.size())) { LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: entries is full!\n", __func__); nMessageIDRet = ERR_ENTRIES_FULL; return false; @@ -711,21 +716,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}; @@ -738,9 +746,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 (static_cast(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; From 87ea7a9f0f159825f8ea28805fdb36485d02e563 Mon Sep 17 00:00:00 2001 From: pasta Date: Sun, 9 Aug 2026 13:21:46 -0500 Subject: [PATCH 5/8] fix: serialize CoinJoin timeout transitions CheckTimeout() reset the pool while CheckPool() could be finalizing or committing the very same session on the message-handling thread: HasTimedOut() was tested without any lock and the reset could land between CheckPool()'s decision and its execution, clearing a live session mid-step. CheckTimeout() now takes the same cs_check_pool guard as CheckPool() - with TRY_LOCK, so a contended scheduler tick is skipped instead of blocking the message-handling thread - which makes timeout resets and finalize/commit single-flight. Offender selection also moves under cs_coinjoin, into SelectCollateralToCharge(), and into the same lock scope that closes the corresponding admission path: CheckTimeout() selects and resets atomically, and CreateFinalTransaction() selects and transitions to SIGNING atomically, so an entry that crossed the cutoff on time can no longer be charged as missing. The collateral is consumed only after the lock is released, because ConsumeCollateral() takes cs_main and mempool submission must not run under cs_coinjoin. --- src/coinjoin/server.cpp | 178 ++++++++++++++++------------- src/coinjoin/server.h | 24 ++-- src/test/coinjoin_inouts_tests.cpp | 47 +++++++- 3 files changed, 159 insertions(+), 90 deletions(-) diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index 9e51f37da144..1b3a93ade609 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -338,10 +338,8 @@ void CCoinJoinServer::CheckPool() case Action::None: return; case Action::ChargeAndFinalize: - ChargeFees(); - [[fallthrough]]; case Action::Finalize: - CreateFinalTransaction(session_id); + CreateFinalTransaction(session_id, action == Action::ChargeAndFinalize); return; case Action::Commit: LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckPool -- SIGNING\n"); @@ -350,43 +348,56 @@ void CCoinJoinServer::CheckPool() } } -void CCoinJoinServer::CreateFinalTransaction(int session_id) +void CCoinJoinServer::CreateFinalTransaction(int session_id, bool charge_fees) { 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; - } + CTransactionRef collateral_to_charge; + { + LOCK(cs_coinjoin); - CMutableTransaction txNew; + // Finalizing a session that is already gone would put it back into SIGNING and reject every + // new one until that timed out. Requiring the accepting state also makes selecting an + // offender and closing entry admission one atomic operation. + if (nSessionID != session_id || nState != POOL_STATE_ACCEPTING_ENTRIES) { + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::CreateFinalTransaction -- session %d is gone or no longer accepting entries\n", + session_id); + return; + } - // make our new transaction - for (const auto& entry : vecEntries) { - for (const auto& txout : entry.vecTxOut) { - txNew.vout.push_back(txout); + if (charge_fees) { + collateral_to_charge = SelectCollateralToCharge(); } - for (const auto& txdsin : entry.vecTxDSIn) { - txNew.vin.push_back(txdsin); + + CMutableTransaction txNew; + + // make our new transaction + for (const auto& entry : vecEntries) { + for (const auto& txout : entry.vecTxOut) { + txNew.vout.push_back(txout); + } + for (const auto& txdsin : entry.vecTxDSIn) { + txNew.vin.push_back(txdsin); + } } - } - sort(txNew.vin.begin(), txNew.vin.end(), CompareInputBIP69()); - sort(txNew.vout.begin(), txNew.vout.end(), CompareOutputBIP69()); + sort(txNew.vin.begin(), txNew.vin.end(), CompareInputBIP69()); + sort(txNew.vout.begin(), txNew.vout.end(), CompareOutputBIP69()); + + finalMutableTransaction = txNew; + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateFinalTransaction -- finalMutableTransaction=%s", /* Continued */ + txNew.ToString()); - finalMutableTransaction = txNew; - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CreateFinalTransaction -- finalMutableTransaction=%s", /* Continued */ - txNew.ToString()); + // request signatures from clients + SetState(POOL_STATE_SIGNING); + RelayFinalTransaction(CTransaction(finalMutableTransaction)); + } - // request signatures from clients - SetState(POOL_STATE_SIGNING); - RelayFinalTransaction(CTransaction(finalMutableTransaction)); + if (collateral_to_charge) { + ConsumeCollateral(collateral_to_charge); + } } void CCoinJoinServer::CommitFinalTransaction(int session_id) @@ -462,71 +473,67 @@ void CCoinJoinServer::CommitFinalTransaction(int session_id) // transaction for the client to be able to enter the pool. This transaction is kept by the Masternode // until the transaction is either complete or fails. // -void CCoinJoinServer::ChargeFees() const +/* + * Select one offender while cs_coinjoin still binds the state, entries and collaterals to the + * same session. The caller must close the relevant admission path before releasing the lock and + * consuming the returned collateral, so a late entry or signature cannot make the snapshot stale. + */ +CTransactionRef CCoinJoinServer::SelectCollateralToCharge() const { - AssertLockNotHeld(cs_coinjoin); + AssertLockHeld(cs_coinjoin); //we don't need to charge collateral for every offence. - if (GetRand(/*nMax=*/100) > 33) return; + if (GetRand(/*nMax=*/100) > 33) return {}; std::vector vecOffendersCollaterals; - size_t nSessionCollaterals{0}; - PoolState state{POOL_STATE_IDLE}; + const PoolState state{nState}; + const size_t nSessionCollaterals{m_session_collaterals.size()}; - { - LOCK(cs_coinjoin); - // 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 send transaction), found " - "offence\n"); - vecOffendersCollaterals.push_back(txCollateral); - } + 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::SelectCollateralToCharge -- 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); - } + } + } 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::SelectCollateralToCharge -- found uncooperative node (didn't sign), " + "found offence\n"); + vecOffendersCollaterals.push_back(entry.txCollateral); } } } } // no offences found - if (vecOffendersCollaterals.empty()) return; + if (vecOffendersCollaterals.empty()) return {}; //mostly offending? Charge sometimes - if (vecOffendersCollaterals.size() >= nSessionCollaterals - 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() >= nSessionCollaterals) return; + if (vecOffendersCollaterals.size() >= nSessionCollaterals) return {}; //charge one of the offenders randomly Shuffle(vecOffendersCollaterals.begin(), vecOffendersCollaterals.end(), FastRandomContext()); LogPrint(BCLog::COINJOIN, /* Continued */ - "CCoinJoinServer::ChargeFees -- found uncooperative node (didn't %s transaction), charging fees: %s", + "CCoinJoinServer::SelectCollateralToCharge -- found uncooperative node (didn't %s transaction), charging " + "fees: %s", (state == POOL_STATE_SIGNING) ? "sign" : "send", vecOffendersCollaterals[0]->ToString()); - ConsumeCollateral(vecOffendersCollaterals[0]); + return vecOffendersCollaterals[0]; } /* @@ -586,13 +593,28 @@ void CCoinJoinServer::CheckTimeout() { m_queueman.CheckQueue(); - // Too early to do anything - if (!CCoinJoinServer::HasTimedOut()) return; + // CheckPool can be finalizing or committing on the message-handling thread. Skipping this tick + // keeps timeout reset and finalization/commit single-flight without blocking the scheduler. + TRY_LOCK(cs_check_pool, lock_check_pool); + if (!lock_check_pool) return; - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckTimeout -- %s timed out -- resetting\n", - (nState == POOL_STATE_SIGNING) ? "Signing" : "Session"); - ChargeFees(); - WITH_LOCK(cs_coinjoin, SetNull()); + CTransactionRef collateral_to_charge; + { + LOCK(cs_coinjoin); + + // Too early to do anything. Recheck while holding the lock so selecting an offender and + // closing the session form one atomic cutoff for late entries and signatures. + if (!CCoinJoinServer::HasTimedOut()) return; + + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckTimeout -- %s timed out -- resetting\n", + (nState == POOL_STATE_SIGNING) ? "Signing" : "Session"); + collateral_to_charge = SelectCollateralToCharge(); + SetNull(); + } + + if (collateral_to_charge) { + ConsumeCollateral(collateral_to_charge); + } } /* @@ -755,7 +777,7 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag // 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) { + if (nSessionID != session_id || nState != POOL_STATE_ACCEPTING_ENTRIES) { LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: session %d is gone!\n", __func__, session_id); nMessageIDRet = ERR_SESSION; return false; diff --git a/src/coinjoin/server.h b/src/coinjoin/server.h index 8782affc1a35..1e9d9dcb60d0 100644 --- a/src/coinjoin/server.h +++ b/src/coinjoin/server.h @@ -26,11 +26,16 @@ class CNode; class CTxMemPool; class UniValue; +namespace coinjoin_inouts_tests { +class TestableCoinJoinServer; +} /** Used to keep track of current status of mixing pool */ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler { + friend class coinjoin_inouts_tests::TestableCoinJoinServer; + private: CoinJoinQueueManager m_queueman; @@ -88,11 +93,12 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler 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. + /// Serializes CheckPool() against itself and against CheckTimeout(). 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. CheckTimeout() uses the same + /// guard so it cannot reset a session during finalization or commit. Production paths always + /// acquire it with TRY_LOCK, so a contended caller skips the round rather than blocking msghand. Mutex cs_check_pool; /// Add a clients entry to the pool @@ -100,8 +106,8 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler /// Add signature to a txin bool AddScriptSig(const CTxIn& txin) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); - /// Charge fees to bad actors (Charge clients a fee if they're abusive) - void ChargeFees() const EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + /// Choose one bad actor whose collateral should be consumed, if any. + CTransactionRef SelectCollateralToCharge() const EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); /// Rarely charge fees to pay miners void ChargeRandomFees() const EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Consume collateral in cases when peer misbehaved @@ -110,7 +116,7 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler /// Check for process void CheckPool() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin, !cs_check_pool); - void CreateFinalTransaction(int session_id) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void CreateFinalTransaction(int session_id, bool charge_fees) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); void CommitFinalTransaction(int session_id) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Is this nDenom and txCollateral acceptable? @@ -161,7 +167,7 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler void Schedule(CScheduler& scheduler) override; bool HasTimedOut() const; - void CheckTimeout() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void CheckTimeout() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin, !cs_check_pool); void CheckForCompleteQueue() EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); void GetJsonInfo(UniValue& obj) const; diff --git a/src/test/coinjoin_inouts_tests.cpp b/src/test/coinjoin_inouts_tests.cpp index f0d49250f53d..4c33ab62ae83 100644 --- a/src/test/coinjoin_inouts_tests.cpp +++ b/src/test/coinjoin_inouts_tests.cpp @@ -26,7 +26,9 @@ #include #include #include +#include #include +#include #include BOOST_FIXTURE_TEST_SUITE(coinjoin_inouts_tests, TestingSetup) @@ -171,9 +173,9 @@ BOOST_AUTO_TEST_CASE(entry_addscriptsig_matches_and_rejects) } } -// Test-only subclass exposing the minimal seams needed to observe how -// ProcessDSSIGNFINALTX treats messages from participants vs. non-participants -// without standing up a full DKG-backed signing session. +// Test-only subclass exposing the minimal seams needed to exercise server lifecycle behavior +// without standing up a full DKG-backed signing session. The helpers only establish preconditions +// and invoke the production paths; the behavior under test is not reproduced here. class TestableCoinJoinServer : public CCoinJoinServer { public: @@ -188,6 +190,21 @@ class TestableCoinJoinServer : public CCoinJoinServer LOCK(cs_coinjoin); vecEntries.push_back(std::move(entry)); } + + void SeedTimedOutSession() + { + LOCK(cs_coinjoin); + nSessionID = 1; + nState = POOL_STATE_ACCEPTING_ENTRIES; + nTimeLastSuccessfulStep = GetTime() - COINJOIN_QUEUE_TIMEOUT; + } + + void HoldPoolCheck(std::latch& locked, std::latch& release) + { + LOCK(cs_check_pool); + locked.count_down(); + release.wait(); + } }; static std::unique_ptr MakePeer(NodeId id, uint32_t ipv4) @@ -287,6 +304,30 @@ BOOST_AUTO_TEST_CASE(server_signfinaltx_participant_oversized_count_is_rejected_ BOOST_CHECK_EQUAL(server.GetEntriesCount(), 1); } +BOOST_AUTO_TEST_CASE(server_timeout_does_not_reset_during_pool_check) +{ + CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey()); + TestableCoinJoinServer server(m_node.peerman.get(), *Assert(m_node.chainman), *Assert(m_node.connman), + *Assert(m_node.dmnman), *Assert(m_node.dstxman), *Assert(m_node.mn_metaman), + *Assert(m_node.mempool), mn_activeman, *Assert(m_node.mn_sync), + *Assert(m_node.llmq_ctx->isman)); + server.SeedTimedOutSession(); + + std::latch locked{1}; + std::latch release{1}; + std::thread pool_check{[&] { server.HoldPoolCheck(locked, release); }}; + locked.wait(); + + server.CheckTimeout(); + BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_ACCEPTING_ENTRIES}); + + release.count_down(); + pool_check.join(); + + server.CheckTimeout(); + BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_IDLE}); +} + BOOST_AUTO_TEST_CASE(entry_deserializes_vectors_through_wire_cap) { const size_t wire_cap{CoinJoin::GetMaxPoolInputOutputCount()}; From 5bf90631acec87912c764c946bfa45290c6595c5 Mon Sep 17 00:00:00 2001 From: pasta Date: Sun, 9 Aug 2026 21:37:57 -0500 Subject: [PATCH 6/8] fix(coinjoin): validate entries against session snapshot AddEntry() validated a submission against the live nSessionDenom while holding no session lock: IsValidInOuts() runs long cs_main work, and a scheduler-thread SetNull() in that window zeroes the denomination, so every output of an honest, on-time entry compared unequal to denom 0 and the ERR_DENOM path consumed that participant's collateral for a reset it could not have known about. IsValidInOuts() now takes the denomination as a parameter and AddEntry() passes the snapshot captured under cs_coinjoin alongside the session id it already revalidates before committing, so validation, punishment and commit are all bound to the same session. AddEntry() also rejects submissions up front when the pool is no longer accepting entries, instead of relying on the entries-full bound alone. --- src/coinjoin/client.cpp | 3 +-- src/coinjoin/coinjoin.cpp | 9 +++++---- src/coinjoin/coinjoin.h | 7 ++++--- src/coinjoin/server.cpp | 28 ++++++++++++++++++++-------- src/test/coinjoin_inouts_tests.cpp | 30 ++++++++++++++++++++++++++++++ 5 files changed, 60 insertions(+), 17 deletions(-) diff --git a/src/coinjoin/client.cpp b/src/coinjoin/client.cpp index 13be545cd2d8..2a3583a10cb8 100644 --- a/src/coinjoin/client.cpp +++ b/src/coinjoin/client.cpp @@ -475,7 +475,7 @@ bool CCoinJoinClientSession::SignFinalTransaction(CNode& peer, Chainstate& activ // Make sure all inputs/outputs are valid PoolMessage nMessageID{MSG_NOERR}; if (!IsValidInOuts(active_chainstate, m_isman, mempool, finalMutableTransaction.vin, finalMutableTransaction.vout, - nMessageID, nullptr)) { + nSessionDenom.load(), nMessageID, nullptr)) { WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::%s -- ERROR! IsValidInOuts() failed: %s\n", __func__, CoinJoin::GetMessageByID(nMessageID).translated); UnlockCoins(); keyHolderStorage.ReturnAll(); @@ -2004,4 +2004,3 @@ UniValue CCoinJoinClientManager::getJsonInfo() const obj.pushKV("sessions", arrSessions); return obj; } - diff --git a/src/coinjoin/coinjoin.cpp b/src/coinjoin/coinjoin.cpp index 3711dbc89f01..9a4bbf6d76ec 100644 --- a/src/coinjoin/coinjoin.cpp +++ b/src/coinjoin/coinjoin.cpp @@ -206,8 +206,8 @@ std::string CCoinJoinBaseSession::GetStateString() const bool CCoinJoinBaseSession::IsValidInOuts(Chainstate& active_chainstate, const llmq::CInstantSendManager& isman, const CTxMemPool& mempool, const std::vector& vin, - const std::vector& vout, PoolMessage& nMessageIDRet, - bool* fConsumeCollateralRet) const + const std::vector& vout, int session_denom, + PoolMessage& nMessageIDRet, bool* fConsumeCollateralRet) { std::set setScripPubKeys; nMessageIDRet = MSG_NOERR; @@ -221,9 +221,10 @@ bool CCoinJoinBaseSession::IsValidInOuts(Chainstate& active_chainstate, const ll } auto checkTxOut = [&](const CTxOut& txout) { - if (int nDenom = CoinJoin::AmountToDenomination(txout.nValue); nDenom != nSessionDenom) { + if (int nDenom = CoinJoin::AmountToDenomination(txout.nValue); nDenom != session_denom) { LogPrint(BCLog::COINJOIN, "CCoinJoinBaseSession::IsValidInOuts -- ERROR: incompatible denom %d (%s) != nSessionDenom %d (%s)\n", - nDenom, CoinJoin::DenominationToString(nDenom), nSessionDenom, CoinJoin::DenominationToString(nSessionDenom)); + nDenom, CoinJoin::DenominationToString(nDenom), session_denom, + CoinJoin::DenominationToString(session_denom)); nMessageIDRet = ERR_DENOM; if (fConsumeCollateralRet) *fConsumeCollateralRet = true; return false; diff --git a/src/coinjoin/coinjoin.h b/src/coinjoin/coinjoin.h index 863774b84dfb..8b9b18196f47 100644 --- a/src/coinjoin/coinjoin.h +++ b/src/coinjoin/coinjoin.h @@ -337,9 +337,10 @@ class CCoinJoinBaseSession virtual void SetNull() EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); - bool IsValidInOuts(Chainstate& active_chainstate, const llmq::CInstantSendManager& isman, - const CTxMemPool& mempool, const std::vector& vin, const std::vector& vout, - PoolMessage& nMessageIDRet, bool* fConsumeCollateralRet) const; + static bool IsValidInOuts(Chainstate& active_chainstate, const llmq::CInstantSendManager& isman, + const CTxMemPool& mempool, const std::vector& vin, + const std::vector& vout, int session_denom, PoolMessage& nMessageIDRet, + bool* fConsumeCollateralRet); public: // Atomic because the message-handling and scheduler threads write it while those threads and diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index 1b3a93ade609..1a72eb4bf6f1 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -699,13 +699,25 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag // 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}; + int session_id{0}; + int session_denom{0}; + { + LOCK(cs_coinjoin); + session_id = nSessionID; + session_denom = nSessionDenom; - // Cheap gate before the cs_main work below; the authoritative check is at commit time. - if (WITH_LOCK(cs_coinjoin, return static_cast(GetEntriesCountLocked()) >= m_session_collaterals.size())) { - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: entries is full!\n", __func__); - nMessageIDRet = ERR_ENTRIES_FULL; - return false; + if (nState != POOL_STATE_ACCEPTING_ENTRIES) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: session is not accepting entries!\n", __func__); + nMessageIDRet = ERR_SESSION; + return false; + } + + // Cheap gate before the cs_main work below; the authoritative check is at commit time. + if (static_cast(GetEntriesCountLocked()) >= m_session_collaterals.size()) { + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: entries is full!\n", __func__); + nMessageIDRet = ERR_ENTRIES_FULL; + return false; + } } if (entry.vecTxDSIn.size() > COINJOIN_ENTRY_MAX_SIZE || entry.vecTxOut.size() > COINJOIN_ENTRY_MAX_SIZE) { @@ -759,8 +771,8 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag } bool fConsumeCollateral{false}; - if (!IsValidInOuts(m_chainman.ActiveChainstate(), m_isman, mempool, vin, entry.vecTxOut, nMessageIDRet, - &fConsumeCollateral)) { + if (!IsValidInOuts(m_chainman.ActiveChainstate(), m_isman, mempool, vin, entry.vecTxOut, session_denom, + nMessageIDRet, &fConsumeCollateral)) { LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR! IsValidInOuts() failed: %s\n", __func__, CoinJoin::GetMessageByID(nMessageIDRet).translated); if (fConsumeCollateral) { ConsumeCollateral(entry.txCollateral); diff --git a/src/test/coinjoin_inouts_tests.cpp b/src/test/coinjoin_inouts_tests.cpp index 4c33ab62ae83..9883760b2039 100644 --- a/src/test/coinjoin_inouts_tests.cpp +++ b/src/test/coinjoin_inouts_tests.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include @@ -205,6 +206,13 @@ class TestableCoinJoinServer : public CCoinJoinServer locked.count_down(); release.wait(); } + + bool ValidateInOuts(const std::vector& vin, const std::vector& vout, int session_denom, + PoolMessage& message, bool& consume_collateral) + { + return IsValidInOuts(m_chainman.ActiveChainstate(), m_isman, mempool, vin, vout, session_denom, message, + &consume_collateral); + } }; static std::unique_ptr MakePeer(NodeId id, uint32_t ipv4) @@ -328,6 +336,28 @@ BOOST_AUTO_TEST_CASE(server_timeout_does_not_reset_during_pool_check) BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_IDLE}); } +BOOST_AUTO_TEST_CASE(server_validation_uses_session_denom_snapshot) +{ + CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey()); + TestableCoinJoinServer server(m_node.peerman.get(), *Assert(m_node.chainman), *Assert(m_node.connman), + *Assert(m_node.dmnman), *Assert(m_node.dstxman), *Assert(m_node.mn_metaman), + *Assert(m_node.mempool), mn_activeman, *Assert(m_node.mn_sync), + *Assert(m_node.llmq_ctx->isman)); + + const int session_denom{CoinJoin::AmountToDenomination(CoinJoin::GetSmallestDenomination())}; + const std::vector vin{CTxIn{COutPoint{uint256::ONE, 0}}}; + const std::vector vout{CTxOut{CoinJoin::GetSmallestDenomination(), P2PKHScript()}}; + PoolMessage message{MSG_NOERR}; + bool consume_collateral{false}; + + // The live session denomination is zero, as it is after a concurrent SetNull(). Validation + // must use the denomination captured before the reset instead of treating this as a punishable + // denomination mismatch. The deliberately absent input makes validation stop with ERR_MISSING_TX. + BOOST_CHECK(!server.ValidateInOuts(vin, vout, session_denom, message, consume_collateral)); + BOOST_CHECK_EQUAL(message, ERR_MISSING_TX); + BOOST_CHECK(!consume_collateral); +} + BOOST_AUTO_TEST_CASE(entry_deserializes_vectors_through_wire_cap) { const size_t wire_cap{CoinJoin::GetMaxPoolInputOutputCount()}; From 4f364ab23b89765375a02482afbbe1a3f9546c7e Mon Sep 17 00:00:00 2001 From: pasta Date: Sun, 9 Aug 2026 22:33:20 -0500 Subject: [PATCH 7/8] fix(coinjoin): prioritize progress over timeout reset The scheduler calls CheckPool() and CheckTimeout() separately, leaving a gap in which the final collateral, entry, or signature can arrive after CheckPool() takes its snapshot. The message thread then skips its own CheckPool() while the scheduler holds cs_check_pool, and an unconditional timeout reset would discard a session that can now advance. Recheck readiness, finalizability, and signature completeness under cs_coinjoin before resetting. Actionable work takes priority and is picked up by the next scheduler tick. --- src/coinjoin/server.cpp | 14 ++++++++ src/test/coinjoin_inouts_tests.cpp | 56 ++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index 1a72eb4bf6f1..6b526d44c953 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -606,6 +606,20 @@ void CCoinJoinServer::CheckTimeout() // closing the session form one atomic cutoff for late entries and signatures. if (!CCoinJoinServer::HasTimedOut()) return; + // CheckForCompleteQueue() and CheckPool() run before this method on the scheduler thread, + // but a final collateral, entry, or signature can arrive after their snapshots. The + // message-handling thread then skips its own CheckPool() if this scheduler tick still holds + // cs_check_pool. Give a session that can now advance priority over resetting it; the next + // scheduler tick will perform the transition, finalization, or commit. + const int entries{GetEntriesCountLocked()}; + const bool can_advance{ + (nState == POOL_STATE_QUEUE && IsSessionReady()) || + (nState == POOL_STATE_ACCEPTING_ENTRIES && + (static_cast(entries) == m_session_collaterals.size() || + entries >= CoinJoin::GetMinPoolParticipants())) || + (nState == POOL_STATE_SIGNING && IsSignaturesComplete())}; + if (can_advance) return; + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CheckTimeout -- %s timed out -- resetting\n", (nState == POOL_STATE_SIGNING) ? "Signing" : "Session"); collateral_to_charge = SelectCollateralToCharge(); diff --git a/src/test/coinjoin_inouts_tests.cpp b/src/test/coinjoin_inouts_tests.cpp index 9883760b2039..04dd6ed50bd1 100644 --- a/src/test/coinjoin_inouts_tests.cpp +++ b/src/test/coinjoin_inouts_tests.cpp @@ -198,6 +198,40 @@ class TestableCoinJoinServer : public CCoinJoinServer nSessionID = 1; nState = POOL_STATE_ACCEPTING_ENTRIES; nTimeLastSuccessfulStep = GetTime() - COINJOIN_QUEUE_TIMEOUT; + for (int i = 0; i < CoinJoin::GetMinPoolParticipants(); ++i) { + CMutableTransaction collateral; + collateral.vin.emplace_back(COutPoint{uint256::ONE, static_cast(i)}); + m_session_collaterals.Add(collateral); + } + } + + void SeedTimedOutActionableSession(PoolState state, bool has_missing_entry = false) + { + LOCK(cs_coinjoin); + SetNull(); + + nSessionID = 1; + nState = state; + nTimeLastSuccessfulStep = GetTime() - + (state == POOL_STATE_SIGNING ? COINJOIN_SIGNING_TIMEOUT : COINJOIN_QUEUE_TIMEOUT); + + for (int i = 0; i < CoinJoin::GetMinPoolParticipants(); ++i) { + CMutableTransaction collateral; + collateral.vin.emplace_back(COutPoint{uint256::ONE, static_cast(i)}); + m_session_collaterals.Add(collateral); + + if (state == POOL_STATE_QUEUE) continue; + + CTxDSIn txdsin{CTxIn{COutPoint{uint256::TWO, static_cast(i)}}, P2PKHScript(), 0}; + txdsin.fHasSig = state == POOL_STATE_SIGNING; + vecEntries.emplace_back(std::vector{txdsin}, std::vector{}, CTransaction{collateral}); + } + + if (has_missing_entry) { + CMutableTransaction collateral; + collateral.vin.emplace_back(COutPoint{uint256::ONE, static_cast(CoinJoin::GetMinPoolParticipants())}); + m_session_collaterals.Add(collateral); + } } void HoldPoolCheck(std::latch& locked, std::latch& release) @@ -336,6 +370,28 @@ BOOST_AUTO_TEST_CASE(server_timeout_does_not_reset_during_pool_check) BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_IDLE}); } +BOOST_AUTO_TEST_CASE(server_timeout_does_not_reset_actionable_session) +{ + CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey()); + TestableCoinJoinServer server(m_node.peerman.get(), *Assert(m_node.chainman), *Assert(m_node.connman), + *Assert(m_node.dmnman), *Assert(m_node.dstxman), *Assert(m_node.mn_metaman), + *Assert(m_node.mempool), mn_activeman, *Assert(m_node.mn_sync), + *Assert(m_node.llmq_ctx->isman)); + + for (const auto state : {POOL_STATE_QUEUE, POOL_STATE_ACCEPTING_ENTRIES, POOL_STATE_SIGNING}) { + BOOST_TEST_CONTEXT("state=" << state) + { + server.SeedTimedOutActionableSession(state); + server.CheckTimeout(); + BOOST_CHECK_EQUAL(server.GetState(), int{state}); + } + } + + server.SeedTimedOutActionableSession(POOL_STATE_ACCEPTING_ENTRIES, /*has_missing_entry=*/true); + server.CheckTimeout(); + BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_ACCEPTING_ENTRIES}); +} + BOOST_AUTO_TEST_CASE(server_validation_uses_session_denom_snapshot) { CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey()); From f7c5e1d648a6a2a28a2e741e56b126aebb215cb3 Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 10 Aug 2026 09:49:57 -0500 Subject: [PATCH 8/8] fix(coinjoin): bind completion tail to session snapshot Completion relay previously read and mutated live session state after CommitFinalTransaction() released cs_coinjoin. If all old participants disconnected, RelayCompletedTransaction() could reset the session, allowing a replacement session to open before random charging and the unconditional tail reset; those operations could then charge or clear the replacement. Capture the committed transaction, participants, and collaterals under cs_coinjoin. Relay and charge only those snapshots, keep completion notification side-effect-free, and reset only the matching signing session. The invalid-transaction path now also notifies captured participants before reset. --- src/coinjoin/server.cpp | 55 +++++++++++++++++------------- src/coinjoin/server.h | 6 ++-- src/test/coinjoin_inouts_tests.cpp | 46 +++++++++++++++++++++++++ 3 files changed, 81 insertions(+), 26 deletions(-) diff --git a/src/coinjoin/server.cpp b/src/coinjoin/server.cpp index 6b526d44c953..5e0d4011a448 100644 --- a/src/coinjoin/server.cpp +++ b/src/coinjoin/server.cpp @@ -405,6 +405,8 @@ void CCoinJoinServer::CommitFinalTransaction(int session_id) AssertLockNotHeld(cs_coinjoin); CTransactionRef finalTransaction; + std::vector participants; + std::vector collaterals; { LOCK(cs_coinjoin); // Committing a session that is already gone would push a cleared finalMutableTransaction @@ -415,6 +417,9 @@ void CCoinJoinServer::CommitFinalTransaction(int session_id) return; } finalTransaction = MakeTransactionRef(finalMutableTransaction); + participants.reserve(vecEntries.size()); + std::ranges::transform(vecEntries, std::back_inserter(participants), [](const auto& entry) { return entry.addr; }); + collaterals = m_session_collaterals.txs(); } uint256 hashTx = finalTransaction->GetHash(); @@ -428,9 +433,9 @@ void CCoinJoinServer::CommitFinalTransaction(int session_id) if (!lockMain || !ATMPIfSaneFee(m_chainman, finalTransaction)) { LogPrint(BCLog::COINJOIN, /* Continued */ "CCoinJoinServer::CommitFinalTransaction -- ATMPIfSaneFee() error: Transaction not valid\n"); - WITH_LOCK(cs_coinjoin, SetNull()); // not much we can do in this case, just notify clients - RelayCompletedTransaction(ERR_INVALID_TX); + RelayCompletedTransaction(session_id, participants, ERR_INVALID_TX); + ResetSigningSessionIfCurrent(session_id); return; } } @@ -451,14 +456,14 @@ void CCoinJoinServer::CommitFinalTransaction(int session_id) m_peer_manager->PeerRelayInv(inv); // Tell the clients it was successful - RelayCompletedTransaction(MSG_SUCCESS); + RelayCompletedTransaction(session_id, participants, MSG_SUCCESS); // Randomly charge clients - ChargeRandomFees(); + ChargeRandomFees(collaterals); // Reset LogPrint(BCLog::COINJOIN, "CCoinJoinServer::CommitFinalTransaction -- COMPLETED -- RESETTING\n"); - WITH_LOCK(cs_coinjoin, SetNull()); + ResetSigningSessionIfCurrent(session_id); } // @@ -548,17 +553,11 @@ CTransactionRef CCoinJoinServer::SelectCollateralToCharge() const stop these kinds of attacks 1 in 10 successful transactions are charged. This adds up to a cost of 0.001DRK per transaction on average. */ -void CCoinJoinServer::ChargeRandomFees() const +void CCoinJoinServer::ChargeRandomFees(const std::vector& collaterals) const { AssertLockNotHeld(cs_coinjoin); - std::vector session_collaterals; - { - LOCK(cs_coinjoin); - session_collaterals = m_session_collaterals.txs(); - } - - for (const auto& txCollateral : session_collaterals) { + for (const auto& txCollateral : collaterals) { if (GetRand(/*nMax=*/100) > 10) return; LogPrint(BCLog::COINJOIN, /* Continued */ "CCoinJoinServer::ChargeRandomFees -- charging random fees, txCollateral=%s", txCollateral->ToString()); @@ -1098,28 +1097,36 @@ void CCoinJoinServer::RelayStatus(PoolStatusUpdate nStatusUpdate, PoolMessage nM } } -void CCoinJoinServer::RelayCompletedTransaction(PoolMessage nMessageID) +void CCoinJoinServer::RelayCompletedTransaction(int session_id, const std::vector& participants, + PoolMessage nMessageID) { AssertLockNotHeld(cs_coinjoin); - LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- nSessionID: %d nSessionDenom: %d (%s)\n", - __func__, nSessionID, nSessionDenom, CoinJoin::DenominationToString(nSessionDenom)); + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- nSessionID: %d\n", __func__, session_id); - // final mixing tx with empty signatures should be relayed to mixing participants only - LOCK(cs_coinjoin); - for (const auto& entry : vecEntries) { - bool fOk = connman.ForNode(entry.addr, [&nMessageID, this](CNode* pnode) { + for (const auto& addr : participants) { + const bool fOk = connman.ForNode(addr, [&nMessageID, session_id, this](CNode* pnode) { CNetMsgMaker msgMaker(pnode->GetCommonVersion()); - connman.PushMessage(pnode, msgMaker.Make(NetMsgType::DSCOMPLETE, nSessionID.load(), nMessageID)); + connman.PushMessage(pnode, msgMaker.Make(NetMsgType::DSCOMPLETE, session_id, nMessageID)); return true; }); if (!fOk) { - // no such node? maybe client disconnected or our own connection went down - RelayStatus(STATUS_REJECTED); - break; + LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- participant disconnected before completion\n", __func__); } } } +void CCoinJoinServer::ResetSigningSessionIfCurrent(int session_id) +{ + AssertLockNotHeld(cs_coinjoin); + LOCK(cs_coinjoin); + if (nSessionID != session_id || nState != POOL_STATE_SIGNING) { + LogPrint(BCLog::COINJOIN, /* Continued */ + "CCoinJoinServer::%s -- signing session %d is no longer current, not resetting\n", __func__, session_id); + return; + } + SetNull(); +} + void CCoinJoinServer::SetState(PoolState nStateNew) { AssertLockHeld(cs_coinjoin); diff --git a/src/coinjoin/server.h b/src/coinjoin/server.h index 1e9d9dcb60d0..61c1f3ffda09 100644 --- a/src/coinjoin/server.h +++ b/src/coinjoin/server.h @@ -109,7 +109,7 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler /// Choose one bad actor whose collateral should be consumed, if any. CTransactionRef SelectCollateralToCharge() const EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); /// Rarely charge fees to pay miners - void ChargeRandomFees() const EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void ChargeRandomFees(const std::vector& collaterals) const EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); /// Consume collateral in cases when peer misbehaved void ConsumeCollateral(const CTransactionRef& txref) const; @@ -141,7 +141,9 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler void RelayFinalTransaction(const CTransaction& txFinal) EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); void PushStatus(CNode& peer, PoolStatusUpdate nStatusUpdate, PoolMessage nMessageID) const; void RelayStatus(PoolStatusUpdate nStatusUpdate, PoolMessage nMessageID = MSG_NOERR) EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin); - void RelayCompletedTransaction(PoolMessage nMessageID) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void RelayCompletedTransaction(int session_id, const std::vector& participants, PoolMessage nMessageID) + EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); + void ResetSigningSessionIfCurrent(int session_id) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); void ProcessDSACCEPT(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin); void ProcessDSQUEUE(NodeId from, CDataStream& vRecv); diff --git a/src/test/coinjoin_inouts_tests.cpp b/src/test/coinjoin_inouts_tests.cpp index 04dd6ed50bd1..60e3cfbe0353 100644 --- a/src/test/coinjoin_inouts_tests.cpp +++ b/src/test/coinjoin_inouts_tests.cpp @@ -192,6 +192,27 @@ class TestableCoinJoinServer : public CCoinJoinServer vecEntries.push_back(std::move(entry)); } + void SeedCompletionSession(int session_id, const CService& addr, PoolState state = POOL_STATE_SIGNING) + { + LOCK(cs_coinjoin); + SetNull(); + nSessionID = session_id; + nState = state; + + if (state == POOL_STATE_SIGNING) { + CCoinJoinEntry entry; + entry.addr = addr; + vecEntries.push_back(std::move(entry)); + } + } + + void RelayCompletion(int session_id, const CService& addr) + { + RelayCompletedTransaction(session_id, {addr}, MSG_SUCCESS); + } + + void ResetForSession(int session_id) { ResetSigningSessionIfCurrent(session_id); } + void SeedTimedOutSession() { LOCK(cs_coinjoin); @@ -346,6 +367,31 @@ BOOST_AUTO_TEST_CASE(server_signfinaltx_participant_oversized_count_is_rejected_ BOOST_CHECK_EQUAL(server.GetEntriesCount(), 1); } +BOOST_AUTO_TEST_CASE(server_completion_does_not_reset_an_unreachable_or_replacement_session) +{ + CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey()); + TestableCoinJoinServer server(m_node.peerman.get(), *Assert(m_node.chainman), *Assert(m_node.connman), + *Assert(m_node.dmnman), *Assert(m_node.dstxman), *Assert(m_node.mn_metaman), + *Assert(m_node.mempool), mn_activeman, *Assert(m_node.mn_sync), + *Assert(m_node.llmq_ctx->isman)); + + auto participant = MakePeer(/*id=*/7, /*ipv4=*/0x0a000001); + server.SeedCompletionSession(/*session_id=*/1, participant->addr); + + // The participant is deliberately absent from connman. Failing to deliver DSCOMPLETE must not + // reset live session data; only CommitFinalTransaction owns the completion reset. + server.RelayCompletion(/*session_id=*/1, participant->addr); + BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_SIGNING}); + BOOST_CHECK_EQUAL(server.GetEntriesCount(), 1); + + // A delayed tail operation from session 1 must never clear a replacement queue, even if the + // random wire session ID happens to be reused. + server.SeedCompletionSession(/*session_id=*/1, participant->addr, POOL_STATE_QUEUE); + server.ResetForSession(/*session_id=*/1); + BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_QUEUE}); + BOOST_CHECK_EQUAL(server.GetEntriesCount(), 0); +} + BOOST_AUTO_TEST_CASE(server_timeout_does_not_reset_during_pool_check) { CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey());