refactor: migrate CDKGSessionManager::ProcessMessage() and friends

kwvg · kwvg · commit ef4a0bba1e5f · 2025-08-15T17:41:02.000Z
diff --git a/src/llmq/dkgsessionhandler.cpp b/src/llmq/dkgsessionhandler.cpp
@@ -58,7 +58,7 @@ CDKGSessionHandler::CDKGSessionHandler(CBLSWorker& _blsWorker, CChainState& chai
 
 CDKGSessionHandler::~CDKGSessionHandler() = default;
 
-void CDKGPendingMessages::PushPendingMessage(NodeId from, CDataStream& vRecv, PeerManager& peerman)
+MessageProcessingResult CDKGPendingMessages::PushPendingMessage(NodeId from, CDataStream& vRecv)
 {
     // this will also consume the data, even if we bail out early
     auto pm = std::make_shared<CDataStream>(std::move(vRecv));
@@ -67,25 +67,27 @@ void CDKGPendingMessages::PushPendingMessage(NodeId from, CDataStream& vRecv, Pe
     hw.write(AsWritableBytes(Span{*pm}));
     uint256 hash = hw.GetHash();
 
+    MessageProcessingResult ret{};
     if (from != -1) {
-        WITH_LOCK(::cs_main, peerman.EraseObjectRequest(from, CInv(invType, hash)));
+        ret.m_to_erase = CInv{invType, hash};
     }
 
     LOCK(cs_messages);
 
     if (messagesPerNode[from] >= maxMessagesPerNode) {
         // TODO ban?
         LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from);
-        return;
+        return ret;
     }
     messagesPerNode[from]++;
 
     if (!seenMessages.emplace(hash).second) {
         LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from);
-        return;
+        return ret;
     }
 
     pendingMessages.emplace_back(std::make_pair(from, std::move(pm)));
+    return ret;
 }
 
 std::list<CDKGPendingMessages::BinaryMessage> CDKGPendingMessages::PopPendingMessages(size_t maxCount)
@@ -150,19 +152,19 @@ void CDKGSessionHandler::UpdatedBlockTip(const CBlockIndex* pindexNew)
              params.name, quorumIndex, currentHeight, pQuorumBaseBlockIndex->nHeight, ToUnderlying(oldPhase), ToUnderlying(phase));
 }
 
-void CDKGSessionHandler::ProcessMessage(const CNode& pfrom, PeerManager& peerman, const std::string& msg_type,
-                                        CDataStream& vRecv)
+MessageProcessingResult CDKGSessionHandler::ProcessMessage(NodeId from, std::string_view msg_type, CDataStream& vRecv)
 {
     // We don't handle messages in the calling thread as deserialization/processing of these would block everything
     if (msg_type == NetMsgType::QCONTRIB) {
-        pendingContributions.PushPendingMessage(pfrom.GetId(), vRecv, peerman);
+        return pendingContributions.PushPendingMessage(from, vRecv);
     } else if (msg_type == NetMsgType::QCOMPLAINT) {
-        pendingComplaints.PushPendingMessage(pfrom.GetId(), vRecv, peerman);
+        return pendingComplaints.PushPendingMessage(from, vRecv);
     } else if (msg_type == NetMsgType::QJUSTIFICATION) {
-        pendingJustifications.PushPendingMessage(pfrom.GetId(), vRecv, peerman);
+        return pendingJustifications.PushPendingMessage(from, vRecv);
     } else if (msg_type == NetMsgType::QPCOMMITMENT) {
-        pendingPrematureCommitments.PushPendingMessage(pfrom.GetId(), vRecv, peerman);
+        return pendingPrematureCommitments.PushPendingMessage(from, vRecv);
     }
+    return {};
 }
 
 void CDKGSessionHandler::StartThread(CConnman& connman, PeerManager& peerman)
diff --git a/src/llmq/dkgsessionhandler.h b/src/llmq/dkgsessionhandler.h
@@ -6,6 +6,7 @@
 #define BITCOIN_LLMQ_DKGSESSIONHANDLER_H
 
 #include <net.h> // for NodeId
+#include <net_processing.h>
 
 #include <atomic>
 #include <list>
@@ -14,6 +15,7 @@
 #include <optional>
 #include <set>
 #include <string>
+#include <string_view>
 #include <thread>
 #include <vector>
 
@@ -24,7 +26,6 @@ class CChainState;
 class CConnman;
 class CDeterministicMNManager;
 class CMasternodeMetaMan;
-class CNode;
 class CSporkManager;
 class PeerManager;
 
@@ -64,18 +65,18 @@ class CDKGPendingMessages
     using BinaryMessage = std::pair<NodeId, std::shared_ptr<CDataStream>>;
 
 private:
-    const int invType;
+    const uint32_t invType;
     const size_t maxMessagesPerNode;
     mutable Mutex cs_messages;
     std::list<BinaryMessage> pendingMessages GUARDED_BY(cs_messages);
     std::map<NodeId, size_t> messagesPerNode GUARDED_BY(cs_messages);
     std::set<uint256> seenMessages GUARDED_BY(cs_messages);
 
 public:
-    explicit CDKGPendingMessages(size_t _maxMessagesPerNode, int _invType) :
+    explicit CDKGPendingMessages(size_t _maxMessagesPerNode, uint32_t _invType) :
             invType(_invType), maxMessagesPerNode(_maxMessagesPerNode) {};
 
-    void PushPendingMessage(NodeId from, CDataStream& vRecv, PeerManager& peerman);
+    [[nodiscard]] MessageProcessingResult PushPendingMessage(NodeId from, CDataStream& vRecv);
     std::list<BinaryMessage> PopPendingMessages(size_t maxCount);
     bool HasSeen(const uint256& hash) const;
     void Misbehaving(NodeId from, int score, PeerManager& peerman);
@@ -86,7 +87,7 @@ class CDKGPendingMessages
     {
         CDataStream ds(SER_NETWORK, PROTOCOL_VERSION);
         ds << msg;
-        PushPendingMessage(from, ds, peerman);
+        peerman.PostProcessMessage(PushPendingMessage(from, ds), from);
     }
 
     // Might return nullptr messages, which indicates that deserialization failed for some reason
@@ -165,7 +166,7 @@ class CDKGSessionHandler
     ~CDKGSessionHandler();
 
     void UpdatedBlockTip(const CBlockIndex *pindexNew);
-    void ProcessMessage(const CNode& pfrom, PeerManager& peerman, const std::string& msg_type, CDataStream& vRecv);
+    [[nodiscard]] MessageProcessingResult ProcessMessage(NodeId from, std::string_view msg_type, CDataStream& vRecv);
 
     void StartThread(CConnman& connman, PeerManager& peerman);
     void StopThread();
diff --git a/src/llmq/dkgsessionmgr.cpp b/src/llmq/dkgsessionmgr.cpp
@@ -92,8 +92,8 @@ void CDKGSessionManager::UpdatedBlockTip(const CBlockIndex* pindexNew, bool fIni
     }
 }
 
-PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman, bool is_masternode,
-                                              const std::string& msg_type, CDataStream& vRecv)
+MessageProcessingResult CDKGSessionManager::ProcessMessage(CNode& pfrom, bool is_masternode, std::string_view msg_type,
+                                                           CDataStream& vRecv)
 {
     static Mutex cs_indexedQuorumsCache;
     static std::map<Consensus::LLMQType, unordered_lru_cache<uint256, int, StaticSaltedHasher>> indexedQuorumsCache GUARDED_BY(cs_indexedQuorumsCache);
@@ -112,19 +112,19 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman
     if (msg_type == NetMsgType::QWATCH) {
         if (!is_masternode) {
             // non-masternodes should never receive this
-            return tl::unexpected{10};
+            return MisbehavingError{10};
         }
         pfrom.qwatch = true;
         return {};
     }
 
     if ((!is_masternode && !IsWatchQuorumsEnabled())) {
         // regular non-watching nodes should never receive any of these
-        return tl::unexpected{10};
+        return MisbehavingError{10};
     }
 
     if (vRecv.empty()) {
-        return tl::unexpected{100};
+        return MisbehavingError{100};
     }
 
     Consensus::LLMQType llmqType;
@@ -137,7 +137,7 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman
     const auto& llmq_params_opt = Params().GetLLMQ(llmqType);
     if (!llmq_params_opt.has_value()) {
         LogPrintf("CDKGSessionManager -- invalid llmqType [%d]\n", ToUnderlying(llmqType));
-        return tl::unexpected{100};
+        return MisbehavingError{100};
     }
     const auto& llmq_params = llmq_params_opt.value();
 
@@ -158,12 +158,12 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman
         if (pQuorumBaseBlockIndex == nullptr) {
             LogPrintf("CDKGSessionManager -- unknown quorumHash %s\n", quorumHash.ToString());
             // NOTE: do not insta-ban for this, we might be lagging behind
-            return tl::unexpected{10};
+            return MisbehavingError{10};
         }
 
         if (!IsQuorumTypeEnabled(llmqType, pQuorumBaseBlockIndex->pprev)) {
             LogPrintf("CDKGSessionManager -- llmqType [%d] quorums aren't active\n", ToUnderlying(llmqType));
-            return tl::unexpected{100};
+            return MisbehavingError{100};
         }
 
         quorumIndex = pQuorumBaseBlockIndex->nHeight % llmq_params.dkgInterval;
@@ -172,19 +172,18 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman
 
         if (quorumIndex > quorumIndexMax) {
             LogPrintf("CDKGSessionManager -- invalid quorumHash %s\n", quorumHash.ToString());
-            return tl::unexpected{100};
+            return MisbehavingError{100};
         }
 
         if (!dkgSessionHandlers.count(std::make_pair(llmqType, quorumIndex))) {
             LogPrintf("CDKGSessionManager -- no session handlers for quorumIndex [%d]\n", quorumIndex);
-            return tl::unexpected{100};
+            return MisbehavingError{100};
         }
     }
 
     assert(quorumIndex != -1);
     WITH_LOCK(cs_indexedQuorumsCache, indexedQuorumsCache[llmqType].insert(quorumHash, quorumIndex));
-    dkgSessionHandlers.at(std::make_pair(llmqType, quorumIndex)).ProcessMessage(pfrom, peerman, msg_type, vRecv);
-    return {};
+    return dkgSessionHandlers.at(std::make_pair(llmqType, quorumIndex)).ProcessMessage(pfrom.GetId(), msg_type, vRecv);
 }
 
 bool CDKGSessionManager::AlreadyHave(const CInv& inv) const
diff --git a/src/llmq/dkgsessionmgr.h b/src/llmq/dkgsessionmgr.h
@@ -12,6 +12,7 @@
 
 #include <map>
 #include <memory>
+#include <string_view>
 
 template <class T>
 class CBLSIESMultiRecipientObjects;
@@ -88,8 +89,8 @@ class CDKGSessionManager
 
     void UpdatedBlockTip(const CBlockIndex *pindexNew, bool fInitialDownload);
 
-    PeerMsgRet ProcessMessage(CNode& pfrom, PeerManager& peerman, bool is_masternode, const std::string& msg_type,
-                              CDataStream& vRecv);
+    [[nodiscard]] MessageProcessingResult ProcessMessage(CNode& pfrom, bool is_masternode, std::string_view msg_type,
+                                                         CDataStream& vRecv);
     bool AlreadyHave(const CInv& inv) const;
     bool GetContribution(const uint256& hash, CDKGContribution& ret) const;
     bool GetComplaint(const uint256& hash, CDKGComplaint& ret) const;
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
@@ -5273,7 +5273,7 @@ void PeerManagerImpl::ProcessMessage(
         ProcessPeerMsgRet(m_govman.ProcessMessage(pfrom, m_connman, *this, msg_type, vRecv), pfrom);
         ProcessPeerMsgRet(CMNAuth::ProcessMessage(pfrom, peer->m_their_services, m_connman, m_mn_metaman, m_mn_activeman, m_mn_sync, m_dmnman->GetListAtChainTip(), msg_type, vRecv), pfrom);
         PostProcessMessage(m_llmq_ctx->quorum_block_processor->ProcessMessage(pfrom, msg_type, vRecv), pfrom.GetId());
-        ProcessPeerMsgRet(m_llmq_ctx->qdkgsman->ProcessMessage(pfrom, *this, is_masternode, msg_type, vRecv), pfrom);
+        PostProcessMessage(m_llmq_ctx->qdkgsman->ProcessMessage(pfrom, is_masternode, msg_type, vRecv), pfrom.GetId());
         PostProcessMessage(m_llmq_ctx->qman->ProcessMessage(pfrom, m_connman, msg_type, vRecv), pfrom.GetId());
         m_llmq_ctx->shareman->ProcessMessage(pfrom, *this, m_sporkman, msg_type, vRecv);
         PostProcessMessage(m_llmq_ctx->sigman->ProcessMessage(pfrom.GetId(), msg_type, vRecv), pfrom.GetId());

Original file line number	Diff line number	Diff line change
`@@ -92,8 +92,8 @@ void CDKGSessionManager::UpdatedBlockTip(const CBlockIndex* pindexNew, bool fIni`
`92`	`92`	`}`
`93`	`93`	`}`
`94`	`94`
`95`		`-PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman, bool is_masternode,`
`96`		`- const std::string& msg_type, CDataStream& vRecv)`
	`95`	`+MessageProcessingResult CDKGSessionManager::ProcessMessage(CNode& pfrom, bool is_masternode, std::string_view msg_type,`
	`96`	`+ CDataStream& vRecv)`
`97`	`97`	`{`
`98`	`98`	`static Mutex cs_indexedQuorumsCache;`
`99`	`99`	`static std::map<Consensus::LLMQType, unordered_lru_cache<uint256, int, StaticSaltedHasher>> indexedQuorumsCache GUARDED_BY(cs_indexedQuorumsCache);`
`@@ -112,19 +112,19 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman`
`112`	`112`	`if (msg_type == NetMsgType::QWATCH) {`
`113`	`113`	`if (!is_masternode) {`
`114`	`114`	`// non-masternodes should never receive this`
`115`		`- return tl::unexpected{10};`
	`115`	`+ return MisbehavingError{10};`
`116`	`116`	`}`
`117`	`117`	`pfrom.qwatch = true;`
`118`	`118`	`return {};`
`119`	`119`	`}`
`120`	`120`
`121`	`121`	`if ((!is_masternode && !IsWatchQuorumsEnabled())) {`
`122`	`122`	`// regular non-watching nodes should never receive any of these`
`123`		`- return tl::unexpected{10};`
	`123`	`+ return MisbehavingError{10};`
`124`	`124`	`}`
`125`	`125`
`126`	`126`	`if (vRecv.empty()) {`
`127`		`- return tl::unexpected{100};`
	`127`	`+ return MisbehavingError{100};`
`128`	`128`	`}`
`129`	`129`
`130`	`130`	`Consensus::LLMQType llmqType;`
`@@ -137,7 +137,7 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman`
`137`	`137`	`const auto& llmq_params_opt = Params().GetLLMQ(llmqType);`
`138`	`138`	`if (!llmq_params_opt.has_value()) {`
`139`	`139`	`LogPrintf("CDKGSessionManager -- invalid llmqType [%d]\n", ToUnderlying(llmqType));`
`140`		`- return tl::unexpected{100};`
	`140`	`+ return MisbehavingError{100};`
`141`	`141`	`}`
`142`	`142`	`const auto& llmq_params = llmq_params_opt.value();`
`143`	`143`
`@@ -158,12 +158,12 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman`
`158`	`158`	`if (pQuorumBaseBlockIndex == nullptr) {`
`159`	`159`	`LogPrintf("CDKGSessionManager -- unknown quorumHash %s\n", quorumHash.ToString());`
`160`	`160`	`// NOTE: do not insta-ban for this, we might be lagging behind`
`161`		`- return tl::unexpected{10};`
	`161`	`+ return MisbehavingError{10};`
`162`	`162`	`}`
`163`	`163`
`164`	`164`	`if (!IsQuorumTypeEnabled(llmqType, pQuorumBaseBlockIndex->pprev)) {`
`165`	`165`	`LogPrintf("CDKGSessionManager -- llmqType [%d] quorums aren't active\n", ToUnderlying(llmqType));`
`166`		`- return tl::unexpected{100};`
	`166`	`+ return MisbehavingError{100};`
`167`	`167`	`}`
`168`	`168`
`169`	`169`	`quorumIndex = pQuorumBaseBlockIndex->nHeight % llmq_params.dkgInterval;`
`@@ -172,19 +172,18 @@ PeerMsgRet CDKGSessionManager::ProcessMessage(CNode& pfrom, PeerManager& peerman`
`172`	`172`
`173`	`173`	`if (quorumIndex > quorumIndexMax) {`
`174`	`174`	`LogPrintf("CDKGSessionManager -- invalid quorumHash %s\n", quorumHash.ToString());`
`175`		`- return tl::unexpected{100};`
	`175`	`+ return MisbehavingError{100};`
`176`	`176`	`}`
`177`	`177`
`178`	`178`	`if (!dkgSessionHandlers.count(std::make_pair(llmqType, quorumIndex))) {`
`179`	`179`	`LogPrintf("CDKGSessionManager -- no session handlers for quorumIndex [%d]\n", quorumIndex);`
`180`		`- return tl::unexpected{100};`
	`180`	`+ return MisbehavingError{100};`
`181`	`181`	`}`
`182`	`182`	`}`
`183`	`183`
`184`	`184`	`assert(quorumIndex != -1);`
`185`	`185`	`WITH_LOCK(cs_indexedQuorumsCache, indexedQuorumsCache[llmqType].insert(quorumHash, quorumIndex));`
`186`		`- dkgSessionHandlers.at(std::make_pair(llmqType, quorumIndex)).ProcessMessage(pfrom, peerman, msg_type, vRecv);`
`187`		`- return {};`
	`186`	`+ return dkgSessionHandlers.at(std::make_pair(llmqType, quorumIndex)).ProcessMessage(pfrom.GetId(), msg_type, vRecv);`
`188`	`187`	`}`
`189`	`188`
`190`	`189`	`bool CDKGSessionManager::AlreadyHave(const CInv& inv) const`