fix: bound pending sig share queue

src/llmq/net_signing.cpp

@@ -320,8 +320,11 @@ void NetSigning::WorkThreadDispatcher()
             }
         }
 
-        // Collect pending sig shares synchronously and dispatch each batch to a worker for parallel BLS verification
-        while (!workInterrupt) {
+        // Collect pending sig shares synchronously and dispatch each batch to a worker for parallel BLS verification.
+        // Batches awaiting verification are bounded so that under flood shares back up in the capped
+        // pending maps instead of migrating into the unbounded worker pool task queue.
+        static constexpr int MAX_UNVERIFIED_BATCHES{4};
+        while (!workInterrupt && unverified_batches < MAX_UNVERIFIED_BATCHES) {
             std::unordered_map<NodeId, std::vector<CSigShare>> sigSharesByNodes;
             std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher> quorums;
 
@@ -332,7 +335,13 @@ void NetSigning::WorkThreadDispatcher()
                 break;
             }
 
+            ++unverified_batches;
             worker_pool.push([this, sigSharesByNodes = std::move(sigSharesByNodes), quorums = std::move(quorums)](int) mutable {
+                // Ensures unverified_batches is decremented on every exit path, including exceptions.
+                struct UnverifiedBatchGuard {
+                    std::atomic<int>& count;
+                    ~UnverifiedBatchGuard() { --count; }
+                } guard{unverified_batches};
                 ProcessPendingSigShares(std::move(sigSharesByNodes), std::move(quorums));
             });
 

src/llmq/net_signing.h

@@ -15,6 +15,7 @@
 
 #include <thread>
 
+#include <atomic>
 #include <memory>
 
 class CSporkManager;
@@ -72,6 +73,7 @@ class NetSigning final : public NetHandler, public CValidationInterface
     std::thread shares_cleaning_thread;
     std::thread shares_dispatcher_thread;
     mutable ctpl::thread_pool worker_pool;
+    std::atomic<int> unverified_batches{0};
 
     CThreadInterrupt workInterrupt;
 };

src/llmq/signing_shares.cpp

@@ -31,6 +31,11 @@ namespace {
 constexpr size_t MAX_SESSIONS_PER_PEER_FACTOR{4};
 constexpr size_t MIN_SESSIONS_PER_PEER{100};
 
+// Incoming QSIGSHARE/QBSIGSHARES traffic is cheap to admit but drains only at BLS verification
+// speed, so unverified shares are bounded and over-cap shares dropped without misbehaviour scoring.
+constexpr size_t MAX_PENDING_SIG_SHARES_PER_NODE{1000};
+constexpr size_t MAX_PENDING_SIG_SHARES_TOTAL{10000};
+
 size_t GetMaxSessionsForPeer(const Consensus::LLMQParams& params)
 {
     return std::max<size_t>(size_t(params.size) * MAX_SESSIONS_PER_PEER_FACTOR, MIN_SESSIONS_PER_PEER);
@@ -422,7 +427,7 @@ bool CSigSharesManager::ProcessMessageBatchedSigShares(const CNode& pfrom, const
     LOCK(cs);
     auto& nodeState = nodeStates[pfrom.GetId()];
     for (const auto& s : sigSharesToProcess) {
-        nodeState.pendingIncomingSigShares.Add(s.GetKey(), s);
+        TryAddPendingIncomingSigShare(pfrom.GetId(), nodeState, s);
     }
     return true;
 }
@@ -467,14 +472,39 @@ bool CSigSharesManager::ProcessMessageSigShare(NodeId fromId, const CSigShare& s
         }
 
         auto& nodeState = nodeStates[fromId];
-        nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare);
+        TryAddPendingIncomingSigShare(fromId, nodeState, sigShare);
     }
 
     LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- signHash=%s, id=%s, msgHash=%s, member=%d, node=%d\n", __func__,
              signHash.ToString(), sigShare.getId().ToString(), sigShare.getMsgHash().ToString(), sigShare.getQuorumMember(), fromId);
     return true;
 }
 
+bool CSigSharesManager::TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesNodeState& nodeState,
+                                                      const CSigShare& sigShare)
+{
+    AssertLockHeld(cs);
+
+    if (nodeState.banned) {
+        return false;
+    }
+    if (nodeState.pendingIncomingSigShares.Size() >= MAX_PENDING_SIG_SHARES_PER_NODE) {
+        LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- per-node pending sig shares cap reached (%d), dropping sigShare. node=%d\n",
+                 __func__, MAX_PENDING_SIG_SHARES_PER_NODE, nodeId);
+        return false;
+    }
+    size_t total{0};
+    for (const auto& [_, ns] : nodeStates) {
+        total += ns.pendingIncomingSigShares.Size();
+    }
+    if (total >= MAX_PENDING_SIG_SHARES_TOTAL) {
+        LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- global pending sig shares cap reached (%d), dropping sigShare. node=%d\n",
+                 __func__, MAX_PENDING_SIG_SHARES_TOTAL, nodeId);
+        return false;
+    }
+    return nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare);
+}
+
 bool CSigSharesManager::CollectPendingSigSharesToVerify(
     size_t maxUniqueSessions, std::unordered_map<NodeId, std::vector<CSigShare>>& retSigShares,
     std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>& retQuorums)
@@ -1425,6 +1455,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId)
         sigSharesRequested.Erase(k);
     });
     nodeState.requestedSigShares.Clear();
+    nodeState.pendingIncomingSigShares.Clear();
     nodeState.banned = true;
 }
 

src/llmq/signing_shares.h

@@ -9,7 +9,6 @@
 #include <evo/types.h>
 #include <llmq/signhash.h>
 #include <llmq/signing.h>
-#include <util/std23.h>
 
 #include <random.h>
 #include <saltedhasher.h>
@@ -160,49 +159,124 @@ class CBatchedSigShares
     [[nodiscard]] std::string ToInvString() const;
 };
 
+/**
+ * Two-level (signHash -> quorumMember) map with a running entry count, so Size() is O(1)
+ * instead of a fold over all sign hash buckets. All structural mutations go through the
+ * counted methods; Buckets() is for lookups and in-place value updates only.
+ */
 template<typename T>
-class SigShareMap
+class CountedBucketMap
 {
+public:
+    using BucketMap = Uint256HashMap<std::unordered_map<uint16_t, T>>;
+
 private:
-    Uint256HashMap<std::unordered_map<uint16_t, T>> internalMap;
+    BucketMap m_data;
+    size_t m_num_entries{0};
 
 public:
-    bool Add(const SigShareKey& k, const T& v)
+    BucketMap& Buckets() { return m_data; }
+    const BucketMap& Buckets() const { return m_data; }
+    [[nodiscard]] size_t Size() const { return m_num_entries; }
+
+    bool Emplace(const SigShareKey& k, const T& v)
     {
-        auto& m = internalMap[k.first];
-        return m.emplace(k.second, v).second;
+        if (!m_data[k.first].emplace(k.second, v).second) {
+            return false;
+        }
+        ++m_num_entries;
+        return true;
     }
 
     void Erase(const SigShareKey& k)
     {
-        auto it = internalMap.find(k.first);
-        if (it == internalMap.end()) {
+        auto it = m_data.find(k.first);
+        if (it == m_data.end()) {
             return;
         }
-        it->second.erase(k.second);
+        m_num_entries -= it->second.erase(k.second);
         if (it->second.empty()) {
-            internalMap.erase(it);
+            m_data.erase(it);
+        }
+    }
+
+    void EraseBucket(const uint256& signHash)
+    {
+        auto it = m_data.find(signHash);
+        if (it == m_data.end()) {
+            return;
+        }
+        m_num_entries -= it->second.size();
+        m_data.erase(it);
+    }
+
+    template<typename F>
+    void EraseIf(F&& f)
+    {
+        for (auto it = m_data.begin(); it != m_data.end(); ) {
+            SigShareKey k;
+            k.first = it->first;
+            for (auto jt = it->second.begin(); jt != it->second.end(); ) {
+                k.second = jt->first;
+                if (f(k, jt->second)) {
+                    jt = it->second.erase(jt);
+                    --m_num_entries;
+                } else {
+                    ++jt;
+                }
+            }
+            if (it->second.empty()) {
+                it = m_data.erase(it);
+            } else {
+                ++it;
+            }
         }
     }
 
     void Clear()
     {
-        internalMap.clear();
+        m_data.clear();
+        m_num_entries = 0;
+    }
+};
+
+template<typename T>
+class SigShareMap
+{
+private:
+    CountedBucketMap<T> internalMap;
+
+public:
+    bool Add(const SigShareKey& k, const T& v)
+    {
+        return internalMap.Emplace(k, v);
+    }
+
+    void Erase(const SigShareKey& k)
+    {
+        internalMap.Erase(k);
+    }
+
+    void Clear()
+    {
+        internalMap.Clear();
     }
 
     [[nodiscard]] bool Has(const SigShareKey& k) const
     {
-        auto it = internalMap.find(k.first);
-        if (it == internalMap.end()) {
+        auto& m = internalMap.Buckets();
+        auto it = m.find(k.first);
+        if (it == m.end()) {
             return false;
         }
         return it->second.count(k.second) != 0;
     }
 
     T* Get(const SigShareKey& k)
     {
-        auto it = internalMap.find(k.first);
-        if (it == internalMap.end()) {
+        auto& m = internalMap.Buckets();
+        auto it = m.find(k.first);
+        if (it == m.end()) {
             return nullptr;
         }
 
@@ -226,72 +300,58 @@ class SigShareMap
 
     const T* GetFirst() const
     {
-        if (internalMap.empty()) {
+        auto& m = internalMap.Buckets();
+        if (m.empty()) {
             return nullptr;
         }
-        return &internalMap.begin()->second.begin()->second;
+        return &m.begin()->second.begin()->second;
     }
 
     [[nodiscard]] size_t Size() const
     {
-        return std23::ranges::fold_left(internalMap, size_t{0},
-                                        [](size_t s, const auto& p) { return s + p.second.size(); });
+        return internalMap.Size();
     }
 
     [[nodiscard]] size_t CountForSignHash(const uint256& signHash) const
     {
-        auto it = internalMap.find(signHash);
-        if (it == internalMap.end()) {
+        auto& m = internalMap.Buckets();
+        auto it = m.find(signHash);
+        if (it == m.end()) {
             return 0;
         }
         return it->second.size();
     }
 
     [[nodiscard]] bool Empty() const
     {
-        return internalMap.empty();
+        return internalMap.Buckets().empty();
     }
 
     const std::unordered_map<uint16_t, T>* GetAllForSignHash(const uint256& signHash) const
     {
-        auto it = internalMap.find(signHash);
-        if (it == internalMap.end()) {
+        auto& m = internalMap.Buckets();
+        auto it = m.find(signHash);
+        if (it == m.end()) {
             return nullptr;
         }
         return &it->second;
     }
 
     void EraseAllForSignHash(const uint256& signHash)
     {
-        internalMap.erase(signHash);
+        internalMap.EraseBucket(signHash);
     }
 
     template<typename F>
     void EraseIf(F&& f)
     {
-        for (auto it = internalMap.begin(); it != internalMap.end(); ) {
-            SigShareKey k;
-            k.first = it->first;
-            for (auto jt = it->second.begin(); jt != it->second.end(); ) {
-                k.second = jt->first;
-                if (f(k, jt->second)) {
-                    jt = it->second.erase(jt);
-                } else {
-                    ++jt;
-                }
-            }
-            if (it->second.empty()) {
-                it = internalMap.erase(it);
-            } else {
-                ++it;
-            }
-        }
+        internalMap.EraseIf(f);
     }
 
     template<typename F>
     void ForEach(F&& f)
     {
-        for (auto& p : internalMap) {
+        for (auto& p : internalMap.Buckets()) {
             SigShareKey k;
             k.first = p.first;
             for (auto& p2 : p.second) {
@@ -484,6 +544,8 @@ class CSigSharesManager : public llmq::CRecoveredSigsListener
     bool GetSessionInfoByRecvId(NodeId nodeId, uint32_t sessionId, CSigSharesNodeState::SessionInfo& retInfo)
         EXCLUSIVE_LOCKS_REQUIRED(!cs);
     static CSigShare RebuildSigShare(const CSigSharesNodeState::SessionInfo& session, const std::pair<uint16_t, CBLSLazySignature>& in);
+    bool TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesNodeState& nodeState, const CSigShare& sigShare)
+        EXCLUSIVE_LOCKS_REQUIRED(cs);
 
     void RemoveSigSharesForSession(const uint256& signHash) EXCLUSIVE_LOCKS_REQUIRED(cs);