diff --git a/doc/release-notes-35406.md b/doc/release-notes-35406.md new file mode 100644 index 00000000000..178a079132b --- /dev/null +++ b/doc/release-notes-35406.md @@ -0,0 +1,8 @@ +P2P and network changes +----------------------- + +- The private-broadcast queue (transactions submitted via `sendrawtransaction` +when `-privatebroadcast` is enabled and not yet echoed back from the network) +is now capped at 10,000 entries. When full, new submissions are rejected. It is +up to the caller to inspect the queue via `getprivatebroadcastinfo` and free +up space when stuck via `abortprivatebroadcast`. (#35406) diff --git a/src/common/messages.cpp b/src/common/messages.cpp index 700f4f03cd6..82ad310bb70 100644 --- a/src/common/messages.cpp +++ b/src/common/messages.cpp @@ -143,6 +143,8 @@ bilingual_str TransactionErrorString(const TransactionError err) return Untranslated("Unspendable output exceeds maximum configured by user (maxburnamount)"); case TransactionError::INVALID_PACKAGE: return Untranslated("Transaction rejected due to invalid package"); + case TransactionError::PRIVATE_BROADCAST_FULL: + return Untranslated("Private broadcast queue is full"); } // no default case, so the compiler can warn about missing cases assert(false); } diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 3b38a70c6c5..27f0a63c558 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -540,7 +540,7 @@ public: 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 InitiateTxBroadcastPrivate(const CTransactionRef& tx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + node::TransactionError InitiateTxBroadcastPrivate(const CTransactionRef& tx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SetBestBlock(int height, std::chrono::seconds time) override { m_best_height = height; @@ -1660,8 +1660,9 @@ void PeerManagerImpl::ReattemptPrivateBroadcast(CScheduler& scheduler) size_t num_for_rebroadcast{0}; const auto stale_txs = m_tx_for_private_broadcast.GetStale(); if (!stale_txs.empty()) { - LOCK(cs_main); for (const auto& stale_tx : stale_txs) { + // Only hold lock per single submission + LOCK(cs_main); auto mempool_acceptable = m_chainman.ProcessTransaction(stale_tx, /*test_accept=*/true); if (mempool_acceptable.m_result_type == MempoolAcceptResult::ResultType::VALID) { LogDebug(BCLog::PRIVBROADCAST, @@ -2293,15 +2294,22 @@ void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wt } } -void PeerManagerImpl::InitiateTxBroadcastPrivate(const CTransactionRef& tx) +node::TransactionError PeerManagerImpl::InitiateTxBroadcastPrivate(const CTransactionRef& tx) { const auto txstr{strprintf("txid=%s, wtxid=%s", tx->GetHash().ToString(), tx->GetWitnessHash().ToString())}; - if (m_tx_for_private_broadcast.Add(tx)) { + switch (m_tx_for_private_broadcast.Add(tx)) { + case PrivateBroadcast::AddResult::Added: LogDebug(BCLog::PRIVBROADCAST, "Requesting %d new connections due to %s", NUM_PRIVATE_BROADCAST_PER_TX, txstr); m_connman.m_private_broadcast.NumToOpenAdd(NUM_PRIVATE_BROADCAST_PER_TX); - } else { + return node::TransactionError::OK; + case PrivateBroadcast::AddResult::AlreadyPresent: LogDebug(BCLog::PRIVBROADCAST, "Ignoring unnecessary request to schedule an already scheduled transaction: %s", txstr); - } + return node::TransactionError::OK; + case PrivateBroadcast::AddResult::QueueFull: + LogDebug(BCLog::PRIVBROADCAST, "Rejecting private broadcast, queue full (cap=%u): %s", PrivateBroadcast::MAX_TRANSACTIONS, txstr); + return node::TransactionError::PRIVATE_BROADCAST_FULL; + } // no default case, so the compiler can warn about missing cases + assert(false); } void PeerManagerImpl::RelayAddress(NodeId originator, diff --git a/src/net_processing.h b/src/net_processing.h index 630656e2af6..f29adc0157f 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -147,8 +148,10 @@ public: /** * Initiate a private transaction broadcast. This is done * asynchronously via short-lived connections to peers on privacy networks. + * @retval node::TransactionError::OK The transaction is scheduled for private broadcast (or was already scheduled). + * @retval node::TransactionError::PRIVATE_BROADCAST_FULL Rejected because the private broadcast queue is full. */ - virtual void InitiateTxBroadcastPrivate(const CTransactionRef& tx) = 0; + [[nodiscard]] virtual node::TransactionError InitiateTxBroadcastPrivate(const CTransactionRef& tx) = 0; /** Send ping message to all peers */ virtual void SendPings() = 0; diff --git a/src/node/transaction.cpp b/src/node/transaction.cpp index bff82f21560..e7877c69855 100644 --- a/src/node/transaction.cpp +++ b/src/node/transaction.cpp @@ -133,8 +133,7 @@ TransactionError BroadcastTransaction(NodeContext& node, node.peerman->InitiateTxBroadcastToAll(txid, wtxid); break; case TxBroadcast::NO_MEMPOOL_PRIVATE_BROADCAST: - node.peerman->InitiateTxBroadcastPrivate(tx); - break; + return node.peerman->InitiateTxBroadcastPrivate(tx); } return TransactionError::OK; diff --git a/src/node/types.h b/src/node/types.h index 5124b983112..0a022bec987 100644 --- a/src/node/types.h +++ b/src/node/types.h @@ -25,6 +25,7 @@ enum class TransactionError { MAX_FEE_EXCEEDED, MAX_BURN_EXCEEDED, INVALID_PACKAGE, + PRIVATE_BROADCAST_FULL, }; /** diff --git a/src/private_broadcast.cpp b/src/private_broadcast.cpp index 1d78a6ebef0..c7df778fdca 100644 --- a/src/private_broadcast.cpp +++ b/src/private_broadcast.cpp @@ -3,17 +3,23 @@ // file COPYING or https://opensource.org/license/mit/. #include + #include #include -bool PrivateBroadcast::Add(const CTransactionRef& tx) +[[nodiscard]] PrivateBroadcast::AddResult PrivateBroadcast::Add(const CTransactionRef& tx) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) { LOCK(m_mutex); - const bool inserted{m_transactions.try_emplace(tx).second}; - return inserted; + // Re-adding an already-tracked transaction is a no-op regardless of the cap. + if (m_transactions.contains(tx)) return AddResult::AlreadyPresent; + + if (m_transactions.size() >= m_max_transactions) return AddResult::QueueFull; + + m_transactions.try_emplace(tx); + return AddResult::Added; } std::optional PrivateBroadcast::Remove(const CTransactionRef& tx) diff --git a/src/private_broadcast.h b/src/private_broadcast.h index ae456ef8b85..cf955506e23 100644 --- a/src/private_broadcast.h +++ b/src/private_broadcast.h @@ -38,6 +38,15 @@ public: /// after it is broadcast, then we consider it stale / for rebroadcasting. static constexpr auto STALE_DURATION{1min}; + /// Maximum number of transactions tracked simultaneously. + /// Additions that would exceed this are rejected (see Add()). + static constexpr size_t MAX_TRANSACTIONS{10'000}; + + /// @param[in] max_transactions Cap on the number of simultaneously tracked + /// transactions. Defaults to MAX_TRANSACTIONS. + explicit PrivateBroadcast(size_t max_transactions = MAX_TRANSACTIONS) + : m_max_transactions{max_transactions} {} + struct PeerSendInfo { CService address; NodeClock::time_point sent; @@ -50,13 +59,23 @@ public: std::vector peers; }; + /// Outcome of Add(). + enum class AddResult { + //! The transaction was newly added. + Added, + //! The transaction was already present; no change. + AlreadyPresent, + //! Rejected: the queue is already at MAX_TRANSACTIONS. + QueueFull, + }; + /** * Add a transaction to the storage. * @param[in] tx The transaction to add. - * @retval true The transaction was added. - * @retval false The transaction was already present. + * @return Whether the transaction was newly added, was already present, or + * was rejected because the queue is full (see AddResult). */ - bool Add(const CTransactionRef& tx) + [[nodiscard]] AddResult Add(const CTransactionRef& tx) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex); /** @@ -196,6 +215,8 @@ private: const NodeClock::time_point time_added{NodeClock::now()}; std::vector send_statuses; }; + /// Cap on the number of simultaneously tracked transactions (see Add()). + const size_t m_max_transactions; mutable Mutex m_mutex; std::unordered_map m_transactions GUARDED_BY(m_mutex); diff --git a/src/rpc/mempool.cpp b/src/rpc/mempool.cpp index 0358aeae938..d18bc686665 100644 --- a/src/rpc/mempool.cpp +++ b/src/rpc/mempool.cpp @@ -60,6 +60,9 @@ static RPCMethod sendrawtransaction() "dedicated, short-lived connections to Tor or I2P peers or IPv4/IPv6 peers\n" "via the Tor network. This conceals the transaction's origin. The transaction\n" "will only enter the local mempool when it is received back from the network.\n" + "The private broadcast queue is bounded: when it is full, this RPC fails and\n" + "the transaction is not scheduled, until an existing one completes or is\n" + "aborted. Use getprivatebroadcastinfo to inspect the queue and abortprivatebroadcast to abort.\n" "\nA specific exception, RPC_TRANSACTION_ALREADY_IN_UTXO_SET, may throw if the transaction cannot be added to the mempool.\n" diff --git a/src/rpc/util.cpp b/src/rpc/util.cpp index 410bb741c15..46ab8ea7e19 100644 --- a/src/rpc/util.cpp +++ b/src/rpc/util.cpp @@ -395,6 +395,8 @@ RPCErrorCode RPCErrorFromTransactionError(TransactionError terr) return RPC_TRANSACTION_REJECTED; case TransactionError::ALREADY_IN_UTXO_SET: return RPC_VERIFY_ALREADY_IN_UTXO_SET; + case TransactionError::PRIVATE_BROADCAST_FULL: + return RPC_OUT_OF_MEMORY; default: break; } return RPC_TRANSACTION_ERROR; diff --git a/src/test/fuzz/private_broadcast.cpp b/src/test/fuzz/private_broadcast.cpp index 4db24f8e178..ae834ce6ac3 100644 --- a/src/test/fuzz/private_broadcast.cpp +++ b/src/test/fuzz/private_broadcast.cpp @@ -38,7 +38,8 @@ FUZZ_TARGET(private_broadcast) FuzzedDataProvider fdp(buffer.data(), buffer.size()); FakeNodeClock clock_ctx{ConsumeTime(fdp)}; - PrivateBroadcast pb; + const size_t cap{fdp.ConsumeIntegralInRange(1, 12)}; + PrivateBroadcast pb{cap}; // Random transaction that the test generated and passed to Add(). Trimmed when Remove() is called. // The values are the number of times a transaction was picked for sending. @@ -64,15 +65,20 @@ FUZZ_TARGET(private_broadcast) fdp, [&] { // Add() CTransactionRef tx; - bool from_transactions{false}; if (transactions.empty() || fdp.ConsumeBool()) { tx = MakeTransactionRef(ConsumeTransaction(fdp, std::nullopt)); } else { tx = PickIterator(fdp, transactions)->first; - from_transactions = true; } - if (pb.Add(tx)) { - Assert(!from_transactions); + + const bool present_before{transactions.contains(tx)}; + const auto res{pb.Add(tx)}; + if (present_before) { + Assert(res == PrivateBroadcast::AddResult::AlreadyPresent); + } else if (transactions.size() >= cap) { + Assert(res == PrivateBroadcast::AddResult::QueueFull); + } else { + Assert(res == PrivateBroadcast::AddResult::Added); transactions.emplace(tx, 0); } }, diff --git a/src/test/private_broadcast_tests.cpp b/src/test/private_broadcast_tests.cpp index 8e7d13551ae..932ad8ee484 100644 --- a/src/test/private_broadcast_tests.cpp +++ b/src/test/private_broadcast_tests.cpp @@ -44,15 +44,15 @@ BOOST_AUTO_TEST_CASE(basic) // Make a transaction and add it. const auto tx1{MakeDummyTx(/*id=*/1, /*num_witness=*/0)}; - BOOST_CHECK(pb.Add(tx1)); - BOOST_CHECK(!pb.Add(tx1)); + BOOST_CHECK(pb.Add(tx1) == PrivateBroadcast::AddResult::Added); + BOOST_CHECK(pb.Add(tx1) == PrivateBroadcast::AddResult::AlreadyPresent); // Make another transaction with same txid, different wtxid and add it. const auto tx2{MakeDummyTx(/*id=*/1, /*num_witness=*/1)}; BOOST_REQUIRE(tx1->GetHash() == tx2->GetHash()); BOOST_REQUIRE(tx1->GetWitnessHash() != tx2->GetWitnessHash()); - BOOST_CHECK(pb.Add(tx2)); + BOOST_CHECK(pb.Add(tx2) == PrivateBroadcast::AddResult::Added); const auto find_tx_info{[](auto& infos, const CTransactionRef& tx) -> const PrivateBroadcast::TxBroadcastInfo& { const auto it{std::ranges::find(infos, tx->GetWitnessHash(), [](const auto& info) { return info.tx->GetWitnessHash(); })}; BOOST_REQUIRE(it != infos.end()); @@ -146,7 +146,7 @@ BOOST_AUTO_TEST_CASE(stale_unpicked_tx) PrivateBroadcast pb; const auto tx{MakeDummyTx(/*id=*/42, /*num_witness=*/0)}; - BOOST_REQUIRE(pb.Add(tx)); + BOOST_REQUIRE(pb.Add(tx) == PrivateBroadcast::AddResult::Added); // Unpicked transactions use the longer INITIAL_STALE_DURATION. BOOST_CHECK_EQUAL(pb.GetStale().size(), 0); @@ -158,4 +158,56 @@ BOOST_AUTO_TEST_CASE(stale_unpicked_tx) BOOST_CHECK_EQUAL(stale_state[0], tx); } +BOOST_AUTO_TEST_CASE(rejection_at_cap) +{ + PrivateBroadcast pb; + constexpr size_t num_cap{PrivateBroadcast::MAX_TRANSACTIONS}; + constexpr size_t num_over{5}; + + // Fill the queue exactly to the cap; every distinct Add() succeeds. + std::vector txs; + txs.reserve(num_cap); + for (size_t i{0}; i < num_cap; ++i) { + auto tx{MakeDummyTx(/*id=*/static_cast(i), /*num_witness=*/0)}; + BOOST_REQUIRE(pb.Add(tx) == PrivateBroadcast::AddResult::Added); + txs.push_back(std::move(tx)); + } + BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap); + + // Further distinct transactions are rejected, and the queue is unchanged. + for (size_t i{0}; i < num_over; ++i) { + const auto tx{MakeDummyTx(/*id=*/static_cast(num_cap + i), /*num_witness=*/0)}; + BOOST_CHECK(pb.Add(tx) == PrivateBroadcast::AddResult::QueueFull); + } + BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap); + + // Nothing was evicted: all originally-added transactions are still present. + const auto infos{pb.GetBroadcastInfo()}; + std::set present_wtxids; + for (const auto& info : infos) { + present_wtxids.insert(info.tx->GetWitnessHash().ToUint256()); + } + BOOST_CHECK_EQUAL(present_wtxids.size(), infos.size()); + for (size_t i{0}; i < num_cap; ++i) { + BOOST_CHECK_MESSAGE(present_wtxids.contains(txs[i]->GetWitnessHash().ToUint256()), + "tx index " << i << " should still be present"); + } + + // Re-adding an already-present tx is AlreadyPresent even at the cap (not QueueFull). + BOOST_CHECK(pb.Add(txs[0]) == PrivateBroadcast::AddResult::AlreadyPresent); + BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap); + + // Removing one frees exactly one slot for a new transaction. + BOOST_REQUIRE(pb.Remove(txs[0]).has_value()); + BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap - 1); + const auto fresh{MakeDummyTx(/*id=*/0xffffffff, /*num_witness=*/0)}; + BOOST_CHECK(pb.Add(fresh) == PrivateBroadcast::AddResult::Added); + BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap); + + // A previously-removed tx can be added again as a brand-new entry + BOOST_REQUIRE(pb.Remove(fresh).has_value()); + BOOST_CHECK(pb.Add(txs[0]) == PrivateBroadcast::AddResult::Added); + BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap); +} + BOOST_AUTO_TEST_SUITE_END() diff --git a/test/functional/p2p_private_broadcast_cap.py b/test/functional/p2p_private_broadcast_cap.py new file mode 100755 index 00000000000..154bf8b5c7d --- /dev/null +++ b/test/functional/p2p_private_broadcast_cap.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +# Copyright (c) 2026-present 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 the private-broadcast queue size cap: submissions beyond the cap are +rejected (the queue is not modified), rather than evicting existing entries. +""" + +from test_framework.test_framework import BitcoinTestFramework +from test_framework.util import assert_equal, assert_raises_rpc_error +from test_framework.wallet import MiniWallet + + +# Must match PrivateBroadcast::MAX_TRANSACTIONS +MAX_TRANSACTIONS = 10_000 +OVER_CAP = 5 + + +class PrivateBroadcastCapTest(BitcoinTestFramework): + def set_test_params(self): + self.num_nodes = 1 + # -privatebroadcast is incompatible with the framework's default + # -connect=0; allow autoconnect (no actual peers will succeed though). + self.disable_autoconnect = False + self.extra_args = [[ + "-privatebroadcast", + # Fake I2P reachability so the privatebroadcast startup precondition passes. + "-i2psam=127.0.0.1:1", + "-proxy=127.0.0.1:1", + ]] + + def setup_network(self): + # Skip the framework's default connect_nodes loop. We have a single + # node and don't need any peer connections. + self.setup_nodes() + + def run_test(self): + node = self.nodes[0] + wallet = MiniWallet(node) + # Mature one coinbase to spend. + self.generate(wallet, 101) + + # Build a parent that fans out to MAX_TRANSACTIONS + OVER_CAP outputs. + # Inject it directly via generateblock since -privatebroadcast bypasses + # the mempool + utxo = wallet.get_utxo() + parent = wallet.create_self_transfer_multi( + utxos_to_spend=[utxo], + num_outputs=MAX_TRANSACTIONS + OVER_CAP, + fee_per_output=500, + ) + self.generateblock(node, wallet.get_address(), [parent["hex"]]) + + children = [wallet.create_self_transfer(utxo_to_spend=u) + for u in parent["new_utxos"]] + assert_equal(len(children), MAX_TRANSACTIONS + OVER_CAP) + + # Fill the queue exactly to the cap; every distinct submission succeeds. + self.log.info(f"Filling private broadcast queue to cap ({MAX_TRANSACTIONS} txns)") + for child in children[:MAX_TRANSACTIONS]: + node.sendrawtransaction(child["hex"]) + + pbinfo = node.getprivatebroadcastinfo() + assert_equal(len(pbinfo["transactions"]), MAX_TRANSACTIONS) + present_wtxids = {t["wtxid"] for t in pbinfo["transactions"]} + for i, child in enumerate(children[:MAX_TRANSACTIONS]): + assert child["wtxid"] in present_wtxids, \ + f"tx index {i} (wtxid={child['wtxid']}) should be in the queue" + + # Further distinct submissions are rejected with an RPC error, and the + # queue is left unchanged (nothing evicted to make room). + self.log.info(f"Submitting {OVER_CAP} more; each should be rejected (queue full)") + for child in children[MAX_TRANSACTIONS:]: + assert_raises_rpc_error(-7, "Private broadcast queue is full", + node.sendrawtransaction, child["hex"]) + + assert_equal(pbinfo["transactions"], node.getprivatebroadcastinfo()["transactions"]) + + self.log.info("Checking abortprivatebroadcast frees a slot for a new submission") + abort_res = node.abortprivatebroadcast(children[1]["txid"]) + assert_equal([t["wtxid"] for t in abort_res["removed_transactions"]], + [children[1]["wtxid"]]) + wtxids = {t["wtxid"] for t in node.getprivatebroadcastinfo()["transactions"]} + assert_equal(len(wtxids), MAX_TRANSACTIONS - 1) + assert children[1]["wtxid"] not in wtxids, "aborted tx should be gone from the queue" + + new_child = children[MAX_TRANSACTIONS] # first previously-rejected tx + node.sendrawtransaction(new_child["hex"]) + wtxids = {t["wtxid"] for t in node.getprivatebroadcastinfo()["transactions"]} + assert_equal(len(wtxids), MAX_TRANSACTIONS) + assert new_child["wtxid"] in wtxids, "freed slot should now hold the new tx" + assert children[1]["wtxid"] not in wtxids, "aborted tx should not reappear" + + # Re-submitting an already-queued transaction is a no-op, not an error, + # even when the queue is full. + self.log.info("Re-submitting an already-queued tx should not error") + node.sendrawtransaction(children[0]["hex"]) + wtxids = {t["wtxid"] for t in node.getprivatebroadcastinfo()["transactions"]} + assert_equal(len(wtxids), MAX_TRANSACTIONS) + assert children[0]["wtxid"] in wtxids, "re-submitted tx should remain queued" + +if __name__ == "__main__": + PrivateBroadcastCapTest(__file__).main() diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py index 3fa177fea8f..c570f5b71ed 100755 --- a/test/functional/test_runner.py +++ b/test/functional/test_runner.py @@ -316,6 +316,7 @@ BASE_SCRIPTS = [ 'feature_minchainwork.py', 'rpc_estimatefee.py', 'p2p_private_broadcast.py', + 'p2p_private_broadcast_cap.py', 'rpc_getblockstats.py', 'feature_port.py', 'feature_bind_port_externalip.py',