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()