Merge #7415: fix: bound pending sig share queue

bdca48c refactor: use CountedBucketMap for counting total amount of pending sigs (Konstantin Akimov)
355da45 fix: cap sig-share verification batch by share count (PastaClaw)
08d9533 fix: cap amount of unverified batches by 4 (Konstantin Akimov)
e2862fe fix: the actual cap enforcement for pending signatures (Konstantin Akimov)
4666ef3 test: add regression tests for CountedBucketMap (Konstantin Akimov)
ec15f90 refactor: use dedicated util structure to count internal objects in signing-shares (Konstantin Akimov)

Pull request description:

  ## Issue being fixed or feature implemented

  Pending incoming LLMQ sig shares are admitted cheaply and verified later by worker threads. Direct and batched share traffic could grow retained pending verification work without per-node/global backpressure.

  This replaces closed PR #7413 and keeps the same scope: resource/backpressure hardening for LLMQ signing, with no consensus or user-facing behavior change.

  ## What was done?

  - Add per-node and global caps for pending incoming sig shares.
  - Track pending counts explicitly so cap checks do not require repeated scans.
  - Gate `QSIGSHARE` and `QBSIGSHARES` pending admission; over-cap shares are dropped without misbehavior scoring.
  - Clear pending incoming shares when a peer is banned and skip banned node states during collection.
  - Refuse new pending work for already-banned node states.
  - Add unit coverage for cap/drop/count cleanup behavior.

  ## How Has This Been Tested?

  - `./autogen.sh`
  - `./configure --without-gui --disable-bench --disable-fuzz-binary`
  - `make -C src test/test_dash -j$(sysctl -n hw.ncpu)`
  - `src/test/test_dash --run_test=llmq_utils_tests`
  - `git diff --check upstream/develop..HEAD`
  - `COMMIT_RANGE=upstream/develop..HEAD test/lint/lint-whitespace.py`

  ## Breaking Changes

  None.

  ## 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
  - [ ] I have made corresponding changes to the documentation
  - [ ] I have assigned this pull request to a milestone _(for repository code-owners and collaborators only)_

ACKs for top commit:
  kwvg:
    utACK bdca48c

Tree-SHA512: e2403635a608aab5488dbf10b481a778e735a6e5b065eec5a4cc444818f61e89b3a688f82c14aaceae07fd714432d39c013769a54aa66e31bb8abddb3531aaf3

Author: PastaPastaPasta

src/llmq/net_signing.cpp

@@ -320,19 +320,31 @@ 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.
+        //
+        // Each batch is bounded by actual share count (not unique-session count), so at most
+        // MAX_UNVERIFIED_BATCHES * MAX_SHARES_PER_BATCH shares can be sitting in / on the worker pool at once.
+        static constexpr int MAX_UNVERIFIED_BATCHES{4};
+        static constexpr size_t MAX_SHARES_PER_BATCH{32};
+        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;
 
-            const size_t nMaxBatchSize{32};
-            bool more_work = m_shares_manager->CollectPendingSigSharesToVerify(nMaxBatchSize, sigSharesByNodes, quorums);
+            bool more_work = m_shares_manager->CollectPendingSigSharesToVerify(MAX_SHARES_PER_BATCH, sigSharesByNodes, quorums);
 
             if (sigSharesByNodes.empty()) {
                 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,16 +472,43 @@ 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) {
+        // the size of nodeStates is limited by DEFAULT_MAX_PEER_CONNECTIONS(125) so it should not be performance issue
+        // The name of variable is intentionally mentioned in comment to make this code snippet relevant for possible changes in future
+        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,
+    size_t maxShares, std::unordered_map<NodeId, std::vector<CSigShare>>& retSigShares,
     std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>& retQuorums)
 {
     bool more_work{false};
@@ -487,16 +519,19 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify(
             return false;
         }
 
-        // This will iterate node states in random order and pick one sig share at a time. This avoids processing
-        // of large batches at once from the same node while other nodes also provided shares. If we wouldn't do this,
-        // other nodes would be able to poison us with a large batch with N-1 valid shares and the last one being
-        // invalid, making batch verification fail and revert to per-share verification, which in turn would slow down
-        // the whole verification process
-        std::unordered_set<std::pair<NodeId, uint256>, StaticSaltedHasher> uniqueSignHashes;
+        // Iterate node states in random order and pick one sig share at a time. This ensures no single peer can
+        // dominate a batch and that a large flood from one peer cannot poison batch verification (an N-1 valid /
+        // 1 invalid batch would fall back to per-share verification and slow the whole pipeline).
+        //
+        // The batch is bounded by the number of shares actually added (maxShares), not by the count of unique
+        // (nodeId, signHash) sessions. Bounding by sessions could otherwise let a single session inflate the
+        // batch to the full pending-share cap, and, together with the in-flight batch cap, keep tens of thousands
+        // of shares outside the pending accounting.
+        size_t sharesAdded{0};
         IterateNodesRandom(
             nodeStates,
             [&]() {
-                return uniqueSignHashes.size() < maxUniqueSessions;
+                return sharesAdded < maxShares;
                 // TODO: remove NO_THREAD_SAFETY_ANALYSIS
                 // using here template IterateNodesRandom makes impossible to use lock annotation
             },
@@ -508,8 +543,8 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify(
 
                 AssertLockHeld(cs);
                 if (const bool alreadyHave = this->sigShares.Has(sigShare.GetKey()); !alreadyHave) {
-                    uniqueSignHashes.emplace(nodeId, sigShare.GetSignHash());
                     retSigShares[nodeId].emplace_back(sigShare);
+                    ++sharesAdded;
                 }
                 ns.pendingIncomingSigShares.Erase(sigShare.GetKey());
                 return !ns.pendingIncomingSigShares.Empty();
@@ -1426,6 +1461,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId)
         sigSharesRequested.Erase(k);
     });
     nodeState.requestedSigShares.Clear();
+    nodeState.pendingIncomingSigShares.Clear();
     nodeState.banned = true;
 }