From 46c8c471dcf55f0e631cd4281d2413a63e7b3775 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Sun, 12 Jul 2026 09:06:17 +1000 Subject: [PATCH 01/13] net_processing: bump last_inv_sequence for bip35 messages explicitly This avoids relying on the bump that occurs through the normal INV process, which is confusing. --- src/net_processing.cpp | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 27f0a63c558..3397b388cb3 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -6109,6 +6109,10 @@ bool PeerManagerImpl::SendMessages(CNode& node) // Respond to BIP35 mempool requests if (fSendTrickle && tx_relay->m_send_mempool) { auto vtxinfo = m_mempool.infoAll(); + + // Ensure we'll respond to GETDATA requests for anything we're about to announce + tx_relay->m_last_inv_sequence = WITH_LOCK(m_mempool.cs, return m_mempool.GetSequence()); + tx_relay->m_send_mempool = false; const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; From 026f70e05f96f5795c888ed69ba9128169ec5b7f Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Thu, 19 Feb 2026 20:44:42 +1000 Subject: [PATCH 02/13] net_processing: Remove per-peer rate-limiting Per-peer rate limiting introduces storage and compute costs proportional to the number of peers. This has caused severe bugs in the past, and continues to be a risk in the event of periods of extremely high rates of transaction submission. Avoid these problems by always completely emptying the m_tx_inventory_to_send queue when processing it. Note that this increases the potential size of INV messages we send for normal tx relay from ~1000 (limited by INVENTORY_BROADCAST_MAX) to potentially 50000 (limited by MAX_INV_SZ). --- src/net_processing.cpp | 21 ++++++--------------- 1 file changed, 6 insertions(+), 15 deletions(-) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 3397b388cb3..b5b2375fe7f 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -169,13 +169,9 @@ static constexpr auto INBOUND_INVENTORY_BROADCAST_INTERVAL{5s}; static constexpr auto OUTBOUND_INVENTORY_BROADCAST_INTERVAL{2s}; /** Maximum rate of inventory items to send per second. * Limits the impact of low-fee transaction floods. */ -static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND{14}; +[[maybe_unused]] static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND{14}; /** Target number of tx inventory items to send per transmission. */ -static constexpr unsigned int INVENTORY_BROADCAST_TARGET = INVENTORY_BROADCAST_PER_SECOND * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL); -/** Maximum number of inventory items to send per transmission. */ -static constexpr unsigned int INVENTORY_BROADCAST_MAX = 1000; -static_assert(INVENTORY_BROADCAST_MAX >= INVENTORY_BROADCAST_TARGET, "INVENTORY_BROADCAST_MAX too low"); -static_assert(INVENTORY_BROADCAST_MAX <= node::MAX_PEER_TX_ANNOUNCEMENTS, "INVENTORY_BROADCAST_MAX too high"); +[[maybe_unused]] static constexpr unsigned int INVENTORY_BROADCAST_TARGET = INVENTORY_BROADCAST_PER_SECOND * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL); /** Average delay between feefilter broadcasts in seconds. */ static constexpr auto AVG_FEEFILTER_BROADCAST_INTERVAL{10min}; /** Maximum feefilter broadcast delay after significant change. */ @@ -6074,7 +6070,7 @@ bool PeerManagerImpl::SendMessages(CNode& node) std::vector vInv; { LOCK(peer.m_block_inv_mutex); - vInv.reserve(std::max(peer.m_blocks_for_inv_relay.size(), INVENTORY_BROADCAST_TARGET)); + vInv.reserve(peer.m_blocks_for_inv_relay.size()); // Add blocks for (const uint256& hash : peer.m_blocks_for_inv_relay) { @@ -6152,16 +6148,12 @@ bool PeerManagerImpl::SendMessages(CNode& node) } const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; // Topologically and fee-rate sort the inventory we send for privacy and priority reasons. - // A heap is used so that not all items need sorting if only a few are being sent. + // A heap is used even though all items are being processed/sent. CompareInvMempoolOrder compareInvMempoolOrder(&m_mempool); std::make_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); - // No reason to drain out at many times the network's capacity, - // especially since we have many peers and some will draw much shorter delays. - unsigned int nRelayedTransactions = 0; LOCK(tx_relay->m_bloom_filter_mutex); - size_t broadcast_max{INVENTORY_BROADCAST_TARGET + (tx_relay->m_tx_inventory_to_send.size()/1000)*5}; - broadcast_max = std::min(INVENTORY_BROADCAST_MAX, broadcast_max); - while (!vInvTx.empty() && nRelayedTransactions < broadcast_max) { + vInv.reserve(std::min(MAX_INV_SZ, tx_relay->m_tx_inventory_to_send.size())); + while (!vInvTx.empty()) { // Fetch the top element from the heap std::pop_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); std::set::iterator it = vInvTx.back(); @@ -6191,7 +6183,6 @@ bool PeerManagerImpl::SendMessages(CNode& node) if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*txinfo.tx)) continue; // Send vInv.push_back(inv); - nRelayedTransactions++; if (vInv.size() == MAX_INV_SZ) { MakeAndPushMessage(node, NetMsgType::INV, vInv); vInv.clear(); From 6cfc65d2103e9fd09deea833011bf0d80136f4b1 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Thu, 19 Feb 2026 19:36:25 +1000 Subject: [PATCH 03/13] txmempool: Add ExtractBestByMiningScoreWithTopology Add a method for (partially) sorting a batch of transactions (specified as a std::vector of wtxids) per mempool order, designed for transaction relay. --- src/txmempool.cpp | 58 +++++++++++++++++++++++++++++++++++++++++++++++ src/txmempool.h | 13 +++++++++++ 2 files changed, 71 insertions(+) diff --git a/src/txmempool.cpp b/src/txmempool.cpp index 86cbc1adc43..4bb86c9bace 100644 --- a/src/txmempool.cpp +++ b/src/txmempool.cpp @@ -553,6 +553,64 @@ void CTxMemPool::check(const CCoinsViewCache& active_coins_tip, int64_t spendhei assert(innerUsage == cachedInnerUsage); } +std::vector CTxMemPool::ExtractBestByMiningScoreWithTopology(std::vector& wtxids, size_t n_to_sort) const +{ + /* This function takes a vector of `wtxids`, and returns the + * best mempool entries corresponding to those `wtxids` (by mining + * score/topology). It updates the input `wtxids` so that multiple + * calls with the same vector will drain that vector to empty. + * + * It operates under the following constraints: + * - wtxids that do not correspond to a mempool entry are dropped + * - the return vector contains no duplicates, either with itself + * or with the updated `wtxids` input. + * - the return vector will have `n_to_sort` entries (or `wtxids` + will become empty). + * - the `wtxids` vector will be reduced by at least `n_to_sort` + * entries (or will become empty). + */ + + auto cmp = [&](const auto& a, const auto& b) EXCLUSIVE_LOCKS_REQUIRED(cs) noexcept { return m_txgraph->CompareMainOrder(*a, *b) < 0; }; + + std::vector res; + + n_to_sort = std::min(wtxids.size(), n_to_sort); + if (n_to_sort > 0) { + res.reserve(wtxids.size()); + std::sort(wtxids.begin(), wtxids.end()); + for (auto it = wtxids.begin(); it != wtxids.end(); ++it) { + // skip duplicates + auto itnext = it + 1; + if (itnext != wtxids.end() && *it == *itnext) continue; + + if (auto i{GetIter(*it)}; i.has_value()) { + res.push_back(i.value()); + } + } + wtxids.clear(); + + if (!res.empty()) { + auto begin = res.begin(); + auto end = res.end(); + auto middle = end; + if (n_to_sort >= res.size()) { + // use regular sort when sorting everything + std::sort(begin, end, cmp); + } else { + middle = begin + n_to_sort; + std::partial_sort(begin, middle, end, cmp); + } + auto it = middle; + while (it != end) { + wtxids.push_back((*it)->GetTx().GetWitnessHash()); + ++it; + } + res.erase(middle, end); + } + } + return res; +} + bool CTxMemPool::CompareMiningScoreWithTopology(const Wtxid& hasha, const Wtxid& hashb) const { /* Return `true` if hasha should be considered sooner than hashb, namely when: diff --git a/src/txmempool.h b/src/txmempool.h index 1a5405d5bf0..dcc9ed874e7 100644 --- a/src/txmempool.h +++ b/src/txmempool.h @@ -331,6 +331,19 @@ public: void removeForReorg(CChain& chain, std::function filter_final_and_mature) EXCLUSIVE_LOCKS_REQUIRED(cs, cs_main); void removeForBlock(const std::vector& vtx, unsigned int nBlockHeight) EXCLUSIVE_LOCKS_REQUIRED(cs); + /** Look up wtxids in the mempool and (partially) sort by mining score. + * + * The @p n_to_sort best entries are removed from @p wtxids and their + * corresponding txiter entries are returned. In addition wtxids + * that are duplicates or were not found in the mempool are silently + * dropped from @p wtxids. The returned vector is ordered from best + * to worst (by CompareMainOrder). Entries remaining in @p wtxids + * are in unspecified order. + * + * Note that the returned `txiter` values may become invalidated once + * mempool.cs is released. + */ + std::vector ExtractBestByMiningScoreWithTopology(std::vector& wtxids, size_t n_to_sort) const EXCLUSIVE_LOCKS_REQUIRED(cs); bool CompareMiningScoreWithTopology(const Wtxid& hasha, const Wtxid& hashb) const; bool isSpent(const COutPoint& outpoint) const; unsigned int GetTransactionsUpdated() const; From e1b7490fbc91c9a119d659461fa426810000f0ae Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Tue, 2 Jun 2026 16:10:40 +1000 Subject: [PATCH 04/13] net_processing: Replace CompareInvMempoolOrder Remove CompareInvMempoolOrder, replacing it with the new ExtractBestByMiningScoreWithTopology. The trickle send code is reworked accordingly. --- src/net_processing.cpp | 92 +++++++++++++++++------------------------- 1 file changed, 37 insertions(+), 55 deletions(-) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index b5b2375fe7f..b3e32499781 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -295,11 +295,11 @@ struct Peer { * us or we have announced to the peer. We use this to avoid announcing * the same (w)txid to a peer that already has the transaction. */ CRollingBloomFilter m_tx_inventory_known_filter GUARDED_BY(m_tx_inventory_mutex){50000, 0.000001}; - /** Set of wtxids we still have to announce. For non-wtxid-relay peers, + /** Vector of wtxids we still have to announce. For non-wtxid-relay peers, * we retrieve the txid from the corresponding mempool transaction when * constructing the `inv` message. We use the mempool to sort transactions * in dependency order before relay, so this does not have to be sorted. */ - std::set m_tx_inventory_to_send GUARDED_BY(m_tx_inventory_mutex); + std::vector m_tx_inventory_to_send GUARDED_BY(m_tx_inventory_mutex); /** Whether the peer has requested us to send our complete mempool. Only * permitted if the peer has NetPermissionFlags::Mempool or we advertise * NODE_BLOOM. See BIP35. */ @@ -2285,7 +2285,7 @@ void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wt const uint256& hash{peer.m_wtxid_relay ? wtxid.ToUint256() : txid.ToUint256()}; if (!tx_relay->m_tx_inventory_known_filter.contains(hash)) { - tx_relay->m_tx_inventory_to_send.insert(wtxid); + tx_relay->m_tx_inventory_to_send.push_back(wtxid); } } } @@ -5627,22 +5627,6 @@ void PeerManagerImpl::MaybeSendFeefilter(CNode& pto, Peer& peer, std::chrono::mi } } -namespace { -class CompareInvMempoolOrder -{ - const CTxMemPool* m_mempool; -public: - explicit CompareInvMempoolOrder(CTxMemPool* mempool) : m_mempool{mempool} {} - - bool operator()(std::set::iterator a, std::set::iterator b) - { - /* As std::make_heap produces a max-heap, we want the entries with the - * higher mining score to sort later. */ - return m_mempool->CompareMiningScoreWithTopology(*b, *a); - } -}; -} // namespace - bool PeerManagerImpl::RejectIncomingTxs(const CNode& peer) const { // block-relay-only peers may never send txs to us @@ -6112,6 +6096,9 @@ bool PeerManagerImpl::SendMessages(CNode& node) tx_relay->m_send_mempool = false; const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; + // we'll send everything in the mempool momentarily, so this is redundant + tx_relay->m_tx_inventory_to_send.clear(); + LOCK(tx_relay->m_bloom_filter_mutex); for (const auto& txinfo : vtxinfo) { @@ -6120,7 +6107,6 @@ bool PeerManagerImpl::SendMessages(CNode& node) const auto inv = peer.m_wtxid_relay ? CInv{MSG_WTX, wtxid.ToUint256()} : CInv{MSG_TX, txid.ToUint256()}; - tx_relay->m_tx_inventory_to_send.erase(wtxid); // Don't send transactions that peers will not put into their mempool if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) { @@ -6140,47 +6126,47 @@ bool PeerManagerImpl::SendMessages(CNode& node) // Determine transactions to relay if (fSendTrickle) { - // Produce a vector with all candidates for sending - std::vector::iterator> vInvTx; - vInvTx.reserve(tx_relay->m_tx_inventory_to_send.size()); - for (std::set::iterator it = tx_relay->m_tx_inventory_to_send.begin(); it != tx_relay->m_tx_inventory_to_send.end(); it++) { - vInvTx.push_back(it); - } - const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; // Topologically and fee-rate sort the inventory we send for privacy and priority reasons. - // A heap is used even though all items are being processed/sent. - CompareInvMempoolOrder compareInvMempoolOrder(&m_mempool); - std::make_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); - LOCK(tx_relay->m_bloom_filter_mutex); - vInv.reserve(std::min(MAX_INV_SZ, tx_relay->m_tx_inventory_to_send.size())); - while (!vInvTx.empty()) { - // Fetch the top element from the heap - std::pop_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); - std::set::iterator it = vInvTx.back(); - vInvTx.pop_back(); - auto wtxid = *it; - // Remove it from the to-be-sent set - tx_relay->m_tx_inventory_to_send.erase(it); - // Not in the mempool anymore? don't bother sending it. - auto txinfo = m_mempool.info(wtxid); - if (!txinfo.tx) { - continue; + // (sorted from higher priority to lowest, skipping low fee) + const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; + + auto inv_tx = [&]() EXCLUSIVE_LOCKS_REQUIRED(tx_relay->m_tx_inventory_mutex) { + auto& invs = tx_relay->m_tx_inventory_to_send; + std::vector res; + + if (invs.size() == 0) return res; + + // if previous allocations were excessive, shrink to the current size + if (invs.capacity() > 2 * invs.size()) invs.shrink_to_fit(); + + LOCK(m_mempool.cs); + auto txiters = m_mempool.ExtractBestByMiningScoreWithTopology(invs, invs.size()); + res.reserve(txiters.size()); + for (auto txiter : txiters) { + if (txiter->GetFee() < filterrate.GetFee(txiter->GetTxSize())) { + continue; // higher feerate CPFP txs may follow, so just skip, don't stop + } + res.push_back(txiter->GetSharedTx()); } + // Ensure we'll respond to GETDATA requests for anything we're about to announce + tx_relay->m_last_inv_sequence = m_mempool.GetSequence(); + return res; + }(); + + LOCK(tx_relay->m_bloom_filter_mutex); + vInv.reserve(std::min(MAX_INV_SZ, vInv.size() + inv_tx.size())); + for (auto& tx : inv_tx) { // `TxRelay::m_tx_inventory_known_filter` contains either txids or wtxids // depending on whether our peer supports wtxid-relay. Therefore, first // construct the inv and then use its hash for the filter check. const auto inv = peer.m_wtxid_relay ? - CInv{MSG_WTX, wtxid.ToUint256()} : - CInv{MSG_TX, txinfo.tx->GetHash().ToUint256()}; + CInv{MSG_WTX, tx->GetWitnessHash().ToUint256()} : + CInv{MSG_TX, tx->GetHash().ToUint256()}; // Check if not in the filter already if (tx_relay->m_tx_inventory_known_filter.contains(inv.hash)) { continue; } - // Peer told you to not send transactions at that feerate? Don't bother sending it. - if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) { - continue; - } - if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*txinfo.tx)) continue; + if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*tx)) continue; // Send vInv.push_back(inv); if (vInv.size() == MAX_INV_SZ) { @@ -6189,10 +6175,6 @@ bool PeerManagerImpl::SendMessages(CNode& node) } tx_relay->m_tx_inventory_known_filter.insert(inv.hash); } - - // Ensure we'll respond to GETDATA requests for anything we've just announced - LOCK(m_mempool.cs); - tx_relay->m_last_inv_sequence = m_mempool.GetSequence(); } } if (!vInv.empty()) From 749bb447f819428d4836c9f205d5212283fa3eac Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Thu, 19 Feb 2026 21:07:10 +1000 Subject: [PATCH 05/13] txmempool: Drop CompareMiningScoreWithTopology Now unused; replaced by ExtractBestByMiningScoreWithTopology. --- src/txmempool.cpp | 25 ++++--------------------- src/txmempool.h | 1 - 2 files changed, 4 insertions(+), 22 deletions(-) diff --git a/src/txmempool.cpp b/src/txmempool.cpp index 4bb86c9bace..e5ec3b32cab 100644 --- a/src/txmempool.cpp +++ b/src/txmempool.cpp @@ -458,7 +458,7 @@ void CTxMemPool::check(const CCoinsViewCache& active_coins_tip, int64_t spendhei assert(diagram.size() <= score_with_topo.size() + 1); assert(diagram.size() >= 1); - std::optional last_wtxid = std::nullopt; + std::optional last_iter = std::nullopt; auto diagram_iter = diagram.cbegin(); for (const auto& it : score_with_topo) { @@ -480,11 +480,10 @@ void CTxMemPool::check(const CCoinsViewCache& active_coins_tip, int64_t spendhei innerUsage += it->DynamicMemoryUsage(); const CTransaction& tx = it->GetTx(); - // CompareMiningScoreWithTopology should agree with GetSortedScoreWithTopology() - if (last_wtxid) { - assert(CompareMiningScoreWithTopology(*last_wtxid, tx.GetWitnessHash())); + if (last_iter) { + assert(m_txgraph->CompareMainOrder(**last_iter, *it) < 0); } - last_wtxid = tx.GetWitnessHash(); + last_iter = it; std::set setParentCheck; std::set setParentsStored; @@ -611,22 +610,6 @@ std::vector CTxMemPool::ExtractBestByMiningScoreWithTopology return res; } -bool CTxMemPool::CompareMiningScoreWithTopology(const Wtxid& hasha, const Wtxid& hashb) const -{ - /* Return `true` if hasha should be considered sooner than hashb, namely when: - * a is not in the mempool but b is, or - * both are in the mempool but a is sorted before b in the total mempool ordering - * (which takes dependencies and (chunk) feerates into account). - */ - LOCK(cs); - auto j{GetIter(hashb)}; - if (!j.has_value()) return false; - auto i{GetIter(hasha)}; - if (!i.has_value()) return true; - - return m_txgraph->CompareMainOrder(*i.value(), *j.value()) < 0; -} - std::vector CTxMemPool::GetSortedScoreWithTopology() const { std::vector iters; diff --git a/src/txmempool.h b/src/txmempool.h index dcc9ed874e7..d2324d1c79d 100644 --- a/src/txmempool.h +++ b/src/txmempool.h @@ -344,7 +344,6 @@ public: * mempool.cs is released. */ std::vector ExtractBestByMiningScoreWithTopology(std::vector& wtxids, size_t n_to_sort) const EXCLUSIVE_LOCKS_REQUIRED(cs); - bool CompareMiningScoreWithTopology(const Wtxid& hasha, const Wtxid& hashb) const; bool isSpent(const COutPoint& outpoint) const; unsigned int GetTransactionsUpdated() const; void AddTransactionsUpdated(unsigned int n); From 7927650e56d2cebc26983b332eec35944f29971a Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Thu, 19 Feb 2026 19:35:50 +1000 Subject: [PATCH 06/13] util/tokenbucket.h: Provide a generic TokenBucket class This is a simple token bucket parameterized on clock type, used in the following commit. --- src/test/util_tests.cpp | 132 ++++++++++++++++++++++++++++++++++++++++ src/util/tokenbucket.h | 69 +++++++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 src/util/tokenbucket.h diff --git a/src/test/util_tests.cpp b/src/test/util_tests.cpp index e26234f8696..15c54d412ce 100644 --- a/src/test/util_tests.cpp +++ b/src/test/util_tests.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include #include @@ -1930,4 +1931,135 @@ BOOST_AUTO_TEST_CASE(gib_string_literal_test) BOOST_CHECK_EQUAL(32_GiB, 32768_MiB); } +BOOST_AUTO_TEST_CASE(token_bucket_initial_value) +{ + // Initial value is clamped to cap + util::TokenBucket b1(/*rate=*/1, /*value=*/100, /*cap=*/10); + BOOST_CHECK_EQUAL(b1.value(), 10); + + // Initial value below cap is kept as-is + util::TokenBucket b2(/*rate=*/1, /*value=*/5, /*cap=*/10); + BOOST_CHECK_EQUAL(b2.value(), 5); +} + +BOOST_AUTO_TEST_CASE(token_bucket_first_increment) +{ + // First increment establishes the time baseline but does not refill + util::TokenBucket b(/*rate=*/100, /*value=*/0, /*cap=*/1000); + b.increment(NodeClock::time_point{10s}); + BOOST_CHECK_EQUAL(b.value(), 0); + + // Second increment refills based on elapsed time + b.increment(NodeClock::time_point{15s}); + BOOST_CHECK_EQUAL(b.value(), 500); // 100/s * 5s +} + +BOOST_AUTO_TEST_CASE(token_bucket_refill_caps) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/90, /*cap=*/100); + b.increment(NodeClock::time_point{1s}); + b.increment(NodeClock::time_point{100s}); // would add 990, but cap is 100 + BOOST_CHECK_EQUAL(b.value(), 100); +} + +BOOST_AUTO_TEST_CASE(token_bucket_time_backwards) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/50, /*cap=*/200); + b.increment(NodeClock::time_point{10s}); + b.increment(NodeClock::time_point{5s}); // backwards, no change + BOOST_CHECK_EQUAL(b.value(), 50); + b.increment(NodeClock::time_point{15s}); // forwards takes backwards into account + BOOST_CHECK_EQUAL(b.value(), 150); +} + +BOOST_AUTO_TEST_CASE(token_bucket_decrement_no_debt) +{ + // Default debt=0: returns false at exactly 0 + util::TokenBucket b(/*rate=*/1, /*value=*/3, /*cap=*/10); + BOOST_CHECK(b.decrement(1)); // 3 -> 2 + BOOST_CHECK(b.decrement(1)); // 2 -> 1 + BOOST_CHECK(!b.decrement(1)); // 1 -> 0, at floor + BOOST_CHECK_EQUAL(b.value(), 0); + BOOST_CHECK(!b.decrement(1)); // 0 -> -1, despite being at floor + BOOST_CHECK_EQUAL(b.value(), -1); +} + +BOOST_AUTO_TEST_CASE(token_bucket_decrement_with_debt) +{ + util::TokenBucket b(/*rate=*/1, /*value=*/2, /*cap=*/10); + BOOST_CHECK(b.decrement(1, -3)); // 2 -> 1 + BOOST_CHECK(b.decrement(1, -3)); // 1 -> 0 + BOOST_CHECK(b.decrement(1, -3)); // 0 -> -1, still above -3 + BOOST_CHECK(b.decrement(1, -3)); // -1 -> -2, still above -3 + BOOST_CHECK(!b.decrement(1, -3)); // -2 -> -3, at floor + BOOST_CHECK_EQUAL(b.value(), -3); +} + +BOOST_AUTO_TEST_CASE(token_bucket_drain_and_refill) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/20, /*cap=*/100); + b.decrement(20); // drain to 0 + BOOST_CHECK_EQUAL(b.value(), 0); + + b.increment(NodeClock::time_point{1s}); + b.increment(NodeClock::time_point{4s}); // +30 + BOOST_CHECK_EQUAL(b.value(), 30); +} + + +BOOST_AUTO_TEST_CASE(token_bucket_first_increment_at_epoch) +{ + // The first increment establishes the baseline (no refill) even when it + // lands exactly on the clock epoch; later increments then refill normally. + util::TokenBucket b(/*rate=*/100, /*value=*/0, /*cap=*/1000); + b.increment(NodeClock::time_point{0s}); + BOOST_CHECK_EQUAL(b.value(), 0); + b.increment(NodeClock::time_point{5s}); + BOOST_CHECK_EQUAL(b.value(), 500); // 100/s * 5s +} + +BOOST_AUTO_TEST_CASE(token_bucket_at_cap_advances_baseline) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/100, /*cap=*/100); + BOOST_CHECK_EQUAL(b.value(), 100); // already at cap + b.increment(NodeClock::time_point{1s}); // baseline established at 1s + b.increment(NodeClock::time_point{100s}); // 99s spent at the cap; baseline -> 100s + BOOST_CHECK_EQUAL(b.value(), 100); + + b.decrement(100); // drain to 0 + BOOST_CHECK_EQUAL(b.value(), 0); + + // refill doesn't "bank" the extra 99s we were at cap + b.increment(NodeClock::time_point{101s}); + BOOST_CHECK_EQUAL(b.value(), 10); + + // And when real time genuinely elapses, a single increment refills straight + // back to the cap immediately. + b.increment(NodeClock::time_point{200s}); // 99s elapsed -> +990, clamped to cap + BOOST_CHECK_EQUAL(b.value(), 100); +} + +BOOST_AUTO_TEST_CASE(token_bucket_fractional_refill) +{ + // Sub-second elapsed time accumulates fractional tokens via double math. + util::TokenBucket b(/*rate=*/10, /*value=*/0, /*cap=*/100); + b.increment(NodeClock::time_point{1s}); + b.increment(NodeClock::time_point{1250ms}); // 10/s * 0.25s = 2.5 + BOOST_CHECK_EQUAL(b.value(), 2.5); +} + +BOOST_AUTO_TEST_CASE(token_bucket_refill_from_debt) +{ + // Refilling from a negative (debt) balance accrues normally and still + // clamps to the cap rather than to debt + increment. + util::TokenBucket b(/*rate=*/10, /*value=*/0, /*cap=*/100); + BOOST_CHECK(!b.decrement(50)); // -> -50, below floor 0 + BOOST_CHECK_EQUAL(b.value(), -50); + b.increment(NodeClock::time_point{1s}); // baseline + b.increment(NodeClock::time_point{4s}); // +30 -> -20 + BOOST_CHECK_EQUAL(b.value(), -20); + b.increment(NodeClock::time_point{100s}); // +960 but clamped to cap + BOOST_CHECK_EQUAL(b.value(), 100); +} + BOOST_AUTO_TEST_SUITE_END() diff --git a/src/util/tokenbucket.h b/src/util/tokenbucket.h new file mode 100644 index 00000000000..ae8e8b53ad9 --- /dev/null +++ b/src/util/tokenbucket.h @@ -0,0 +1,69 @@ +// Copyright (c) The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#ifndef BITCOIN_UTIL_TOKENBUCKET_H +#define BITCOIN_UTIL_TOKENBUCKET_H + +#include + +namespace util { + +/** A token bucket rate limiter. + * + * Tokens are added at a steady rate (m_rate per second) up to a capacity + * cap (m_cap). Tokens are removed by calling decrement(), which returns + * false if the bucket is emptied. + * + * Typical usage: + * bucket.increment(now); // refill based on elapsed time + * if (bucket.value() >= 1) bucket.decrement(1); // consume a token + */ +template +class TokenBucket +{ +public: + using clock = Clock; + using time_point = typename Clock::time_point; + using duration = typename Clock::duration; + + const double m_rate{1}; //!< Tokens added per second + const double m_cap{0}; //!< Maximum token balance + + /** @param rate Tokens added per second. + * @param value Initial token balance (clamped to cap). + * @param cap Maximum token balance. */ + TokenBucket(double rate, double value, double cap) : m_rate{rate}, m_cap{cap}, m_value{std::min(value, cap)} {} + + /** Refill tokens based on elapsed time since last call. No refill + * occurs on the first call (establishes the time baseline). */ + void increment(const time_point& now) + { + if (now > m_last_updated) { + if (m_value < m_cap && m_last_updated > MIN_TIME) { + double inc = m_rate * std::chrono::duration_cast(now - m_last_updated).count(); + m_value = std::min(m_cap, m_value + inc); + } + } + m_last_updated = now; + } + + /** Consume n tokens. Returns false if the balance dropped to/below the given floor. */ + bool decrement(double n = 1.0, double floor = 0.0) + { + m_value -= n; + return (m_value > floor); + } + + /** Current token balance. */ + double value() const { return m_value; } + +private: + static constexpr time_point MIN_TIME{time_point::min()}; + time_point m_last_updated{MIN_TIME}; + double m_value{0}; +}; + +} // namespace util + +#endif // BITCOIN_UTIL_TOKENBUCKET_H From df31ee57aaf5d27dc740553f600fe948bcae3e61 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Sun, 12 Jul 2026 07:53:06 +1000 Subject: [PATCH 07/13] net_processing: add a global delay queue for sending txs Without the per-peer rate limiting, nodes can act as an amplifier for transaction spam -- receiving many transactions from one node, but relaying each of them to over 100 other nodes. Limit the impact of this by providing a global rate limit. This is implemented using dual token buckets, one that consumes a token for every transaction, and one that consumes a token for every serialized byte. This rate limits both per-tx resource usage (eg INV messages) and overall relay bandwidth. Main bucket parameters: * Count: 14tx/s rate, 420tx (30s) capacity * Size: 12MB/600s rate (4-6 blocks per target block interval), 50MB capacity The size bucket is expected to be large enough to almost never have an impact in normal usage, even during transaction storms, and is primarily intended to mitigate attack-like scenarios. Outbound connections get a separate pair of buckets, with rates boosted by a 2.5x multiplier. This avoids the excessive memory and CPU usage due to the 100x multiplier from the queues being per-peer. Note that this also reduces the size of INV messages we send for general tx relay back to a more reasonable level of under 600 txs in 99.999% of cases. --- src/net_processing.cpp | 198 +++++++++++++++++++++++++++---- test/functional/mempool_limit.py | 3 + 2 files changed, 176 insertions(+), 25 deletions(-) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index b3e32499781..ddeac0da94a 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -59,6 +59,7 @@ #include #include #include +#include #include #include @@ -172,6 +173,12 @@ static constexpr auto OUTBOUND_INVENTORY_BROADCAST_INTERVAL{2s}; [[maybe_unused]] static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND{14}; /** Target number of tx inventory items to send per transmission. */ [[maybe_unused]] static constexpr unsigned int INVENTORY_BROADCAST_TARGET = INVENTORY_BROADCAST_PER_SECOND * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL); +/** Multiplier for the inventory bucket rate for outbounds */ +static constexpr double OUTBOUND_INVENTORY_BUCKET_MULTIPLIER{Ticks(INBOUND_INVENTORY_BROADCAST_INTERVAL) / Ticks(OUTBOUND_INVENTORY_BROADCAST_INTERVAL)}; +/** Delay between checking inventory bucket and backlog */ +static constexpr auto INVENTORY_BUCKET_CHECK_DELAY{100ms}; +/** Empty backlog target capacity */ +static constexpr size_t INVENTORY_BUCKET_BACKLOG_CAPACITY{300}; /** Average delay between feefilter broadcasts in seconds. */ static constexpr auto AVG_FEEFILTER_BROADCAST_INTERVAL{10min}; /** Maximum feefilter broadcast delay after significant change. */ @@ -494,6 +501,59 @@ struct CNodeState { int64_t m_last_block_announcement{0}; }; +struct InvToSendBucket { + const double count_floor{0}; + std::vector backlog; + util::TokenBucket size_bucket; + util::TokenBucket count_bucket; + + /* Initialization rationale: + * + * Count bucket: Fills at rate*mult, total/initial capacity of 30s with mult=1 + * Size bucket: Fills at 12MB every 600s, times mult so expected to be 6 times + * the rate at which blocks can confirm transactions, but at least 3 times that in + * the worst case. High limit to avoid triggering even with large spikes, but a + * modest initial value to ensure that frequent node restarts don't raise the limit + * too much. + * Count floor: In order to avoid sorting the global backlog too often, we ensure + * that we always remove at least an average INV message's number of transactions + * each time we do work. (Or 50kB if the size bucket is the limiting factor) + */ + + static constexpr double SIZE_INIT{12'000'000}; // 12 MB initially + static constexpr double SIZE_CAP{50'000'000}; // 50 MB maximum + static constexpr double SIZE_REFILL{20'000}; // 20kB/s = 12MB/600s + + static constexpr double INBOUND_COUNT_SECONDS{30}; // cap/initial at 30s/mult worth of txs + + InvToSendBucket(unsigned int rate, double mult) + : count_floor{-1.0 * INVENTORY_BROADCAST_TARGET}, + size_bucket(/*rate=*/SIZE_REFILL * mult, /*value=*/SIZE_INIT, /*cap=*/SIZE_CAP), + count_bucket(/*rate=*/rate * mult, /*value=*/rate * INBOUND_COUNT_SECONDS, /*cap=*/rate * INBOUND_COUNT_SECONDS) + { + } + + bool avail() const + { + return !backlog.empty() && size_bucket.value() > 0 && count_bucket.value() > 0; + } + + void increment(NodeClock::time_point now) + { + size_bucket.increment(now); + count_bucket.increment(now); + } + + std::vector TakeForProcessing(CTxMemPool& mempool) EXCLUSIVE_LOCKS_REQUIRED(mempool.cs); + + bool decrement(double size) + { + bool size_ok = size_bucket.decrement(size, /*floor=*/-50e3); + bool count_ok = count_bucket.decrement(1, /*floor=*/count_floor); + return size_ok && count_ok; + } +}; + class PeerManagerImpl final : public PeerManager { public: @@ -520,9 +580,9 @@ public: void FinalizeNode(const CNode& node) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_headers_presync_mutex, !m_tx_download_mutex); bool HasAllDesirableServiceFlags(ServiceFlags services) const override; bool ProcessMessages(CNode& node, std::atomic& interrupt) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); bool SendMessages(CNode& node) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); /** Implement PeerManager */ void StartScheduledTasks(CScheduler& scheduler) override; @@ -535,7 +595,7 @@ public: std::vector GetPrivateBroadcastInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); std::vector AbortPrivateBroadcast(const uint256& id) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SendPings() override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); node::TransactionError InitiateTxBroadcastPrivate(const CTransactionRef& tx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SetBestBlock(int height, std::chrono::seconds time) override { @@ -549,7 +609,7 @@ public: private: void ProcessMessage(Peer& peer, CNode& pfrom, const std::string& msg_type, DataStream& vRecv, NodeClock::time_point time_received, const std::atomic& interruptMsgProc) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); /** Consider evicting an outbound peer based on the amount of time they've been behind our tip */ void ConsiderEviction(CNode& pto, Peer& peer, std::chrono::seconds time_in_seconds) EXCLUSIVE_LOCKS_REQUIRED(cs_main, g_msgproc_mutex); @@ -558,7 +618,7 @@ private: void EvictExtraOutboundPeers(NodeClock::time_point now) EXCLUSIVE_LOCKS_REQUIRED(cs_main); /** Retrieve unbroadcast transactions from the mempool and reattempt sending to peers */ - void ReattemptInitialBroadcast(CScheduler& scheduler) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void ReattemptInitialBroadcast(CScheduler& scheduler) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); /** Rebroadcast stale private transactions (already broadcast but not received back from the network). */ void ReattemptPrivateBroadcast(CScheduler& scheduler); @@ -616,13 +676,13 @@ private: /** Handle a transaction whose result was MempoolAcceptResult::ResultType::VALID. * Updates m_txrequest, m_orphanage, and vExtraTxnForCompact. Also queues the tx for relay. */ void ProcessValidTx(NodeId nodeid, const CTransactionRef& tx, const std::list& replaced_transactions) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex, !m_inv_to_send_mutex); /** Handle the results of package validation: calls ProcessValidTx and ProcessInvalidTx for * individual transactions, and caches rejection for the package as a group. */ void ProcessPackageResult(const node::PackageToValidate& package_to_validate, const PackageMempoolAcceptResult& package_result) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex, !m_inv_to_send_mutex); /** * Reconsider orphan transactions after a parent has been accepted to the mempool. @@ -636,7 +696,7 @@ private: * will be empty. */ bool ProcessOrphanTx(Peer& peer) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); /** Process a single headers message from a peer. * @@ -1101,6 +1161,13 @@ private: /// The transactions to be broadcast privately. PrivateBroadcast m_tx_for_private_broadcast; + + mutable Mutex m_inv_to_send_mutex ACQUIRED_BEFORE(m_mempool.cs); + InvToSendBucket m_inbound_inv_bucket GUARDED_BY(m_inv_to_send_mutex); + InvToSendBucket m_outbound_inv_bucket GUARDED_BY(m_inv_to_send_mutex); + std::atomic m_next_inv_bucket_check{NodeClock::time_point::min()}; + + void ProcessInvBacklog(NodeClock::time_point now, bool backlog_bumped=false) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); }; const CNodeState* PeerManagerImpl::State(NodeId pnode) const @@ -2039,7 +2106,9 @@ PeerManagerImpl::PeerManagerImpl(CConnman& connman, AddrMan& addrman, m_mempool(pool), m_txdownloadman(node::TxDownloadOptions{pool, m_rng, opts.deterministic_rng}), m_warnings{warnings}, - m_opts{opts} + m_opts{opts}, + m_inbound_inv_bucket(/*rate=*/INVENTORY_BROADCAST_PER_SECOND, /*mult=*/1.0), + m_outbound_inv_bucket(/*rate=*/INVENTORY_BROADCAST_PER_SECOND, /*mult=*/OUTBOUND_INVENTORY_BUCKET_MULTIPLIER) { // While Erlay support is incomplete, it must be enabled explicitly via -txreconciliation. // This argument can go away after Erlay support is complete. @@ -2266,28 +2335,105 @@ void PeerManagerImpl::SendPings() for(auto& it : m_peer_map) it.second->m_ping_queued = true; } -void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) +std::vector InvToSendBucket::TakeForProcessing(CTxMemPool& mempool) { - for (const PeerRef& peer_ref : GetAllPeers()) { - if (!peer_ref) continue; - Peer& peer{*peer_ref}; + AssertLockHeld(mempool.cs); - auto tx_relay = peer.GetTxRelay(); - if (!tx_relay) continue; + size_t n_to_take = static_cast(std::max(count_bucket.value() - count_floor, 0)); - LOCK(tx_relay->m_tx_inventory_mutex); - // Only queue transactions for announcement once the version handshake - // is completed. The time of arrival for these transactions is - // otherwise at risk of leaking to a spy, if the spy is able to - // distinguish transactions received during the handshake from the rest - // in the announcement. - if (tx_relay->m_next_inv_send_time == 0s) continue; + std::vector best; - const uint256& hash{peer.m_wtxid_relay ? wtxid.ToUint256() : txid.ToUint256()}; - if (!tx_relay->m_tx_inventory_known_filter.contains(hash)) { - tx_relay->m_tx_inventory_to_send.push_back(wtxid); + auto itervec = mempool.ExtractBestByMiningScoreWithTopology(backlog, n_to_take); + bool tokens_left = true; + for (auto txiter : itervec) { + auto& wtxid = txiter->GetTx().GetWitnessHash(); + if (tokens_left) { + best.push_back(wtxid); + if (!decrement(txiter->GetTx().ComputeTotalSize())) { + tokens_left = false; + } + } else { + backlog.push_back(wtxid); } } + + // if the backlog is now empty, consider shrinking it if it's oversized + if (backlog.empty() && backlog.capacity() > INVENTORY_BUCKET_BACKLOG_CAPACITY) { + std::vector dummy; + dummy.reserve(INVENTORY_BUCKET_BACKLOG_CAPACITY); + dummy.swap(backlog); + } + + return best; +} + +void PeerManagerImpl::ProcessInvBacklog(NodeClock::time_point now, bool backlog_bumped) +{ + // Don't run the body of this function unless it's been a little + // while since the last run, or we just added a new tx to the backlog. + if (!backlog_bumped && now <= m_next_inv_bucket_check.load()) return; + m_next_inv_bucket_check = now + INVENTORY_BUCKET_CHECK_DELAY; + + LOCK(m_inv_to_send_mutex); + m_inbound_inv_bucket.increment(now); + m_outbound_inv_bucket.increment(now); + + // Early exit to skip pointlessly touching mempool lock + bool in_avail = m_inbound_inv_bucket.avail(); + bool out_avail = m_outbound_inv_bucket.avail(); + if (!in_avail && !out_avail) return; + + std::vector for_inbound; + std::vector for_outbound; + + { + LOCK(m_mempool.cs); + if (in_avail) for_inbound = m_inbound_inv_bucket.TakeForProcessing(m_mempool); + if (out_avail) for_outbound = m_outbound_inv_bucket.TakeForProcessing(m_mempool); + } + + if (!for_inbound.empty() || !for_outbound.empty()) { + bool any_inbound_connected = false; + bool any_outbound_connected = false; + for (const PeerRef& peer_ref : GetAllPeers()) { + if (!peer_ref) continue; + Peer& peer{*peer_ref}; + auto tx_relay = peer.GetTxRelay(); + if (!tx_relay) continue; + + LOCK(tx_relay->m_tx_inventory_mutex); + // Only queue transactions for announcement once the version handshake + // is completed. The time of arrival for these transactions is + // otherwise at risk of leaking to a spy, if the spy is able to + // distinguish transactions received during the handshake from the rest + // in the announcement. + if (tx_relay->m_next_inv_send_time == 0s) continue; + if (peer.m_is_inbound) { + any_inbound_connected = true; + } else { + any_outbound_connected = true; + } + for (auto& i : (peer.m_is_inbound ? for_inbound : for_outbound)) { + tx_relay->m_tx_inventory_to_send.push_back(i); + } + } + + // if the node has no in/outbound connections, clear the corresponding backlog entirely + // this reduces wasted memory, and avoids having the bucket artificially empty for when + // future peers do connect. + if (!any_inbound_connected) m_inbound_inv_bucket.backlog.clear(); + if (!any_outbound_connected) m_outbound_inv_bucket.backlog.clear(); + } +} + +void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid&, const Wtxid& wtxid) +{ + { + LOCK(m_inv_to_send_mutex); + m_inbound_inv_bucket.backlog.push_back(wtxid); + m_outbound_inv_bucket.backlog.push_back(wtxid); + } + ProcessInvBacklog(NodeClock::now(), /*backlog_bumped=*/true); } node::TransactionError PeerManagerImpl::InitiateTxBroadcastPrivate(const CTransactionRef& tx) @@ -5855,6 +6001,8 @@ bool PeerManagerImpl::SendMessages(CNode& node) MaybeSendSendHeaders(node, peer); + ProcessInvBacklog(now); + { LOCK(cs_main); diff --git a/test/functional/mempool_limit.py b/test/functional/mempool_limit.py index 4770e155de8..2e554bb80c5 100755 --- a/test/functional/mempool_limit.py +++ b/test/functional/mempool_limit.py @@ -5,6 +5,7 @@ """Test mempool limiting together/eviction with the wallet.""" from decimal import Decimal +import time from test_framework.mempool_util import ( fill_mempool, @@ -205,7 +206,9 @@ class MempoolLimitTest(BitcoinTestFramework): self.log.info('Check that mempoolminfee is minrelaytxfee') assert_equal(node.getmempoolinfo()['minrelaytxfee'], node.getmempoolinfo()["mempoolminfee"]) + node.setmocktime(int(time.time())-3600) fill_mempool(self, node) + node.setmocktime(0) # bump time forward so the rate limit buckets refresh and don't block broadcast # Deliberately try to create a tx with a fee less than the minimum mempool fee to assert that it does not get added to the mempool self.log.info('Create a mempool tx that will not pass mempoolminfee') From 6307bd034bfb07e45139e80b6177686c02b74d91 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Sat, 6 Jun 2026 04:11:55 +1000 Subject: [PATCH 08/13] net_processing: Provide a 30bpm heartbeat log while inv backlog is in use --- src/net_processing.cpp | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index ddeac0da94a..bc90b0c513b 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -179,6 +179,10 @@ static constexpr double OUTBOUND_INVENTORY_BUCKET_MULTIPLIER{Ticks m_next_inv_bucket_check{NodeClock::time_point::min()}; + std::optional m_next_inv_bucket_heartbeat GUARDED_BY(m_inv_to_send_mutex); void ProcessInvBacklog(NodeClock::time_point now, bool backlog_bumped=false) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); }; @@ -2378,6 +2383,27 @@ void PeerManagerImpl::ProcessInvBacklog(NodeClock::time_point now, bool backlog_ m_inbound_inv_bucket.increment(now); m_outbound_inv_bucket.increment(now); + // Regular heartbeat logging when there's a backlog + if (!m_next_inv_bucket_heartbeat.has_value()) { + if (m_inbound_inv_bucket.backlog.size() >= INVENTORY_BUCKET_BACKLOG_HEARTBEAT_MIN || m_outbound_inv_bucket.backlog.size() >= INVENTORY_BUCKET_BACKLOG_HEARTBEAT_MIN) { + m_next_inv_bucket_heartbeat = now; + } + } + if (m_next_inv_bucket_heartbeat.has_value() && now >= *m_next_inv_bucket_heartbeat) { + LogDebug(BCLog::NET, "Transaction rate-limiting backlog inbound=%d itok=%.1f isz=%.1f outbound=%d otok=%.1f osz=%.1f", + m_inbound_inv_bucket.backlog.size(), + m_inbound_inv_bucket.count_bucket.value(), + m_inbound_inv_bucket.size_bucket.value(), + m_outbound_inv_bucket.backlog.size(), + m_outbound_inv_bucket.count_bucket.value(), + m_outbound_inv_bucket.size_bucket.value()); + if (m_inbound_inv_bucket.backlog.empty() && m_outbound_inv_bucket.backlog.empty()) { + m_next_inv_bucket_heartbeat = std::nullopt; + } else { + m_next_inv_bucket_heartbeat = now + INVENTORY_BUCKET_BACKLOG_HEARTBEAT; + } + } + // Early exit to skip pointlessly touching mempool lock bool in_avail = m_inbound_inv_bucket.avail(); bool out_avail = m_outbound_inv_bucket.avail(); From 74a47a5207135a7a481cb36edfa64907a5d66563 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Thu, 19 Feb 2026 19:50:28 +1000 Subject: [PATCH 09/13] init: add -txsendrate configuration parameter Adds a debug-only configuration option to set the target transaction/second rate for relay to inbound connections. This is mostly intended to be set to artificially low values to aid in testing behaviour when a backlog occurs, but is also available in case the default 14tx/s target is somehow too low in practice. --- src/init.cpp | 4 ++++ src/net_processing.cpp | 11 +++-------- src/net_processing.h | 4 ++++ src/node/peerman_args.cpp | 4 ++++ src/node/txdownloadman.h | 2 +- 5 files changed, 16 insertions(+), 9 deletions(-) diff --git a/src/init.cpp b/src/init.cpp index 290f0936807..4eee0ac537d 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -710,6 +710,10 @@ void SetupServerArgs(ArgsManager& argsman, bool can_listen_ipc) OptionsCategory::NODE_RELAY); argsman.AddArg("-minrelaytxfee=", strprintf("Fees (in %s/kvB) smaller than this are considered zero fee for relaying, mining and transaction creation (default: %s)", CURRENCY_UNIT, FormatMoney(DEFAULT_MIN_RELAY_TX_FEE)), ArgsManager::ALLOW_ANY, OptionsCategory::NODE_RELAY); + argsman.AddArg("-txsendrate=", + strprintf("Set the maximum ongoing rate for sending transactions to (inbound) peers (default: %u tx/s)", + DEFAULT_TX_SEND_RATE), + ArgsManager::ALLOW_ANY | ArgsManager::DEBUG_ONLY, OptionsCategory::NODE_RELAY); argsman.AddArg("-privatebroadcast", strprintf( "Broadcast transactions submitted via sendrawtransaction RPC using short-lived " diff --git a/src/net_processing.cpp b/src/net_processing.cpp index bc90b0c513b..56c79c2ca4e 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -168,11 +168,6 @@ static constexpr auto INBOUND_INVENTORY_BROADCAST_INTERVAL{5s}; * Use a smaller delay as there is less privacy concern for them. * Blocks and peers with NetPermissionFlags::NoBan permission bypass this. */ static constexpr auto OUTBOUND_INVENTORY_BROADCAST_INTERVAL{2s}; -/** Maximum rate of inventory items to send per second. - * Limits the impact of low-fee transaction floods. */ -[[maybe_unused]] static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND{14}; -/** Target number of tx inventory items to send per transmission. */ -[[maybe_unused]] static constexpr unsigned int INVENTORY_BROADCAST_TARGET = INVENTORY_BROADCAST_PER_SECOND * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL); /** Multiplier for the inventory bucket rate for outbounds */ static constexpr double OUTBOUND_INVENTORY_BUCKET_MULTIPLIER{Ticks(INBOUND_INVENTORY_BROADCAST_INTERVAL) / Ticks(OUTBOUND_INVENTORY_BROADCAST_INTERVAL)}; /** Delay between checking inventory bucket and backlog */ @@ -531,7 +526,7 @@ struct InvToSendBucket { static constexpr double INBOUND_COUNT_SECONDS{30}; // cap/initial at 30s/mult worth of txs InvToSendBucket(unsigned int rate, double mult) - : count_floor{-1.0 * INVENTORY_BROADCAST_TARGET}, + : count_floor{-1.0 * rate * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL)}, size_bucket(/*rate=*/SIZE_REFILL * mult, /*value=*/SIZE_INIT, /*cap=*/SIZE_CAP), count_bucket(/*rate=*/rate * mult, /*value=*/rate * INBOUND_COUNT_SECONDS, /*cap=*/rate * INBOUND_COUNT_SECONDS) { @@ -2112,8 +2107,8 @@ PeerManagerImpl::PeerManagerImpl(CConnman& connman, AddrMan& addrman, m_txdownloadman(node::TxDownloadOptions{pool, m_rng, opts.deterministic_rng}), m_warnings{warnings}, m_opts{opts}, - m_inbound_inv_bucket(/*rate=*/INVENTORY_BROADCAST_PER_SECOND, /*mult=*/1.0), - m_outbound_inv_bucket(/*rate=*/INVENTORY_BROADCAST_PER_SECOND, /*mult=*/OUTBOUND_INVENTORY_BUCKET_MULTIPLIER) + m_inbound_inv_bucket(/*rate=*/m_opts.tx_send_rate, /*mult=*/1.0), + m_outbound_inv_bucket(/*rate=*/m_opts.tx_send_rate, /*mult=*/OUTBOUND_INVENTORY_BUCKET_MULTIPLIER) { // While Erlay support is incomplete, it must be enabled explicitly via -txreconciliation. // This argument can go away after Erlay support is complete. diff --git a/src/net_processing.h b/src/net_processing.h index f29adc0157f..8e6c4b87f85 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -42,6 +42,8 @@ static constexpr bool DEFAULT_TXRECONCILIATION_ENABLE{false}; /** Default number of non-mempool transactions to keep around for block reconstruction. Includes orphan, replaced, and rejected transactions. */ static const uint32_t DEFAULT_BLOCK_RECONSTRUCTION_EXTRA_TXN{100}; +/** Default maximum per-second rate for sending transaction inventory to peers. */ +static constexpr unsigned int DEFAULT_TX_SEND_RATE{14}; static const bool DEFAULT_PEERBLOOMFILTERS = false; static const bool DEFAULT_PEERBLOCKFILTERS = false; /** Maximum number of outstanding CMPCTBLOCK requests for the same block. */ @@ -96,6 +98,8 @@ public: uint32_t max_headers_result{MAX_HEADERS_RESULTS}; //! Whether private broadcast is used for sending transactions. bool private_broadcast{DEFAULT_PRIVATE_BROADCAST}; + //! Maximum per-second rate for sending transaction inventory to peers. + unsigned int tx_send_rate{DEFAULT_TX_SEND_RATE}; }; static std::unique_ptr make(CConnman& connman, AddrMan& addrman, diff --git a/src/node/peerman_args.cpp b/src/node/peerman_args.cpp index 9745d69d5ae..47413af0ce5 100644 --- a/src/node/peerman_args.cpp +++ b/src/node/peerman_args.cpp @@ -24,6 +24,10 @@ void ApplyArgsManOptions(const ArgsManager& argsman, PeerManager::Options& optio if (auto value{argsman.GetBoolArg("-blocksonly")}) options.ignore_incoming_txs = *value; + if (auto value{argsman.GetIntArg("-txsendrate")}) { + options.tx_send_rate = uint32_t(std::clamp(*value, 1, 1000)); + } + if (auto value{argsman.GetBoolArg("-privatebroadcast")}) options.private_broadcast = *value; } diff --git a/src/node/txdownloadman.h b/src/node/txdownloadman.h index 2cc1ec2c279..bef1d162d22 100644 --- a/src/node/txdownloadman.h +++ b/src/node/txdownloadman.h @@ -25,7 +25,7 @@ class TxDownloadManagerImpl; static constexpr int32_t MAX_PEER_TX_REQUEST_IN_FLIGHT = 100; /** Maximum number of transactions to consider for requesting, per peer. It provides a reasonable DoS limit to * per-peer memory usage spent on announcements, while covering peers continuously sending INVs at the maximum - * rate (by our own policy, see INVENTORY_BROADCAST_PER_SECOND) for several minutes, while not receiving + * rate (by our own policy, see DEFAULT_TX_SEND_RATE) for several minutes, while not receiving * the actual transaction (from any peer) in response to requests for them. */ static constexpr int32_t MAX_PEER_TX_ANNOUNCEMENTS = 5000; /** How long to delay requesting transactions via txids, if we have wtxid-relaying peers */ From 4842903ac122c8683a810a1b5da4cc230580fe90 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Thu, 19 Feb 2026 20:13:52 +1000 Subject: [PATCH 10/13] rpc: report -txsendrate and bucket info via getnetworkinfo Add `tx_send_rate` and `inv_buckets` fields to getnetworkinfo. The `inv_buckets` field has separate `inbound` and `outbound` entries, reporting backlog count, count tokens, and size tokens. Useful for monitoring relay behavior. --- src/net_processing.cpp | 15 ++++++++++++++- src/net_processing.h | 9 +++++++++ src/rpc/net.cpp | 22 ++++++++++++++++++++++ 3 files changed, 45 insertions(+), 1 deletion(-) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 56c79c2ca4e..11af703cbbc 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -551,6 +551,15 @@ struct InvToSendBucket { bool count_ok = count_bucket.decrement(1, /*floor=*/count_floor); return size_ok && count_ok; } + + PeerManagerInfo::InvBucketInfo info() const + { + return { + .backlog_count = backlog.size(), + .count_bucket = count_bucket.value(), + .size_bucket = size_bucket.value(), + }; + } }; class PeerManagerImpl final : public PeerManager @@ -590,7 +599,7 @@ public: EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); bool GetNodeStateStats(NodeId nodeid, CNodeStateStats& stats) const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); std::vector GetOrphanTransactions() override EXCLUSIVE_LOCKS_REQUIRED(!m_tx_download_mutex); - PeerManagerInfo GetInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + PeerManagerInfo GetInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); std::vector GetPrivateBroadcastInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); std::vector AbortPrivateBroadcast(const uint256& id) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SendPings() override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); @@ -1936,10 +1945,14 @@ std::vector PeerManagerImpl::GetOrphanTransaction PeerManagerInfo PeerManagerImpl::GetInfo() const { + LOCK(m_inv_to_send_mutex); return PeerManagerInfo{ .median_outbound_time_offset = m_outbound_time_offsets.Median(), .ignores_incoming_txs = m_opts.ignore_incoming_txs, .private_broadcast = m_opts.private_broadcast, + .tx_send_rate = m_opts.tx_send_rate, + .inbound_bucket = m_inbound_inv_bucket.info(), + .outbound_bucket = m_outbound_inv_bucket.info(), }; } diff --git a/src/net_processing.h b/src/net_processing.h index 8e6c4b87f85..ff0d2baa8fa 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -72,9 +72,18 @@ struct CNodeStateStats { }; struct PeerManagerInfo { + struct InvBucketInfo { + size_t backlog_count{0}; + double count_bucket{0}; + double size_bucket{0}; + }; + std::chrono::seconds median_outbound_time_offset{0s}; bool ignores_incoming_txs{false}; bool private_broadcast{DEFAULT_PRIVATE_BROADCAST}; + unsigned int tx_send_rate{0}; + InvBucketInfo inbound_bucket; + InvBucketInfo outbound_bucket; }; class PeerManager : public CValidationInterface, public NetEventsInterface diff --git a/src/rpc/net.cpp b/src/rpc/net.cpp index ba1080ed11f..2d20ff16106 100644 --- a/src/rpc/net.cpp +++ b/src/rpc/net.cpp @@ -659,6 +659,16 @@ static RPCMethod getnetworkinfo() }}, {RPCResult::Type::BOOL, "localrelay", "true if transaction relay is requested from peers"}, {RPCResult::Type::NUM, "timeoffset", "the time offset"}, + {RPCResult::Type::NUM, "tx_send_rate", "configured target for maximum number of transactions per second to send to inbound peers"}, + {RPCResult::Type::OBJ_DYN, "inv_buckets", "", { + {RPCResult::Type::OBJ, "inbound/outbound", "connection direction", + { + {RPCResult::Type::NUM, "backlog", "number of queued txs to announce"}, + {RPCResult::Type::NUM, "count_tok", "tokens available to be consumed per-transaction"}, + {RPCResult::Type::NUM, "size_tok", "tokens available to be consumed per-byte"}, + } + } + }}, {RPCResult::Type::NUM, "connections", "the total number of connections"}, {RPCResult::Type::NUM, "connections_in", "the number of inbound connections"}, {RPCResult::Type::NUM, "connections_out", "the number of outbound connections"}, @@ -716,6 +726,18 @@ static RPCMethod getnetworkinfo() auto peerman_info{node.peerman->GetInfo()}; obj.pushKV("localrelay", !peerman_info.ignores_incoming_txs); obj.pushKV("timeoffset", Ticks(peerman_info.median_outbound_time_offset)); + obj.pushKV("tx_send_rate", peerman_info.tx_send_rate); + auto buckjson = [&](const auto& buckinfo) { + UniValue b{UniValue::VOBJ}; + b.pushKV("backlog", buckinfo.backlog_count); + b.pushKV("count_tok", buckinfo.count_bucket); + b.pushKV("size_tok", buckinfo.size_bucket); + return b; + }; + UniValue invbuckets{UniValue::VOBJ}; + invbuckets.pushKV("inbound", buckjson(peerman_info.inbound_bucket)); + invbuckets.pushKV("outbound", buckjson(peerman_info.outbound_bucket)); + obj.pushKV("inv_buckets", invbuckets); } if (node.connman) { obj.pushKV("networkactive", node.connman->GetNetworkActive()); From 5cde66341a66c37454a4b516016243060cdc8f56 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Fri, 20 Feb 2026 05:31:22 +1000 Subject: [PATCH 11/13] tests: basic functional test for tx rate limiting --- test/functional/p2p_tx_relay_rate_limit.py | 114 +++++++++++++++++++++ test/functional/test_runner.py | 1 + 2 files changed, 115 insertions(+) create mode 100755 test/functional/p2p_tx_relay_rate_limit.py diff --git a/test/functional/p2p_tx_relay_rate_limit.py b/test/functional/p2p_tx_relay_rate_limit.py new file mode 100755 index 00000000000..9c61244412b --- /dev/null +++ b/test/functional/p2p_tx_relay_rate_limit.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +# Copyright (c) The Bitcoin Core developers +# Distributed under the MIT software license, see the accompanying +# file COPYING or http://www.opensource.org/licenses/mit-license.php. +"""Test transaction relay rate limiting via token buckets. + +With -txsendrate=R, the inbound count bucket has capacity R*30. A broadcast +transaction is relayed immediately while the bucket has tokens; once it is +exhausted the excess transactions queue in a global backlog and drain as the +bucket refills (R tokens/second as mocktime advances). +""" + +from decimal import Decimal +import time + +from test_framework.blocktools import COINBASE_MATURITY +from test_framework.p2p import P2PTxInvStore +from test_framework.test_framework import BitcoinTestFramework +from test_framework.util import assert_equal +from test_framework.wallet import MiniWallet + +SEND_RATE = 2 # -txsendrate value +BUCKET_CAP = SEND_RATE * 30 # count bucket capacity (60) +NUM_TXS = 80 # total transactions to submit + + +class TxRelayRateLimitTest(BitcoinTestFramework): + def set_test_params(self): + self.num_nodes = 1 + self.extra_args = [[f'-txsendrate={SEND_RATE}']] + + def inbound_backlog(self, node): + return node.getnetworkinfo()['inv_buckets']['inbound']['backlog'] + + def run_test(self): + node = self.nodes[0] + wallet = MiniWallet(node) + + node.setmocktime(int(time.time())) + + # Mine enough blocks for mature coinbase UTXOs. + self.generate(wallet, COINBASE_MATURITY + NUM_TXS + 50) + + # Connect an inbound peer (negotiates wtxid relay by default) + peer = node.add_p2p_connection(P2PTxInvStore()) + + # Advance time so the peer's trickle timer initializes + node.bumpmocktime(10) + peer.sync_with_ping() + assert_equal(len(peer.get_invs()), 0) + + # Verify the configured send rate + assert_equal(node.getnetworkinfo()['tx_send_rate'], SEND_RATE) + + self.test_rate_limit_and_rbf(node, wallet, peer) + + def test_rate_limit_and_rbf(self, node, wallet, peer): + self.log.info(f"Submitting {NUM_TXS} transactions at frozen time (bucket capacity {BUCKET_CAP})") + + # Prepare an RBF pair: original and replacement spending the same UTXO. + # Both are created upfront so we can submit the replacement later. + rbf_utxo = wallet.get_utxo() + tx_rbf_orig = wallet.create_self_transfer(utxo_to_spend=rbf_utxo) + tx_rbf_repl = wallet.create_self_transfer(utxo_to_spend=rbf_utxo, fee_rate=Decimal("0.009")) + + # Submit NUM_TXS transactions at frozen time. Each broadcast is relayed + # immediately while the count bucket has tokens, so the first BUCKET_CAP + # are handed straight to the peer and the rest queue in the backlog. The + # RBF original is placed in the backlogged tail. + RBF_INDEX = NUM_TXS - 5 + for i in range(NUM_TXS): + if i == RBF_INDEX: + node.sendrawtransaction(tx_rbf_orig['hex']) + else: + wallet.send_self_transfer(from_node=node) + + # The excess beyond the bucket capacity is backlogged. Nothing is + # announced yet -- the per-peer trickle timer hasn't fired (frozen time). + self.log.info(f"Backlog after burst: {self.inbound_backlog(node)}") + assert_equal(self.inbound_backlog(node), NUM_TXS - BUCKET_CAP) + assert_equal(len(peer.get_invs()), 0) + + # RBF the backlogged original while time is still frozen, so the + # replacement also queues in the backlog (the bucket is exhausted). The + # original's wtxid stays in the backlog vector for now but is dropped + # when the backlog is processed, since it is no longer in the mempool. + self.log.info("RBF'ing a backlogged transaction") + node.sendrawtransaction(tx_rbf_repl['hex']) + assert_equal(self.inbound_backlog(node), NUM_TXS - BUCKET_CAP + 1) + + # Advance time so the bucket refills and the backlog drains. Loop until + # the backlog is empty and every surviving tx has trickled out. + self.log.info("Advancing time to drain the backlog") + for _ in range(30): + if self.inbound_backlog(node) == 0 and len(peer.get_invs()) == NUM_TXS: + break + node.bumpmocktime(4) + peer.sync_with_ping() + + announced = set(peer.get_invs()) + self.log.info(f"Total announced: {len(announced)}") + + # Every surviving transaction is announced: the burst minus the dropped + # RBF original, plus the replacement. + assert_equal(self.inbound_backlog(node), 0) + assert_equal(len(announced), NUM_TXS) + assert int(tx_rbf_orig['wtxid'], 16) not in announced + assert int(tx_rbf_repl['wtxid'], 16) in announced + + self.log.info("Rate limiting and RBF backlog cleanup test passed") + + +if __name__ == '__main__': + TxRelayRateLimitTest(__file__).main() diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py index c570f5b71ed..8dfa8a3e226 100755 --- a/test/functional/test_runner.py +++ b/test/functional/test_runner.py @@ -275,6 +275,7 @@ BASE_SCRIPTS = [ 'wallet_importprunedfunds.py', 'p2p_leak_tx.py --v1transport', 'p2p_leak_tx.py --v2transport', + 'p2p_tx_relay_rate_limit.py', 'p2p_eviction.py', 'p2p_outbound_eviction.py', 'p2p_ibd_stalling.py --v1transport', From 12b0dc33c4af6fc0eb48ae4265c6c050c4d122d7 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Tue, 28 Apr 2026 16:53:06 +1000 Subject: [PATCH 12/13] doc: Add release note for -txsendrate etc --- doc/release-notes-34628.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 doc/release-notes-34628.md diff --git a/doc/release-notes-34628.md b/doc/release-notes-34628.md new file mode 100644 index 00000000000..884ac0734d2 --- /dev/null +++ b/doc/release-notes-34628.md @@ -0,0 +1,13 @@ +P2P and network changes +----------------------- + +- To reduce memory and CPU usage during periods of high transaction + volume, rate-limiting of outgoing transaction relay has been changed + to use a global backlog instead of being done on a per-peer basis. The + default rate-limit remains as 14 tx/s (boosted by 2.5x for outbound + peers), though this can be changed via the `-txsendrate` configuration + option. An additional bandwidth rate-limit has also been introduced + at 12MB of transactions per 10 minutes, with a high burst rate. The + size of the global backlog and the token bucket values for the rate + limits can be queried via the `getnetworkinfo` RPC. (#34628) + From 349c72ee00a06581aa8cddaced6377c49a81d511 Mon Sep 17 00:00:00 2001 From: Anthony Towns Date: Sat, 30 May 2026 01:43:35 +1000 Subject: [PATCH 13/13] net_processing: Drop unnecessary txid arg from InitiateTxBroadcastToAll --- src/net_processing.cpp | 10 +++++----- src/net_processing.h | 6 +++--- src/node/transaction.cpp | 2 +- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 11af703cbbc..5b57c2f805f 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -603,7 +603,7 @@ public: std::vector GetPrivateBroadcastInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); std::vector AbortPrivateBroadcast(const uint256& id) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SendPings() override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); + void InitiateTxBroadcastToAll(const Wtxid& wtxid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); node::TransactionError InitiateTxBroadcastPrivate(const CTransactionRef& tx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SetBestBlock(int height, std::chrono::seconds time) override { @@ -1713,7 +1713,7 @@ void PeerManagerImpl::ReattemptInitialBroadcast(CScheduler& scheduler) CTransactionRef tx = m_mempool.get(txid); if (tx != nullptr) { - InitiateTxBroadcastToAll(txid, tx->GetWitnessHash()); + InitiateTxBroadcastToAll(tx->GetWitnessHash()); } else { m_mempool.RemoveUnbroadcastTx(txid, true); } @@ -2460,7 +2460,7 @@ void PeerManagerImpl::ProcessInvBacklog(NodeClock::time_point now, bool backlog_ } } -void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid&, const Wtxid& wtxid) +void PeerManagerImpl::InitiateTxBroadcastToAll(const Wtxid& wtxid) { { LOCK(m_inv_to_send_mutex); @@ -3372,7 +3372,7 @@ void PeerManagerImpl::ProcessValidTx(NodeId nodeid, const CTransactionRef& tx, c tx->GetWitnessHash().ToString(), m_mempool.size(), m_mempool.DynamicMemoryUsage() / 1000); - InitiateTxBroadcastToAll(tx->GetHash(), tx->GetWitnessHash()); + InitiateTxBroadcastToAll(tx->GetWitnessHash()); for (const CTransactionRef& removedTx : replaced_transactions) { AddToCompactExtraTransactions(removedTx); @@ -4694,7 +4694,7 @@ void PeerManagerImpl::ProcessMessage(Peer& peer, CNode& pfrom, const std::string } else { LogInfo("Force relaying tx %s (wtxid=%s) from peer=%d\n", txid.ToString(), wtxid.ToString(), pfrom.GetId()); - InitiateTxBroadcastToAll(txid, wtxid); + InitiateTxBroadcastToAll(wtxid); } } diff --git a/src/net_processing.h b/src/net_processing.h index ff0d2baa8fa..a381a6d80bc 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -152,11 +152,11 @@ public: /** * Initiate a transaction broadcast to eligible peers. - * Queue the witness transaction id to `Peer::TxRelay::m_tx_inventory_to_send` - * for each peer. Later, depending on `Peer::TxRelay::m_next_inv_send_time` and if + * Queue the witness transaction id to the inbound and outbound inv backlogs. + * Later, depending on `-txsendrate`, `Peer::TxRelay::m_next_inv_send_time` and if * the transaction is in the mempool, an `INV` about it may be sent to the peer. */ - virtual void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) = 0; + virtual void InitiateTxBroadcastToAll(const Wtxid& wtxid) = 0; /** * Initiate a private transaction broadcast. This is done diff --git a/src/node/transaction.cpp b/src/node/transaction.cpp index e7877c69855..d331ae05aa6 100644 --- a/src/node/transaction.cpp +++ b/src/node/transaction.cpp @@ -130,7 +130,7 @@ TransactionError BroadcastTransaction(NodeContext& node, case TxBroadcast::MEMPOOL_NO_BROADCAST: break; case TxBroadcast::MEMPOOL_AND_BROADCAST_TO_ALL: - node.peerman->InitiateTxBroadcastToAll(txid, wtxid); + node.peerman->InitiateTxBroadcastToAll(wtxid); break; case TxBroadcast::NO_MEMPOOL_PRIVATE_BROADCAST: return node.peerman->InitiateTxBroadcastPrivate(tx);