diff --git a/src/llmq/net_signing.cpp b/src/llmq/net_signing.cpp index d433072b4a13..b8f937b634bb 100644 --- a/src/llmq/net_signing.cpp +++ b/src/llmq/net_signing.cpp @@ -257,6 +257,10 @@ void NetSigning::WorkThreadSigning() constexpr auto CLEANUP_INTERVAL{5s}; if (cleanupThrottler.TryCleanup(CLEANUP_INTERVAL)) { m_sig_manager.Cleanup(); + // Drop pending recovered sigs queued by banned peers so a flood's backlog does not + // persist after the peer is banned (RemoveBannedNodeStates only cleans the sig-shares + // subsystem, not m_sig_manager's pending recovered sigs). + m_sig_manager.RemoveNodesIf([this](NodeId node_id) { return m_peer_manager->PeerIsBanned(node_id); }); } // TODO Wakeup when pending signing is needed? diff --git a/src/llmq/signing.cpp b/src/llmq/signing.cpp index e44411f01d12..efdcd13cd747 100644 --- a/src/llmq/signing.cpp +++ b/src/llmq/signing.cpp @@ -394,7 +394,36 @@ void CSigningManager::VerifyAndProcessRecoveredSig(NodeId from, std::shared_ptr< return; } + // Backpressure: bound the pending queue so a peer cannot enqueue faster than we drain and + // exhaust memory. Drop silently (no misbehaviour) — verification is deferred to a single + // worker thread, so an honest peer can legitimately outrun the drain during a burst. + if (pendingRecoveredSigsCount >= MAX_PENDING_RECSIGS_TOTAL) { + LogPrint(BCLog::LLMQ, "CSigningManager::%s -- global pending recovered sigs cap reached (%d), dropping sig from node=%d\n", + __func__, MAX_PENDING_RECSIGS_TOTAL, from); + return; + } + if (auto it = pendingRecoveredSigs.find(from); + it != pendingRecoveredSigs.end() && it->second.size() >= MAX_PENDING_RECSIGS_PER_NODE) { + LogPrint(BCLog::LLMQ, "CSigningManager::%s -- per-node pending recovered sigs cap reached (%d), dropping sig from node=%d\n", + __func__, MAX_PENDING_RECSIGS_PER_NODE, from); + return; + } + pendingRecoveredSigs[from].emplace_back(std::move(recoveredSig)); + ++pendingRecoveredSigsCount; +} + +void CSigningManager::RemoveNodesIf(const std::function& predicate) +{ + LOCK(cs_pending); + for (auto it = pendingRecoveredSigs.begin(); it != pendingRecoveredSigs.end();) { + if (predicate(it->first)) { + pendingRecoveredSigsCount -= it->second.size(); + it = pendingRecoveredSigs.erase(it); + } else { + ++it; + } + } } bool CSigningManager::CollectPendingRecoveredSigsToVerify( @@ -411,6 +440,7 @@ bool CSigningManager::CollectPendingRecoveredSigsToVerify( // TODO: refactor it to remove duplicated code with `CSigSharesManager::CollectPendingSigSharesToVerify` std::unordered_set, StaticSaltedHasher> uniqueSignHashes; + size_t erasedCount{0}; IterateNodesRandom(pendingRecoveredSigs, [&]() { return uniqueSignHashes.size() < maxUniqueSessions; }, [&](NodeId nodeId, std::list>& ns) { @@ -425,8 +455,21 @@ bool CSigningManager::CollectPendingRecoveredSigsToVerify( retSigShares[nodeId].emplace_back(recSig); } ns.erase(ns.begin()); + ++erasedCount; return !ns.empty(); }, rnd); + pendingRecoveredSigsCount -= erasedCount; + + // Prune drained (now-empty) node entries so the map only holds nodes with pending sigs. + // This keeps VerifyAndProcessRecoveredSig's global-cap check cheap and reclaims the queues + // of disconnected nodes once drained, without waiting for them to be banned. + for (auto it = pendingRecoveredSigs.begin(); it != pendingRecoveredSigs.end();) { + if (it->second.empty()) { + it = pendingRecoveredSigs.erase(it); + } else { + ++it; + } + } if (retSigShares.empty()) { return false; diff --git a/src/llmq/signing.h b/src/llmq/signing.h index 0d9cc71bf7a4..d91926edf6f2 100644 --- a/src/llmq/signing.h +++ b/src/llmq/signing.h @@ -14,6 +14,7 @@ #include #include +#include #include #include #include @@ -159,6 +160,15 @@ class CRecoveredSigsListener [[nodiscard]] virtual RecoveredSigResult HandleNewRecoveredSig(const CRecoveredSig& recoveredSig) = 0; }; +// Backpressure bounds for the pending (not-yet-verified) recovered sig queue. Verification of an +// incoming recovered sig is deferred to a single worker thread doing batched BLS verification, +// while ingestion happens on the network threads without any crypto cost. Without a bound, a peer +// (or several) can enqueue faster than the queue drains, growing memory without limit. These caps +// bound the queue; over-cap messages are dropped silently (no misbehaviour) since an honest peer +// can legitimately relay recovered sigs faster than we drain them during a burst. +static constexpr size_t MAX_PENDING_RECSIGS_PER_NODE{1000}; +static constexpr size_t MAX_PENDING_RECSIGS_TOTAL{10000}; + class CSigningManager { private: @@ -169,6 +179,9 @@ class CSigningManager mutable Mutex cs_pending; // Incoming and not verified yet std::unordered_map>> pendingRecoveredSigs GUARDED_BY(cs_pending); + // Running total of entries across all of pendingRecoveredSigs, kept in sync with it so the + // MAX_PENDING_RECSIGS_TOTAL check doesn't need to rescan the map on every incoming message. + size_t pendingRecoveredSigsCount GUARDED_BY(cs_pending){0}; Uint256HashMap> pendingReconstructedRecoveredSigs GUARDED_BY(cs_pending); FastRandomContext rnd GUARDED_BY(cs_pending); @@ -207,6 +220,9 @@ class CSigningManager size_t maxUniqueSessions, std::unordered_map>>& retSigShares, std::unordered_map, CBLSPublicKey, StaticSaltedHasher>& ret_pubkeys) EXCLUSIVE_LOCKS_REQUIRED(!cs_pending); + // Drop the pending (not-yet-verified) recovered sigs of any node matching the predicate, e.g. + // banned peers. Without this, a flooded peer's backlog would persist even after it is banned. + void RemoveNodesIf(const std::function& predicate) EXCLUSIVE_LOCKS_REQUIRED(!cs_pending); [[nodiscard]] std::vector GetListeners() const EXCLUSIVE_LOCKS_REQUIRED(!cs_listeners); // Returns true if recovered sigs should be send to listeners [[nodiscard]] bool ProcessRecoveredSig(const std::shared_ptr& recoveredSig)