perf: deserialize DKG messages once

src/Makefile.test.include

@@ -315,6 +315,7 @@ test_fuzz_fuzz_SOURCES = \
  test/fuzz/decode_tx.cpp \
  test/fuzz/descriptor_parse.cpp \
  test/fuzz/deserialize.cpp \
+ test/fuzz/dkg_message_framing.cpp \
  test/fuzz/eval_script.cpp \
  test/fuzz/fee_rate.cpp \
  test/fuzz/fees.cpp \

src/llmq/dkgmessages.h

@@ -17,6 +17,13 @@
 #include <vector>
 
 namespace llmq {
+/**
+ * @warning The wire encodings below are walked a second time, by hand, at
+ * network intake: see CheckDKGMessageWireStructure() in llmq/net_dkg.h. That
+ * walk validates framing without materializing BLS objects, so any change to
+ * the (Un)serialize implementations in this file must be mirrored there.
+ * src/test/fuzz/dkg_message_framing.cpp guards the two against divergence.
+ */
 class CDKGContribution
 {
 public:

src/llmq/dkgsessionhandler.cpp

@@ -12,7 +12,7 @@
 namespace llmq {
 CDKGSessionHandler::CDKGSessionHandler(const Consensus::LLMQParams& _params) :
     params{_params},
-    // we allow size*2 messages as we need to make sure we see bad behavior (double messages)
+    // we allow size*2 messages per sender as we need to make sure we see bad behavior (double messages)
     pendingContributions{(size_t)_params.size * 2},
     pendingComplaints{(size_t)_params.size * 2},
     pendingJustifications{(size_t)_params.size * 2},
@@ -25,22 +25,40 @@ CDKGSessionHandler::CDKGSessionHandler(const Consensus::LLMQParams& _params) :
 
 CDKGSessionHandler::~CDKGSessionHandler() = default;
 
-void CDKGPendingMessages::PushPendingMessage(NodeId from, std::shared_ptr<CDataStream> pm, const uint256& hash)
+void CDKGPendingMessages::PushPendingMessage(NodeId from, const uint256& sender_protx,
+                                             std::shared_ptr<CDataStream> pm, const uint256& hash)
 {
     LOCK(cs_messages);
 
-    if (messagesPerNode[from] >= maxMessagesPerNode) {
-        // TODO ban?
-        LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from);
+    // Check duplicates before charging the per-sender quota so a peer that
+    // resends the same hash cannot exhaust its budget with dupes.
+    if (seenMessages.count(hash) != 0) {
+        LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from);
         return;
     }
-    messagesPerNode[from]++;
 
-    if (!seenMessages.emplace(hash).second) {
-        LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from);
+    const bool is_remote = from != -1;
+    if (is_remote) {
+        const auto sender_it = messagesPerSender.find(sender_protx);
+        if (sender_it != messagesPerSender.end() && sender_it->second >= maxMessagesPerSender) {
+            // TODO ban?
+            LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages from %s, peer=%d\n", __func__,
+                     sender_protx.ToString(), from);
+            return;
+        }
+    }
+
+    if ((is_remote && pendingRemoteMessageCount >= maxPendingRemoteMessages) ||
+        pendingMessages.size() >= maxPendingMessages) {
+        LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- pending queue full, peer=%d\n", __func__, from);
         return;
     }
+    if (is_remote) {
+        messagesPerSender[sender_protx]++;
+        pendingRemoteMessageCount++;
+    }
 
+    seenMessages.emplace(hash);
     pendingMessages.emplace_back(std::make_pair(from, std::move(pm)));
 }
 
@@ -50,25 +68,29 @@ std::list<CDKGPendingMessages::BinaryMessage> CDKGPendingMessages::PopPendingMes
 
     std::list<BinaryMessage> ret;
     while (!pendingMessages.empty() && ret.size() < maxCount) {
+        if (pendingMessages.front().first != -1) {
+            pendingRemoteMessageCount--;
+        }
         ret.emplace_back(std::move(pendingMessages.front()));
         pendingMessages.pop_front();
     }
 
     return ret;
 }
 
-bool CDKGPendingMessages::HasSeen(const uint256& hash) const
+void CDKGPendingMessages::Clear()
 {
     LOCK(cs_messages);
-    return seenMessages.count(hash) != 0;
+    pendingMessages.clear();
+    pendingRemoteMessageCount = 0;
+    messagesPerSender.clear();
+    seenMessages.clear();
 }
 
-void CDKGPendingMessages::Clear()
+bool CDKGPendingMessages::HasSeen(const uint256& hash) const
 {
     LOCK(cs_messages);
-    pendingMessages.clear();
-    messagesPerNode.clear();
-    seenMessages.clear();
+    return seenMessages.count(hash) != 0;
 }
 
 void CDKGSessionHandler::ClearPendingMessages()

src/llmq/dkgsessionhandler.h

@@ -6,18 +6,16 @@
 #define BITCOIN_LLMQ_DKGSESSIONHANDLER_H
 
 #include <net.h> // for NodeId
+#include <saltedhasher.h>
 #include <sync.h>
+#include <uint256.h>
 
 #include <list>
-#include <map>
 #include <memory>
-#include <optional>
-#include <string_view>
 #include <vector>
 
 class CDataStream;
 class CBlockIndex;
-class uint256;
 
 namespace Consensus {
 struct LLMQParams;
@@ -47,61 +45,61 @@ enum class QuorumPhase {
  * main handler thread, we push them into a CDKGPendingMessages object and later pop+deserialize them in the DKG phase
  * handler thread.
  *
- * Each message type has it's own instance of this class.
+ * Each message type has its own instance of this class.
  */
 class CDKGPendingMessages
 {
 public:
     using BinaryMessage = std::pair<NodeId, std::shared_ptr<CDataStream>>;
 
 private:
-    const size_t maxMessagesPerNode;
+    const size_t maxMessagesPerSender;
+    const size_t maxPendingRemoteMessages;
+    const size_t maxPendingMessages;
     mutable Mutex cs_messages;
     std::list<BinaryMessage> pendingMessages GUARDED_BY(cs_messages);
-    std::map<NodeId, size_t> messagesPerNode GUARDED_BY(cs_messages);
+    size_t pendingRemoteMessageCount GUARDED_BY(cs_messages){0};
+    // Keyed by the sender's MNAuth-verified proTxHash, not by NodeId: a peer that
+    // reconnects gets a fresh NodeId but keeps the same proTxHash, so its quota
+    // survives the reconnect instead of being reset.
+    //
+    // Entries deliberately live for the whole round rather than being released on
+    // pop: the quota is cumulative, so that a sender cannot regain retention slots
+    // simply by waiting for the worker to drain the queue. Size is therefore not
+    // bounded by the queue caps but by the number of distinct senders that get a
+    // message accepted in one round, which MNAuth pins to the registered
+    // masternode set (see CMNAuth::ProcessMessage: the proTxHash must resolve in
+    // the deterministic MN list and carry a valid operator-key signature).
+    Uint256HashMap<size_t> messagesPerSender GUARDED_BY(cs_messages);
     Uint256HashSet seenMessages GUARDED_BY(cs_messages);
 
 public:
-    explicit CDKGPendingMessages(size_t _maxMessagesPerNode) :
-        maxMessagesPerNode(_maxMessagesPerNode) {};
+    explicit CDKGPendingMessages(size_t _maxMessagesPerSender) :
+        maxMessagesPerSender(_maxMessagesPerSender),
+        // Belt-and-braces bound on live queue occupancy. The per-sender quota is
+        // the primary limit; this only caps the total across distinct senders.
+        maxPendingRemoteMessages(_maxMessagesPerSender * 2),
+        // Reserve one slot for the message produced by this node during the
+        // matching phase.
+        maxPendingMessages(maxPendingRemoteMessages + 1)
+    {
+    }
 
     /**
      * Enqueue a serialized DKG message under @p from with content hash @p hash.
+     * @p sender_protx is the sender's MNAuth-verified proTxHash and keys the
+     * per-sender quota; pass a null hash for messages this node produced itself
+     * (@p from == -1), which are exempt from that quota.
      * Caller is responsible for hashing the payload and (for real peers)
      * routing the erase-request to PeerManager. Drops the message silently on
-     * per-node capacity overflow or duplicate hash.
+     * per-sender or queue-wide capacity overflow, or duplicate hash.
      */
-    void PushPendingMessage(NodeId from, std::shared_ptr<CDataStream> pm, const uint256& hash)
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
+    void PushPendingMessage(NodeId from, const uint256& sender_protx, std::shared_ptr<CDataStream> pm,
+                            const uint256& hash) EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
 
     std::list<BinaryMessage> PopPendingMessages(size_t maxCount) EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
     bool HasSeen(const uint256& hash) const EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
     void Clear() EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
-
-    // Might return nullptr messages, which indicates that deserialization failed for some reason
-    template <typename Message>
-    std::vector<std::pair<NodeId, std::shared_ptr<Message>>> PopAndDeserializeMessages(size_t maxCount)
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_messages)
-    {
-        auto binaryMessages = PopPendingMessages(maxCount);
-        if (binaryMessages.empty()) {
-            return {};
-        }
-
-        std::vector<std::pair<NodeId, std::shared_ptr<Message>>> ret;
-        ret.reserve(binaryMessages.size());
-        for (const auto& bm : binaryMessages) {
-            auto msg = std::make_shared<Message>();
-            try {
-                *bm.second >> *msg;
-            } catch (...) {
-                msg = nullptr;
-            }
-            ret.emplace_back(std::make_pair(bm.first, std::move(msg)));
-        }
-
-        return ret;
-    }
 };
 
 /**