Skip to content

fix: bound pending sig share queue #7415

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
PastaPastaPasta merged 6 commits into from
Jul 11, 2026
Merged

fix: bound pending sig share queue #7415

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ec15f90
refactor: use dedicated util structure to count internal objects in s…
knst Jul 9, 2026
4666ef3
test: add regression tests for CountedBucketMap
knst Jul 9, 2026
e2862fe
fix: the actual cap enforcement for pending signatures
knst Jul 9, 2026
08d9533
fix: cap amount of unverified batches by 4
knst Jul 9, 2026
355da45
fix: cap sig-share verification batch by share count
thepastaclaw Jul 9, 2026
bdca48c
refactor: use CountedBucketMap for counting total amount of pending sigs
knst Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions src/llmq/net_signing.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Bound verification work by share count

Fresh evidence beyond the earlier queue-cap concern is that CollectPendingSigSharesToVerify(32, ...) limits unique (nodeId, signHash) pairs, not actual shares; if peers fill the pending maps with many quorum members for the same few sessions, one collected “batch” can contain up to the 10k pending-share cap. This line then allows four such batches to be queued/in flight after they have been erased from the capped pending maps, so flood traffic can still move tens of thousands of BLS shares into verification work despite the intended global cap. Cap the number of shares collected per dispatch or keep queued/in-flight shares charged against the cap.

Useful? React with 👍 / 👎.

@thepastaclaw thepastaclaw Jul 9, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 355da45d4a4. CollectPendingSigSharesToVerify now caps each dispatch by actual shares added (MAX_SHARES_PER_BATCH = 32) instead of unique sessions. With four unverified batches, at most 128 shares can sit in/on the worker pool outside the 10,000-share pending maps; randomized peer round-robin fairness is preserved. Focused LLMQ tests and neighboring suites pass.

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));
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.

Expand Down
2 changes: 2 additions & 0 deletions src/llmq/net_signing.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

#include <thread>

#include <atomic>
#include <memory>

class CSporkManager;
Expand Down Expand Up @@ -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;
};
Expand Down
66 changes: 51 additions & 15 deletions src/llmq/signing_shares.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -201,14 +206,14 @@ bool CSigSharesNodeState::GetSessionInfoByRecvId(uint32_t sessionId, SessionInfo
return true;
}

void CSigSharesNodeState::RemoveSession(const uint256& signHash)
size_t CSigSharesNodeState::RemoveSession(const uint256& signHash)
{
if (const auto it = sessions.find(signHash); it != sessions.end()) {
sessionByRecvId.erase(it->second.recvSessionId);
sessions.erase(it);
}
requestedSigShares.EraseAllForSignHash(signHash);
pendingIncomingSigShares.EraseAllForSignHash(signHash);
return pendingIncomingSigShares.EraseAllForSignHash(signHash);
}

//////////////////////
Expand Down Expand Up @@ -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);

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve request state when admission fails

When TryAddPendingIncomingSigShare returns false because the per-node or global pending-share cap is full, this loop silently drops a requested QBSIGSHARES response even though line 401 already removed the peer's requestedSigShares entry and the request path clears session.announced.inv once the request is sent. In that over-cap scenario the share is not verified and is not requested from the same peer again, so recovery can stall if no other peer advertises that quorum member; keep the request/announcement state until the share is admitted, or requeue it on admission failure.

Useful? React with 👍 / 👎.

}
return true;
}
Expand Down Expand Up @@ -467,16 +472,41 @@ 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;
}
if (m_pending_sig_shares_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;
}

@thepastaclaw thepastaclaw Jul 9, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Suggestion: Global pending-sig-share cap check is O(peers) per admission

TryAddPendingIncomingSigShare iterates every entry in nodeStates while holding cs to compute the global total on every QSIGSHARE/QBSIGSHARES admission. The per-node check is O(1) thanks to CountedBucketMap::Size(), but this loop reintroduces O(peers) work on the hot admission path. Under a flood with many connected peers, cs is held longer per accepted share, partially defeating the O(1) counting refactor and making the backpressure path itself scale with peer count. Maintain a manager-level running total that is updated alongside per-node Add/Erase operations so the global cap remains O(1). Correctness is unaffected, so this is non-blocking.

source: ['claude']

@thepastaclaw thepastaclaw Jul 9, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 355da45d4a4 as part of the same accounting correction. A manager-level m_pending_sig_shares_total is now updated on every add/erase/session removal/node removal/ban path, so the global admission cap is O(1) instead of walking all peers under cs. The erase primitives return exact removed counts, with regression coverage.

if (!nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare)) {
return false;
}
++m_pending_sig_shares_total;
return true;
}
Comment on lines +483 to +508

@thepastaclaw thepastaclaw Jul 10, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Suggestion: Cap/drop/banned enforcement in TryAddPendingIncomingSigShare has no direct unit coverage

The PR's stated purpose is bounding pending incoming sig shares, but the enforcement logic in TryAddPendingIncomingSigShare — the per-node cap at MAX_PENDING_SIG_SHARES_PER_NODE (line 491), the global cap at MAX_PENDING_SIG_SHARES_TOTAL (line 502), and the banned-state refusal (line 488) — is not exercised by any added test. A grep of llmq_utils_tests.cpp finds no reference to TryAddPendingIncomingSigShare, MAX_PENDING_SIG_SHARES, or the banned path; the new tests only cover the supporting CountedBucketMap size tracking and RemoveSession behavior. So the machinery the caps rely on (Size() accuracy) is tested, but the threshold/drop-without-misbehavior/banned-refusal behavior that is the point of the change is not. TryAddPendingIncomingSigShare is private and needs a CSigSharesManager plus quorum setup to drive, which makes this materially harder than the node-state-level tests already added, so this is a suggestion rather than a blocker.

source: ['claude']


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};
Expand All @@ -487,16 +517,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
},
Expand All @@ -508,10 +541,10 @@ 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());
m_pending_sig_shares_total -= ns.pendingIncomingSigShares.Erase(sigShare.GetKey());
return !ns.pendingIncomingSigShares.Empty();
},
rnd);
Expand Down Expand Up @@ -1366,6 +1399,7 @@ void CSigSharesManager::Cleanup()
AssertLockHeld(cs);
sigSharesRequested.Erase(k);
});
m_pending_sig_shares_total -= it->second.pendingIncomingSigShares.Size();
nodeStates.erase(nodeId);
}
}
Expand All @@ -1375,7 +1409,7 @@ void CSigSharesManager::RemoveSigSharesForSession(const uint256& signHash)
AssertLockHeld(cs);

for (auto& [_, nodeState] : nodeStates) {
nodeState.RemoveSession(signHash);
m_pending_sig_shares_total -= nodeState.RemoveSession(signHash);
}

sigSharesRequested.EraseAllForSignHash(signHash);
Expand All @@ -1397,6 +1431,7 @@ void CSigSharesManager::RemoveNodesIf(std::function<bool(NodeId)> predicate)
AssertLockHeld(cs);
sigSharesRequested.Erase(k);
});
m_pending_sig_shares_total -= it->second.pendingIncomingSigShares.Size();
it = nodeStates.erase(it);
} else {
++it;
Expand Down Expand Up @@ -1425,6 +1460,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId)
sigSharesRequested.Erase(k);
});
nodeState.requestedSigShares.Clear();
m_pending_sig_shares_total -= nodeState.pendingIncomingSigShares.Clear();
nodeState.banned = true;
}

Expand Down
Loading
Loading