From b26434df46b79e1bb87e3e0aa3c857ec6272398b Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 3 Aug 2026 09:58:51 -0500 Subject: [PATCH] fix: bound DKG pending queues across NodeId reconnects CDKGPendingMessages limited intake per NodeId, but NodeId is an ephemeral per-connection identifier. Reconnecting peers could therefore obtain a fresh quota while previously queued payloads remained resident; observer-mode handlers could retain that state indefinitely. Bound each message-type queue by actual serialized payload bytes derived from its quorum parameters while retaining the per-connection count limit. Reject duplicates and invalid sizes before applying capacity pressure, evict the oldest payload from the peer with the largest occupancy when necessary, and keep locally generated DKG messages exempt. Add a protocol-handler finalization hook that releases a disconnecting peer's unprocessed payloads, quota, and corresponding seen hashes. Cover byte accounting, reconnect bounds, duplicate replay, eviction, local messages, and disconnect cleanup with focused unit tests. --- src/Makefile.test.include | 1 + src/llmq/dkgsessionhandler.cpp | 202 ++++++++++++++++++++++++++-- src/llmq/dkgsessionhandler.h | 79 ++++++++++- src/llmq/net_dkg.cpp | 51 ++----- src/llmq/net_dkg.h | 1 + src/net_processing.cpp | 7 + src/net_processing.h | 4 + src/test/llmq_dkg_pending_tests.cpp | 167 +++++++++++++++++++++++ 8 files changed, 456 insertions(+), 56 deletions(-) create mode 100644 src/test/llmq_dkg_pending_tests.cpp diff --git a/src/Makefile.test.include b/src/Makefile.test.include index 5e9d47f8c4c7..0b9209ee7443 100644 --- a/src/Makefile.test.include +++ b/src/Makefile.test.include @@ -137,6 +137,7 @@ BITCOIN_TESTS =\ test/limitedmap_tests.cpp \ test/llmq_blockprocessor_tests.cpp \ test/llmq_dkg_tests.cpp \ + test/llmq_dkg_pending_tests.cpp \ test/llmq_chainlock_tests.cpp \ test/llmq_commitment_tests.cpp \ test/llmq_hash_tests.cpp \ diff --git a/src/llmq/dkgsessionhandler.cpp b/src/llmq/dkgsessionhandler.cpp index c9258ff353f9..fd24fb9980f1 100644 --- a/src/llmq/dkgsessionhandler.cpp +++ b/src/llmq/dkgsessionhandler.cpp @@ -4,19 +4,70 @@ #include +#include +#include #include +#include +#include #include +#include +#include #include namespace llmq { +size_t MaxDKGMessageSize(std::string_view msg_type, const Consensus::LLMQParams& params) +{ + constexpr size_t COMPACT{5}; + constexpr size_t PREFIX{1 + 32 + 32}; + constexpr size_t PUBKEY{BLS_CURVE_PUBKEY_SIZE}; + constexpr size_t SIG{BLS_CURVE_SIG_SIZE}; + constexpr size_t SECKEY{BLS_CURVE_SECKEY_SIZE}; + constexpr size_t BLOB{COMPACT + 128}; + constexpr size_t SLACK{1024}; + constexpr size_t HARD_CEILING{size_t{1} << 20}; + + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + const size_t threshold = params.threshold > 0 ? static_cast(params.threshold) : 0; + + size_t cap{0}; + if (msg_type == NetMsgType::QCONTRIB) { + cap = PREFIX + (COMPACT + threshold * PUBKEY) + (PUBKEY + 32 + COMPACT + size * BLOB) + SIG; + } else if (msg_type == NetMsgType::QJUSTIFICATION) { + cap = PREFIX + (COMPACT + size * (4 + SECKEY)) + SIG; + } else if (msg_type == NetMsgType::QCOMPLAINT) { + cap = PREFIX + 2 * (COMPACT + (size + 7) / 8) + SIG; + } else if (msg_type == NetMsgType::QPCOMMITMENT) { + cap = PREFIX + (COMPACT + (size + 7) / 8) + PUBKEY + 32 + 2 * SIG; + } else { + return HARD_CEILING; + } + cap += SLACK; + return std::min(cap, HARD_CEILING); +} + +namespace { +size_t MaxMessagesPerNode(const Consensus::LLMQParams& params) +{ + return params.size > 0 ? static_cast(params.size) * 2 : 0; +} + +size_t MaxPendingBytes(std::string_view msg_type, const Consensus::LLMQParams& params) +{ + // A round needs at most one message per member. Keep the existing doubled + // allowance for equivocation evidence, but apply it once to the whole queue + // instead of once for each ephemeral NodeId. + return MaxMessagesPerNode(params) * MaxDKGMessageSize(msg_type, params); +} +} // namespace + CDKGSessionHandler::CDKGSessionHandler(const Consensus::LLMQParams& _params) : params{_params}, // we allow size*2 messages as we need to make sure we see bad behavior (double messages) - pendingContributions{(size_t)_params.size * 2}, - pendingComplaints{(size_t)_params.size * 2}, - pendingJustifications{(size_t)_params.size * 2}, - pendingPrematureCommitments{(size_t)_params.size * 2} + pendingContributions{MaxMessagesPerNode(_params), MaxPendingBytes(NetMsgType::QCONTRIB, _params)}, + pendingComplaints{MaxMessagesPerNode(_params), MaxPendingBytes(NetMsgType::QCOMPLAINT, _params)}, + pendingJustifications{MaxMessagesPerNode(_params), MaxPendingBytes(NetMsgType::QJUSTIFICATION, _params)}, + pendingPrematureCommitments{MaxMessagesPerNode(_params), MaxPendingBytes(NetMsgType::QPCOMMITMENT, _params)} { if (params.type == Consensus::LLMQType::LLMQ_NONE) { throw std::runtime_error("Can't initialize CDKGSessionHandler with LLMQ_NONE type."); @@ -25,23 +76,92 @@ CDKGSessionHandler::CDKGSessionHandler(const Consensus::LLMQParams& _params) : CDKGSessionHandler::~CDKGSessionHandler() = default; +std::list::iterator CDKGPendingMessages::EraseEntry( + std::list::iterator it) +{ + seenMessages.erase(it->hash); + if (it->from >= 0) pendingBytes -= it->bytes; + if (auto qit = queuedBytesPerNode.find(it->from); qit != queuedBytesPerNode.end()) { + qit->second -= it->bytes; + if (qit->second == 0) queuedBytesPerNode.erase(qit); + } + return pendingMessages.erase(it); +} + +bool CDKGPendingMessages::EvictGreediestNode() +{ + // Prefer the peer pinning the most payload memory. + std::optional victim; + size_t victim_bytes{0}; + for (const auto& [node, bytes] : queuedBytesPerNode) { + if (node < 0) continue; // never evict our own messages + if (bytes > victim_bytes) { + victim = node; + victim_bytes = bytes; + } + } + if (!victim.has_value()) return false; + + // Drop that peer's oldest message: it is the least likely to still be + // relevant to the current phase. + for (auto it = pendingMessages.begin(); it != pendingMessages.end(); ++it) { + if (it->from == *victim) { + LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- byte cap reached (%d), evicting oldest of peer=%d (%d bytes queued)\n", + __func__, maxPendingBytes, *victim, victim_bytes); + EraseEntry(it); + return true; + } + } + return false; +} + void CDKGPendingMessages::PushPendingMessage(NodeId from, std::shared_ptr pm, const uint256& hash) { LOCK(cs_messages); - if (messagesPerNode[from] >= maxMessagesPerNode) { + if (pm == nullptr) return; + + // Our own messages (from < 0) are produced by our phase handler, are a + // handful per round, and are the only way our contribution reaches the + // quorum. They must never be dropped by a peer-driven bound. + const bool is_own = from < 0; + + // A duplicate must be side-effect free. In particular, it must not consume + // the per-peer quota or evict a different peer from a full queue. + if (seenMessages.count(hash) != 0) { + LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from); + return; + } + + if (!is_own && messagesPerNode[from] >= maxMessagesPerNode) { // TODO ban? LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from); return; } - messagesPerNode[from]++; - if (!seenMessages.emplace(hash).second) { - LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from); + const size_t bytes = pm->size(); + if (!is_own && (bytes == 0 || bytes > maxPendingBytes)) { + LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- invalid message size (%d, cap %d), peer=%d\n", + __func__, bytes, maxPendingBytes, from); return; } - pendingMessages.emplace_back(std::make_pair(from, std::move(pm))); + // NodeId is ephemeral, so the queue-wide retention guarantee must be + // independent of how many times a peer reconnects. Account actual payload + // bytes and evict until the new message fits. + while (!is_own && pendingBytes > maxPendingBytes - bytes) { + if (!EvictGreediestNode()) { + return; + } + } + + seenMessages.emplace(hash); + pendingMessages.emplace_back(PendingMessage{from, std::move(pm), bytes, hash}); + if (!is_own) { + messagesPerNode[from]++; + pendingBytes += bytes; + queuedBytesPerNode[from] += bytes; + } } std::list CDKGPendingMessages::PopPendingMessages(size_t maxCount) @@ -50,7 +170,17 @@ std::list CDKGPendingMessages::PopPendingMes std::list ret; while (!pendingMessages.empty() && ret.size() < maxCount) { - ret.emplace_back(std::move(pendingMessages.front())); + auto& front = pendingMessages.front(); + ret.emplace_back(front.from, std::move(front.msg)); + // Popped messages are handed to the DKG session; their content hash stays + // in seenMessages so AlreadyHave() keeps suppressing re-requests. + if (front.from >= 0) { + pendingBytes -= front.bytes; + } + if (auto qit = queuedBytesPerNode.find(front.from); qit != queuedBytesPerNode.end()) { + qit->second -= front.bytes; + if (qit->second == 0) queuedBytesPerNode.erase(qit); + } pendingMessages.pop_front(); } @@ -67,10 +197,54 @@ void CDKGPendingMessages::Clear() { LOCK(cs_messages); pendingMessages.clear(); + pendingBytes = 0; messagesPerNode.clear(); + queuedBytesPerNode.clear(); seenMessages.clear(); } +void CDKGPendingMessages::RemoveNode(NodeId nodeId) +{ + // Own/local enqueues use from=-1 and are not tied to a peer disconnect. + if (nodeId < 0) { + return; + } + + LOCK(cs_messages); + messagesPerNode.erase(nodeId); + + // Runs under ::cs_main (via PeerManagerImpl::FinalizeNode), so skip the list + // scan entirely for the overwhelmingly common case of a peer that never + // queued a DKG message. + if (queuedBytesPerNode.find(nodeId) == queuedBytesPerNode.end()) { + return; + } + + for (auto it = pendingMessages.begin(); it != pendingMessages.end();) { + if (it->from == nodeId) { + // Free the content-hash slot too; otherwise a reconnecting attacker + // can grow seenMessages without bound even after payloads are dropped + // (especially in observer mode where Clear() never runs). The hash is + // stored alongside the payload, so no re-hashing happens here. + it = EraseEntry(it); + } else { + ++it; + } + } +} + +size_t CDKGPendingMessages::Size() const +{ + LOCK(cs_messages); + return pendingMessages.size(); +} + +size_t CDKGPendingMessages::SizeBytes() const +{ + LOCK(cs_messages); + return pendingBytes; +} + void CDKGSessionHandler::ClearPendingMessages() { pendingContributions.Clear(); @@ -78,4 +252,12 @@ void CDKGSessionHandler::ClearPendingMessages() pendingJustifications.Clear(); pendingPrematureCommitments.Clear(); } + +void CDKGSessionHandler::RemoveNode(NodeId nodeId) +{ + pendingContributions.RemoveNode(nodeId); + pendingComplaints.RemoveNode(nodeId); + pendingJustifications.RemoveNode(nodeId); + pendingPrematureCommitments.RemoveNode(nodeId); +} } // namespace llmq diff --git a/src/llmq/dkgsessionhandler.h b/src/llmq/dkgsessionhandler.h index be55bfcbaa8a..7c3dae773db9 100644 --- a/src/llmq/dkgsessionhandler.h +++ b/src/llmq/dkgsessionhandler.h @@ -7,6 +7,7 @@ #include // for NodeId #include +#include // PendingMessage stores a uint256 by value #include #include @@ -17,7 +18,6 @@ class CDataStream; class CBlockIndex; -class uint256; namespace Consensus { struct LLMQParams; @@ -41,6 +41,9 @@ enum class QuorumPhase { Idle, }; +//! Upper bound used both for DKG intake and pending-queue byte budgets. +size_t MaxDKGMessageSize(std::string_view msg_type, const Consensus::LLMQParams& params); + /** * Acts as a FIFO queue for incoming DKG messages. The reason we need this is that deserialization of these messages * is too slow to be processed in the main message handler thread. So, instead of processing them directly from the @@ -48,6 +51,21 @@ enum class QuorumPhase { * handler thread. * * Each message type has it's own instance of this class. + * + * Retention is bounded both per NodeId (@ref maxMessagesPerNode) and by serialized + * payload bytes across all NodeIds (@ref maxPendingBytes). The byte bound is + * required because NodeId is ephemeral: a reconnecting peer receives a fresh id + * and would otherwise obtain a fresh full per-node quota forever. @ref RemoveNode + * drops a disconnected peer's counter and still-queued payloads so live peers can + * use the capacity. + * + * To avoid letting the first peer to fill the queue reserve it indefinitely, + * overflow evicts the oldest payload belonging to the peer that currently + * occupies the most bytes. Duplicate hashes are rejected before this policy + * runs, so a replay cannot evict anything. Locally generated messages + * (@c from < 0) bypass the bound entirely — they are produced by our own phase + * handler, are a handful per round, and are the only path by which our own + * contribution reaches the quorum. */ class CDKGPendingMessages { @@ -55,21 +73,53 @@ class CDKGPendingMessages using BinaryMessage = std::pair>; private: + struct PendingMessage { + NodeId from; + std::shared_ptr msg; + size_t bytes; + //! Content hash as inserted into @ref seenMessages. Stored so eviction + //! and disconnect cleanup never have to re-hash the payload (which would + //! put unbounded SHA256 work under ::cs_main via FinalizeNode). + uint256 hash; + }; + const size_t maxMessagesPerNode; + const size_t maxPendingBytes; mutable Mutex cs_messages; - std::list pendingMessages GUARDED_BY(cs_messages); + std::list pendingMessages GUARDED_BY(cs_messages); + size_t pendingBytes GUARDED_BY(cs_messages){0}; + //! Per-connection lifetime quota counter. Deliberately not decremented when a + //! message is popped: the quota is per connection, not per queue slot, so a + //! peer cannot regain capacity by getting its messages processed. Released + //! only by @ref RemoveNode and @ref Clear. std::map messagesPerNode GUARDED_BY(cs_messages); + //! Live queued bytes per NodeId, kept in sync with @ref pendingMessages. + //! Used to pick the greediest peer on overflow and to skip the list scan in + //! @ref RemoveNode when the disconnecting peer has nothing queued. + std::map queuedBytesPerNode GUARDED_BY(cs_messages); Uint256HashSet seenMessages GUARDED_BY(cs_messages); + //! Erase one queue entry, keeping byte accounting and seenMessages in sync. + //! Returns the iterator following the erased entry. + std::list::iterator EraseEntry(std::list::iterator it) + EXCLUSIVE_LOCKS_REQUIRED(cs_messages); + //! Drop the oldest queued message of the peer occupying the most bytes. + //! Never evicts our own (from < 0) messages. Returns false if nothing could + //! be evicted (i.e. the queue holds only own messages). + bool EvictGreediestNode() EXCLUSIVE_LOCKS_REQUIRED(cs_messages); + public: - explicit CDKGPendingMessages(size_t _maxMessagesPerNode) : - maxMessagesPerNode(_maxMessagesPerNode) {}; + explicit CDKGPendingMessages(size_t _maxMessagesPerNode, size_t _maxPendingBytes) : + maxMessagesPerNode(_maxMessagesPerNode), + maxPendingBytes(_maxPendingBytes) {}; /** * Enqueue a serialized DKG message under @p from with content hash @p hash. * Caller is responsible for hashing the payload and (for real peers) * routing the erase-request to PeerManager. Drops the message silently on - * per-node capacity overflow or duplicate hash. + * per-node capacity overflow, an individually over-budget payload, or duplicate + * hash. On queue-wide byte-capacity overflow, the greediest peer is evicted + * instead (see the class comment). */ void PushPendingMessage(NodeId from, std::shared_ptr pm, const uint256& hash) EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); @@ -78,6 +128,24 @@ class CDKGPendingMessages bool HasSeen(const uint256& hash) const EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); void Clear() EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); + /** + * Drop @p nodeId's per-node counter, any still-queued payloads from that + * peer, and the corresponding content hashes in @ref seenMessages. Called + * on disconnect so a reconnecting attacker cannot pin memory (or grow + * seenMessages) under abandoned NodeIds. + * + * Only *un-processed* messages are dropped. Anything already popped by the + * phase handler has been accepted into the DKG session state and is + * unaffected, so disconnecting a peer never invalidates a contribution the + * session already took — the DKG completes as before. + */ + void RemoveNode(NodeId nodeId) EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); + + //! Number of still-queued binary messages (for tests / diagnostics). + size_t Size() const EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); + //! Serialized payload bytes currently retained for remote peers. + size_t SizeBytes() const EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); + // Might return nullptr messages, which indicates that deserialization failed for some reason template std::vector>> PopAndDeserializeMessages(size_t maxCount) @@ -123,6 +191,7 @@ class CDKGSessionHandler virtual ~CDKGSessionHandler(); void ClearPendingMessages(); + void RemoveNode(NodeId nodeId); public: virtual bool GetContribution(const uint256& hash, CDKGContribution& ret) const { return false; } diff --git a/src/llmq/net_dkg.cpp b/src/llmq/net_dkg.cpp index 5eddc10c1a9d..da5d1ef16644 100644 --- a/src/llmq/net_dkg.cpp +++ b/src/llmq/net_dkg.cpp @@ -31,47 +31,6 @@ namespace llmq { namespace { -// Upper bound on the serialized size of a well-formed DKG message of the given -// type for the given quorum params. Used to reject oversized payloads at intake -// before any deserialization or retention, which closes the low-cost memory -// amplification window (a legitimate message is bounded by quorum params, far -// below the 3 MiB transport cap). Generous slack is added and the result is -// clamped to a hard ceiling so a future params change can never silently re-open -// the full transport window. -size_t MaxDKGMessageSize(std::string_view msg_type, const Consensus::LLMQParams& params) -{ - constexpr size_t COMPACT = 5; // max CompactSize for any realistic count - constexpr size_t PREFIX = 1 + 32 + 32; // llmqType + quorumHash + proTxHash - constexpr size_t PUBKEY = BLS_CURVE_PUBKEY_SIZE; // 48 - constexpr size_t SIG = BLS_CURVE_SIG_SIZE; // 96 - constexpr size_t SECKEY = BLS_CURVE_SECKEY_SIZE; // 32 - constexpr size_t BLOB = COMPACT + 128; // encrypted seckey blob, generous - constexpr size_t SLACK = 1024; - constexpr size_t HARD_CEILING = size_t{1} << 20; // 1 MiB - - const size_t size = params.size > 0 ? static_cast(params.size) : 0; - const size_t threshold = params.threshold > 0 ? static_cast(params.threshold) : 0; - - size_t cap = 0; - if (msg_type == NetMsgType::QCONTRIB) { - // llmqType/quorumHash/proTxHash + vvec + contributions(IES) + sig - cap = PREFIX + (COMPACT + threshold * PUBKEY) + (PUBKEY + 32 + COMPACT + size * BLOB) + SIG; - } else if (msg_type == NetMsgType::QJUSTIFICATION) { - // ... + contributions(index u32 + seckey) + sig - cap = PREFIX + (COMPACT + size * (4 + SECKEY)) + SIG; - } else if (msg_type == NetMsgType::QCOMPLAINT) { - // ... + 2 dynamic bitsets (badMembers, complainForMembers) + sig - cap = PREFIX + 2 * (COMPACT + (size + 7) / 8) + SIG; - } else if (msg_type == NetMsgType::QPCOMMITMENT) { - // ... + validMembers bitset + quorumPublicKey + quorumVvecHash + quorumSig + sig - cap = PREFIX + (COMPACT + (size + 7) / 8) + PUBKEY + 32 + 2 * SIG; - } else { - return HARD_CEILING; - } - cap += SLACK; - return cap < HARD_CEILING ? cap : HARD_CEILING; -} - // Cheap, param-only structural validation of a pushed DKG message, run at intake // before retention. Deserializes a COPY of the payload (leaving the caller's bytes // intact for the pending queue and its inventory hash) and checks only safe upper @@ -601,6 +560,16 @@ bool NetDKG::ProcessGetData(CNode& pfrom, const CInv& inv, const CNetMsgMaker& m return false; } +void NetDKG::FinalizeNode(NodeId nodeid) +{ + // Release per-NodeId DKG queue state on disconnect. Required for both + // active and observer mode: observer mode never runs ClearPendingMessages, + // and active mode only clears at the start of the next DKG round. + m_qdkgsman.ForEachHandler([nodeid](CDKGSessionHandler& handler) { + handler.RemoveNode(nodeid); + }); +} + void NetDKG::Start() { if (m_active == nullptr) return; diff --git a/src/llmq/net_dkg.h b/src/llmq/net_dkg.h index 81019392400e..eb635a10d848 100644 --- a/src/llmq/net_dkg.h +++ b/src/llmq/net_dkg.h @@ -70,6 +70,7 @@ class NetDKG final : public NetHandler EXCLUSIVE_LOCKS_REQUIRED(!cs_indexed_quorums_cache); bool AlreadyHave(const CInv& inv) override; bool ProcessGetData(CNode& pfrom, const CInv& inv, const CNetMsgMaker& msgMaker) override; + void FinalizeNode(NodeId nodeid) override; /** * Drives one phase-handler thread per ActiveDKGSessionHandler in active mode; * no-op in observer mode (no curSession to drive). diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 97554d4ecfde..cf7019299078 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -1795,6 +1795,13 @@ void PeerManagerImpl::FinalizeNode(const CNode& node) { m_addrman.Connected(node.addr); } + // Let protocol handlers free any state keyed on the ephemeral NodeId. + // Still under cs_main (LOCK scope is the whole function); handlers must only + // take their own locks and must not re-enter PeerManager. + for (const auto& handler : m_handlers) { + handler->FinalizeNode(nodeid); + } + LogPrint(BCLog::NET, "Cleared nodestate for peer=%d\n", nodeid); } diff --git a/src/net_processing.h b/src/net_processing.h index 761da8001c07..924309c9bca5 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -130,6 +130,10 @@ class NetHandler virtual void Interrupt() {} virtual void Schedule(CScheduler& scheduler) {} + //! Called when a peer is fully disconnected. Handlers may free any + //! per-NodeId state keyed on the ephemeral connection id. + virtual void FinalizeNode(NodeId /*nodeid*/) {} + virtual void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv) {} // It returns true, if NetHandler has a responsibility about having this type of inventory and has corresponding data. diff --git a/src/test/llmq_dkg_pending_tests.cpp b/src/test/llmq_dkg_pending_tests.cpp new file mode 100644 index 000000000000..3e838ef253fe --- /dev/null +++ b/src/test/llmq_dkg_pending_tests.cpp @@ -0,0 +1,167 @@ +// Copyright (c) 2026 The Dash Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#include +#include +#include +#include +#include +#include + +#include + +#include +#include + +using namespace llmq; + +BOOST_FIXTURE_TEST_SUITE(llmq_dkg_pending_tests, BasicTestingSetup) + +namespace { +std::shared_ptr MakePayload(uint32_t salt, size_t bytes = 8) +{ + auto pm = std::make_shared(SER_NETWORK, PROTOCOL_VERSION); + *pm << salt; + pm->resize(bytes); + return pm; +} + +uint256 MakeHash(uint32_t salt) +{ + CHashWriter hw(SER_GETHASH, 0); + hw << salt; + return hw.GetHash(); +} + +void PushUnique(CDKGPendingMessages& pending, NodeId from, uint32_t salt, size_t bytes = 8) +{ + pending.PushPendingMessage(from, MakePayload(salt, bytes), MakeHash(salt)); +} +} // namespace + +BOOST_AUTO_TEST_CASE(pending_messages_per_node_cap) +{ + constexpr size_t max_per_node{3}; + CDKGPendingMessages pending{max_per_node, /*max_pending_bytes=*/1024}; + + for (uint32_t i = 0; i < max_per_node + 5; ++i) { + PushUnique(pending, /*from=*/1, i); + } + + BOOST_CHECK_EQUAL(pending.Size(), max_per_node); + BOOST_CHECK_EQUAL(pending.SizeBytes(), max_per_node * 8); +} + +// A fresh NodeId on every reconnect must not grant fresh capacity in this queue. +BOOST_AUTO_TEST_CASE(pending_messages_byte_bounded_across_node_ids) +{ + constexpr size_t byte_cap{24}; + CDKGPendingMessages pending{/*max_messages_per_node=*/100, byte_cap}; + + for (NodeId node = 1; node <= 20; ++node) { + PushUnique(pending, node, static_cast(node), /*bytes=*/8); + BOOST_CHECK_LE(pending.SizeBytes(), byte_cap); + } + + BOOST_CHECK_EQUAL(pending.SizeBytes(), byte_cap); + BOOST_CHECK_EQUAL(pending.Size(), 3); +} + +BOOST_AUTO_TEST_CASE(pending_messages_oversized_payload_rejected) +{ + CDKGPendingMessages pending{/*max_messages_per_node=*/10, /*max_pending_bytes=*/16}; + + PushUnique(pending, /*from=*/1, /*salt=*/1, /*bytes=*/17); + pending.PushPendingMessage(/*from=*/2, std::make_shared(SER_NETWORK, PROTOCOL_VERSION), MakeHash(2)); + BOOST_CHECK_EQUAL(pending.Size(), 0); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 0); +} + +// Duplicate replay must be side-effect free even when the queue is full. The +// old draft evicted before checking seenMessages, allowing replay-only griefing. +BOOST_AUTO_TEST_CASE(pending_messages_duplicate_does_not_evict) +{ + CDKGPendingMessages pending{/*max_messages_per_node=*/10, /*max_pending_bytes=*/16}; + PushUnique(pending, /*from=*/1, /*salt=*/1, /*bytes=*/8); + PushUnique(pending, /*from=*/2, /*salt=*/2, /*bytes=*/8); + + pending.PushPendingMessage(/*from=*/3, MakePayload(/*salt=*/99, /*bytes=*/8), MakeHash(/*salt=*/1)); + + BOOST_CHECK_EQUAL(pending.Size(), 2); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 16); + const auto msgs = pending.PopPendingMessages(/*maxCount=*/10); + BOOST_REQUIRE_EQUAL(msgs.size(), 2); + BOOST_CHECK_EQUAL(msgs.front().first, 1); + BOOST_CHECK_EQUAL(msgs.back().first, 2); +} + +BOOST_AUTO_TEST_CASE(pending_messages_remove_node_releases_bytes) +{ + CDKGPendingMessages pending{/*max_messages_per_node=*/10, /*max_pending_bytes=*/24}; + PushUnique(pending, /*from=*/1, /*salt=*/1, /*bytes=*/8); + PushUnique(pending, /*from=*/2, /*salt=*/2, /*bytes=*/16); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 24); + + pending.RemoveNode(/*nodeId=*/2); + BOOST_CHECK_EQUAL(pending.Size(), 1); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 8); + + PushUnique(pending, /*from=*/3, /*salt=*/3, /*bytes=*/16); + BOOST_CHECK_EQUAL(pending.Size(), 2); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 24); +} + +// Capacity pressure is charged by bytes, so a peer pinning most of the memory +// gives up its oldest payload when a new peer arrives. +BOOST_AUTO_TEST_CASE(pending_messages_evicts_greediest_peer_by_bytes) +{ + CDKGPendingMessages pending{/*max_messages_per_node=*/10, /*max_pending_bytes=*/32}; + PushUnique(pending, /*from=*/1, /*salt=*/1, /*bytes=*/8); + PushUnique(pending, /*from=*/1, /*salt=*/2, /*bytes=*/8); + PushUnique(pending, /*from=*/1, /*salt=*/3, /*bytes=*/8); + PushUnique(pending, /*from=*/2, /*salt=*/4, /*bytes=*/8); + + PushUnique(pending, /*from=*/3, /*salt=*/5, /*bytes=*/4); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 28); + + const auto msgs = pending.PopPendingMessages(/*maxCount=*/10); + BOOST_REQUIRE_EQUAL(msgs.size(), 4); + BOOST_CHECK_EQUAL(std::count_if(msgs.begin(), msgs.end(), [](const auto& msg) { return msg.first == 1; }), 2); + BOOST_CHECK_EQUAL(std::count_if(msgs.begin(), msgs.end(), [](const auto& msg) { return msg.first == 2; }), 1); + BOOST_CHECK_EQUAL(std::count_if(msgs.begin(), msgs.end(), [](const auto& msg) { return msg.first == 3; }), 1); +} + +BOOST_AUTO_TEST_CASE(pending_messages_own_message_survives_full_queue) +{ + CDKGPendingMessages pending{/*max_messages_per_node=*/10, /*max_pending_bytes=*/16}; + PushUnique(pending, /*from=*/1, /*salt=*/1, /*bytes=*/16); + PushUnique(pending, /*from=*/-1, /*salt=*/2, /*bytes=*/8); + + BOOST_CHECK_EQUAL(pending.Size(), 2); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 16); + const auto msgs = pending.PopPendingMessages(/*maxCount=*/10); + BOOST_CHECK(std::any_of(msgs.begin(), msgs.end(), [](const auto& msg) { return msg.first < 0; })); +} + +BOOST_AUTO_TEST_CASE(pending_messages_remove_node_only_drops_unprocessed) +{ + CDKGPendingMessages pending{/*max_messages_per_node=*/4, /*max_pending_bytes=*/64}; + + for (uint32_t i = 0; i < 4; ++i) { + PushUnique(pending, /*from=*/7, i); + } + auto processed = pending.PopPendingMessages(/*maxCount=*/2); + BOOST_REQUIRE_EQUAL(processed.size(), 2); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 16); + + pending.RemoveNode(/*nodeId=*/7); + BOOST_CHECK_EQUAL(pending.Size(), 0); + BOOST_CHECK_EQUAL(pending.SizeBytes(), 0); + BOOST_CHECK_EQUAL(processed.size(), 2); + + PushUnique(pending, /*from=*/7, /*salt=*/5); + BOOST_CHECK_EQUAL(pending.Size(), 1); +} + +BOOST_AUTO_TEST_SUITE_END()