Skip to content

fix: bound pending sig share queue #7413

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

Closed
PastaPastaPasta wants to merge 1 commit into from
Closed

fix: bound pending sig share queue #7413

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
91 changes: 84 additions & 7 deletions src/llmq/signing_shares.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -201,14 +206,52 @@ 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()) {
sessionByRecvId.erase(it->second.recvSessionId);
sessions.erase(it);
}
requestedSigShares.EraseAllForSignHash(signHash);
pendingIncomingSigShares.EraseAllForSignHash(signHash);
ErasePendingIncomingSigSharesForSession(signHash);
}

//////////////////////
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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__,
Expand All @@ -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<std::pair<NodeId, uint256>, StaticSaltedHasher> uniqueSignHashes;
size_t erasedCount{0};
IterateNodesRandom(
nodeStates,
[&]() {
Expand All @@ -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();
Expand All @@ -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;
}
Expand All @@ -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();
});
}

Expand Down Expand Up @@ -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};
Expand Down Expand Up @@ -1366,6 +1436,7 @@ void CSigSharesManager::Cleanup()
AssertLockHeld(cs);
sigSharesRequested.Erase(k);
});
pendingIncomingSigSharesCount -= it->second.GetPendingIncomingSigSharesCount();
nodeStates.erase(nodeId);
}
}
Expand All @@ -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);
Expand All @@ -1397,6 +1470,7 @@ void CSigSharesManager::RemoveNodesIf(std::function<bool(NodeId)> predicate)
AssertLockHeld(cs);
sigSharesRequested.Erase(k);
});
pendingIncomingSigSharesCount -= it->second.GetPendingIncomingSigSharesCount();
it = nodeStates.erase(it);
} else {
++it;
Expand Down Expand Up @@ -1425,6 +1499,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId)
sigSharesRequested.Erase(k);
});
nodeState.requestedSigShares.Clear();
pendingIncomingSigSharesCount -= nodeState.ClearPendingIncomingSigShares();
nodeState.banned = true;
}

Expand All @@ -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<CRecoveredSig> CSigSharesManager::SignAndProcessSingleShare(PendingSignatureData work)
Expand Down
15 changes: 15 additions & 0 deletions src/llmq/signing_shares.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,12 @@ constexpr uint32_t UNINITIALIZED_SESSION_ID{std::numeric_limits<uint32_t>::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:
Expand Down Expand Up @@ -340,6 +346,7 @@ class CSigSharesNodeState
std::unordered_map<uint32_t, Session*> sessionByRecvId;

SigShareMap<CSigShare> pendingIncomingSigShares;
size_t pendingIncomingSigSharesCount{0};
SigShareMap<int64_t> requestedSigShares;

bool banned{false};
Expand All @@ -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);
};
Expand Down Expand Up @@ -405,6 +417,7 @@ class CSigSharesManager : public llmq::CRecoveredSigsListener
Uint256HashMap<int64_t> timeSeenForSessions GUARDED_BY(cs);

std::unordered_map<NodeId, CSigSharesNodeState> nodeStates GUARDED_BY(cs);
size_t pendingIncomingSigSharesCount GUARDED_BY(cs){0};
SigShareMap<std::pair<NodeId, int64_t>> sigSharesRequested GUARDED_BY(cs);
SigShareMap<bool> sigSharesQueuedToAnnounce GUARDED_BY(cs);

Expand Down Expand Up @@ -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<uint16_t, CBLSLazySignature>& in);
bool TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesNodeState& nodeState, const CSigShare& sigShare)
EXCLUSIVE_LOCKS_REQUIRED(cs);

void RemoveSigSharesForSession(const uint256& signHash) EXCLUSIVE_LOCKS_REQUIRED(cs);

Expand Down
48 changes: 48 additions & 0 deletions src/test/llmq_utils_tests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading