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)