fix: bound DKG pending message queue across NodeId reconnects

CDKGPendingMessages bounded retention only per NodeId, an ephemeral per-connection counter, and never erased the entry. A peer that disconnects and reconnects therefore got a fresh quota while its previously queued payloads and seen-hashes stayed resident, so the queue grew linearly in the number of connections with no global ceiling. For llmq_400_85 that is roughly 56 MB per connection cycle, repeatable indefinitely. Victims are not limited to masternodes: in observer mode the handler still enqueues, never pops and never clears, so retention there is permanent.

Add a global cap across all NodeIds and release a disconnecting peer's payloads and seen-hashes via a finalize hook. Filed as a draft: the cap is still message-count-based rather than byte-based, and EvictGreediestNode runs before the duplicate check, so a malicious masternode can replay seen hashes to evict honest peers' queued messages without occupying a slot itself. Both are unresolved.

Author: PastaPastaPasta

src/Makefile.test.include

@@ -137,6 +137,7 @@ BITCOIN_TESTS =\
   test/limitedmap_tests.cpp \
   test/llmq_blockprocessor_tests.cpp \
   test/llmq_dkg_tests.cpp \
+  test/llmq_dkg_pending_tests.cpp \
   test/llmq_chainlock_tests.cpp \
   test/llmq_commitment_tests.cpp \
   test/llmq_hash_tests.cpp \

src/llmq/dkgsessionhandler.cpp

@@ -5,8 +5,10 @@
 #include <llmq/dkgsessionhandler.h>
 
 #include <logging.h>
+#include <streams.h>
 #include <uint256.h>
 
+#include <optional>
 #include <stdexcept>
 
 namespace llmq {
@@ -25,23 +27,82 @@ CDKGSessionHandler::CDKGSessionHandler(const Consensus::LLMQParams& _params) :
 
 CDKGSessionHandler::~CDKGSessionHandler() = default;
 
+std::list<CDKGPendingMessages::PendingMessage>::iterator CDKGPendingMessages::EraseEntry(
+    std::list<PendingMessage>::iterator it)
+{
+    seenMessages.erase(it->hash);
+    if (auto qit = queuedPerNode.find(it->from); qit != queuedPerNode.end() && --qit->second == 0) {
+        queuedPerNode.erase(qit);
+    }
+    return pendingMessages.erase(it);
+}
+
+bool CDKGPendingMessages::EvictGreediestNode()
+{
+    // Pick the peer holding the most slots, so a flooder is always its own
+    // victim: a peer that queued one message can only be evicted once every
+    // other peer is down to one as well.
+    std::optional<NodeId> victim;
+    size_t victim_count{0};
+    for (const auto& [node, count] : queuedPerNode) {
+        if (node < 0) continue; // never evict our own messages
+        if (count > victim_count) {
+            victim = node;
+            victim_count = count;
+        }
+    }
+    if (!victim.has_value()) return false;
+
+    // Drop that peer's oldest message: it is the least likely to still be
+    // relevant to the current phase.
+    for (auto it = pendingMessages.begin(); it != pendingMessages.end(); ++it) {
+        if (it->from == *victim) {
+            LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- queue full (%d), evicting oldest of peer=%d (%d queued)\n",
+                     __func__, maxMessagesTotal, *victim, victim_count);
+            EraseEntry(it);
+            return true;
+        }
+    }
+    return false;
+}
+
 void CDKGPendingMessages::PushPendingMessage(NodeId from, std::shared_ptr<CDataStream> pm, const uint256& hash)
 {
     LOCK(cs_messages);
 
-    if (messagesPerNode[from] >= maxMessagesPerNode) {
+    // Our own messages (from < 0) are produced by our phase handler, are a
+    // handful per round, and are the only way our contribution reaches the
+    // quorum. They must never be dropped by a peer-driven bound.
+    const bool is_own = from < 0;
+
+    if (!is_own && messagesPerNode[from] >= maxMessagesPerNode) {
         // TODO ban?
         LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from);
         return;
     }
-    messagesPerNode[from]++;
+
+    // Global bound: NodeId is ephemeral (reconnect => fresh id => fresh per-node
+    // quota), so a per-node-only cap cannot bound total queue memory. Enforce it
+    // by evicting the greediest peer rather than rejecting the newest message,
+    // so one flooding peer cannot starve the DKG for honest members.
+    while (pendingMessages.size() >= maxMessagesTotal) {
+        if (!EvictGreediestNode()) {
+            // Only own messages left; they are bounded by our own phase handler.
+            break;
+        }
+    }
+
+    if (!is_own) {
+        messagesPerNode[from]++;
+    }
 
     if (!seenMessages.emplace(hash).second) {
         LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from);
         return;
     }
 
-    pendingMessages.emplace_back(std::make_pair(from, std::move(pm)));
+    pendingMessages.emplace_back(PendingMessage{from, std::move(pm), hash});
+    queuedPerNode[from]++;
 }
 
 std::list<CDKGPendingMessages::BinaryMessage> CDKGPendingMessages::PopPendingMessages(size_t maxCount)
@@ -50,7 +111,13 @@ std::list<CDKGPendingMessages::BinaryMessage> CDKGPendingMessages::PopPendingMes
 
     std::list<BinaryMessage> ret;
     while (!pendingMessages.empty() && ret.size() < maxCount) {
-        ret.emplace_back(std::move(pendingMessages.front()));
+        auto& front = pendingMessages.front();
+        ret.emplace_back(front.from, std::move(front.msg));
+        // Popped messages are handed to the DKG session; their content hash stays
+        // in seenMessages so AlreadyHave() keeps suppressing re-requests.
+        if (auto qit = queuedPerNode.find(front.from); qit != queuedPerNode.end() && --qit->second == 0) {
+            queuedPerNode.erase(qit);
+        }
         pendingMessages.pop_front();
     }
 
@@ -68,14 +135,59 @@ void CDKGPendingMessages::Clear()
     LOCK(cs_messages);
     pendingMessages.clear();
     messagesPerNode.clear();
+    queuedPerNode.clear();
     seenMessages.clear();
 }
 
+void CDKGPendingMessages::RemoveNode(NodeId nodeId)
+{
+    // Own/local enqueues use from=-1 and are not tied to a peer disconnect.
+    if (nodeId < 0) {
+        return;
+    }
+
+    LOCK(cs_messages);
+    messagesPerNode.erase(nodeId);
+
+    // Runs under ::cs_main (via PeerManagerImpl::FinalizeNode), so skip the list
+    // scan entirely for the overwhelmingly common case of a peer that never
+    // queued a DKG message.
+    if (queuedPerNode.find(nodeId) == queuedPerNode.end()) {
+        return;
+    }
+
+    for (auto it = pendingMessages.begin(); it != pendingMessages.end();) {
+        if (it->from == nodeId) {
+            // Free the content-hash slot too; otherwise a reconnecting attacker
+            // can grow seenMessages without bound even after payloads are dropped
+            // (especially in observer mode where Clear() never runs). The hash is
+            // stored alongside the payload, so no re-hashing happens here.
+            it = EraseEntry(it);
+        } else {
+            ++it;
+        }
+    }
+}
+
+size_t CDKGPendingMessages::Size() const
+{
+    LOCK(cs_messages);
+    return pendingMessages.size();
+}
+
 void CDKGSessionHandler::ClearPendingMessages()
 {
     pendingContributions.Clear();
     pendingComplaints.Clear();
     pendingJustifications.Clear();
     pendingPrematureCommitments.Clear();
 }
+
+void CDKGSessionHandler::RemoveNode(NodeId nodeId)
+{
+    pendingContributions.RemoveNode(nodeId);
+    pendingComplaints.RemoveNode(nodeId);
+    pendingJustifications.RemoveNode(nodeId);
+    pendingPrematureCommitments.RemoveNode(nodeId);
+}
 } // namespace llmq

src/llmq/dkgsessionhandler.h

@@ -7,6 +7,7 @@
 
 #include <net.h> // for NodeId
 #include <sync.h>
+#include <uint256.h> // PendingMessage stores a uint256 by value
 
 #include <list>
 #include <map>
@@ -17,7 +18,6 @@
 
 class CDataStream;
 class CBlockIndex;
-class uint256;
 
 namespace Consensus {
 struct LLMQParams;
@@ -48,28 +48,79 @@ enum class QuorumPhase {
  * handler thread.
  *
  * Each message type has it's own instance of this class.
+ *
+ * Retention is bounded both per NodeId (@ref maxMessagesPerNode) and globally
+ * (@ref maxMessagesTotal = maxMessagesPerNode * MAX_PENDING_NODES_FACTOR). The
+ * global bound is required because NodeId is ephemeral: a reconnecting peer
+ * receives a fresh id and would otherwise obtain a fresh full per-node quota
+ * forever. @ref RemoveNode drops a disconnected peer's counter and still-queued
+ * payloads so live peers can use the capacity.
+ *
+ * The global bound must not become a way for one peer to stall the DKG for
+ * everyone, so it is enforced by evicting the peer that currently occupies the
+ * most slots rather than by rejecting whoever pushes next: a flooder is always
+ * its own victim and honest members keep their place in the queue. Locally
+ * generated messages (@c from < 0) bypass the bound entirely — they are produced
+ * by our own phase handler, are a handful per round, and are the only path by
+ * which our own contribution reaches the quorum.
  */
 class CDKGPendingMessages
 {
 public:
     using BinaryMessage = std::pair<NodeId, std::shared_ptr<CDataStream>>;
 
+    //! Cap total retained messages at this multiple of the per-node limit. A
+    //! legitimate round needs at most maxMessagesPerNode messages in total (one
+    //! per member, doubled to catch double-sends), so this leaves generous
+    //! headroom for honest traffic while still bounding what peers can pin.
+    static constexpr size_t MAX_PENDING_NODES_FACTOR = 8;
+
 private:
+    struct PendingMessage {
+        NodeId from;
+        std::shared_ptr<CDataStream> msg;
+        //! Content hash as inserted into @ref seenMessages. Stored so eviction
+        //! and disconnect cleanup never have to re-hash the payload (which would
+        //! put unbounded SHA256 work under ::cs_main via FinalizeNode).
+        uint256 hash;
+    };
+
     const size_t maxMessagesPerNode;
+    const size_t maxMessagesTotal;
     mutable Mutex cs_messages;
-    std::list<BinaryMessage> pendingMessages GUARDED_BY(cs_messages);
+    std::list<PendingMessage> pendingMessages GUARDED_BY(cs_messages);
+    //! Per-connection lifetime quota counter. Deliberately not decremented when a
+    //! message is popped: the quota is per connection, not per queue slot, so a
+    //! peer cannot regain capacity by getting its messages processed. Released
+    //! only by @ref RemoveNode and @ref Clear.
     std::map<NodeId, size_t> messagesPerNode GUARDED_BY(cs_messages);
+    //! Live queue occupancy per NodeId, kept in sync with @ref pendingMessages.
+    //! Used to pick the greediest peer on overflow and to skip the list scan in
+    //! @ref RemoveNode when the disconnecting peer has nothing queued.
+    std::map<NodeId, size_t> queuedPerNode GUARDED_BY(cs_messages);
     Uint256HashSet seenMessages GUARDED_BY(cs_messages);
 
+    //! Erase one queue entry, keeping queuedPerNode and seenMessages in sync.
+    //! Returns the iterator following the erased entry.
+    std::list<PendingMessage>::iterator EraseEntry(std::list<PendingMessage>::iterator it)
+        EXCLUSIVE_LOCKS_REQUIRED(cs_messages);
+    //! Drop the oldest queued message of the peer occupying the most slots.
+    //! Never evicts our own (from < 0) messages. Returns false if nothing could
+    //! be evicted (i.e. the queue holds only own messages).
+    bool EvictGreediestNode() EXCLUSIVE_LOCKS_REQUIRED(cs_messages);
+
 public:
     explicit CDKGPendingMessages(size_t _maxMessagesPerNode) :
-        maxMessagesPerNode(_maxMessagesPerNode) {};
+        maxMessagesPerNode(_maxMessagesPerNode),
+        maxMessagesTotal(_maxMessagesPerNode * MAX_PENDING_NODES_FACTOR) {};
 
     /**
      * Enqueue a serialized DKG message under @p from with content hash @p hash.
      * 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-node capacity overflow or duplicate hash. On global capacity overflow
+     * the greediest *other* peer is evicted instead (see the class comment), so
+     * a flooder cannot deny the queue to honest members.
      */
     void PushPendingMessage(NodeId from, std::shared_ptr<CDataStream> pm, const uint256& hash)
         EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
@@ -78,6 +129,22 @@ class CDKGPendingMessages
     bool HasSeen(const uint256& hash) const EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
     void Clear() EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
 
+    /**
+     * Drop @p nodeId's per-node counter, any still-queued payloads from that
+     * peer, and the corresponding content hashes in @ref seenMessages. Called
+     * on disconnect so a reconnecting attacker cannot pin memory (or grow
+     * seenMessages) under abandoned NodeIds.
+     *
+     * Only *un-processed* messages are dropped. Anything already popped by the
+     * phase handler has been accepted into the DKG session state and is
+     * unaffected, so disconnecting a peer never invalidates a contribution the
+     * session already took — the DKG completes as before.
+     */
+    void RemoveNode(NodeId nodeId) EXCLUSIVE_LOCKS_REQUIRED(!cs_messages);
+
+    //! Number of still-queued binary messages (for tests / diagnostics).
+    size_t Size() const 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)
@@ -123,6 +190,7 @@ class CDKGSessionHandler
     virtual ~CDKGSessionHandler();
 
     void ClearPendingMessages();
+    void RemoveNode(NodeId nodeId);
 
 public:
     virtual bool GetContribution(const uint256& hash, CDKGContribution& ret) const { return false; }

src/llmq/net_dkg.cpp

@@ -595,6 +595,16 @@ bool NetDKG::ProcessGetData(CNode& pfrom, const CInv& inv, CConnman& connman, co
     return false;
 }
 
+void NetDKG::FinalizeNode(NodeId nodeid)
+{
+    // Release per-NodeId DKG queue state on disconnect. Required for both
+    // active and observer mode: observer mode never runs ClearPendingMessages,
+    // and active mode only clears at the start of the next DKG round.
+    m_qdkgsman.ForEachHandler([nodeid](CDKGSessionHandler& handler) {
+        handler.RemoveNode(nodeid);
+    });
+}
+
 void NetDKG::Start()
 {
     if (m_active == nullptr) return;

src/llmq/net_dkg.h

@@ -70,6 +70,7 @@ class NetDKG final : public NetHandler
         EXCLUSIVE_LOCKS_REQUIRED(!cs_indexed_quorums_cache);
     bool AlreadyHave(const CInv& inv) override;
     bool ProcessGetData(CNode& pfrom, const CInv& inv, CConnman& connman, const CNetMsgMaker& msgMaker) override;
+    void FinalizeNode(NodeId nodeid) override;
     /**
      * Drives one phase-handler thread per ActiveDKGSessionHandler in active mode;
      * no-op in observer mode (no curSession to drive).

src/net_processing.cpp

@@ -1795,6 +1795,13 @@ void PeerManagerImpl::FinalizeNode(const CNode& node) {
         m_addrman.Connected(node.addr);
     }
 
+    // Let protocol handlers free any state keyed on the ephemeral NodeId.
+    // Still under cs_main (LOCK scope is the whole function); handlers must only
+    // take their own locks and must not re-enter PeerManager.
+    for (const auto& handler : m_handlers) {
+        handler->FinalizeNode(nodeid);
+    }
+
     LogPrint(BCLog::NET, "Cleared nodestate for peer=%d\n", nodeid);
 }
 

src/net_processing.h

@@ -130,6 +130,10 @@ class NetHandler
     virtual void Interrupt() {}
     virtual void Schedule(CScheduler& scheduler) {}
 
+    //! Called when a peer is fully disconnected. Handlers may free any
+    //! per-NodeId state keyed on the ephemeral connection id.
+    virtual void FinalizeNode(NodeId /*nodeid*/) {}
+
     virtual void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv) {}
 
     // It returns true, if NetHandler has a responsibility about having this type of inventory and has corresponding data.