Skip to content

fix: use std::atomic to protect connected manager/signer pointers, check before queueman access, update P2P message map in tests #6842

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
PastaPastaPasta merged 4 commits into from
Sep 23, 2025
Merged

fix: use std::atomic to protect connected manager/signer pointers, check before queueman access, update P2P message map in tests #6842

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b7700b3
trivial: use `std::atomic` to protect connected signer pointers
kwvg Sep 9, 2025
0444e59
trivial: use `std::atomic` to protect connected manager pointers
kwvg Sep 9, 2025
a432a95
fix: check if `queueman` is initialized before accessing it
kwvg Sep 9, 2025
24f9357
fix: update P2P message map in functional test framework
kwvg Sep 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 16 additions & 14 deletions src/chainlock/chainlock.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,17 +64,18 @@ CChainLocksHandler::~CChainLocksHandler()

void CChainLocksHandler::Start(const llmq::CInstantSendManager& isman)
{
if (m_signer) {
m_signer->Start();
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->Start();
}
scheduler->scheduleEvery(
[&]() {
auto signer = m_signer.load(std::memory_order_acquire);
CheckActiveState();
EnforceBestChainLock();
Cleanup();
// regularly retry signing the current chaintip as it might have failed before due to missing islocks
if (m_signer) {
m_signer->TrySignChainTip(isman);
if (signer) {
signer->TrySignChainTip(isman);
}
},
std::chrono::seconds{5});
Expand All @@ -83,8 +84,8 @@ void CChainLocksHandler::Start(const llmq::CInstantSendManager& isman)
void CChainLocksHandler::Stop()
{
scheduler->stop();
if (m_signer) {
m_signer->Stop();
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->Stop();
}
}

Expand Down Expand Up @@ -218,11 +219,12 @@ void CChainLocksHandler::UpdatedBlockTip(const llmq::CInstantSendManager& isman)
if (bool expected = false; tryLockChainTipScheduled.compare_exchange_strong(expected, true)) {
scheduler->scheduleFromNow(
[&]() {
auto signer = m_signer.load(std::memory_order_acquire);
CheckActiveState();
EnforceBestChainLock();
Cleanup();
if (m_signer) {
m_signer->TrySignChainTip(isman);
if (signer) {
signer->TrySignChainTip(isman);
}
tryLockChainTipScheduled = false;
},
Expand Down Expand Up @@ -274,16 +276,16 @@ void CChainLocksHandler::BlockConnected(const std::shared_ptr<const CBlock>& pbl
}

// We need this information later when we try to sign a new tip, so that we can determine if all included TXs are safe.
if (m_signer) {
m_signer->UpdateBlockHashTxidMap(pindex->GetBlockHash(), pblock->vtx);
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->UpdateBlockHashTxidMap(pindex->GetBlockHash(), pblock->vtx);
}
}

void CChainLocksHandler::BlockDisconnected(const std::shared_ptr<const CBlock>& pblock,
gsl::not_null<const CBlockIndex*> pindexDisconnected)
{
if (m_signer) {
m_signer->EraseFromBlockHashTxidMap(pindexDisconnected->GetBlockHash());
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->EraseFromBlockHashTxidMap(pindexDisconnected->GetBlockHash());
}
}

Expand Down Expand Up @@ -451,8 +453,8 @@ void CChainLocksHandler::Cleanup()
}
}

if (m_signer) {
const auto cleanup_txes{m_signer->Cleanup()};
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
const auto cleanup_txes{signer->Cleanup()};
LOCK(cs);
for (const auto& tx : cleanup_txes) {
for (const auto& txid : *tx) {
Expand Down
8 changes: 4 additions & 4 deletions src/chainlock/chainlock.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ class CChainLocksHandler final : public chainlock::ChainLockSignerParent
std::unique_ptr<CScheduler> scheduler;
std::unique_ptr<std::thread> scheduler_thread;

chainlock::ChainLockSigner* m_signer{nullptr};
std::atomic<chainlock::ChainLockSigner*> m_signer{nullptr};

mutable Mutex cs;
std::atomic<bool> tryLockChainTipScheduled{false};
Expand All @@ -73,10 +73,10 @@ class CChainLocksHandler final : public chainlock::ChainLockSignerParent
void ConnectSigner(gsl::not_null<chainlock::ChainLockSigner*> signer)
{
// Prohibit double initialization
assert(m_signer == nullptr);
m_signer = signer;
assert(m_signer.load(std::memory_order_acquire) == nullptr);
m_signer.store(signer, std::memory_order_release);
}
void DisconnectSigner() { m_signer = nullptr; }
void DisconnectSigner() { m_signer.store(nullptr, std::memory_order_release); }

void Start(const llmq::CInstantSendManager& isman);
void Stop();
Expand Down
27 changes: 19 additions & 8 deletions src/evo/mnhftx.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,11 +201,12 @@ static bool extractSignals(const ChainstateManager& chainman, const llmq::CQuoru

std::optional<CMNHFManager::Signals> CMNHFManager::ProcessBlock(const CBlock& block, const CBlockIndex* const pindex, bool fJustCheck, BlockValidationState& state)
{
assert(m_chainman && m_qman);
auto chainman = Assert(m_chainman.load(std::memory_order_acquire));
auto qman = Assert(m_qman.load(std::memory_order_acquire));

try {
std::vector<uint8_t> new_signals;
if (!extractSignals(*m_chainman, *m_qman, block, pindex, new_signals, state)) {
if (!extractSignals(*chainman, *qman, block, pindex, new_signals, state)) {
// state is set inside extractSignals
return std::nullopt;
}
Expand Down Expand Up @@ -252,11 +253,12 @@ std::optional<CMNHFManager::Signals> CMNHFManager::ProcessBlock(const CBlock& bl

bool CMNHFManager::UndoBlock(const CBlock& block, const CBlockIndex* const pindex)
{
assert(m_chainman && m_qman);
auto chainman = Assert(m_chainman.load(std::memory_order_acquire));
auto qman = Assert(m_qman.load(std::memory_order_acquire));

std::vector<uint8_t> excluded_signals;
BlockValidationState state;
if (!extractSignals(*m_chainman, *m_qman, block, pindex, excluded_signals, state)) {
if (!extractSignals(*chainman, *qman, block, pindex, excluded_signals, state)) {
LogPrintf("CMNHFManager::%s: failed to extract signals\n", __func__);
return false;
}
Expand Down Expand Up @@ -372,19 +374,28 @@ void CMNHFManager::AddSignal(const CBlockIndex* const pindex, int bit)
void CMNHFManager::ConnectManagers(gsl::not_null<ChainstateManager*> chainman, gsl::not_null<llmq::CQuorumManager*> qman)
{
// Do not allow double-initialization
assert(m_chainman == nullptr && m_qman == nullptr);
m_chainman = chainman;
m_qman = qman;
assert(m_chainman.load(std::memory_order_acquire) == nullptr);
m_chainman.store(chainman, std::memory_order_release);
assert(m_qman.load(std::memory_order_acquire) == nullptr);
m_qman.store(qman, std::memory_order_release);
}

void CMNHFManager::DisconnectManagers()
{
m_chainman.store(nullptr, std::memory_order_release);
m_qman.store(nullptr, std::memory_order_release);
}

bool CMNHFManager::ForceSignalDBUpdate()
{
auto chainman = Assert(m_chainman.load(std::memory_order_acquire));

// force ehf signals db update
auto dbTx = m_evoDb.BeginTransaction();

const bool last_legacy = bls::bls_legacy_scheme.load();
bls::bls_legacy_scheme.store(false);
GetSignalsStage(m_chainman->ActiveChainstate().m_chain.Tip());
GetSignalsStage(chainman->ActiveChainstate().m_chain.Tip());
bls::bls_legacy_scheme.store(last_legacy);

dbTx->Commit();
Expand Down
19 changes: 11 additions & 8 deletions src/evo/mnhftx.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,16 +5,19 @@
#ifndef BITCOIN_EVO_MNHFTX_H
#define BITCOIN_EVO_MNHFTX_H

#include <bls/bls.h>
#include <gsl/pointers.h>
#include <saltedhasher.h>
#include <sync.h>
#include <threadsafety.h>
#include <versionbits.h>

#include <bls/bls.h>
#include <unordered_lru_cache.h>

#include <gsl/pointers.h>
#include <univalue.h>

#include <atomic>
#include <optional>
#include <saltedhasher.h>
#include <unordered_lru_cache.h>
#include <versionbits.h>

class BlockValidationState;
class CBlock;
Expand Down Expand Up @@ -91,8 +94,8 @@ class CMNHFManager : public AbstractEHFManager
{
private:
CEvoDB& m_evoDb;
ChainstateManager* m_chainman{nullptr};
llmq::CQuorumManager* m_qman{nullptr};
std::atomic<ChainstateManager*> m_chainman{nullptr};
std::atomic<llmq::CQuorumManager*> m_qman{nullptr};

static constexpr size_t MNHFCacheSize = 1000;
Mutex cs_cache;
Expand Down Expand Up @@ -144,7 +147,7 @@ class CMNHFManager : public AbstractEHFManager
*
* @pre Must be called before LLMQContext (containing llmq::CQuorumManager) is destroyed.
*/
void DisconnectManagers() { m_chainman = nullptr; m_qman = nullptr; };
void DisconnectManagers();

bool ForceSignalDBUpdate() EXCLUSIVE_LOCKS_REQUIRED(::cs_main);

Expand Down
33 changes: 17 additions & 16 deletions src/instantsend/instantsend.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,15 +75,15 @@ void CInstantSendManager::Start(PeerManager& peerman)

workThread = std::thread(&util::TraceThread, "isman", [this, &peerman] { WorkThreadMain(peerman); });

if (m_signer) {
m_signer->Start();
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->Start();
}
}

void CInstantSendManager::Stop()
{
if (m_signer) {
m_signer->Stop();
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->Stop();
}

// make sure to call InterruptWorkerThread() first
Expand Down Expand Up @@ -348,8 +348,8 @@ MessageProcessingResult CInstantSendManager::ProcessInstantSendLock(NodeId from,
LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- txid=%s, islock=%s: processing islock, peer=%d\n",
__func__, islock->txid.ToString(), hash.ToString(), from);

if (m_signer) {
m_signer->ClearLockFromQueue(islock);
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->ClearLockFromQueue(islock);
}
if (db.KnownInstantSendLock(hash)) {
return {};
Expand Down Expand Up @@ -449,8 +449,8 @@ void CInstantSendManager::TransactionAddedToMempool(const CTransactionRef& tx)
}

if (islock == nullptr) {
if (m_signer) {
m_signer->ProcessTx(*tx, false, Params().GetConsensus());
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->ProcessTx(*tx, false, Params().GetConsensus());
}
// TX is not locked, so make sure it is tracked
AddNonLockedTx(tx, nullptr);
Expand Down Expand Up @@ -491,8 +491,8 @@ void CInstantSendManager::BlockConnected(const std::shared_ptr<const CBlock>& pb
}

if (!IsLocked(tx->GetHash()) && !has_chainlock) {
if (m_signer) {
m_signer->ProcessTx(*tx, true, Params().GetConsensus());
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->ProcessTx(*tx, true, Params().GetConsensus());
}
// TX is not locked, so make sure it is tracked
AddNonLockedTx(tx, pindex);
Expand Down Expand Up @@ -597,16 +597,16 @@ void CInstantSendManager::RemoveNonLockedTx(const uint256& txid, bool retryChild
void CInstantSendManager::RemoveConflictedTx(const CTransaction& tx)
{
RemoveNonLockedTx(tx.GetHash(), false);
if (m_signer) {
m_signer->ClearInputsFromQueue(GetIdsFromLockable(tx.vin));
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->ClearInputsFromQueue(GetIdsFromLockable(tx.vin));
}
}

void CInstantSendManager::TruncateRecoveredSigsForInputs(const instantsend::InstantSendLock& islock)
{
auto ids = GetIdsFromLockable(islock.inputs);
if (m_signer) {
m_signer->ClearInputsFromQueue(ids);
if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
signer->ClearInputsFromQueue(ids);
}
for (const auto& id : ids) {
sigman.TruncateRecoveredSig(Params().GetConsensus().llmqTypeDIP0024InstantSend, id);
Expand Down Expand Up @@ -925,7 +925,8 @@ void CInstantSendManager::WorkThreadMain(PeerManager& peerman)
for (auto& [node_id, mpr] : peer_activity) {
peerman.PostProcessMessage(std::move(mpr), node_id);
}
if (!m_signer) return more_work;
auto signer = m_signer.load(std::memory_order_acquire);
if (!signer) return more_work;
// Construct set of non-locked transactions that are pending to retry
std::vector<CTransactionRef> txns{};
{
Expand All @@ -942,7 +943,7 @@ void CInstantSendManager::WorkThreadMain(PeerManager& peerman)
}
}
// Retry processing them
m_signer->ProcessPendingRetryLockTxs(txns);
signer->ProcessPendingRetryLockTxs(txns);
return more_work;
}();

Expand Down
8 changes: 4 additions & 4 deletions src/instantsend/instantsend.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ class CInstantSendManager final : public instantsend::InstantSendSignerParent
CTxMemPool& mempool;
const CMasternodeSync& m_mn_sync;

instantsend::InstantSendSigner* m_signer{nullptr};
std::atomic<instantsend::InstantSendSigner*> m_signer{nullptr};

std::thread workThread;
CThreadInterrupt workInterrupt;
Expand Down Expand Up @@ -97,10 +97,10 @@ class CInstantSendManager final : public instantsend::InstantSendSignerParent
void ConnectSigner(gsl::not_null<instantsend::InstantSendSigner*> signer)
{
// Prohibit double initialization
assert(m_signer == nullptr);
m_signer = signer;
assert(m_signer.load(std::memory_order_acquire) == nullptr);
m_signer.store(signer, std::memory_order_release);
}
void DisconnectSigner() { m_signer = nullptr; }
void DisconnectSigner() { m_signer.store(nullptr, std::memory_order_release); }

void Start(PeerManager& peerman);
void Stop();
Expand Down
14 changes: 8 additions & 6 deletions src/net_processing.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2276,11 +2276,11 @@ bool PeerManagerImpl::AlreadyHave(const CInv& inv)
case MSG_ISDLOCK:
return m_llmq_ctx->isman->AlreadyHave(inv);
case MSG_DSQ:
return
#ifdef ENABLE_WALLET
return m_cj_ctx->queueman->HasQueue(inv.hash) || (m_active_ctx && m_active_ctx->cj_server->HasQueue(inv.hash));
#else
return m_active_ctx && m_active_ctx->cj_server->HasQueue(inv.hash);
#endif
(m_cj_ctx->queueman && m_cj_ctx->queueman->HasQueue(inv.hash)) ||
#endif // ENABLE_WALLET
(m_active_ctx && m_active_ctx->cj_server->HasQueue(inv.hash));
}


Expand Down Expand Up @@ -2885,7 +2885,7 @@ void PeerManagerImpl::ProcessGetData(CNode& pfrom, Peer& peer, const std::atomic
if (!push && inv.type == MSG_DSQ) {
auto opt_dsq = m_active_ctx ? m_active_ctx->cj_server->GetQueueFromHash(inv.hash) : std::nullopt;
#ifdef ENABLE_WALLET
if (!opt_dsq.has_value()) {
if (m_cj_ctx->queueman && !opt_dsq.has_value()) {
opt_dsq = m_cj_ctx->queueman->GetQueueFromHash(inv.hash);
}
#endif
Expand Down Expand Up @@ -5273,7 +5273,9 @@ void PeerManagerImpl::ProcessMessage(
{
//probably one the extensions
#ifdef ENABLE_WALLET
PostProcessMessage(m_cj_ctx->queueman->ProcessMessage(pfrom.GetId(), m_connman, *this, msg_type, vRecv), pfrom.GetId());
if (m_cj_ctx->queueman) {
PostProcessMessage(m_cj_ctx->queueman->ProcessMessage(pfrom.GetId(), m_connman, *this, msg_type, vRecv), pfrom.GetId());
}
m_cj_ctx->walletman->ForEachCJClientMan([this, &pfrom, &msg_type, &vRecv](std::unique_ptr<CCoinJoinClientManager>& clientman) {
clientman->ProcessMessage(pfrom, m_chainman.ActiveChainstate(), m_connman, m_mempool, msg_type, vRecv);
});
Expand Down
4 changes: 3 additions & 1 deletion src/rpc/coinjoin.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -474,7 +474,9 @@ static RPCHelpMan getcoinjoininfo()
#ifdef ENABLE_WALLET
CCoinJoinClientOptions::GetJsonInfo(obj);

obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
if (node.cj_ctx->queueman) {
obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
}
Comment on lines +477 to +479
Copy link

@coderabbitai coderabbitai bot Sep 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent

🧩 Analysis chain

Also guard node.cj_ctx and keep queue_size stable to avoid NPE and API drift

node.cj_ctx itself can be nullptr in some runtime configurations; dereferencing it before checking risks a crash. Additionally, omitting queue_size changes the response schema. Push a default 0 when unavailable.

Apply this diff:

-    if (node.cj_ctx->queueman) {
-        obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
-    }
+    if (node.cj_ctx && node.cj_ctx->queueman) {
+        obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
+    } else {
+        // Preserve schema stability even if queue manager isn't initialized
+        obj.pushKV("queue_size", 0);
+    }

Run to find any other unguarded dereferences:

🏁 Script executed:

#!/usr/bin/env bash
set -euo pipefail
rg -nP -C2 '\bcj_ctx->queueman\b' --glob 'src/**'

Length of output: 815

Guard node.cj_ctx dereferences in both RPC and init paths

  • In src/rpc/coinjoin.cpp (lines 477–479), apply:

    -    if (node.cj_ctx->queueman) {
-        obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
-    }
+    if (node.cj_ctx && node.cj_ctx->queueman) {
+        obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
+    } else {
+        obj.pushKV("queue_size", 0);
+    }

  • In src/init.cpp (line 2290), wrap the scheduler call to avoid dereferencing a null context:

    -        node.scheduler->scheduleEvery(
-            std::bind(&CCoinJoinClientQueueManager::DoMaintenance, std::ref(*node.cj_ctx->queueman)),
-            std::chrono::seconds{1});
+        if (node.cj_ctx && node.cj_ctx->queueman) {
+            node.scheduler->scheduleEvery(
+                std::bind(&CCoinJoinClientQueueManager::DoMaintenance, std::ref(*node.cj_ctx->queueman)),
+                std::chrono::seconds{1});
+        }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
if (node.cj_ctx->queueman) {
obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
}
// File: src/rpc/coinjoin.cpp (around lines 477–479)
- if (node.cj_ctx->queueman) {
- obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
if (node.cj_ctx && node.cj_ctx->queueman) {
obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
} else {
obj.pushKV("queue_size", 0);
}
Suggested change
if (node.cj_ctx->queueman) {
obj.pushKV("queue_size", node.cj_ctx->queueman->GetQueueSize());
}
// File: src/init.cpp (around line 2290)
- node.scheduler->scheduleEvery(
- std::bind(&CCoinJoinClientQueueManager::DoMaintenance, std::ref(*node.cj_ctx->queueman)),
if (node.cj_ctx && node.cj_ctx->queueman) {
node.scheduler->scheduleEvery(
std::bind(&CCoinJoinClientQueueManager::DoMaintenance, std::ref(*node.cj_ctx->queueman)),
std::chrono::seconds{1});
}
🤖 Prompt for AI Agents 
In src/rpc/coinjoin.cpp around lines 477-479 and src/init.cpp around line 2290,
the code dereferences node.cj_ctx without null-checks; update both locations to
first check that node.cj_ctx is non-null before accessing its members: in
coinjoin.cpp, only call node.cj_ctx->queueman and pushKV("queue_size", ...) if
node.cj_ctx != nullptr (and ensure queueman itself is checked as already done),
and in init.cpp wrap the scheduler call at line 2290 with a guard that verifies
cj_ctx is non-null before invoking the scheduler to avoid null dereference
during initialization.


const std::shared_ptr<const CWallet> wallet = GetWalletForJSONRPCRequest(request);
if (!wallet) {
Expand Down
Loading
Loading