From 5410afe8ef9af247594e90427615ecceaaf98fe0 Mon Sep 17 00:00:00 2001 From: Pasta Date: Tue, 7 Jul 2026 11:57:09 -0500 Subject: [PATCH] fix: bound pending sig share queue --- src/llmq/signing_shares.cpp | 91 ++++++++++++++++++++++++++++++++--- src/llmq/signing_shares.h | 15 ++++++ src/test/llmq_utils_tests.cpp | 48 ++++++++++++++++++ 3 files changed, 147 insertions(+), 7 deletions(-) diff --git a/src/llmq/signing_shares.cpp b/src/llmq/signing_shares.cpp index 4849e2940ee9..896f633c6160 100644 --- a/src/llmq/signing_shares.cpp +++ b/src/llmq/signing_shares.cpp @@ -167,6 +167,11 @@ size_t CSigSharesNodeState::GetAnnouncementSessionCount(Consensus::LLMQType llmq }); } +size_t CSigSharesNodeState::GetPendingIncomingSigSharesCount() const +{ + return pendingIncomingSigSharesCount; +} + CSigSharesNodeState::Session* CSigSharesNodeState::GetSessionBySignHash(const uint256& signHash) { auto it = sessions.find(signHash); @@ -201,6 +206,44 @@ bool CSigSharesNodeState::GetSessionInfoByRecvId(uint32_t sessionId, SessionInfo return true; } +bool CSigSharesNodeState::AddPendingIncomingSigShare(const llmq::CSigShare& sigShare) +{ + if (pendingIncomingSigSharesCount >= MAX_PENDING_SIG_SHARES_PER_NODE) { + return false; + } + if (!pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare)) { + return false; + } + ++pendingIncomingSigSharesCount; + return true; +} + +bool CSigSharesNodeState::ErasePendingIncomingSigShare(const SigShareKey& key) +{ + if (!pendingIncomingSigShares.Has(key)) { + return false; + } + pendingIncomingSigShares.Erase(key); + --pendingIncomingSigSharesCount; + return true; +} + +size_t CSigSharesNodeState::ClearPendingIncomingSigShares() +{ + const size_t removed = pendingIncomingSigSharesCount; + pendingIncomingSigShares.Clear(); + pendingIncomingSigSharesCount = 0; + return removed; +} + +size_t CSigSharesNodeState::ErasePendingIncomingSigSharesForSession(const uint256& signHash) +{ + const size_t removed = pendingIncomingSigShares.CountForSignHash(signHash); + pendingIncomingSigShares.EraseAllForSignHash(signHash); + pendingIncomingSigSharesCount -= removed; + return removed; +} + void CSigSharesNodeState::RemoveSession(const uint256& signHash) { if (const auto it = sessions.find(signHash); it != sessions.end()) { @@ -208,7 +251,7 @@ void CSigSharesNodeState::RemoveSession(const uint256& signHash) sessions.erase(it); } requestedSigShares.EraseAllForSignHash(signHash); - pendingIncomingSigShares.EraseAllForSignHash(signHash); + ErasePendingIncomingSigSharesForSession(signHash); } ////////////////////// @@ -422,7 +465,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,7 +510,7 @@ 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__, @@ -493,6 +536,7 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify( // invalid, making batch verification fail and revert to per-share verification, which in turn would slow down // the whole verification process std::unordered_set, StaticSaltedHasher> uniqueSignHashes; + size_t erasedCount{0}; IterateNodesRandom( nodeStates, [&]() { @@ -501,7 +545,7 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify( // using here template IterateNodesRandom makes impossible to use lock annotation }, [&](NodeId nodeId, CSigSharesNodeState& ns) NO_THREAD_SAFETY_ANALYSIS { - if (ns.pendingIncomingSigShares.Empty()) { + if (ns.banned || ns.pendingIncomingSigShares.Empty()) { return false; } const auto& sigShare = *ns.pendingIncomingSigShares.GetFirst(); @@ -511,11 +555,15 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify( uniqueSignHashes.emplace(nodeId, sigShare.GetSignHash()); retSigShares[nodeId].emplace_back(sigShare); } - ns.pendingIncomingSigShares.Erase(sigShare.GetKey()); + if (ns.ErasePendingIncomingSigShare(sigShare.GetKey())) { + ++erasedCount; + } return !ns.pendingIncomingSigShares.Empty(); }, rnd); + pendingIncomingSigSharesCount -= erasedCount; + if (retSigShares.empty()) { return false; } @@ -524,7 +572,7 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify( more_work = std::any_of(nodeStates.begin(), nodeStates.end(), [](const auto& entry) { const auto& ns = entry.second; - return !ns.pendingIncomingSigShares.Empty(); + return !ns.banned && !ns.pendingIncomingSigShares.Empty(); }); } @@ -1230,6 +1278,28 @@ CSigShare CSigSharesManager::RebuildSigShare(const CSigSharesNodeState::SessionI return sigShare; } +bool CSigSharesManager::TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesNodeState& nodeState, const CSigShare& sigShare) +{ + AssertLockHeld(cs); + + if (pendingIncomingSigSharesCount >= 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; + } + if (nodeState.GetPendingIncomingSigSharesCount() >= 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 (!nodeState.AddPendingIncomingSigShare(sigShare)) { + return false; + } + ++pendingIncomingSigSharesCount; + return true; +} + void CSigSharesManager::Cleanup() { constexpr auto CLEANUP_INTERVAL{5s}; @@ -1366,6 +1436,7 @@ void CSigSharesManager::Cleanup() AssertLockHeld(cs); sigSharesRequested.Erase(k); }); + pendingIncomingSigSharesCount -= it->second.GetPendingIncomingSigSharesCount(); nodeStates.erase(nodeId); } } @@ -1375,7 +1446,9 @@ void CSigSharesManager::RemoveSigSharesForSession(const uint256& signHash) AssertLockHeld(cs); for (auto& [_, nodeState] : nodeStates) { + const size_t pending_before{nodeState.GetPendingIncomingSigSharesCount()}; nodeState.RemoveSession(signHash); + pendingIncomingSigSharesCount -= pending_before - nodeState.GetPendingIncomingSigSharesCount(); } sigSharesRequested.EraseAllForSignHash(signHash); @@ -1397,6 +1470,7 @@ void CSigSharesManager::RemoveNodesIf(std::function predicate) AssertLockHeld(cs); sigSharesRequested.Erase(k); }); + pendingIncomingSigSharesCount -= it->second.GetPendingIncomingSigSharesCount(); it = nodeStates.erase(it); } else { ++it; @@ -1425,6 +1499,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId) sigSharesRequested.Erase(k); }); nodeState.requestedSigShares.Clear(); + pendingIncomingSigSharesCount -= nodeState.ClearPendingIncomingSigShares(); nodeState.banned = true; } @@ -1444,7 +1519,9 @@ bool CSigSharesManager::IsAnyPendingProcessing() const LOCK(cs); // Check if there's work, spawn a helper if so return std::any_of(nodeStates.begin(), nodeStates.end(), - [](const auto& entry) { return !entry.second.pendingIncomingSigShares.Empty(); }); + [](const auto& entry) { + return !entry.second.banned && !entry.second.pendingIncomingSigShares.Empty(); + }); } std::shared_ptr CSigSharesManager::SignAndProcessSingleShare(PendingSignatureData work) diff --git a/src/llmq/signing_shares.h b/src/llmq/signing_shares.h index ef8aa42edc0f..b2b6e395cc0a 100644 --- a/src/llmq/signing_shares.h +++ b/src/llmq/signing_shares.h @@ -46,6 +46,12 @@ constexpr uint32_t UNINITIALIZED_SESSION_ID{std::numeric_limits::max() // 400 is the maximum quorum size, so this is also the maximum number of sigs we need to support constexpr size_t MAX_MSGS_TOTAL_BATCHED_SIGS{400}; +// Backpressure bounds for the pending (not-yet-verified) sig-share queue. Incoming QSIGSHARE and +// QBSIGSHARES traffic is cheap to admit but later consumes memory and BLS verification worker +// capacity, so over-cap shares are dropped silently without scoring misbehaviour. +static constexpr size_t MAX_PENDING_SIG_SHARES_PER_NODE{1000}; +static constexpr size_t MAX_PENDING_SIG_SHARES_TOTAL{10000}; + class CSigShare : virtual public CSigBase { protected: @@ -340,6 +346,7 @@ class CSigSharesNodeState std::unordered_map sessionByRecvId; SigShareMap pendingIncomingSigShares; + size_t pendingIncomingSigSharesCount{0}; SigShareMap requestedSigShares; bool banned{false}; @@ -350,9 +357,14 @@ class CSigSharesNodeState [[nodiscard]] size_t GetSessionCount() const; [[nodiscard]] size_t GetSessionCount(Consensus::LLMQType llmqType) const; [[nodiscard]] size_t GetAnnouncementSessionCount(Consensus::LLMQType llmqType) const; + [[nodiscard]] size_t GetPendingIncomingSigSharesCount() const; Session* GetSessionBySignHash(const uint256& signHash); Session* GetSessionByRecvId(uint32_t sessionId); bool GetSessionInfoByRecvId(uint32_t sessionId, SessionInfo& retInfo); + bool AddPendingIncomingSigShare(const CSigShare& sigShare); + bool ErasePendingIncomingSigShare(const SigShareKey& key); + size_t ClearPendingIncomingSigShares(); + size_t ErasePendingIncomingSigSharesForSession(const uint256& signHash); void RemoveSession(const uint256& signHash); }; @@ -405,6 +417,7 @@ class CSigSharesManager : public llmq::CRecoveredSigsListener Uint256HashMap timeSeenForSessions GUARDED_BY(cs); std::unordered_map nodeStates GUARDED_BY(cs); + size_t pendingIncomingSigSharesCount GUARDED_BY(cs){0}; SigShareMap> sigSharesRequested GUARDED_BY(cs); SigShareMap sigSharesQueuedToAnnounce GUARDED_BY(cs); @@ -484,6 +497,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& in); + bool TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesNodeState& nodeState, const CSigShare& sigShare) + EXCLUSIVE_LOCKS_REQUIRED(cs); void RemoveSigSharesForSession(const uint256& signHash) EXCLUSIVE_LOCKS_REQUIRED(cs); diff --git a/src/test/llmq_utils_tests.cpp b/src/test/llmq_utils_tests.cpp index ea51601a8b5a..741bc5396db2 100644 --- a/src/test/llmq_utils_tests.cpp +++ b/src/test/llmq_utils_tests.cpp @@ -103,6 +103,54 @@ BOOST_AUTO_TEST_CASE(sig_ses_ann_limit_is_per_llmq_type) BOOST_CHECK_EQUAL(node_state.GetSessionCount(Consensus::LLMQType::LLMQ_400_60), 1U); } +BOOST_AUTO_TEST_CASE(pending_sig_shares_per_node_cap_drops_without_growth) +{ + CSigSharesNodeState node_state; + + for (size_t i{0}; i < MAX_PENDING_SIG_SHARES_PER_NODE; ++i) { + BOOST_CHECK(node_state.AddPendingIncomingSigShare(MakeSigShare(i + 1))); + } + + BOOST_CHECK_EQUAL(node_state.GetPendingIncomingSigSharesCount(), MAX_PENDING_SIG_SHARES_PER_NODE); + BOOST_CHECK(!node_state.AddPendingIncomingSigShare(MakeSigShare(MAX_PENDING_SIG_SHARES_PER_NODE + 1))); + BOOST_CHECK_EQUAL(node_state.GetPendingIncomingSigSharesCount(), MAX_PENDING_SIG_SHARES_PER_NODE); +} + +BOOST_AUTO_TEST_CASE(pending_sig_shares_counts_unique_entries) +{ + CSigSharesNodeState node_state; + const CSigShare sig_share1{MakeSigShare(1)}; + const CSigShare sig_share2{MakeSigShare(2)}; + + BOOST_CHECK(node_state.AddPendingIncomingSigShare(sig_share1)); + BOOST_CHECK(!node_state.AddPendingIncomingSigShare(sig_share1)); + BOOST_CHECK(node_state.AddPendingIncomingSigShare(sig_share2)); + BOOST_CHECK_EQUAL(node_state.GetPendingIncomingSigSharesCount(), 2U); + + BOOST_CHECK(node_state.ErasePendingIncomingSigShare(sig_share1.GetKey())); + BOOST_CHECK(!node_state.ErasePendingIncomingSigShare(sig_share1.GetKey())); + BOOST_CHECK_EQUAL(node_state.GetPendingIncomingSigSharesCount(), 1U); + + BOOST_CHECK_EQUAL(node_state.ClearPendingIncomingSigShares(), 1U); + BOOST_CHECK_EQUAL(node_state.GetPendingIncomingSigSharesCount(), 0U); +} + +BOOST_AUTO_TEST_CASE(pending_sig_shares_session_removal_updates_count) +{ + CSigSharesNodeState node_state; + const CSigShare sig_share1{MakeSigShare(1)}; + const CSigShare sig_share2{MakeSigShare(2)}; + + BOOST_CHECK(node_state.AddPendingIncomingSigShare(sig_share1)); + BOOST_CHECK(node_state.AddPendingIncomingSigShare(sig_share2)); + BOOST_CHECK_EQUAL(node_state.GetPendingIncomingSigSharesCount(), 2U); + + node_state.RemoveSession(sig_share1.GetSignHash()); + BOOST_CHECK_EQUAL(node_state.GetPendingIncomingSigSharesCount(), 1U); + BOOST_CHECK(!node_state.pendingIncomingSigShares.Has(sig_share1.GetKey())); + BOOST_CHECK(node_state.pendingIncomingSigShares.Has(sig_share2.GetKey())); +} + BOOST_AUTO_TEST_CASE(deterministic_outbound_connection_test) { // Test deterministic behavior