Merge #7063: refactor: extract CActiveMasternodeManager from LLMQContext (2/n, CQuorumManager handler separation)

d41e5bd chore: apply review suggestions (Kittywhiskers Van Gogh)
8a3ad09 refactor: abstract away parent implementation from handler (Kittywhiskers Van Gogh)
718ee50 refactor: streamline quorum cache logic (Kittywhiskers Van Gogh)
1360d9d refactor: drop unnecessary `CConnman` argument in handlers (Kittywhiskers Van Gogh)
2a57a44 refactor: move quorum manager to separate source file (Kittywhiskers Van Gogh)
50a5f26 refactor: pull handler initialization out of `CQuorumManager` (Kittywhiskers Van Gogh)
61da16d refactor: separate observer/common routines into dedicated class (Kittywhiskers Van Gogh)
e8b771e move-only: move observer/common routines to `llmq/observer` (Kittywhiskers Van Gogh)
4078de0 refactor: disentangle watch-only and masternode mode conn checking (Kittywhiskers Van Gogh)
a2d909b refactor: pull deletable quorums routine to dedicated function (Kittywhiskers Van Gogh)
eaee1a8 refactor: disentangle watch-only and masternode mode thread triggers (Kittywhiskers Van Gogh)
e6d8e69 refactor: pull data recovery lambda to a dedicated function (Kittywhiskers Van Gogh)
1fe85be refactor: separate observer/participant logic into dedicated class (Kittywhiskers Van Gogh)
ba24b4e move-only: move observer/participant logic to `active/quorums.cpp` (Kittywhiskers Van Gogh)
de0ce4e refactor: extract masternode-mode specific `SetSecretKeyShare()` (Kittywhiskers Van Gogh)
201ccfc refactor: extract `ENCRYPTED_CONTRIBUTIONS` processing case (Kittywhiskers Van Gogh)

Pull request description:

  ## Additional Information

  * Depends on #7062

  * Dependency for #7065

  * To enforce the split between masternode mode (which can participate in quorums and seek quorum data) and watch-only mode (which can **only** seek quorum data), threading logic is split between `StartDataRecoveryThread()` and `StartVvecSyncThread()`, they both call the same underlying `DataRecoveryThread()` but one has access to masternode-specific parameters and the other does not.

    This becomes relevant as the entities are split out and the access to specific parameters are enforced by the relevant class members outright not existing.

    * It is recommended to use `TryStartVvecSyncThread()` as it will not start threads if no data is actually needed, calling `StartVvecSyncThread()` directly bypasses this check.

  * `CQuorumManager` exposes both `IsWatching()` and `IsMasternode()` to allow P2P code and interfaces to query the node's state (this is most relevant in `PeerManager` which can trivially detect masternode mode but not watch-only status).
    * Watch-only nodes cannot be masternodes but masternodes can **also** be watch-only (the term "observer" has been used in the codebase where possible as watch-only becomes a bit of a misnomer in this case but is the established term) nodes.

  * The `CQuorumManager` cache warmer was one of the tasks allocated to the common worker pool, as the rest of the activities are managed by the observer context, rather than keeping the worker pool in the quorum manager and then exposing it through the interface for the sake of one task, the worker pool has been moved to the observer context and we have a regular thread for the quorum manager instead.

  ## Breaking Changes

  None expected.

  ## Checklist

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

ACKs for top commit:
  UdjinM6:
    utACK d41e5bd

Tree-SHA512: 9d4f0c67149dcec10d50a74dafe18b165590b8977203f82f3104645bf5a6e1e109fa901c1aaebf543ebc215aa910bd47b8b88dc993a66363655fb0c6ea57ed73

Author: PastaPastaPasta

src/Makefile.am

@@ -151,6 +151,7 @@ endif
 .PHONY: FORCE check-symbols check-security
 # dash core #
 BITCOIN_CORE_H = \
+  active/quorums.h \
   addrdb.h \
   addressindex.h \
   spentindex.h \
@@ -278,6 +279,7 @@ BITCOIN_CORE_H = \
   llmq/options.h \
   llmq/params.h \
   llmq/quorums.h \
+  llmq/quorumsman.h \
   llmq/signhash.h \
   llmq/signing.h \
   llmq/net_signing.h \
@@ -286,6 +288,7 @@ BITCOIN_CORE_H = \
   llmq/types.h \
   llmq/utils.h \
   llmq/observer/context.h \
+  llmq/observer/quorums.h \
   logging.h \
   logging/timer.h \
   mapport.h \
@@ -479,6 +482,7 @@ libbitcoin_util_a-clientversion.$(OBJEXT): obj/build.h
 libbitcoin_node_a_CPPFLAGS = $(AM_CPPFLAGS) $(BITCOIN_INCLUDES) $(BOOST_CPPFLAGS) $(MINIUPNPC_CPPFLAGS) $(NATPMP_CPPFLAGS) $(EVENT_CFLAGS) $(EVENT_PTHREADS_CFLAGS)
 libbitcoin_node_a_CXXFLAGS = $(AM_CXXFLAGS) $(PIE_FLAGS)
 libbitcoin_node_a_SOURCES = \
+  active/quorums.cpp \
   addrdb.cpp \
   addressindex.cpp \
   addrman.cpp \
@@ -549,12 +553,14 @@ libbitcoin_node_a_SOURCES = \
   llmq/net_signing.cpp \
   llmq/options.cpp \
   llmq/quorums.cpp \
+  llmq/quorumsman.cpp \
   llmq/signhash.cpp \
   llmq/signing.cpp \
   llmq/signing_shares.cpp \
   llmq/snapshot.cpp \
   llmq/utils.cpp \
   llmq/observer/context.cpp \
+  llmq/observer/quorums.cpp \
   mapport.cpp \
   masternode/active/context.cpp \
   masternode/active/notificationinterface.cpp \

src/active/quorums.cpp

@@ -0,0 +1,235 @@
+// Copyright (c) 2018-2025 The Dash Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <active/quorums.h>
+
+#include <bls/bls_ies.h>
+#include <bls/bls_worker.h>
+#include <evo/deterministicmns.h>
+#include <llmq/commitment.h>
+#include <llmq/dkgsessionmgr.h>
+#include <llmq/options.h>
+#include <llmq/quorums.h>
+#include <llmq/utils.h>
+#include <masternode/node.h>
+#include <masternode/sync.h>
+
+#include <chain.h>
+#include <chainparams.h>
+#include <logging.h>
+#include <net.h>
+#include <netmessagemaker.h>
+#include <validation.h>
+
+#include <cxxtimer.hpp>
+
+namespace llmq {
+QuorumParticipant::QuorumParticipant(CBLSWorker& bls_worker, CConnman& connman, CDeterministicMNManager& dmnman,
+                                     QuorumObserverParent& qman, CQuorumSnapshotManager& qsnapman,
+                                     const CActiveMasternodeManager& mn_activeman, const ChainstateManager& chainman,
+                                     const CMasternodeSync& mn_sync, const CSporkManager& sporkman,
+                                     const llmq::QvvecSyncModeMap& sync_map, bool quorums_recovery, bool quorums_watch) :
+    QuorumObserver(connman, dmnman, qman, qsnapman, chainman, mn_sync, sporkman, sync_map, quorums_recovery),
+    m_bls_worker{bls_worker},
+    m_mn_activeman{mn_activeman},
+    m_quorums_watch{quorums_watch}
+{
+}
+
+QuorumParticipant::~QuorumParticipant() = default;
+
+void QuorumParticipant::CheckQuorumConnections(const Consensus::LLMQParams& llmqParams,
+                                               gsl::not_null<const CBlockIndex*> pindexNew) const
+{
+    auto lastQuorums = m_qman.ScanQuorums(llmqParams.type, pindexNew, (size_t)llmqParams.keepOldConnections);
+    auto deletableQuorums = GetQuorumsToDelete(llmqParams, pindexNew);
+
+    const uint256 proTxHash = m_mn_activeman.GetProTxHash();
+    const bool watchOtherISQuorums = llmqParams.type == Params().GetConsensus().llmqTypeDIP0024InstantSend &&
+                                     ranges::any_of(lastQuorums, [&proTxHash](const auto& old_quorum){ return old_quorum->IsMember(proTxHash); });
+
+    for (const auto& quorum : lastQuorums) {
+        if (utils::EnsureQuorumConnections(llmqParams, m_connman, m_sporkman, {m_dmnman, m_qsnapman, m_chainman, quorum->m_quorum_base_block_index},
+                                           m_dmnman.GetListAtChainTip(), proTxHash, /*is_masternode=*/true, m_quorums_watch)) {
+            if (deletableQuorums.erase(quorum->qc->quorumHash) > 0) {
+                LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- llmqType[%d] h[%d] keeping mn quorum connections for quorum: [%d:%s]\n", __func__, ToUnderlying(llmqParams.type), pindexNew->nHeight, quorum->m_quorum_base_block_index->nHeight, quorum->m_quorum_base_block_index->GetBlockHash().ToString());
+            }
+        } else if (watchOtherISQuorums && !quorum->IsMember(proTxHash)) {
+            Uint256HashSet connections;
+            const auto& cindexes = utils::CalcDeterministicWatchConnections(llmqParams.type, quorum->m_quorum_base_block_index, quorum->members.size(), 1);
+            for (auto idx : cindexes) {
+                connections.emplace(quorum->members[idx]->proTxHash);
+            }
+            if (!connections.empty()) {
+                if (!m_connman.HasMasternodeQuorumNodes(llmqParams.type, quorum->m_quorum_base_block_index->GetBlockHash())) {
+                    LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- llmqType[%d] h[%d] adding mn inter-quorum connections for quorum: [%d:%s]\n", __func__, ToUnderlying(llmqParams.type), pindexNew->nHeight, quorum->m_quorum_base_block_index->nHeight, quorum->m_quorum_base_block_index->GetBlockHash().ToString());
+                    m_connman.SetMasternodeQuorumNodes(llmqParams.type, quorum->m_quorum_base_block_index->GetBlockHash(), connections);
+                    m_connman.SetMasternodeQuorumRelayMembers(llmqParams.type, quorum->m_quorum_base_block_index->GetBlockHash(), connections);
+                }
+                if (deletableQuorums.erase(quorum->qc->quorumHash) > 0) {
+                    LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- llmqType[%d] h[%d] keeping mn inter-quorum connections for quorum: [%d:%s]\n", __func__, ToUnderlying(llmqParams.type), pindexNew->nHeight, quorum->m_quorum_base_block_index->nHeight, quorum->m_quorum_base_block_index->GetBlockHash().ToString());
+                }
+            }
+        }
+    }
+
+    for (const auto& quorumHash : deletableQuorums) {
+        LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- removing masternodes quorum connections for quorum %s:\n", __func__, quorumHash.ToString());
+        m_connman.RemoveMasternodeQuorumNodes(llmqParams.type, quorumHash);
+    }
+}
+
+bool QuorumParticipant::SetQuorumSecretKeyShare(CQuorum& quorum, Span<CBLSSecretKey> skContributions) const
+{
+    return quorum.SetSecretKeyShare(m_bls_worker.AggregateSecretKeys(skContributions), m_mn_activeman.GetProTxHash());
+}
+
+size_t QuorumParticipant::GetQuorumRecoveryStartOffset(const CQuorum& quorum, gsl::not_null<const CBlockIndex*> pIndex) const
+{
+    auto mns = m_dmnman.GetListForBlock(pIndex);
+    std::vector<uint256> vecProTxHashes;
+    vecProTxHashes.reserve(mns.GetValidMNsCount());
+    mns.ForEachMN(/*onlyValid=*/true,
+                  [&](const auto& pMasternode) { vecProTxHashes.emplace_back(pMasternode.proTxHash); });
+    std::sort(vecProTxHashes.begin(), vecProTxHashes.end());
+    size_t nIndex{0};
+    {
+        auto my_protx_hash = m_mn_activeman.GetProTxHash();
+        for (const auto i : irange::range(vecProTxHashes.size())) {
+            // cppcheck-suppress useStlAlgorithm
+            if (my_protx_hash == vecProTxHashes[i]) {
+                nIndex = i;
+                break;
+            }
+        }
+    }
+    return nIndex % quorum.qc->validMembers.size();
+}
+
+MessageProcessingResult QuorumParticipant::ProcessContribQGETDATA(bool request_limit_exceeded, CDataStream& vStream,
+                                                                  const CQuorum& quorum, CQuorumDataRequest& request,
+                                                                  gsl::not_null<const CBlockIndex*> block_index)
+{
+    if (request.GetDataMask() & CQuorumDataRequest::ENCRYPTED_CONTRIBUTIONS) {
+        assert(block_index);
+
+        int memberIdx = quorum.GetMemberIndex(request.GetProTxHash());
+        if (memberIdx == -1) {
+            request.SetError(CQuorumDataRequest::Errors::MASTERNODE_IS_NO_MEMBER);
+            return request_limit_exceeded ? MisbehavingError{25, "request limit exceeded"} : MessageProcessingResult{};
+        }
+
+        std::vector<CBLSIESEncryptedObject<CBLSSecretKey>> vecEncrypted;
+        if (!m_qman.GetEncryptedContributions(request.GetLLMQType(), block_index,
+                                              quorum.qc->validMembers, request.GetProTxHash(), vecEncrypted)) {
+            request.SetError(CQuorumDataRequest::Errors::ENCRYPTED_CONTRIBUTIONS_MISSING);
+            return request_limit_exceeded ? MisbehavingError{25, "request limit exceeded"} : MessageProcessingResult{};
+        }
+
+        vStream << vecEncrypted;
+    }
+
+    return {};
+}
+
+MessageProcessingResult QuorumParticipant::ProcessContribQDATA(CNode& pfrom, CDataStream& vStream,
+                                                               CQuorum& quorum, CQuorumDataRequest& request)
+{
+    if (request.GetDataMask() & CQuorumDataRequest::ENCRYPTED_CONTRIBUTIONS) {
+        if (WITH_LOCK(quorum.cs_vvec_shShare, return !quorum.HasVerificationVectorInternal()
+                                                     || quorum.quorumVvec->size() != size_t(quorum.params.threshold))) {
+            // Don't bump score because we asked for it
+            LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- %s: No valid quorum verification vector available, from peer=%d\n", __func__, NetMsgType::QDATA, pfrom.GetId());
+            return {};
+        }
+
+        int memberIdx = quorum.GetMemberIndex(request.GetProTxHash());
+        if (memberIdx == -1) {
+            // Don't bump score because we asked for it
+            LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- %s: Not a member of the quorum, from peer=%d\n", __func__, NetMsgType::QDATA, pfrom.GetId());
+            return {};
+        }
+
+        std::vector<CBLSIESEncryptedObject<CBLSSecretKey>> vecEncrypted;
+        vStream >> vecEncrypted;
+
+        std::vector<CBLSSecretKey> vecSecretKeys;
+        vecSecretKeys.resize(vecEncrypted.size());
+        for (const auto i : irange::range(vecEncrypted.size())) {
+            if (!m_mn_activeman.Decrypt(vecEncrypted[i], memberIdx, vecSecretKeys[i], PROTOCOL_VERSION)) {
+                return MisbehavingError{10, "failed to decrypt"};
+            }
+        }
+
+        if (!quorum.SetSecretKeyShare(m_bls_worker.AggregateSecretKeys(vecSecretKeys), m_mn_activeman.GetProTxHash())) {
+            return MisbehavingError{10, "invalid secret key share received"};
+        }
+    }
+
+    return {};
+}
+
+bool QuorumParticipant::IsMasternode() const
+{
+    // We are only initialized if masternode mode is enabled
+    return true;
+}
+
+bool QuorumParticipant::IsWatching() const
+{
+    // Watch-only mode can co-exist with masternode mode
+    return m_quorums_watch;
+}
+
+void QuorumParticipant::StartDataRecoveryThread(gsl::not_null<const CBlockIndex*> pIndex, CQuorumCPtr pQuorum,
+                                                uint16_t nDataMaskIn) const
+{
+    bool expected = false;
+    if (!pQuorum->fQuorumDataRecoveryThreadRunning.compare_exchange_strong(expected, true)) {
+        LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- Already running\n", __func__);
+        return;
+    }
+
+    workerPool.push([pQuorum = std::move(pQuorum), pIndex, nDataMaskIn, this](int threadId) mutable {
+        const size_t size_offset = GetQuorumRecoveryStartOffset(*pQuorum, pIndex);
+        DataRecoveryThread(pIndex, std::move(pQuorum), nDataMaskIn, m_mn_activeman.GetProTxHash(), size_offset);
+    });
+}
+
+void QuorumParticipant::TriggerQuorumDataRecoveryThreads(gsl::not_null<const CBlockIndex*> block_index) const
+{
+    if (!m_quorums_recovery) {
+        return;
+    }
+
+    LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- Process block %s\n", __func__, block_index->GetBlockHash().ToString());
+
+    const uint256 proTxHash = m_mn_activeman.GetProTxHash();
+
+    for (const auto& params : Params().GetConsensus().llmqs) {
+        auto vecQuorums = m_qman.ScanQuorums(params.type, block_index, params.keepOldConnections);
+        const bool fWeAreQuorumTypeMember = ranges::any_of(vecQuorums, [&proTxHash](const auto& pQuorum) { return pQuorum->IsValidMember(proTxHash); });
+
+        for (auto& pQuorum : vecQuorums) {
+            if (pQuorum->IsValidMember(proTxHash)) {
+                uint16_t nDataMask{0};
+                if (!pQuorum->HasVerificationVector()) {
+                    nDataMask |= CQuorumDataRequest::QUORUM_VERIFICATION_VECTOR;
+                }
+                if (!pQuorum->GetSkShare().IsValid()) {
+                    nDataMask |= CQuorumDataRequest::ENCRYPTED_CONTRIBUTIONS;
+                }
+                if (nDataMask != 0) {
+                    StartDataRecoveryThread(block_index, std::move(pQuorum), nDataMask);
+                } else {
+                    LogPrint(BCLog::LLMQ, "QuorumParticipant::%s -- No data needed from (%d, %s) at height %d\n", __func__,
+                             ToUnderlying(pQuorum->qc->llmqType), pQuorum->qc->quorumHash.ToString(), block_index->nHeight);
+                }
+            } else {
+                TryStartVvecSyncThread(block_index, std::move(pQuorum), fWeAreQuorumTypeMember);
+            }
+        }
+    }
+}
+} // namespace llmq

src/active/quorums.h

@@ -0,0 +1,83 @@
+// Copyright (c) 2018-2025 The Dash Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_ACTIVE_QUORUMS_H
+#define BITCOIN_ACTIVE_QUORUMS_H
+
+#include <llmq/observer/quorums.h>
+#include <llmq/types.h>
+
+#include <consensus/params.h>
+#include <saltedhasher.h>
+#include <span.h>
+#include <sync.h>
+#include <threadsafety.h>
+#include <uint256.h>
+
+#include <map>
+
+class CActiveMasternodeManager;
+class CBlockIndex;
+class CBLSWorker;
+class CConnman;
+class CDeterministicMNManager;
+class CDKGSessionManager;
+class CNode;
+class CSporkManager;
+struct MessageProcessingResult;
+namespace llmq {
+class CQuorum;
+class CQuorumDataRequest;
+class CQuorumSnapshotManager;
+enum class QvvecSyncMode : int8_t;
+} // namespace llmq
+
+namespace llmq {
+class QuorumParticipant final : public QuorumObserver
+{
+private:
+    CBLSWorker& m_bls_worker;
+    const CActiveMasternodeManager& m_mn_activeman;
+    const bool m_quorums_watch{false};
+
+public:
+    QuorumParticipant() = delete;
+    QuorumParticipant(const QuorumParticipant&) = delete;
+    QuorumParticipant& operator=(const QuorumParticipant&) = delete;
+    explicit QuorumParticipant(CBLSWorker& bls_worker, CConnman& connman, CDeterministicMNManager& dmnman,
+                               QuorumObserverParent& qman, CQuorumSnapshotManager& qsnapman,
+                               const CActiveMasternodeManager& mn_activeman, const ChainstateManager& chainman,
+                               const CMasternodeSync& mn_sync, const CSporkManager& sporkman,
+                               const llmq::QvvecSyncModeMap& sync_map, bool quorums_recovery, bool quorums_watch);
+    ~QuorumParticipant();
+
+public:
+    // QuorumObserver
+    bool IsMasternode() const override;
+    bool IsWatching() const override;
+    bool SetQuorumSecretKeyShare(CQuorum& quorum, Span<CBLSSecretKey> skContributions) const override;
+    [[nodiscard]] MessageProcessingResult ProcessContribQGETDATA(bool request_limit_exceeded, CDataStream& vStream,
+                                                                 const CQuorum& quorum, CQuorumDataRequest& request,
+                                                                 gsl::not_null<const CBlockIndex*> block_index) override;
+    [[nodiscard]] MessageProcessingResult ProcessContribQDATA(CNode& pfrom, CDataStream& vStream, CQuorum& quorum,
+                                                              CQuorumDataRequest& request) override;
+
+protected:
+    // QuorumObserver
+    void CheckQuorumConnections(const Consensus::LLMQParams& llmqParams,
+                                gsl::not_null<const CBlockIndex*> pindexNew) const override;
+    void TriggerQuorumDataRecoveryThreads(gsl::not_null<const CBlockIndex*> block_index) const override;
+
+private:
+    /// Returns the start offset for the masternode with the given proTxHash. This offset is applied when picking data
+    /// recovery members of a quorum's memberlist and is calculated based on a list of all member of all active quorums
+    /// for the given llmqType in a way that each member should receive the same number of request if all active
+    /// llmqType members requests data from one llmqType quorum.
+    size_t GetQuorumRecoveryStartOffset(const CQuorum& quorum, gsl::not_null<const CBlockIndex*> pIndex) const;
+
+    void StartDataRecoveryThread(gsl::not_null<const CBlockIndex*> pIndex, CQuorumCPtr pQuorum, uint16_t nDataMaskIn) const;
+};
+} // namespace llmq
+
+#endif // BITCOIN_ACTIVE_QUORUMS_H

src/chainlock/chainlock.cpp

@@ -17,7 +17,7 @@
 #include <validationinterface.h>
 
 #include <instantsend/instantsend.h>
-#include <llmq/quorums.h>
+#include <llmq/quorumsman.h>
 #include <masternode/sync.h>
 #include <spork.h>
 #include <stats/client.h>

src/chainlock/chainlock.h

@@ -38,7 +38,7 @@ namespace llmq {
 class CInstantSendManager;
 class CQuorumManager;
 class CSigningManager;
-enum class VerifyRecSigStatus;
+enum class VerifyRecSigStatus : uint8_t;
 
 class CChainLocksHandler final : public chainlock::ChainLockSignerParent
 {