Merge bitcoin/bitcoin#35406: private broadcast: limit outstanding txs to count of 10,000

4e29de719e private broadcast: add release note for limited cap (Gregory Sanders)
cbf8c107c1 Release cs_main between individual private tx re-attempts (Greg Sanders)
5aea3d0373 private broadcast: limit outstanding txs to count of 10,000 (Gregory Sanders)

Pull request description:

  Add a belt-and-suspenders feature, limit the amount of memory and cpu possible when unlucky or simply misconfigured. The worst case limit is roughly 400kB * 10,000 = 4GB, regardless of usage pattern.

  Before this change, sheer volume of broadcasts, mismatches in standardness rules, or simply fee mismatches may result in unbounded growth of memory usage. As the feature may be expanded in the future, explicit bounds helps reasoning going forward.

ACKs for top commit:
  frankomosh:
    tACK 4e29de719e. Ran private_broadcast_tests and p2p_private_broadcast_cap.py. Great to have an explicit bound as the belt-and-suspenders against unbounded queue growth.
  vasild:
    ACK 4e29de719e
  andrewtoth:
    ACK 4e29de719e
  stickies-v:
    ACK 4e29de719e

Tree-SHA512: 18161755f37d07cca185a09e782dbe2fd0025b8befd4f6660e988865cc3a9b705d41769b816161e8142fe6ce31a56e0288bd78efc25135cedfc47fc855011799

doc/release-notes-35406.md

@@ -0,0 +1,8 @@
+P2P and network changes
+-----------------------
+
+- The private-broadcast queue (transactions submitted via `sendrawtransaction`
+when `-privatebroadcast` is enabled and not yet echoed back from the network)
+is now capped at 10,000 entries. When full, new submissions are rejected. It is
+up to the caller to inspect the queue via `getprivatebroadcastinfo` and free
+up space when stuck via `abortprivatebroadcast`. (#35406)

src/common/messages.cpp

@@ -143,6 +143,8 @@ bilingual_str TransactionErrorString(const TransactionError err)
             return Untranslated("Unspendable output exceeds maximum configured by user (maxburnamount)");
         case TransactionError::INVALID_PACKAGE:
             return Untranslated("Transaction rejected due to invalid package");
+        case TransactionError::PRIVATE_BROADCAST_FULL:
+            return Untranslated("Private broadcast queue is full");
     } // no default case, so the compiler can warn about missing cases
     assert(false);
 }

src/net_processing.cpp

@@ -540,7 +540,7 @@ public:
     std::vector<CTransactionRef> AbortPrivateBroadcast(const uint256& id) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     void SendPings() override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
-    void InitiateTxBroadcastPrivate(const CTransactionRef& tx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
+    node::TransactionError InitiateTxBroadcastPrivate(const CTransactionRef& tx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     void SetBestBlock(int height, std::chrono::seconds time) override
     {
         m_best_height = height;
@@ -1660,8 +1660,9 @@ void PeerManagerImpl::ReattemptPrivateBroadcast(CScheduler& scheduler)
     size_t num_for_rebroadcast{0};
     const auto stale_txs = m_tx_for_private_broadcast.GetStale();
     if (!stale_txs.empty()) {
-        LOCK(cs_main);
         for (const auto& stale_tx : stale_txs) {
+            // Only hold lock per single submission
+            LOCK(cs_main);
             auto mempool_acceptable = m_chainman.ProcessTransaction(stale_tx, /*test_accept=*/true);
             if (mempool_acceptable.m_result_type == MempoolAcceptResult::ResultType::VALID) {
                 LogDebug(BCLog::PRIVBROADCAST,
@@ -2293,15 +2294,22 @@ void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wt
     }
 }
 
-void PeerManagerImpl::InitiateTxBroadcastPrivate(const CTransactionRef& tx)
+node::TransactionError PeerManagerImpl::InitiateTxBroadcastPrivate(const CTransactionRef& tx)
 {
     const auto txstr{strprintf("txid=%s, wtxid=%s", tx->GetHash().ToString(), tx->GetWitnessHash().ToString())};
-    if (m_tx_for_private_broadcast.Add(tx)) {
+    switch (m_tx_for_private_broadcast.Add(tx)) {
+    case PrivateBroadcast::AddResult::Added:
         LogDebug(BCLog::PRIVBROADCAST, "Requesting %d new connections due to %s", NUM_PRIVATE_BROADCAST_PER_TX, txstr);
         m_connman.m_private_broadcast.NumToOpenAdd(NUM_PRIVATE_BROADCAST_PER_TX);
-    } else {
+        return node::TransactionError::OK;
+    case PrivateBroadcast::AddResult::AlreadyPresent:
         LogDebug(BCLog::PRIVBROADCAST, "Ignoring unnecessary request to schedule an already scheduled transaction: %s", txstr);
-    }
+        return node::TransactionError::OK;
+    case PrivateBroadcast::AddResult::QueueFull:
+        LogDebug(BCLog::PRIVBROADCAST, "Rejecting private broadcast, queue full (cap=%u): %s", PrivateBroadcast::MAX_TRANSACTIONS, txstr);
+        return node::TransactionError::PRIVATE_BROADCAST_FULL;
+    } // no default case, so the compiler can warn about missing cases
+    assert(false);
 }
 
 void PeerManagerImpl::RelayAddress(NodeId originator,

src/net_processing.h

@@ -9,6 +9,7 @@
 #include <consensus/amount.h>
 #include <net.h>
 #include <node/txorphanage.h>
+#include <node/types.h>
 #include <private_broadcast.h>
 #include <protocol.h>
 #include <uint256.h>
@@ -147,8 +148,10 @@ public:
     /**
      * Initiate a private transaction broadcast. This is done
      * asynchronously via short-lived connections to peers on privacy networks.
+     * @retval node::TransactionError::OK The transaction is scheduled for private broadcast (or was already scheduled).
+     * @retval node::TransactionError::PRIVATE_BROADCAST_FULL Rejected because the private broadcast queue is full.
      */
-    virtual void InitiateTxBroadcastPrivate(const CTransactionRef& tx) = 0;
+    [[nodiscard]] virtual node::TransactionError InitiateTxBroadcastPrivate(const CTransactionRef& tx) = 0;
 
     /** Send ping message to all peers */
     virtual void SendPings() = 0;

src/node/transaction.cpp

@@ -133,8 +133,7 @@ TransactionError BroadcastTransaction(NodeContext& node,
         node.peerman->InitiateTxBroadcastToAll(txid, wtxid);
         break;
     case TxBroadcast::NO_MEMPOOL_PRIVATE_BROADCAST:
-        node.peerman->InitiateTxBroadcastPrivate(tx);
-        break;
+        return node.peerman->InitiateTxBroadcastPrivate(tx);
     }
 
     return TransactionError::OK;

src/node/types.h

@@ -25,6 +25,7 @@ enum class TransactionError {
     MAX_FEE_EXCEEDED,
     MAX_BURN_EXCEEDED,
     INVALID_PACKAGE,
+    PRIVATE_BROADCAST_FULL,
 };
 
 /**

src/private_broadcast.cpp

@@ -3,17 +3,23 @@
 // file COPYING or https://opensource.org/license/mit/.
 
 #include <private_broadcast.h>
+
 #include <util/check.h>
 
 #include <algorithm>
 
 
-bool PrivateBroadcast::Add(const CTransactionRef& tx)
+[[nodiscard]] PrivateBroadcast::AddResult PrivateBroadcast::Add(const CTransactionRef& tx)
     EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
 {
     LOCK(m_mutex);
-    const bool inserted{m_transactions.try_emplace(tx).second};
-    return inserted;
+    // Re-adding an already-tracked transaction is a no-op regardless of the cap.
+    if (m_transactions.contains(tx)) return AddResult::AlreadyPresent;
+
+    if (m_transactions.size() >= m_max_transactions) return AddResult::QueueFull;
+
+    m_transactions.try_emplace(tx);
+    return AddResult::Added;
 }
 
 std::optional<size_t> PrivateBroadcast::Remove(const CTransactionRef& tx)

src/private_broadcast.h

@@ -38,6 +38,15 @@ public:
     /// after it is broadcast, then we consider it stale / for rebroadcasting.
     static constexpr auto STALE_DURATION{1min};
 
+    /// Maximum number of transactions tracked simultaneously.
+    /// Additions that would exceed this are rejected (see Add()).
+    static constexpr size_t MAX_TRANSACTIONS{10'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} {}
+
     struct PeerSendInfo {
         CService address;
         NodeClock::time_point sent;
@@ -50,13 +59,23 @@ public:
         std::vector<PeerSendInfo> peers;
     };
 
+    /// Outcome of Add().
+    enum class AddResult {
+        //! The transaction was newly added.
+        Added,
+        //! The transaction was already present; no change.
+        AlreadyPresent,
+        //! Rejected: the queue is already at MAX_TRANSACTIONS.
+        QueueFull,
+    };
+
     /**
      * Add a transaction to the storage.
      * @param[in] tx The transaction to add.
-     * @retval true The transaction was added.
-     * @retval false The transaction was already present.
+     * @return Whether the transaction was newly added, was already present, or
+     * was rejected because the queue is full (see AddResult).
      */
-    bool Add(const CTransactionRef& tx)
+    [[nodiscard]] AddResult Add(const CTransactionRef& tx)
         EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
 
     /**
@@ -196,6 +215,8 @@ private:
         const NodeClock::time_point time_added{NodeClock::now()};
         std::vector<SendStatus> send_statuses;
     };
+    /// Cap on the number of simultaneously tracked transactions (see Add()).
+    const size_t m_max_transactions;
     mutable Mutex m_mutex;
     std::unordered_map<CTransactionRef, TxSendStatus, CTransactionRefHash, CTransactionRefComp>
         m_transactions GUARDED_BY(m_mutex);

src/rpc/mempool.cpp

@@ -60,6 +60,9 @@ static RPCMethod sendrawtransaction()
         "dedicated, short-lived connections to Tor or I2P peers or IPv4/IPv6 peers\n"
         "via the Tor network. This conceals the transaction's origin. The transaction\n"
         "will only enter the local mempool when it is received back from the network.\n"
+        "The private broadcast queue is bounded: when it is full, this RPC fails and\n"
+        "the transaction is not scheduled, until an existing one completes or is\n"
+        "aborted. Use getprivatebroadcastinfo to inspect the queue and abortprivatebroadcast to abort.\n"
 
         "\nA specific exception, RPC_TRANSACTION_ALREADY_IN_UTXO_SET, may throw if the transaction cannot be added to the mempool.\n"
 

src/rpc/util.cpp

@@ -395,6 +395,8 @@ RPCErrorCode RPCErrorFromTransactionError(TransactionError terr)
             return RPC_TRANSACTION_REJECTED;
         case TransactionError::ALREADY_IN_UTXO_SET:
             return RPC_VERIFY_ALREADY_IN_UTXO_SET;
+        case TransactionError::PRIVATE_BROADCAST_FULL:
+            return RPC_OUT_OF_MEMORY;
         default: break;
     }
     return RPC_TRANSACTION_ERROR;

src/test/fuzz/private_broadcast.cpp

@@ -38,7 +38,8 @@ FUZZ_TARGET(private_broadcast)
     FuzzedDataProvider fdp(buffer.data(), buffer.size());
     FakeNodeClock clock_ctx{ConsumeTime(fdp)};
 
-    PrivateBroadcast pb;
+    const size_t cap{fdp.ConsumeIntegralInRange<size_t>(1, 12)};
+    PrivateBroadcast pb{cap};
 
     // 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.
@@ -64,15 +65,20 @@ FUZZ_TARGET(private_broadcast)
             fdp,
             [&] { // Add()
                 CTransactionRef tx;
-                bool from_transactions{false};
                 if (transactions.empty() || fdp.ConsumeBool()) {
                     tx = MakeTransactionRef(ConsumeTransaction(fdp, std::nullopt));
                 } else {
                     tx = PickIterator(fdp, transactions)->first;
-                    from_transactions = true;
                 }
-                if (pb.Add(tx)) {
-                    Assert(!from_transactions);
+
+                const bool present_before{transactions.contains(tx)};
+                const auto res{pb.Add(tx)};
+                if (present_before) {
+                    Assert(res == PrivateBroadcast::AddResult::AlreadyPresent);
+                } else if (transactions.size() >= cap) {
+                    Assert(res == PrivateBroadcast::AddResult::QueueFull);
+                } else {
+                    Assert(res == PrivateBroadcast::AddResult::Added);
                     transactions.emplace(tx, 0);
                 }
             },

src/test/private_broadcast_tests.cpp

@@ -44,15 +44,15 @@ BOOST_AUTO_TEST_CASE(basic)
     // Make a transaction and add it.
     const auto tx1{MakeDummyTx(/*id=*/1, /*num_witness=*/0)};
 
-    BOOST_CHECK(pb.Add(tx1));
-    BOOST_CHECK(!pb.Add(tx1));
+    BOOST_CHECK(pb.Add(tx1) == PrivateBroadcast::AddResult::Added);
+    BOOST_CHECK(pb.Add(tx1) == PrivateBroadcast::AddResult::AlreadyPresent);
 
     // Make another transaction with same txid, different wtxid and add it.
     const auto tx2{MakeDummyTx(/*id=*/1, /*num_witness=*/1)};
     BOOST_REQUIRE(tx1->GetHash() == tx2->GetHash());
     BOOST_REQUIRE(tx1->GetWitnessHash() != tx2->GetWitnessHash());
 
-    BOOST_CHECK(pb.Add(tx2));
+    BOOST_CHECK(pb.Add(tx2) == PrivateBroadcast::AddResult::Added);
     const auto find_tx_info{[](auto& infos, const CTransactionRef& tx) -> const PrivateBroadcast::TxBroadcastInfo& {
         const auto it{std::ranges::find(infos, tx->GetWitnessHash(), [](const auto& info) { return info.tx->GetWitnessHash(); })};
         BOOST_REQUIRE(it != infos.end());
@@ -146,7 +146,7 @@ BOOST_AUTO_TEST_CASE(stale_unpicked_tx)
 
     PrivateBroadcast pb;
     const auto tx{MakeDummyTx(/*id=*/42, /*num_witness=*/0)};
-    BOOST_REQUIRE(pb.Add(tx));
+    BOOST_REQUIRE(pb.Add(tx) == PrivateBroadcast::AddResult::Added);
 
     // Unpicked transactions use the longer INITIAL_STALE_DURATION.
     BOOST_CHECK_EQUAL(pb.GetStale().size(), 0);
@@ -158,4 +158,56 @@ BOOST_AUTO_TEST_CASE(stale_unpicked_tx)
     BOOST_CHECK_EQUAL(stale_state[0], tx);
 }
 
+BOOST_AUTO_TEST_CASE(rejection_at_cap)
+{
+    PrivateBroadcast pb;
+    constexpr size_t num_cap{PrivateBroadcast::MAX_TRANSACTIONS};
+    constexpr size_t num_over{5};
+
+    // Fill the queue exactly to the cap; every distinct Add() succeeds.
+    std::vector<CTransactionRef> txs;
+    txs.reserve(num_cap);
+    for (size_t i{0}; i < num_cap; ++i) {
+        auto tx{MakeDummyTx(/*id=*/static_cast<uint32_t>(i), /*num_witness=*/0)};
+        BOOST_REQUIRE(pb.Add(tx) == PrivateBroadcast::AddResult::Added);
+        txs.push_back(std::move(tx));
+    }
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap);
+
+    // Further distinct transactions are rejected, and the queue is unchanged.
+    for (size_t i{0}; i < num_over; ++i) {
+        const auto tx{MakeDummyTx(/*id=*/static_cast<uint32_t>(num_cap + i), /*num_witness=*/0)};
+        BOOST_CHECK(pb.Add(tx) == PrivateBroadcast::AddResult::QueueFull);
+    }
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap);
+
+    // Nothing was evicted: all originally-added transactions are still present.
+    const auto infos{pb.GetBroadcastInfo()};
+    std::set<uint256> present_wtxids;
+    for (const auto& info : infos) {
+        present_wtxids.insert(info.tx->GetWitnessHash().ToUint256());
+    }
+    BOOST_CHECK_EQUAL(present_wtxids.size(), infos.size());
+    for (size_t i{0}; i < num_cap; ++i) {
+        BOOST_CHECK_MESSAGE(present_wtxids.contains(txs[i]->GetWitnessHash().ToUint256()),
+                            "tx index " << i << " should still be present");
+    }
+
+    // Re-adding an already-present tx is AlreadyPresent even at the cap (not QueueFull).
+    BOOST_CHECK(pb.Add(txs[0]) == PrivateBroadcast::AddResult::AlreadyPresent);
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap);
+
+    // Removing one frees exactly one slot for a new transaction.
+    BOOST_REQUIRE(pb.Remove(txs[0]).has_value());
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap - 1);
+    const auto fresh{MakeDummyTx(/*id=*/0xffffffff, /*num_witness=*/0)};
+    BOOST_CHECK(pb.Add(fresh) == PrivateBroadcast::AddResult::Added);
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap);
+
+    // A previously-removed tx can be added again as a brand-new entry
+    BOOST_REQUIRE(pb.Remove(fresh).has_value());
+    BOOST_CHECK(pb.Add(txs[0]) == PrivateBroadcast::AddResult::Added);
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), num_cap);
+}
+
 BOOST_AUTO_TEST_SUITE_END()

test/functional/p2p_private_broadcast_cap.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+# Copyright (c) 2026-present The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test the private-broadcast queue size cap: submissions beyond the cap are
+rejected (the queue is not modified), rather than evicting existing entries.
+"""
+
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import assert_equal, assert_raises_rpc_error
+from test_framework.wallet import MiniWallet
+
+
+# Must match PrivateBroadcast::MAX_TRANSACTIONS
+MAX_TRANSACTIONS = 10_000
+OVER_CAP = 5
+
+
+class PrivateBroadcastCapTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.num_nodes = 1
+        # -privatebroadcast is incompatible with the framework's default
+        # -connect=0; allow autoconnect (no actual peers will succeed though).
+        self.disable_autoconnect = False
+        self.extra_args = [[
+            "-privatebroadcast",
+            # Fake I2P reachability so the privatebroadcast startup precondition passes.
+            "-i2psam=127.0.0.1:1",
+            "-proxy=127.0.0.1:1",
+        ]]
+
+    def setup_network(self):
+        # Skip the framework's default connect_nodes loop. We have a single
+        # node and don't need any peer connections.
+        self.setup_nodes()
+
+    def run_test(self):
+        node = self.nodes[0]
+        wallet = MiniWallet(node)
+        # Mature one coinbase to spend.
+        self.generate(wallet, 101)
+
+        # Build a parent that fans out to MAX_TRANSACTIONS + OVER_CAP outputs.
+        # Inject it directly via generateblock since -privatebroadcast bypasses
+        # the mempool
+        utxo = wallet.get_utxo()
+        parent = wallet.create_self_transfer_multi(
+            utxos_to_spend=[utxo],
+            num_outputs=MAX_TRANSACTIONS + OVER_CAP,
+            fee_per_output=500,
+        )
+        self.generateblock(node, wallet.get_address(), [parent["hex"]])
+
+        children = [wallet.create_self_transfer(utxo_to_spend=u)
+                    for u in parent["new_utxos"]]
+        assert_equal(len(children), MAX_TRANSACTIONS + OVER_CAP)
+
+        # Fill the queue exactly to the cap; every distinct submission succeeds.
+        self.log.info(f"Filling private broadcast queue to cap ({MAX_TRANSACTIONS} txns)")
+        for child in children[:MAX_TRANSACTIONS]:
+            node.sendrawtransaction(child["hex"])
+
+        pbinfo = node.getprivatebroadcastinfo()
+        assert_equal(len(pbinfo["transactions"]), MAX_TRANSACTIONS)
+        present_wtxids = {t["wtxid"] for t in pbinfo["transactions"]}
+        for i, child in enumerate(children[:MAX_TRANSACTIONS]):
+            assert child["wtxid"] in present_wtxids, \
+                f"tx index {i} (wtxid={child['wtxid']}) should be in the queue"
+
+        # Further distinct submissions are rejected with an RPC error, and the
+        # queue is left unchanged (nothing evicted to make room).
+        self.log.info(f"Submitting {OVER_CAP} more; each should be rejected (queue full)")
+        for child in children[MAX_TRANSACTIONS:]:
+            assert_raises_rpc_error(-7, "Private broadcast queue is full",
+                                    node.sendrawtransaction, child["hex"])
+
+        assert_equal(pbinfo["transactions"], node.getprivatebroadcastinfo()["transactions"])
+
+        self.log.info("Checking abortprivatebroadcast frees a slot for a new submission")
+        abort_res = node.abortprivatebroadcast(children[1]["txid"])
+        assert_equal([t["wtxid"] for t in abort_res["removed_transactions"]],
+                     [children[1]["wtxid"]])
+        wtxids = {t["wtxid"] for t in node.getprivatebroadcastinfo()["transactions"]}
+        assert_equal(len(wtxids), MAX_TRANSACTIONS - 1)
+        assert children[1]["wtxid"] not in wtxids, "aborted tx should be gone from the queue"
+
+        new_child = children[MAX_TRANSACTIONS]  # first previously-rejected tx
+        node.sendrawtransaction(new_child["hex"])
+        wtxids = {t["wtxid"] for t in node.getprivatebroadcastinfo()["transactions"]}
+        assert_equal(len(wtxids), MAX_TRANSACTIONS)
+        assert new_child["wtxid"] in wtxids, "freed slot should now hold the new tx"
+        assert children[1]["wtxid"] not in wtxids, "aborted tx should not reappear"
+
+        # Re-submitting an already-queued transaction is a no-op, not an error,
+        # even when the queue is full.
+        self.log.info("Re-submitting an already-queued tx should not error")
+        node.sendrawtransaction(children[0]["hex"])
+        wtxids = {t["wtxid"] for t in node.getprivatebroadcastinfo()["transactions"]}
+        assert_equal(len(wtxids), MAX_TRANSACTIONS)
+        assert children[0]["wtxid"] in wtxids, "re-submitted tx should remain queued"
+
+if __name__ == "__main__":
+    PrivateBroadcastCapTest(__file__).main()

test/functional/test_runner.py

@@ -316,6 +316,7 @@ BASE_SCRIPTS = [
     'feature_minchainwork.py',
     'rpc_estimatefee.py',
     'p2p_private_broadcast.py',
+    'p2p_private_broadcast_cap.py',
     'rpc_getblockstats.py',
     'feature_port.py',
     'feature_bind_port_externalip.py',