net: always complete all initial private broadcast connections #36277

1767 | @@ -1767,7 +1768,8 @@ void PeerManagerImpl::ReattemptPrivateBroadcast(CScheduler& scheduler)
1768 |                  LogDebug(BCLog::PRIVBROADCAST, "Giving up broadcast attempts for txid=%s wtxid=%s: %s",
1769 |                           stale_tx->GetHash().ToString(), stale_tx->GetWitnessHash().ToString(),
1770 |                           mempool_acceptable.m_state.ToString());
1771 | -                m_tx_for_private_broadcast.Remove(stale_tx);
1772 | +                // Mark received so it will get cleaned up.
1773 | +                m_tx_for_private_broadcast.MarkReceived(stale_tx);

---

1856 | -        m_connman.m_private_broadcast.NumToOpenAdd(1);
1857 | +    if (node.IsPrivateBroadcastConn()) {
1858 | +        m_tx_for_private_broadcast.NodeDisconnected(nodeid);
1859 | +        // Retry a connection if we didn't complete the handshake.
1860 | +        if (!node.fSuccessfullyConnected && m_tx_for_private_broadcast.HavePendingTransactions()) {
1861 | +            m_connman.m_private_broadcast.NumToOpenAdd(1);

---

  13 | @@ -14,12 +14,21 @@ PrivateBroadcast::AddResult PrivateBroadcast::Add(const CTransactionRef& tx)
  14 |      EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
  15 |  {
  16 |      LOCK(m_mutex);
  17 | +    // Cleanup finished transactions
  18 | +    std::erase_if(m_transactions, [this](const auto& entry) {
  19 | +        const auto& state{entry.second};
  20 | +        return state.received && !IsPending(state) &&
  21 | +               std::ranges::all_of(state.send_statuses, [](const auto& status) { return status.disconnected; });
  22 | +    });

---

4728 | @@ -4726,15 +4729,10 @@ void PeerManagerImpl::ProcessMessage(Peer& peer, CNode& pfrom, const std::string
4729 |          const uint256& hash = peer.m_wtxid_relay ? wtxid.ToUint256() : txid.ToUint256();
4730 |          AddKnownTx(peer, hash);
4731 |  
4732 | -        if (const auto num_broadcasted{m_tx_for_private_broadcast.Remove(ptx)}) {
4733 | +        if (m_tx_for_private_broadcast.MarkReceived(ptx)) {
4734 |              LogDebug(BCLog::PRIVBROADCAST, "Received our privately broadcast transaction (txid=%s) from the "
4735 | -                                           "network from %s; stopping private broadcast attempts",
4736 | +                                           "network from %s; stopping private broadcast retries",

---

 102 | @@ -100,6 +103,30 @@ class PrivateBroadcast
 103 |      std::optional<size_t> Remove(const CTransactionRef& tx)
 104 |          EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
 105 |  
 106 | +    /**
 107 | +     * Mark a transaction as being received back from the network.
 108 | +     * @param[in] tx Transaction received from the network.
 109 | +     * @return Whether the transaction was found in the storage.
 110 | +     */
 111 | +    bool MarkReceived(const CTransactionRef& tx)

---

1848 | @@ -1847,11 +1849,12 @@ void PeerManagerImpl::FinalizeNode(const CNode& node)
1849 |          LOCK(m_headers_presync_mutex);
1850 |          m_headers_presync_stats.erase(nodeid);
1851 |      }
1852 | -    if (node.IsPrivateBroadcastConn() &&
1853 | -        !m_tx_for_private_broadcast.DidNodeConfirmReception(nodeid) &&

---

 256 | @@ -228,6 +257,10 @@ class PrivateBroadcast
 257 |      struct TxSendStatus {
 258 |          NodeClock::time_point time_added{NodeClock::now()};
 259 |          std::vector<SendStatus> send_statuses;
 260 | +        /// Total number of sends granted, including initial count.
 261 | +        size_t send_limit{INITIAL_BROADCAST_COUNT};
 262 | +        /// Whether the transaction was received back from the network.
 263 | +        bool received{false};

---

 256 | @@ -228,6 +257,10 @@ class PrivateBroadcast
 257 |      struct TxSendStatus {
 258 |          NodeClock::time_point time_added{NodeClock::now()};
 259 |          std::vector<SendStatus> send_statuses;
 260 | +        /// Total number of sends granted, including initial count.
 261 | +        size_t send_limit{INITIAL_BROADCAST_COUNT};

---

 201 | @@ -202,7 +202,7 @@ static constexpr double MAX_ADDR_RATE_PER_SECOND{0.1};
 202 |   *  is exempt from this limit). */
 203 |  static constexpr size_t MAX_ADDR_PROCESSING_TOKEN_BUCKET{MAX_ADDR_TO_SEND};
 204 |  /** For private broadcast, send a transaction to this many peers. */
 205 | -static constexpr size_t NUM_PRIVATE_BROADCAST_PER_TX{3};
 206 | +static constexpr size_t NUM_PRIVATE_BROADCAST_PER_TX{PrivateBroadcast::INITIAL_BROADCAST_COUNT};

---

1758 | @@ -1759,6 +1759,7 @@ void PeerManagerImpl::ReattemptPrivateBroadcast(CScheduler& scheduler)
1759 |              LOCK(cs_main);
1760 |              auto mempool_acceptable = m_chainman.ProcessTransaction(stale_tx, /*test_accept=*/true);
1761 |              if (mempool_acceptable.m_result_type == MempoolAcceptResult::ResultType::VALID) {
1762 | +                if (!m_tx_for_private_broadcast.TryGrantRetry(stale_tx)) continue;

---