From fe7d475d450b9aabd549627c6e61024ae45e7a8c Mon Sep 17 00:00:00 2001 From: Gregory Sanders Date: Tue, 7 Jul 2026 15:22:51 -0400 Subject: [PATCH] private broadcast: bound broadcast attempts per tx to 1k Rather than rebroadcasting forever, bound attempts at private broadcast, report remaining attempts over RPC results, and allow exhausted transactions to be retried when submitted. --- doc/release-notes-35680.md | 13 ++++ src/private_broadcast.cpp | 27 ++++++-- src/private_broadcast.h | 39 ++++++++---- src/rpc/mempool.cpp | 5 +- src/test/fuzz/private_broadcast.cpp | 59 ++++++++++++----- src/test/private_broadcast_tests.cpp | 80 ++++++++++++++++++++++++ test/functional/p2p_private_broadcast.py | 4 +- 7 files changed, 193 insertions(+), 34 deletions(-) create mode 100644 doc/release-notes-35680.md diff --git a/doc/release-notes-35680.md b/doc/release-notes-35680.md new file mode 100644 index 00000000000..c5d19b150c8 --- /dev/null +++ b/doc/release-notes-35680.md @@ -0,0 +1,13 @@ +P2P and network changes +----------------------- + +- Each transaction sent via private broadcast (`-privatebroadcast`) is limited + to 1,000 send attempts. After reaching the limit, broadcasting stops; call + `sendrawtransaction` again to retry. Transactions that reach the limit remain + available through `getprivatebroadcastinfo` and `abortprivatebroadcast`. (#35680) + +Updated RPCs +------------ + +- `getprivatebroadcastinfo` now reports an `attempts_remaining` field for each + transaction. (#35680) diff --git a/src/private_broadcast.cpp b/src/private_broadcast.cpp index b4d61a658ad..04a89e15df2 100644 --- a/src/private_broadcast.cpp +++ b/src/private_broadcast.cpp @@ -7,14 +7,21 @@ #include #include +#include PrivateBroadcast::AddResult PrivateBroadcast::Add(const CTransactionRef& tx) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) { LOCK(m_mutex); - // Re-adding an already-tracked transaction is a no-op regardless of the cap. - if (m_transactions.contains(tx)) return AddResult::AlreadyPresent; + if (const auto it{m_transactions.find(tx)}; it != m_transactions.end()) { + if (IsPending(it->second)) return AddResult::AlreadyPresent; + + // An exhausted transaction can be explicitly retried by adding it again. + it->second.time_added = NodeClock::now(); + it->second.send_statuses.clear(); + return AddResult::Added; + } if (m_transactions.size() >= m_max_transactions) return AddResult::QueueFull; @@ -44,12 +51,13 @@ std::optional PrivateBroadcast::PickTxForSend(const NodeId& wil return std::nullopt; } + auto pending_transactions{m_transactions | std::views::filter([this](const auto& entry) { return IsPending(entry.second); })}; const auto it{std::ranges::max_element( - m_transactions, + pending_transactions, [](const auto& a, const auto& b) { return a < b; }, [](const auto& el) { return DerivePriority(el.second.send_statuses); })}; - if (it != m_transactions.end()) { + if (it != pending_transactions.end()) { auto& [tx, state]{*it}; state.send_statuses.emplace_back(will_send_to_nodeid, will_send_to_address, NodeClock::now()); return tx; @@ -94,7 +102,7 @@ bool PrivateBroadcast::HavePendingTransactions() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) { LOCK(m_mutex); - return !m_transactions.empty(); + return std::ranges::any_of(m_transactions, [this](const auto& entry) { return IsPending(entry.second); }); } std::vector PrivateBroadcast::GetStale() const @@ -104,6 +112,7 @@ std::vector PrivateBroadcast::GetStale() const const auto now{NodeClock::now()}; std::vector stale; for (const auto& [tx, state] : m_transactions) { + if (!IsPending(state)) continue; const Priority p{DerivePriority(state.send_statuses)}; if (p.num_confirmed == 0) { if (state.time_added < now - INITIAL_STALE_DURATION) stale.push_back(tx); @@ -127,12 +136,18 @@ std::vector PrivateBroadcast::GetBroadcastInf for (const auto& status : state.send_statuses) { peers.emplace_back(PeerSendInfo{.address = status.address, .sent = status.picked, .received = status.confirmed}); } - entries.emplace_back(TxBroadcastInfo{.tx = tx, .time_added = state.time_added, .peers = std::move(peers)}); + const size_t attempts_remaining{m_max_send_attempts - std::min(state.send_statuses.size(), m_max_send_attempts)}; + entries.emplace_back(TxBroadcastInfo{.tx = tx, .time_added = state.time_added, .attempts_remaining = attempts_remaining, .peers = std::move(peers)}); } return entries; } +bool PrivateBroadcast::IsPending(const TxSendStatus& status) const +{ + return status.send_statuses.size() < m_max_send_attempts; +} + PrivateBroadcast::Priority PrivateBroadcast::DerivePriority(const std::vector& sent_to) { Priority p; diff --git a/src/private_broadcast.h b/src/private_broadcast.h index cf955506e23..f3ba6ef4054 100644 --- a/src/private_broadcast.h +++ b/src/private_broadcast.h @@ -42,10 +42,18 @@ public: /// Additions that would exceed this are rejected (see Add()). static constexpr size_t MAX_TRANSACTIONS{10'000}; + /// Maximum number of send attempts for a transaction. Once this limit is + /// reached, the transaction remains tracked but is not sent again unless + /// explicitly re-added. + static constexpr size_t MAX_SEND_ATTEMPTS{1'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} {} + /// @param[in] max_send_attempts Cap on the number of send attempts per + /// transaction. Defaults to MAX_SEND_ATTEMPTS. + explicit PrivateBroadcast(size_t max_transactions = MAX_TRANSACTIONS, + size_t max_send_attempts = MAX_SEND_ATTEMPTS) + : m_max_transactions{max_transactions}, m_max_send_attempts{max_send_attempts} {} struct PeerSendInfo { CService address; @@ -56,24 +64,28 @@ public: struct TxBroadcastInfo { CTransactionRef tx; NodeClock::time_point time_added; + /// Number of additional send attempts allowed for this transaction (0 if exhausted). + size_t attempts_remaining; std::vector peers; }; /// Outcome of Add(). enum class AddResult { - //! The transaction was newly added. + //! The transaction was newly added or reset after exhausting its send attempts. Added, - //! The transaction was already present; no change. + //! The transaction was already present with send attempts remaining; no change. AlreadyPresent, //! Rejected: the queue is already at MAX_TRANSACTIONS. QueueFull, }; /** - * Add a transaction to the storage. + * Add a transaction to the storage, or reset an exhausted transaction so it + * can be broadcast again. * @param[in] tx The transaction to add. - * @return Whether the transaction was newly added, was already present, or - * was rejected because the queue is full (see AddResult). + * @return Whether the transaction was newly added or reset, was already + * present with send attempts remaining, or was rejected because the queue is + * full (see AddResult). */ [[nodiscard]] AddResult Add(const CTransactionRef& tx) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex); @@ -97,7 +109,8 @@ public: * transaction to one node would be a privacy leak. * @param[in] will_send_to_address Address of the peer to which this transaction * will be sent. - * @return Most urgent transaction or nullopt if there are no transactions. + * @return Most urgent transaction or nullopt if there are no transactions + * with send attempts remaining. */ std::optional PickTxForSend(const NodeId& will_send_to_nodeid, const CService& will_send_to_address) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex); @@ -127,13 +140,14 @@ public: EXCLUSIVE_LOCKS_REQUIRED(!m_mutex); /** - * Check if there are transactions that need to be broadcast. + * Check if there are transactions with send attempts remaining. */ bool HavePendingTransactions() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex); /** - * Get the transactions that have not been broadcast recently. + * Get the transactions that have not been broadcast recently and have send + * attempts remaining. */ std::vector GetStale() const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex); @@ -212,11 +226,14 @@ private: std::optional GetSendStatusByNode(const NodeId& nodeid) EXCLUSIVE_LOCKS_REQUIRED(m_mutex); struct TxSendStatus { - const NodeClock::time_point time_added{NodeClock::now()}; + NodeClock::time_point time_added{NodeClock::now()}; std::vector send_statuses; }; + bool IsPending(const TxSendStatus& status) const; /// Cap on the number of simultaneously tracked transactions (see Add()). const size_t m_max_transactions; + /// Cap on the number of send attempts per transaction (see PickTxForSend()). + const size_t m_max_send_attempts; 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 c262332942f..fccfac4bc56 100644 --- a/src/rpc/mempool.cpp +++ b/src/rpc/mempool.cpp @@ -146,7 +146,8 @@ static RPCMethod getprivatebroadcastinfo() { return RPCMethod{ "getprivatebroadcastinfo", - "Returns information about transactions that are currently being privately broadcast.\n" + "Returns information about transactions tracked for private broadcast.\n" + "Transactions that have reached the send-attempt limit remain in the result with attempts_remaining=0.\n" "This method is only available when running with -privatebroadcast enabled.\n", {}, RPCResult{ @@ -160,6 +161,7 @@ static RPCMethod getprivatebroadcastinfo() {RPCResult::Type::STR_HEX, "wtxid", "The transaction witness hash in hex"}, {RPCResult::Type::STR_HEX, "hex", "The serialized, hex-encoded transaction data"}, {RPCResult::Type::NUM_TIME, "time_added", "The time this transaction was added to the private broadcast queue (seconds since epoch)"}, + {RPCResult::Type::NUM, "attempts_remaining", "The number of additional private broadcast send attempts allowed for this transaction"}, {RPCResult::Type::ARR, "peers", "Per-peer send and acknowledgment information for this transaction", { {RPCResult::Type::OBJ, "", "", @@ -193,6 +195,7 @@ static RPCMethod getprivatebroadcastinfo() o.pushKV("wtxid", tx_info.tx->GetWitnessHash().ToString()); o.pushKV("hex", EncodeHexTx(*tx_info.tx)); o.pushKV("time_added", TicksSinceEpoch(tx_info.time_added)); + o.pushKV("attempts_remaining", tx_info.attempts_remaining); UniValue peers(UniValue::VARR); for (const auto& peer : tx_info.peers) { UniValue p(UniValue::VOBJ); diff --git a/src/test/fuzz/private_broadcast.cpp b/src/test/fuzz/private_broadcast.cpp index cd93cefa33b..adcb25855d3 100644 --- a/src/test/fuzz/private_broadcast.cpp +++ b/src/test/fuzz/private_broadcast.cpp @@ -16,6 +16,9 @@ #include #include +#include +#include +#include #include struct CTransactionRefHash { @@ -39,20 +42,27 @@ FUZZ_TARGET(private_broadcast) FakeNodeClock clock_ctx{ConsumeTime(fdp)}; const size_t cap{fdp.ConsumeIntegralInRange(1, 12)}; - PrivateBroadcast pb{cap}; + const size_t max_send_attempts{fdp.ConsumeIntegralInRange(1, 12)}; + PrivateBroadcast pb{cap, max_send_attempts}; // 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. std::unordered_map transactions; - // Ids of nodes that were passed to PickTxForSend(). Trimmed when Remove() is called. - std::unordered_set nodes_sent_to; + // Transactions passed to PickTxForSend(), indexed by node id. Trimmed when + // Remove() is called or a transaction is reset by Add(). + std::unordered_map nodes_sent_to; - // A subset of `nodes_sent_to`, node ids passed to NodeConfirmedReception(). Trimmed when Remove() is called. + // A subset of `nodes_sent_to`, node ids passed to NodeConfirmedReception(). + // Trimmed when Remove() is called or a transaction is reset by Add(). std::unordered_set nodes_that_confirmed_reception; NodeId next_nodeid{0}; // Generate unique node ids. + const auto is_pending{[max_send_attempts](const auto& entry) { + return entry.second < max_send_attempts; + }}; + const auto ExistentOrNewNodeId = [&next_nodeid, &fdp](){ if (next_nodeid == 0 || fdp.ConsumeBool()) { return next_nodeid++; @@ -74,7 +84,22 @@ FUZZ_TARGET(private_broadcast) const bool present_before{transactions.contains(tx)}; const auto res{pb.Add(tx)}; if (present_before) { - Assert(res == PrivateBroadcast::AddResult::AlreadyPresent); + auto tx_it{transactions.find(tx)}; + Assert(tx_it != transactions.end()); + if (is_pending(*tx_it)) { + Assert(res == PrivateBroadcast::AddResult::AlreadyPresent); + } else { + Assert(res == PrivateBroadcast::AddResult::Added); + tx_it->second = 0; + for (auto it = nodes_sent_to.begin(); it != nodes_sent_to.end();) { + if (CTransactionRefComp{}(it->second, tx)) { + nodes_that_confirmed_reception.erase(it->first); + it = nodes_sent_to.erase(it); + } else { + ++it; + } + } + } } else if (transactions.size() >= cap) { Assert(res == PrivateBroadcast::AddResult::QueueFull); } else { @@ -93,9 +118,8 @@ FUZZ_TARGET(private_broadcast) // Remove relevant entries from nodes_sent_to[] and nodes_that_confirmed_reception[] if any. for (auto it = nodes_sent_to.begin(); it != nodes_sent_to.end();) { - const NodeId nodeid{*it}; - const auto opt_tx_for_node{pb.GetTxForNode(nodeid)}; - if (opt_tx_for_node.has_value() && opt_tx_for_node.value() == tx) { + const NodeId nodeid{it->first}; + if (CTransactionRefComp{}(it->second, tx)) { it = nodes_sent_to.erase(it); if (nodes_that_confirmed_reception.erase(nodeid) > 0) { ++num_nodes_that_confirmed_tx; @@ -126,17 +150,18 @@ FUZZ_TARGET(private_broadcast) // (fewest sends = highest priority), so PickTxForSend() must return a transaction // with the minimum send count of any in the queue. Ties are broken by state we // don't model, so only check this key. + auto pending_transactions{transactions | std::views::filter(is_pending)}; const size_t min_picked{std::ranges::min_element( - transactions, {}, [](const auto& el) { return el.second; })->second}; + pending_transactions, {}, [](const auto& el) { return el.second; })->second}; const auto picked_it{transactions.find(opt_tx.value())}; Assert(picked_it != transactions.end()); Assert(picked_it->second == min_picked); // picked the least-sent transaction ++picked_it->second; // PickTxForSend() recorded exactly one send - const auto& [_, inserted]{nodes_sent_to.emplace(will_send_to_nodeid)}; + const auto& [_, inserted]{nodes_sent_to.emplace(will_send_to_nodeid, opt_tx.value())}; Assert(inserted); } else { - Assert(transactions.empty()); + Assert(std::ranges::none_of(transactions, is_pending)); } }, [&] { // GetTxForNode() @@ -147,6 +172,7 @@ FUZZ_TARGET(private_broadcast) if (nodes_sent_to.contains(nodeid)) { Assert(opt_tx.has_value()); Assert(transactions.contains(opt_tx.value())); + Assert(opt_tx.value() == nodes_sent_to.at(nodeid)); } else { Assert(!opt_tx.has_value()); } @@ -175,10 +201,10 @@ FUZZ_TARGET(private_broadcast) } }, [&] { // HavePendingTransactions() - if (pb.HavePendingTransactions()) { - Assert(!transactions.empty()); + if (std::ranges::any_of(transactions, is_pending)) { + Assert(pb.HavePendingTransactions()); } else { - Assert(transactions.empty()); + Assert(!pb.HavePendingTransactions()); } }, [&] { // GetStale() @@ -187,7 +213,9 @@ FUZZ_TARGET(private_broadcast) Assert(stale.size() <= transactions.size()); for (const auto& stale_tx : stale) { - Assert(transactions.contains(stale_tx)); + const auto it{transactions.find(stale_tx)}; + Assert(it != transactions.end()); + Assert(is_pending(*it)); } }, [&] { // GetBroadcastInfo() @@ -199,6 +227,7 @@ FUZZ_TARGET(private_broadcast) const auto it{transactions.find(info.tx)}; Assert(it != transactions.end()); Assert(info.peers.size() == it->second); // exactly the sends we recorded + Assert(info.attempts_remaining == max_send_attempts - it->second); } }, [&] { diff --git a/src/test/private_broadcast_tests.cpp b/src/test/private_broadcast_tests.cpp index c65d44c5206..94e68d74c19 100644 --- a/src/test/private_broadcast_tests.cpp +++ b/src/test/private_broadcast_tests.cpp @@ -169,6 +169,86 @@ BOOST_AUTO_TEST_CASE(stale_unpicked_tx) BOOST_CHECK_EQUAL(stale_state[0], tx); } +BOOST_AUTO_TEST_CASE(send_attempt_limit) +{ + FakeNodeClock clock{}; + + constexpr size_t max_attempts{5}; + PrivateBroadcast pb{PrivateBroadcast::MAX_TRANSACTIONS, max_attempts}; + const auto tx{MakeDummyTx(/*id=*/1, /*num_witness=*/0)}; + BOOST_REQUIRE_EQUAL(pb.Add(tx), PrivateBroadcast::AddResult::Added); + + in_addr ipv4_addr; + ipv4_addr.s_addr = 0xa0b0c001; + const CService address{ipv4_addr, 1111}; + + NodeId node_id{0}; + for (size_t attempt{0}; attempt < max_attempts; ++attempt) { + BOOST_CHECK(pb.HavePendingTransactions()); + BOOST_REQUIRE_EQUAL(pb.PickTxForSend(/*will_send_to_nodeid=*/node_id++, address).value(), tx); + } + + // The transaction and its complete send history remain available, but no + // further connections should be opened for it. + BOOST_CHECK(!pb.HavePendingTransactions()); + BOOST_CHECK(!pb.PickTxForSend(/*will_send_to_nodeid=*/node_id++, address).has_value()); + const auto info{pb.GetBroadcastInfo()}; + BOOST_REQUIRE_EQUAL(info.size(), 1); + BOOST_CHECK_EQUAL(info[0].peers.size(), max_attempts); + BOOST_CHECK_EQUAL(info[0].attempts_remaining, 0); + + clock += PrivateBroadcast::INITIAL_STALE_DURATION + 1min; + BOOST_CHECK(pb.GetStale().empty()); + + // An exhausted transaction does not prevent another transaction from being sent. + const auto next_tx{MakeDummyTx(/*id=*/2, /*num_witness=*/0)}; + BOOST_REQUIRE_EQUAL(pb.Add(next_tx), PrivateBroadcast::AddResult::Added); + BOOST_CHECK(pb.HavePendingTransactions()); + BOOST_REQUIRE_EQUAL(pb.PickTxForSend(/*will_send_to_nodeid=*/node_id++, address).value(), next_tx); + + // Re-adding an exhausted transaction resets its state and starts a fresh + // initial stale period. + BOOST_REQUIRE_EQUAL(pb.Add(tx), PrivateBroadcast::AddResult::Added); + BOOST_CHECK(pb.HavePendingTransactions()); + BOOST_CHECK(pb.GetStale().empty()); + const auto reset_info{pb.GetBroadcastInfo()}; + const auto reset_tx_info{std::ranges::find(reset_info, tx->GetWitnessHash(), [](const auto& entry) { return entry.tx->GetWitnessHash(); })}; + BOOST_REQUIRE(reset_tx_info != reset_info.end()); + BOOST_CHECK(reset_tx_info->peers.empty()); + BOOST_CHECK_EQUAL(reset_tx_info->attempts_remaining, max_attempts); + BOOST_REQUIRE_EQUAL(pb.PickTxForSend(/*will_send_to_nodeid=*/node_id++, address).value(), tx); +} + +BOOST_AUTO_TEST_CASE(reset_with_equivalent_transaction_reference) +{ + PrivateBroadcast pb{PrivateBroadcast::MAX_TRANSACTIONS, /*max_send_attempts=*/1}; + const auto tx{MakeDummyTx(/*id=*/1, /*num_witness=*/0)}; + const auto equivalent_tx{MakeDummyTx(/*id=*/1, /*num_witness=*/0)}; + BOOST_REQUIRE(tx != equivalent_tx); + BOOST_REQUIRE(tx->GetWitnessHash() == equivalent_tx->GetWitnessHash()); + + in_addr ipv4_addr; + ipv4_addr.s_addr = 0xa0b0c001; + const CService address{ipv4_addr, 1111}; + BOOST_REQUIRE_EQUAL(pb.Add(tx), PrivateBroadcast::AddResult::Added); + BOOST_REQUIRE_EQUAL(pb.PickTxForSend(/*will_send_to_nodeid=*/0, address).value(), tx); + pb.NodeConfirmedReception(/*nodeid=*/0); + BOOST_CHECK(pb.DidNodeConfirmReception(/*nodeid=*/0)); + BOOST_CHECK(!pb.HavePendingTransactions()); + + // A distinct CTransactionRef with the same WTXID must reset the exhausted + // transaction, including its send and confirmation history. + BOOST_REQUIRE_EQUAL(pb.Add(equivalent_tx), PrivateBroadcast::AddResult::Added); + BOOST_CHECK(pb.HavePendingTransactions()); + BOOST_CHECK(!pb.GetTxForNode(/*nodeid=*/0).has_value()); + BOOST_CHECK(!pb.DidNodeConfirmReception(/*nodeid=*/0)); + const auto info{pb.GetBroadcastInfo()}; + BOOST_REQUIRE_EQUAL(info.size(), 1); + BOOST_CHECK(info[0].tx->GetWitnessHash() == tx->GetWitnessHash()); + BOOST_CHECK(info[0].peers.empty()); + BOOST_CHECK_EQUAL(pb.Remove(equivalent_tx).value(), 0); +} + BOOST_AUTO_TEST_CASE(rejection_at_cap) { PrivateBroadcast pb; diff --git a/test/functional/p2p_private_broadcast.py b/test/functional/p2p_private_broadcast.py index 054058e946b..d6d22bdbbf7 100755 --- a/test/functional/p2p_private_broadcast.py +++ b/test/functional/p2p_private_broadcast.py @@ -47,6 +47,7 @@ from test_framework.wallet import ( P2P_PRIVATE_VERSION = 70016 NUM_PRIVATE_BROADCAST_PER_TX = 3 +MAX_PRIVATE_BROADCAST_ATTEMPTS = 1000 class NoRelayP2PInterface(P2PInterface): @@ -227,7 +228,8 @@ class P2PPrivateBroadcast(BitcoinTestFramework): assert_equal(len(pending), 1) assert_equal(pending[0]["hex"].lower(), tx["hex"].lower()) peers = pending[0]["peers"] - assert len(peers) >= NUM_PRIVATE_BROADCAST_PER_TX + assert_greater_than_or_equal(len(peers), NUM_PRIVATE_BROADCAST_PER_TX) + assert_equal(pending[0]["attempts_remaining"], MAX_PRIVATE_BROADCAST_ATTEMPTS - len(peers)) assert all("address" in p and "sent" in p for p in peers) assert_greater_than_or_equal(sum(1 for p in peers if "received" in p), broadcasts_to_expect)