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

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

src/private_broadcast.cpp

@@ -7,14 +7,21 @@
 #include <util/check.h>
 
 #include <algorithm>
+#include <ranges>
 
 
 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<CTransactionRef> 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<CTransactionRef> PrivateBroadcast::GetStale() const
@@ -104,6 +112,7 @@ std::vector<CTransactionRef> PrivateBroadcast::GetStale() const
     const auto now{NodeClock::now()};
     std::vector<CTransactionRef> 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::TxBroadcastInfo> 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<SendStatus>& sent_to)
 {
     Priority p;

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<PeerSendInfo> 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<CTransactionRef> 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<CTransactionRef> GetStale() const
         EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
@@ -212,11 +226,14 @@ private:
     std::optional<TxAndSendStatusForNode> 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<SendStatus> 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<CTransactionRef, TxSendStatus, CTransactionRefHash, CTransactionRefComp>
         m_transactions GUARDED_BY(m_mutex);

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<std::chrono::seconds>(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);

src/test/fuzz/private_broadcast.cpp

@@ -16,6 +16,9 @@
 #include <util/overflow.h>
 #include <util/time.h>
 
+#include <algorithm>
+#include <ranges>
+#include <unordered_map>
 #include <unordered_set>
 
 struct CTransactionRefHash {
@@ -39,20 +42,27 @@ FUZZ_TARGET(private_broadcast)
     FakeNodeClock clock_ctx{ConsumeTime(fdp)};
 
     const size_t cap{fdp.ConsumeIntegralInRange<size_t>(1, 12)};
-    PrivateBroadcast pb{cap};
+    const size_t max_send_attempts{fdp.ConsumeIntegralInRange<size_t>(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<CTransactionRef, size_t, CTransactionRefHash, CTransactionRefComp> transactions;
 
-    // Ids of nodes that were passed to PickTxForSend(). Trimmed when Remove() is called.
-    std::unordered_set<NodeId> 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<NodeId, CTransactionRef> 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<NodeId> 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);
                 }
             },
             [&] {

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;

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)