From 783fb5fc66311a3c28058ea5f50388dfb2c21714 Mon Sep 17 00:00:00 2001 From: Navid Rahimi Date: Sun, 23 Aug 2026 22:15:31 +0000 Subject: [PATCH] LLMQ: bound pending recovered signatures --- src/llmq/quorums_signing.cpp | 46 +++++- src/llmq/quorums_signing.h | 11 ++ src/llmq/quorums_signing_shares.cpp | 17 +++ src/test/CMakeLists.txt | 1 + src/test/quorums_signing_queue_tests.cpp | 185 +++++++++++++++++++++++ 5 files changed, 259 insertions(+), 1 deletion(-) create mode 100644 src/test/quorums_signing_queue_tests.cpp diff --git a/src/llmq/quorums_signing.cpp b/src/llmq/quorums_signing.cpp index 6ebd710ad6..2a4dc8eaf7 100644 --- a/src/llmq/quorums_signing.cpp +++ b/src/llmq/quorums_signing.cpp @@ -476,8 +476,41 @@ void CSigningManager::ProcessMessageRecoveredSig(CNode* pfrom, const CRecoveredS LogPrint("llmq", "CSigningManager::%s -- signHash=%s, id=%s, msgHash=%s, node=%d\n", __func__, CLLMQUtils::BuildSignHash(recoveredSig).ToString(), recoveredSig.id.ToString(), recoveredSig.msgHash.ToString(), pfrom->GetId()); + PushPendingRecoveredSig(pfrom->id, recoveredSig); +} + +void CSigningManager::PushPendingRecoveredSig(NodeId from, const CRecoveredSig& recoveredSig) +{ LOCK(cs); - pendingRecoveredSigs[pfrom->id].emplace_back(recoveredSig); + + if (pendingRecoveredSigsCount >= MAX_PENDING_RECSIGS_TOTAL) { + LogPrint("llmq", "CSigningManager::%s -- global pending recovered sigs cap reached (%u), dropping sig from node=%d\n", + __func__, static_cast(MAX_PENDING_RECSIGS_TOTAL), from); + return; + } + + auto nodeIt = pendingRecoveredSigs.find(from); + if (nodeIt != pendingRecoveredSigs.end() && nodeIt->second.size() >= MAX_PENDING_RECSIGS_PER_NODE) { + LogPrint("llmq", "CSigningManager::%s -- per-node pending recovered sigs cap reached (%u), dropping sig from node=%d\n", + __func__, static_cast(MAX_PENDING_RECSIGS_PER_NODE), from); + return; + } + + pendingRecoveredSigs[from].emplace_back(recoveredSig); + ++pendingRecoveredSigsCount; +} + +void CSigningManager::RemoveNodesIf(const std::function& predicate) +{ + LOCK(cs); + for (auto it = pendingRecoveredSigs.begin(); it != pendingRecoveredSigs.end();) { + if (predicate(it->first)) { + pendingRecoveredSigsCount -= it->second.size(); + it = pendingRecoveredSigs.erase(it); + } else { + ++it; + } + } } bool CSigningManager::PreVerifyRecoveredSig(NodeId nodeId, const CRecoveredSig& recoveredSig, bool& retBan) @@ -516,6 +549,7 @@ void CSigningManager::CollectPendingRecoveredSigsToVerify( } std::unordered_set, StaticSaltedHasher> uniqueSignHashes; + size_t erasedCount = 0; CLLMQUtils::IterateNodesRandom(pendingRecoveredSigs, [&]() { return uniqueSignHashes.size() < maxUniqueSessions; }, [&](NodeId nodeId, std::list& ns) { @@ -530,8 +564,18 @@ void CSigningManager::CollectPendingRecoveredSigsToVerify( retSigShares[nodeId].emplace_back(recSig); } ns.erase(ns.begin()); + ++erasedCount; return !ns.empty(); }, rnd); + pendingRecoveredSigsCount -= erasedCount; + + for (auto it = pendingRecoveredSigs.begin(); it != pendingRecoveredSigs.end();) { + if (it->second.empty()) { + it = pendingRecoveredSigs.erase(it); + } else { + ++it; + } + } if (retSigShares.empty()) { return; diff --git a/src/llmq/quorums_signing.h b/src/llmq/quorums_signing.h index 92d18e4af0..14b6eac361 100644 --- a/src/llmq/quorums_signing.h +++ b/src/llmq/quorums_signing.h @@ -13,11 +13,17 @@ #include "univalue.h" #include "unordered_lru_cache.h" +#include #include namespace llmq { +static constexpr size_t MAX_PENDING_RECSIGS_PER_NODE{1000}; +static constexpr size_t MAX_PENDING_RECSIGS_TOTAL{10000}; + +struct CSigningManagerTestAccess; + class CRecoveredSig { public: @@ -111,6 +117,7 @@ class CRecoveredSigsListener class CSigningManager { friend class CSigSharesManager; + friend struct CSigningManagerTestAccess; static const int64_t DEFAULT_MAX_RECOVERED_SIGS_AGE = 60 * 60 * 24 * 7; // keep them for a week // when selecting a quorum for signing and verification, we use CQuorumManager::SelectQuorum with this offset as @@ -125,6 +132,8 @@ class CSigningManager // Incoming and not verified yet std::unordered_map> pendingRecoveredSigs; + // Running total across pendingRecoveredSigs, protected by cs. + size_t pendingRecoveredSigsCount{0}; std::list> pendingReconstructedRecoveredSigs; // must be protected by cs @@ -152,6 +161,7 @@ class CSigningManager private: void ProcessMessageRecoveredSig(CNode* pfrom, const CRecoveredSig& recoveredSig, CConnman& connman); + void PushPendingRecoveredSig(NodeId from, const CRecoveredSig& recoveredSig); bool PreVerifyRecoveredSig(NodeId nodeId, const CRecoveredSig& recoveredSig, bool& retBan); void CollectPendingRecoveredSigsToVerify(size_t maxUniqueSessions, @@ -164,6 +174,7 @@ class CSigningManager public: // public interface + void RemoveNodesIf(const std::function& predicate); void RegisterRecoveredSigsListener(CRecoveredSigsListener* l); void UnregisterRecoveredSigsListener(CRecoveredSigsListener* l); diff --git a/src/llmq/quorums_signing_shares.cpp b/src/llmq/quorums_signing_shares.cpp index 0fb2b0fc1d..e6f618ee38 100644 --- a/src/llmq/quorums_signing_shares.cpp +++ b/src/llmq/quorums_signing_shares.cpp @@ -1419,6 +1419,12 @@ void CSigSharesManager::MarkNodeBanned(NodeId nodeId) return; } + if (quorumSigningManager) { + quorumSigningManager->RemoveNodesIf([nodeId](NodeId pendingNodeId) { + return pendingNodeId == nodeId; + }); + } + LOCK(cs); auto it = nodeStates.find(nodeId); if (it == nodeStates.end()) { @@ -1472,6 +1478,17 @@ void CSigSharesManager::WorkThreadMain() // remaining per-node state for banned peers. if (GetTimeMillis() - lastRemoveBannedNodeStatesTime > 30000 /* 30s */) { RemoveBannedNodeStates(); + + std::unordered_set connectedNodes; + g_connman->ForEachNode([&](CNode* pnode) { + if (!pnode->fDisconnect) { + connectedNodes.emplace(pnode->id); + } + }); + quorumSigningManager->RemoveNodesIf([&](NodeId nodeId) { + return !connectedNodes.count(nodeId); + }); + lastRemoveBannedNodeStatesTime = GetTimeMillis(); } diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index 6408280c81..bb3934d74e 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -63,6 +63,7 @@ add_executable(test_firo ${CMAKE_CURRENT_SOURCE_DIR}/net_tests.cpp ${CMAKE_CURRENT_SOURCE_DIR}/pmt_tests.cpp ${CMAKE_CURRENT_SOURCE_DIR}/prevector_tests.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/quorums_signing_queue_tests.cpp ${CMAKE_CURRENT_SOURCE_DIR}/raii_event_tests.cpp ${CMAKE_CURRENT_SOURCE_DIR}/random_tests.cpp ${CMAKE_CURRENT_SOURCE_DIR}/reverselock_tests.cpp diff --git a/src/test/quorums_signing_queue_tests.cpp b/src/test/quorums_signing_queue_tests.cpp new file mode 100644 index 0000000000..8a6e4e5ad4 --- /dev/null +++ b/src/test/quorums_signing_queue_tests.cpp @@ -0,0 +1,185 @@ +// Copyright (c) 2026 The Firo developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#include "dbwrapper.h" +#include "llmq/quorums_signing.h" +#include "llmq/quorums_signing_shares.h" +#include "test/test_bitcoin.h" +#include "validation.h" + +#include + +#include +#include + +namespace llmq +{ + +struct CSigningManagerTestAccess +{ + static void Push(CSigningManager& manager, NodeId from, const CRecoveredSig& recoveredSig) + { + manager.PushPendingRecoveredSig(from, recoveredSig); + } + + static size_t Count(CSigningManager& manager) + { + LOCK(manager.cs); + size_t count = 0; + for (const auto& nodeEntry : manager.pendingRecoveredSigs) { + count += nodeEntry.second.size(); + } + return count; + } + + static size_t Count(CSigningManager& manager, NodeId nodeId) + { + LOCK(manager.cs); + auto it = manager.pendingRecoveredSigs.find(nodeId); + return it == manager.pendingRecoveredSigs.end() ? 0 : it->second.size(); + } + + static size_t TrackedCount(CSigningManager& manager) + { + LOCK(manager.cs); + return manager.pendingRecoveredSigsCount; + } + + static bool HasNode(CSigningManager& manager, NodeId nodeId) + { + LOCK(manager.cs); + return manager.pendingRecoveredSigs.count(nodeId) != 0; + } + + static void DrainKnown(CSigningManager& manager, const CRecoveredSig& recoveredSig) + { + manager.db.WriteRecoveredSig(recoveredSig); + std::unordered_map> recoveredSigsByNode; + std::unordered_map, CQuorumCPtr, StaticSaltedHasher> quorums; + manager.CollectPendingRecoveredSigsToVerify(1, recoveredSigsByNode, quorums); + } +}; + +} // namespace llmq + +namespace +{ + +struct SigningQueueSetup : BasicTestingSetup +{ + CDBWrapper db; + llmq::CSigningManager manager; + llmq::CRecoveredSig recoveredSig{}; + + SigningQueueSetup() : db(boost::filesystem::temp_directory_path() / boost::filesystem::unique_path(), 1 << 20, true, false), + manager(db, true) + { + } +}; + +struct ScopedSigningManager +{ + llmq::CSigningManager* previous; + + explicit ScopedSigningManager(llmq::CSigningManager* manager) : + previous(llmq::quorumSigningManager) + { + llmq::quorumSigningManager = manager; + } + + ~ScopedSigningManager() + { + llmq::quorumSigningManager = previous; + } +}; + +} // namespace + +BOOST_FIXTURE_TEST_SUITE(quorums_signing_queue_tests, SigningQueueSetup) + +BOOST_AUTO_TEST_CASE(pending_recovered_sig_limits) +{ + for (size_t i = 0; i < llmq::MAX_PENDING_RECSIGS_PER_NODE + 1; ++i) { + llmq::CSigningManagerTestAccess::Push(manager, 1, recoveredSig); + } + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::Count(manager, 1), llmq::MAX_PENDING_RECSIGS_PER_NODE); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::TrackedCount(manager), llmq::MAX_PENDING_RECSIGS_PER_NODE); + + llmq::CSigningManager globalManager(db, true); + constexpr NodeId nodesAtGlobalLimit = llmq::MAX_PENDING_RECSIGS_TOTAL / llmq::MAX_PENDING_RECSIGS_PER_NODE; + for (NodeId nodeId = 0; nodeId < nodesAtGlobalLimit; ++nodeId) { + for (size_t i = 0; i < llmq::MAX_PENDING_RECSIGS_PER_NODE; ++i) { + llmq::CSigningManagerTestAccess::Push(globalManager, nodeId, recoveredSig); + } + } + llmq::CSigningManagerTestAccess::Push(globalManager, nodesAtGlobalLimit, recoveredSig); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::Count(globalManager), llmq::MAX_PENDING_RECSIGS_TOTAL); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::TrackedCount(globalManager), llmq::MAX_PENDING_RECSIGS_TOTAL); + BOOST_CHECK(!llmq::CSigningManagerTestAccess::HasNode(globalManager, nodesAtGlobalLimit)); + + globalManager.RemoveNodesIf([](NodeId nodeId) { + return nodeId == 0; + }); + llmq::CSigningManagerTestAccess::Push(globalManager, nodesAtGlobalLimit, recoveredSig); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::Count(globalManager), llmq::MAX_PENDING_RECSIGS_TOTAL - llmq::MAX_PENDING_RECSIGS_PER_NODE + 1); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::TrackedCount(globalManager), llmq::MAX_PENDING_RECSIGS_TOTAL - llmq::MAX_PENDING_RECSIGS_PER_NODE + 1); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::Count(globalManager, nodesAtGlobalLimit), 1U); +} + +BOOST_AUTO_TEST_CASE(pending_recovered_sig_limit_is_concurrent) +{ + constexpr NodeId nodeCount = 12; + constexpr size_t attemptsPerNode = 1200; + std::vector threads; + threads.reserve(nodeCount); + + for (NodeId nodeId = 0; nodeId < nodeCount; ++nodeId) { + threads.emplace_back([&, nodeId] { + for (size_t i = 0; i < attemptsPerNode; ++i) { + llmq::CSigningManagerTestAccess::Push(manager, nodeId, recoveredSig); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::Count(manager), llmq::MAX_PENDING_RECSIGS_TOTAL); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::TrackedCount(manager), llmq::MAX_PENDING_RECSIGS_TOTAL); + for (NodeId nodeId = 0; nodeId < nodeCount; ++nodeId) { + BOOST_CHECK_LE(llmq::CSigningManagerTestAccess::Count(manager, nodeId), llmq::MAX_PENDING_RECSIGS_PER_NODE); + } +} + +BOOST_AUTO_TEST_CASE(pending_recovered_sigs_are_removed_when_banned) +{ + llmq::CSigningManagerTestAccess::Push(manager, 1, recoveredSig); + llmq::CSigningManagerTestAccess::Push(manager, 2, recoveredSig); + + ScopedSigningManager scopedSigningManager(&manager); + llmq::CSigSharesManager sigSharesManager; + { + LOCK(cs_main); + sigSharesManager.MarkNodeBanned(1); + } + + BOOST_CHECK(!llmq::CSigningManagerTestAccess::HasNode(manager, 1)); + BOOST_CHECK(llmq::CSigningManagerTestAccess::HasNode(manager, 2)); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::Count(manager), 1U); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::TrackedCount(manager), 1U); +} + +BOOST_AUTO_TEST_CASE(pending_recovered_sig_drain_prunes_node) +{ + recoveredSig.UpdateHash(); + llmq::CSigningManagerTestAccess::Push(manager, 1, recoveredSig); + + llmq::CSigningManagerTestAccess::DrainKnown(manager, recoveredSig); + + BOOST_CHECK(!llmq::CSigningManagerTestAccess::HasNode(manager, 1)); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::Count(manager), 0U); + BOOST_CHECK_EQUAL(llmq::CSigningManagerTestAccess::TrackedCount(manager), 0U); +} + +BOOST_AUTO_TEST_SUITE_END()