Merge bitcoin/bitcoin#34329: rpc,net: Add private broadcast RPCs

2a1d0db799 doc: Mention private broadcast RPCs in release notes (Andrew Toth)
c3378be10b test: Cover abortprivatebroadcast in p2p_private_broadcast (Andrew Toth)
557260ca14 rpc: Add abortprivatebroadcast (Andrew Toth)
15dff452eb test: Cover getprivatebroadcastinfo in p2p_private_broadcast (Andrew Toth)
996f20c18a rpc: Add getprivatebroadcastinfo (Andrew Toth)
5e64982541 net: Add PrivateBroadcast::GetBroadcastInfo (Andrew Toth)
573bb542be net: Store recipient node address in private broadcast (Andrew Toth)

Pull request description:

  Follow up from #29415

  Sending a transaction via private broadcast does not have any way for a user to track the status of the transaction before it gets returned by another peer. The default logs have been removed as well in #34267. Nor is there any way to abort a transaction once it has been added to the private broadcast queue.

  This adds two new RPCs:
  - `getprivatebroadastinfo` returns information about what transactions are in the private broadcast queue, including all the peers' addresses we have chosen and timestamps.
  - `abortprivatebroadcast` stops broadcasting a transaction in the private broadcast queue.

ACKs for top commit:
  nervana21:
    tACK 2a1d0db799
  achow101:
    ACK 2a1d0db799
  l0rinc:
    ACK 2a1d0db799
  danielabrozzoni:
    tACK 2a1d0db799
  sedited:
    ACK 2a1d0db799

Tree-SHA512: cc8682d0be68a57b42bea6e3d091da2b80995d9e6d3b98644cb120a05c2b48a97c2e211173289b758c4f4e23f1d1a1f9be528a9b8c6644f71d1dd0ae5f673326

doc/release-notes-29415.md

@@ -12,3 +12,8 @@ P2P and network changes
   2. If the originator sends two otherwise unrelated transactions, they
      will not be linkable. This is because a separate connection is used
      for broadcasting each transaction. (#29415)
+
+- New RPCs have been added to introspect and control private broadcast:
+  `getprivatebroadcastinfo` reports transactions currently being privately
+  broadcast, and `abortprivatebroadcast` removes matching
+  transactions from the private broadcast queue.

src/net_processing.cpp

@@ -542,6 +542,8 @@ public:
     bool GetNodeStateStats(NodeId nodeid, CNodeStateStats& stats) const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
     std::vector<node::TxOrphanage::OrphanInfo> GetOrphanTransactions() override EXCLUSIVE_LOCKS_REQUIRED(!m_tx_download_mutex);
     PeerManagerInfo GetInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
+    std::vector<PrivateBroadcast::TxBroadcastInfo> GetPrivateBroadcastInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex);
+    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);
@@ -1855,6 +1857,31 @@ PeerManagerInfo PeerManagerImpl::GetInfo() const
     };
 }
 
+std::vector<PrivateBroadcast::TxBroadcastInfo> PeerManagerImpl::GetPrivateBroadcastInfo() const
+{
+    return m_tx_for_private_broadcast.GetBroadcastInfo();
+}
+
+std::vector<CTransactionRef> PeerManagerImpl::AbortPrivateBroadcast(const uint256& id)
+{
+    const auto snapshot{m_tx_for_private_broadcast.GetBroadcastInfo()};
+    std::vector<CTransactionRef> removed_txs;
+
+    size_t connections_cancelled{0};
+    for (const auto& [tx, _] : snapshot) {
+        if (tx->GetHash().ToUint256() != id && tx->GetWitnessHash().ToUint256() != id) continue;
+        if (const auto peer_acks{m_tx_for_private_broadcast.Remove(tx)}) {
+            removed_txs.push_back(tx);
+            if (NUM_PRIVATE_BROADCAST_PER_TX > *peer_acks) {
+                connections_cancelled += (NUM_PRIVATE_BROADCAST_PER_TX - *peer_acks);
+            }
+        }
+    }
+    m_connman.m_private_broadcast.NumToOpenSub(connections_cancelled);
+
+    return removed_txs;
+}
+
 void PeerManagerImpl::AddToCompactExtraTransactions(const CTransactionRef& tx)
 {
     if (m_opts.max_extra_txs <= 0)
@@ -3531,7 +3558,7 @@ void PeerManagerImpl::PushPrivateBroadcastTx(CNode& node)
 {
     Assume(node.IsPrivateBroadcastConn());
 
-    const auto opt_tx{m_tx_for_private_broadcast.PickTxForSend(node.GetId())};
+    const auto opt_tx{m_tx_for_private_broadcast.PickTxForSend(node.GetId(), CService{node.addr})};
     if (!opt_tx) {
         LogDebug(BCLog::PRIVBROADCAST, "Disconnecting: no more transactions for private broadcast (connected in vain), peer=%d%s", node.GetId(), node.LogIP(fLogIPs));
         node.fDisconnect = true;

src/net_processing.h

@@ -9,8 +9,10 @@
 #include <consensus/amount.h>
 #include <net.h>
 #include <node/txorphanage.h>
+#include <private_broadcast.h>
 #include <protocol.h>
 #include <threadsafety.h>
+#include <uint256.h>
 #include <util/expected.h>
 #include <validationinterface.h>
 
@@ -118,6 +120,21 @@ public:
     /** Get peer manager info. */
     virtual PeerManagerInfo GetInfo() const = 0;
 
+    /** Get info about transactions currently being privately broadcast. */
+    virtual std::vector<PrivateBroadcast::TxBroadcastInfo> GetPrivateBroadcastInfo() const = 0;
+
+    /**
+     * Abort private broadcast attempts for transactions currently being privately broadcast.
+     *
+     * @param[in] id A transaction identifier. It will be matched against both txid and wtxid for
+     *               all transactions in the private broadcast queue.
+     *
+     * @return Transactions removed from the private broadcast queue. If the provided id matches a
+     *         txid that corresponds to multiple transactions with different wtxids, multiple
+     *         transactions may be returned.
+     */
+    virtual std::vector<CTransactionRef> AbortPrivateBroadcast(const uint256& id) = 0;
+
     /**
      * Initiate a transaction broadcast to eligible peers.
      * Queue the witness transaction id to `Peer::TxRelay::m_tx_inventory_to_send`

src/private_broadcast.cpp

@@ -31,7 +31,7 @@ std::optional<size_t> PrivateBroadcast::Remove(const CTransactionRef& tx)
     return std::nullopt;
 }
 
-std::optional<CTransactionRef> PrivateBroadcast::PickTxForSend(const NodeId& will_send_to_nodeid)
+std::optional<CTransactionRef> PrivateBroadcast::PickTxForSend(const NodeId& will_send_to_nodeid, const CService& will_send_to_address)
     EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
 {
     LOCK(m_mutex);
@@ -43,7 +43,7 @@ std::optional<CTransactionRef> PrivateBroadcast::PickTxForSend(const NodeId& wil
 
     if (it != m_transactions.end()) {
         auto& [tx, sent_to]{*it};
-        sent_to.emplace_back(will_send_to_nodeid, NodeClock::now());
+        sent_to.emplace_back(will_send_to_nodeid, will_send_to_address, NodeClock::now());
         return tx;
     }
 
@@ -104,6 +104,25 @@ std::vector<CTransactionRef> PrivateBroadcast::GetStale() const
     return stale;
 }
 
+std::vector<PrivateBroadcast::TxBroadcastInfo> PrivateBroadcast::GetBroadcastInfo() const
+    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+{
+    LOCK(m_mutex);
+    std::vector<TxBroadcastInfo> entries;
+    entries.reserve(m_transactions.size());
+
+    for (const auto& [tx, sent_to] : m_transactions) {
+        std::vector<PeerSendInfo> peers;
+        peers.reserve(sent_to.size());
+        for (const auto& status : sent_to) {
+            peers.emplace_back(PeerSendInfo{.address = status.address, .sent = status.picked, .received = status.confirmed});
+        }
+        entries.emplace_back(TxBroadcastInfo{.tx = tx, .peers = std::move(peers)});
+    }
+
+    return entries;
+}
+
 PrivateBroadcast::Priority PrivateBroadcast::DerivePriority(const std::vector<SendStatus>& sent_to)
 {
     Priority p;

src/private_broadcast.h

@@ -30,6 +30,17 @@
 class PrivateBroadcast
 {
 public:
+    struct PeerSendInfo {
+        CService address;
+        NodeClock::time_point sent;
+        std::optional<NodeClock::time_point> received;
+    };
+
+    struct TxBroadcastInfo {
+        CTransactionRef tx;
+        std::vector<PeerSendInfo> peers;
+    };
+
     /**
      * Add a transaction to the storage.
      * @param[in] tx The transaction to add.
@@ -54,9 +65,11 @@ public:
      * and oldest send/confirm times.
      * @param[in] will_send_to_nodeid Will remember that the returned transaction
      * was picked for sending to this node.
+     * @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.
      */
-    std::optional<CTransactionRef> PickTxForSend(const NodeId& will_send_to_nodeid)
+    std::optional<CTransactionRef> PickTxForSend(const NodeId& will_send_to_nodeid, const CService& will_send_to_address)
         EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
 
     /**
@@ -95,14 +108,21 @@ public:
     std::vector<CTransactionRef> GetStale() const
         EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
 
+    /**
+     * Get stats about all transactions currently being privately broadcast.
+     */
+    std::vector<TxBroadcastInfo> GetBroadcastInfo() const
+        EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
 private:
     /// Status of a transaction sent to a given node.
     struct SendStatus {
         const NodeId nodeid; /// Node to which the transaction will be sent (or was sent).
+        const CService address; /// Address of the node.
         const NodeClock::time_point picked; ///< When was the transaction picked for sending to the node.
         std::optional<NodeClock::time_point> confirmed; ///< When was the transaction reception confirmed by the node (by PONG).
 
-        SendStatus(const NodeId& nodeid, const NodeClock::time_point& picked) : nodeid{nodeid}, picked{picked} {}
+        SendStatus(const NodeId& nodeid, const CService& address, const NodeClock::time_point& picked) : nodeid{nodeid}, address{address}, picked{picked} {}
     };
 
     /// Cumulative stats from all the send attempts for a transaction. Used to prioritize transactions.

src/rpc/mempool.cpp

@@ -137,6 +137,126 @@ static RPCHelpMan sendrawtransaction()
     };
 }
 
+static RPCHelpMan getprivatebroadcastinfo()
+{
+    return RPCHelpMan{
+        "getprivatebroadcastinfo",
+        "Returns information about transactions that are currently being privately broadcast.\n",
+        {},
+        RPCResult{
+            RPCResult::Type::OBJ, "", "",
+            {
+                {RPCResult::Type::ARR, "transactions", "",
+                    {
+                        {RPCResult::Type::OBJ, "", "",
+                            {
+                                {RPCResult::Type::STR_HEX, "txid", "The transaction hash in hex"},
+                                {RPCResult::Type::STR_HEX, "wtxid", "The transaction witness hash in hex"},
+                                {RPCResult::Type::STR_HEX, "hex", "The serialized, hex-encoded transaction data"},
+                                {RPCResult::Type::ARR, "peers", "Per-peer send and acknowledgment information for this transaction",
+                                    {
+                                        {RPCResult::Type::OBJ, "", "",
+                                            {
+                                                {RPCResult::Type::STR, "address", "The address of the peer to which the transaction was sent"},
+                                                {RPCResult::Type::NUM_TIME, "sent", "The time this transaction was picked for sending to this peer via private broadcast (seconds since epoch)"},
+                                                {RPCResult::Type::NUM_TIME, "received", /*optional=*/true, "The time this peer acknowledged reception of the transaction (seconds since epoch)"},
+                                            }},
+                                    }},
+                            }},
+                    }},
+            }},
+        RPCExamples{
+            HelpExampleCli("getprivatebroadcastinfo", "")
+            + HelpExampleRpc("getprivatebroadcastinfo", "")
+        },
+        [&](const RPCHelpMan& self, const JSONRPCRequest& request) -> UniValue
+        {
+            const NodeContext& node{EnsureAnyNodeContext(request.context)};
+            const PeerManager& peerman{EnsurePeerman(node)};
+            const auto txs{peerman.GetPrivateBroadcastInfo()};
+
+            UniValue transactions(UniValue::VARR);
+            for (const auto& tx_info : txs) {
+                UniValue o(UniValue::VOBJ);
+                o.pushKV("txid", tx_info.tx->GetHash().ToString());
+                o.pushKV("wtxid", tx_info.tx->GetWitnessHash().ToString());
+                o.pushKV("hex", EncodeHexTx(*tx_info.tx));
+                UniValue peers(UniValue::VARR);
+                for (const auto& peer : tx_info.peers) {
+                    UniValue p(UniValue::VOBJ);
+                    p.pushKV("address", peer.address.ToStringAddrPort());
+                    p.pushKV("sent", TicksSinceEpoch<std::chrono::seconds>(peer.sent));
+                    if (peer.received.has_value()) {
+                        p.pushKV("received", TicksSinceEpoch<std::chrono::seconds>(*peer.received));
+                    }
+                    peers.push_back(std::move(p));
+                }
+                o.pushKV("peers", std::move(peers));
+                transactions.push_back(std::move(o));
+            }
+
+            UniValue ret(UniValue::VOBJ);
+            ret.pushKV("transactions", std::move(transactions));
+            return ret;
+        },
+    };
+}
+
+static RPCHelpMan abortprivatebroadcast()
+{
+    return RPCHelpMan{
+        "abortprivatebroadcast",
+        "Abort private broadcast attempts for a transaction currently being privately broadcast.\n"
+        "The transaction will be removed from the private broadcast queue.\n",
+        {
+            {"id", RPCArg::Type::STR_HEX, RPCArg::Optional::NO, "A transaction identifier to abort. It will be matched against both txid and wtxid for all transactions in the private broadcast queue.\n"
+                                                                "If the provided id matches a txid that corresponds to multiple transactions with different wtxids, multiple transactions will be removed and returned."},
+        },
+        RPCResult{
+            RPCResult::Type::OBJ, "", "",
+            {
+                {RPCResult::Type::ARR, "removed_transactions", "Transactions removed from the private broadcast queue",
+                    {
+                        {RPCResult::Type::OBJ, "", "",
+                            {
+                                {RPCResult::Type::STR_HEX, "txid", "The transaction hash in hex"},
+                                {RPCResult::Type::STR_HEX, "wtxid", "The transaction witness hash in hex"},
+                                {RPCResult::Type::STR_HEX, "hex", "The serialized, hex-encoded transaction data"},
+                            }},
+                    }},
+            }
+        },
+        RPCExamples{
+            HelpExampleCli("abortprivatebroadcast", "\"id\"")
+            + HelpExampleRpc("abortprivatebroadcast", "\"id\"")
+        },
+        [&](const RPCHelpMan& self, const JSONRPCRequest& request) -> UniValue
+        {
+            const uint256 id{ParseHashV(self.Arg<UniValue>("id"), "id")};
+
+            const NodeContext& node{EnsureAnyNodeContext(request.context)};
+            PeerManager& peerman{EnsurePeerman(node)};
+
+            const auto removed_txs{peerman.AbortPrivateBroadcast(id)};
+            if (removed_txs.empty()) {
+                throw JSONRPCError(RPC_INVALID_ADDRESS_OR_KEY, "Transaction not in private broadcast queue. Check getprivatebroadcastinfo.");
+            }
+
+            UniValue removed_transactions(UniValue::VARR);
+            for (const auto& tx : removed_txs) {
+                UniValue o(UniValue::VOBJ);
+                o.pushKV("txid", tx->GetHash().ToString());
+                o.pushKV("wtxid", tx->GetWitnessHash().ToString());
+                o.pushKV("hex", EncodeHexTx(*tx));
+                removed_transactions.push_back(std::move(o));
+            }
+            UniValue ret(UniValue::VOBJ);
+            ret.pushKV("removed_transactions", std::move(removed_transactions));
+            return ret;
+        },
+    };
+}
+
 static RPCHelpMan testmempoolaccept()
 {
     return RPCHelpMan{
@@ -1329,6 +1449,8 @@ void RegisterMempoolRPCCommands(CRPCTable& t)
 {
     static const CRPCCommand commands[]{
         {"rawtransactions", &sendrawtransaction},
+        {"rawtransactions", &getprivatebroadcastinfo},
+        {"rawtransactions", &abortprivatebroadcast},
         {"rawtransactions", &testmempoolaccept},
         {"blockchain", &getmempoolancestors},
         {"blockchain", &getmempooldescendants},

src/test/fuzz/rpc.cpp

@@ -90,6 +90,7 @@ const std::vector<std::string> RPC_COMMANDS_NOT_SAFE_FOR_FUZZING{
 
 // RPC commands which are safe for fuzzing.
 const std::vector<std::string> RPC_COMMANDS_SAFE_FOR_FUZZING{
+    "abortprivatebroadcast",
     "analyzepsbt",
     "clearbanned",
     "combinepsbt",
@@ -147,6 +148,7 @@ const std::vector<std::string> RPC_COMMANDS_SAFE_FOR_FUZZING{
     "getorphantxs",
     "getpeerinfo",
     "getprioritisedtransactions",
+    "getprivatebroadcastinfo",
     "getrawaddrman",
     "getrawmempool",
     "getrawtransaction",

src/test/private_broadcast_tests.cpp

@@ -7,6 +7,7 @@
 #include <test/util/setup_common.h>
 #include <util/time.h>
 
+#include <algorithm>
 #include <boost/test/unit_test.hpp>
 
 BOOST_FIXTURE_TEST_SUITE(private_broadcast_tests, BasicTestingSetup)
@@ -29,11 +30,15 @@ BOOST_AUTO_TEST_CASE(basic)
 
     PrivateBroadcast pb;
     const NodeId recipient1{1};
+    in_addr ipv4Addr;
+    ipv4Addr.s_addr = 0xa0b0c001;
+    const CService addr1{ipv4Addr, 1111};
 
     // No transactions initially.
-    BOOST_CHECK(!pb.PickTxForSend(/*will_send_to_nodeid=*/recipient1).has_value());
+    BOOST_CHECK(!pb.PickTxForSend(/*will_send_to_nodeid=*/recipient1, /*will_send_to_address=*/addr1).has_value());
     BOOST_CHECK_EQUAL(pb.GetStale().size(), 0);
     BOOST_CHECK(!pb.HavePendingTransactions());
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), 0);
 
     // Make a transaction and add it.
     const auto tx1{MakeDummyTx(/*id=*/1, /*num_witness=*/0)};
@@ -47,16 +52,32 @@ BOOST_AUTO_TEST_CASE(basic)
     BOOST_REQUIRE(tx1->GetWitnessHash() != tx2->GetWitnessHash());
 
     BOOST_CHECK(pb.Add(tx2));
+    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());
+        return *it;
+    }};
+    const auto check_peer_counts{[&](size_t tx1_peer_count, size_t tx2_peer_count) {
+        const auto infos{pb.GetBroadcastInfo()};
+        BOOST_CHECK_EQUAL(infos.size(), 2);
+        BOOST_CHECK_EQUAL(find_tx_info(infos, tx1).peers.size(), tx1_peer_count);
+        BOOST_CHECK_EQUAL(find_tx_info(infos, tx2).peers.size(), tx2_peer_count);
+    }};
 
-    const auto tx_for_recipient1{pb.PickTxForSend(/*will_send_to_nodeid=*/recipient1).value()};
+    check_peer_counts(/*tx1_peer_count=*/0, /*tx2_peer_count=*/0);
+
+    const auto tx_for_recipient1{pb.PickTxForSend(/*will_send_to_nodeid=*/recipient1, /*will_send_to_address=*/addr1).value()};
     BOOST_CHECK(tx_for_recipient1 == tx1 || tx_for_recipient1 == tx2);
 
     // A second pick must return the other transaction.
     const NodeId recipient2{2};
-    const auto tx_for_recipient2{pb.PickTxForSend(/*will_send_to_nodeid=*/recipient2).value()};
+    const CService addr2{ipv4Addr, 2222};
+    const auto tx_for_recipient2{pb.PickTxForSend(/*will_send_to_nodeid=*/recipient2, /*will_send_to_address=*/addr2).value()};
     BOOST_CHECK(tx_for_recipient2 == tx1 || tx_for_recipient2 == tx2);
     BOOST_CHECK_NE(tx_for_recipient1, tx_for_recipient2);
 
+    check_peer_counts(/*tx1_peer_count=*/1, /*tx2_peer_count=*/1);
+
     const NodeId nonexistent_recipient{0};
 
     // Confirm transactions <-> recipients mapping is correct.
@@ -78,6 +99,21 @@ BOOST_AUTO_TEST_CASE(basic)
     BOOST_CHECK(pb.DidNodeConfirmReception(recipient1));
     BOOST_CHECK(!pb.DidNodeConfirmReception(recipient2));
 
+    const auto infos{pb.GetBroadcastInfo()};
+    BOOST_CHECK_EQUAL(infos.size(), 2);
+    {
+        const auto& [tx, peers]{find_tx_info(infos, tx_for_recipient1)};
+        BOOST_CHECK_EQUAL(peers.size(), 1);
+        BOOST_CHECK_EQUAL(peers[0].address.ToStringAddrPort(), addr1.ToStringAddrPort());
+        BOOST_CHECK(peers[0].received.has_value());
+    }
+    {
+        const auto& [tx, peers]{find_tx_info(infos, tx_for_recipient2)};
+        BOOST_CHECK_EQUAL(peers.size(), 1);
+        BOOST_CHECK_EQUAL(peers[0].address.ToStringAddrPort(), addr2.ToStringAddrPort());
+        BOOST_CHECK(!peers[0].received.has_value());
+    }
+
     BOOST_CHECK_EQUAL(pb.GetStale().size(), 1);
     BOOST_CHECK_EQUAL(pb.GetStale()[0], tx_for_recipient2);
 
@@ -90,7 +126,9 @@ BOOST_AUTO_TEST_CASE(basic)
     BOOST_CHECK_EQUAL(pb.Remove(tx_for_recipient2).value(), 0);
     BOOST_CHECK(!pb.Remove(tx_for_recipient2).has_value());
 
-    BOOST_CHECK(!pb.PickTxForSend(/*will_send_to_nodeid=*/nonexistent_recipient).has_value());
+    BOOST_CHECK_EQUAL(pb.GetBroadcastInfo().size(), 0);
+    const CService addr_nonexistent{ipv4Addr, 3333};
+    BOOST_CHECK(!pb.PickTxForSend(/*will_send_to_nodeid=*/nonexistent_recipient, /*will_send_to_address=*/addr_nonexistent).has_value());
 }
 
 BOOST_AUTO_TEST_SUITE_END()

test/functional/p2p_private_broadcast.py

@@ -36,6 +36,7 @@ from test_framework.test_framework import (
 )
 from test_framework.util import (
     assert_equal,
+    assert_greater_than_or_equal,
     assert_not_equal,
     assert_raises_rpc_error,
     p2p_port,
@@ -301,6 +302,16 @@ class P2PPrivateBroadcast(BitcoinTestFramework):
             self.log.info(f"{label}: ok: outbound connection i={i} is private broadcast of txid={tx['txid']}")
             broadcasts_done += 1
 
+        # Verify the tx we just observed is tracked in getprivatebroadcastinfo.
+        pbinfo = self.nodes[0].getprivatebroadcastinfo()
+        pending = [t for t in pbinfo["transactions"] if t["txid"] == tx["txid"] and t["wtxid"] == tx["wtxid"]]
+        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 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), NUM_PRIVATE_BROADCAST_PER_TX)
+
     def run_test(self):
         tx_originator = self.nodes[0]
         tx_receiver = self.nodes[1]
@@ -366,6 +377,30 @@ class P2PPrivateBroadcast(BitcoinTestFramework):
         self.log.info("Waiting for normal broadcast to another peer")
         self.destinations[1]["node"].wait_for_inv([inv])
 
+        self.log.info("Checking getprivatebroadcastinfo no longer reports the transaction after it is received back")
+        pbinfo = tx_originator.getprivatebroadcastinfo()
+        pending = [t for t in pbinfo["transactions"] if t["txid"] == txs[0]["txid"] and t["wtxid"] == txs[0]["wtxid"]]
+        assert_equal(len(pending), 0)
+
+        self.log.info("Checking abortprivatebroadcast removes a pending private-broadcast transaction")
+        tx_abort = wallet.create_self_transfer()
+        tx_originator.sendrawtransaction(hexstring=tx_abort["hex"], maxfeerate=0.1)
+        assert any(t["wtxid"] == tx_abort["wtxid"] for t in tx_originator.getprivatebroadcastinfo()["transactions"])
+        abort_res = tx_originator.abortprivatebroadcast(tx_abort["txid"])
+        assert_equal(len(abort_res["removed_transactions"]), 1)
+        assert_equal(abort_res["removed_transactions"][0]["txid"], tx_abort["txid"])
+        assert_equal(abort_res["removed_transactions"][0]["wtxid"], tx_abort["wtxid"])
+        assert_equal(abort_res["removed_transactions"][0]["hex"].lower(), tx_abort["hex"].lower())
+        assert all(t["wtxid"] != tx_abort["wtxid"] for t in tx_originator.getprivatebroadcastinfo()["transactions"])
+
+        self.log.info("Checking abortprivatebroadcast fails for non-existent transaction")
+        assert_raises_rpc_error(
+            -5,
+            "Transaction not in private broadcast queue",
+            tx_originator.abortprivatebroadcast,
+            "0" * 64,
+        )
+
         self.log.info("Sending a transaction that is already in the mempool")
         skip_destinations = len(self.destinations)
         tx_originator.sendrawtransaction(hexstring=txs[0]["hex"], maxfeerate=0)