Merge bitcoin/bitcoin#35680: private broadcast: bound rebroadcast attempts to 1,000

fe7d475d45 private broadcast: bound broadcast attempts per tx to 1k (Gregory Sanders)

Pull request description:

  Since rebroacasts introduce additional state, bound the state growth by capping the number of rebroadcasts. With ~72 bytes per record, 10k transactions rebroadcasting for ~42 hours will result about 703 MiB allocated with overhead.

ACKs for top commit:
  andrewtoth:
    ACK fe7d475d45
  frankomosh:
    ReACK fe7d475d45
  sedited:
    ACK fe7d475d45

Tree-SHA512: e4ec5156b90ad24d68b561df03ad09bdf0ac7535886ff56891cb698cf64ff0e1e484075b76040bba6194baf874c9237028c82debf7405136447ba5b5faee589c

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);
                 }
             },
             [&] {