net_processing: handle ConnectionType::PRIVATE_BROADCAST connections

diff --git a/src/net.cpp b/src/net.cpp
index c0c054f2..9ee31c46 100644
--- a/src/net.cpp
+++ b/src/net.cpp
@@ -354,7 +354,16 @@ bool CConnman::CheckIncomingNonce(uint64_t nonce)
 {
     LOCK(m_nodes_mutex);
     for (const CNode* pnode : m_nodes) {
-        if (!pnode->fSuccessfullyConnected && !pnode->IsInboundConn() && pnode->GetLocalNonce() == nonce)
+        // Omit private broadcast connections from this check to prevent this privacy attack:
+        // - We connect to a peer in an attempt to privately broadcast a transaction. From our
+        //   VERSION message the peer deducts that this is a short-lived connection for
+        //   broadcasting a transaction, takes our nonce and delays their VERACK.
+        // - The peer starts connecting to (clearnet) nodes and sends them a VERSION message
+        //   which contains our nonce. If the peer manages to connect to us we would disconnect.
+        // - Upon a disconnect, the peer knows our clearnet address. They go back to the short
+        //   lived privacy broadcast connection and continue with VERACK.
+        if (!pnode->fSuccessfullyConnected && !pnode->IsInboundConn() && !pnode->IsPrivateBroadcastConn() &&
+            pnode->GetLocalNonce() == nonce)
             return false;
     }
     return true;
@@ -4048,10 +4057,26 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
     return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
 }
 
+/// Private broadcast connections only need to send certain message types.
+/// Other messages are not needed and may degrade privacy.
+static bool IsOutboundMessageAllowedInPrivateBroadcast(std::string_view type) noexcept
+{
+    return type == NetMsgType::VERSION ||
+           type == NetMsgType::VERACK ||
+           type == NetMsgType::INV ||
+           type == NetMsgType::TX ||
+           type == NetMsgType::PING;
+}
+
 void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 {
     AssertLockNotHeld(m_total_bytes_sent_mutex);
 
+    if (pnode->IsPrivateBroadcastConn() && !IsOutboundMessageAllowedInPrivateBroadcast(msg.m_type)) {
+        LogDebug(BCLog::PRIVBROADCAST, "Omitting send of message '%s', peer=%d%s", msg.m_type, pnode->GetId(), pnode->LogIP(fLogIPs));
+        return;
+    }
+
     if (!m_private_broadcast.m_outbound_tor_ok_at_least_once.load() && !pnode->IsInboundConn() &&
         pnode->addr.IsTor() && msg.m_type == NetMsgType::VERACK) {
         // If we are sending the peer VERACK that means we successfully sent
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
index 2d16f2bd..abfcb673 100644
--- a/src/net_processing.cpp
+++ b/src/net_processing.cpp
@@ -199,6 +199,8 @@ static constexpr size_t MAX_ADDR_PROCESSING_TOKEN_BUCKET{MAX_ADDR_TO_SEND};
 static constexpr uint64_t CMPCTBLOCKS_VERSION{2};
 /** For private broadcast, send a transaction to this many peers. */
 static constexpr size_t NUM_PRIVATE_BROADCAST_PER_TX{3};
+/** Private broadcast connections must complete within this time. Disconnect the peer if it takes longer. */
+static constexpr auto PRIVATE_BROADCAST_MAX_CONNECTION_LIFETIME{3min};
 
 // Internal stuff
 namespace {
@@ -723,8 +725,8 @@ private:
     /** Send a version message to a peer */
     void PushNodeVersion(CNode& pnode, const Peer& peer);
 
-    /** Send a ping message every PING_INTERVAL or if requested via RPC. May
-     *  mark the peer to be disconnected if a ping has timed out.
+    /** Send a ping message every PING_INTERVAL or if requested via RPC (peer.m_ping_queued is true).
+     *  May mark the peer to be disconnected if a ping has timed out.
      *  We use mockable time for ping timeouts, so setmocktime may cause pings
      *  to time out. */
     void MaybeSendPing(CNode& node_to, Peer& peer, std::chrono::microseconds now);
@@ -962,6 +964,14 @@ private:
     void ProcessCompactBlockTxns(CNode& pfrom, Peer& peer, const BlockTransactions& block_transactions)
         EXCLUSIVE_LOCKS_REQUIRED(g_msgproc_mutex, !m_most_recent_block_mutex);
 
+    /**
+     * Schedule an INV for a transaction to be sent to the given peer (via `PushMessage()`).
+     * The transaction is picked from the list of transactions for private broadcast.
+     * It is assumed that the connection to the peer is `ConnectionType::PRIVATE_BROADCAST`.
+     * Avoid calling this for other peers since it will degrade privacy.
+     */
+    void PushPrivateBroadcastTx(CNode& node) EXCLUSIVE_LOCKS_REQUIRED(g_msgproc_mutex, !m_most_recent_block_mutex);
+
     /**
      * When a peer sends us a valid block, instruct it to announce blocks to us
      * using CMPCTBLOCK if possible by adding its nodeid to the end of
@@ -1535,15 +1545,24 @@ void PeerManagerImpl::PushNodeVersion(CNode& pnode, const Peer& peer)
     std::string my_user_agent;
     int my_height;
     bool my_tx_relay;
-
-    const CAddress& addr{pnode.addr};
-    my_services = peer.m_our_services;
-    my_time = count_seconds(GetTime<std::chrono::seconds>());
-    your_services = addr.nServices;
-    your_addr = addr.IsRoutable() && !IsProxy(addr) && addr.IsAddrV1Compatible() ? CService{addr} : CService{};
-    my_user_agent = strSubVersion;
-    my_height = m_best_height;
-    my_tx_relay = !RejectIncomingTxs(pnode);
+    if (pnode.IsPrivateBroadcastConn()) {
+        my_services = NODE_NONE;
+        my_time = 0;
+        your_services = NODE_NONE;
+        your_addr = CService{};
+        my_user_agent = "/pynode:0.0.1/"; // Use a constant other than the default (or user-configured). See https://github.com/bitcoin/bitcoin/pull/27509#discussion_r1214671917
+        my_height = 0;
+        my_tx_relay = false;
+    } else {
+        const CAddress& addr{pnode.addr};
+        my_services = peer.m_our_services;
+        my_time = count_seconds(GetTime<std::chrono::seconds>());
+        your_services = addr.nServices;
+        your_addr = addr.IsRoutable() && !IsProxy(addr) && addr.IsAddrV1Compatible() ? CService{addr} : CService{};
+        my_user_agent = strSubVersion;
+        my_height = m_best_height;
+        my_tx_relay = !RejectIncomingTxs(pnode);
+    }
 
     MakeAndPushMessage(
         pnode,
@@ -1671,16 +1690,23 @@ void PeerManagerImpl::FinalizeNode(const CNode& node)
     }
     } // cs_main
     if (node.fSuccessfullyConnected &&
-        !node.IsBlockOnlyConn() && !node.IsInboundConn()) {
+        !node.IsBlockOnlyConn() && !node.IsPrivateBroadcastConn() && !node.IsInboundConn()) {
         // Only change visible addrman state for full outbound peers.  We don't
         // call Connected() for feeler connections since they don't have
-        // fSuccessfullyConnected set.
+        // fSuccessfullyConnected set. Also don't call Connected() for private broadcast
+        // connections since they could leak information in addrman.
         m_addrman.Connected(node.addr);
     }
     {
         LOCK(m_headers_presync_mutex);
         m_headers_presync_stats.erase(nodeid);
     }
+    if (node.IsPrivateBroadcastConn() &&
+        !m_tx_for_private_broadcast.DidNodeConfirmReception(nodeid) &&
+        m_tx_for_private_broadcast.HavePendingTransactions()) {
+
+        m_connman.m_private_broadcast.NumToOpenAdd(1);
+    }
     LogDebug(BCLog::NET, "Cleared nodestate for peer=%d\n", nodeid);
 }
 
@@ -3459,6 +3485,25 @@ void PeerManagerImpl::LogBlockHeader(const CBlockIndex& index, const CNode& peer
     }
 }
 
+void PeerManagerImpl::PushPrivateBroadcastTx(CNode& node)
+{
+    Assume(node.IsPrivateBroadcastConn());
+
+    const auto opt_tx{m_tx_for_private_broadcast.PickTxForSend(node.GetId())};
+    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;
+        return;
+    }
+    const CTransactionRef& tx{*opt_tx};
+
+    LogInfo("[privatebroadcast] P2P handshake completed, sending INV for txid=%s%s, peer=%d%s",
+            tx->GetHash().ToString(), tx->HasWitness() ? strprintf(", wtxid=%s", tx->GetWitnessHash().ToString()) : "",
+            node.GetId(), node.LogIP(fLogIPs));
+
+    MakeAndPushMessage(node, NetMsgType::INV, std::vector<CInv>{{CInv{MSG_TX, tx->GetHash().ToUint256()}}});
+}
+
 void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type, DataStream& vRecv,
                                      const std::chrono::microseconds time_received,
                                      const std::atomic<bool>& interruptMsgProc)
@@ -3589,6 +3634,17 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
                   peer->m_starting_height, addrMe.ToStringAddrPort(), fRelay, pfrom.GetId(),
                   pfrom.LogIP(fLogIPs), (mapped_as ? strprintf(", mapped_as=%d", mapped_as) : ""));
 
+        if (pfrom.IsPrivateBroadcastConn()) {
+            if (fRelay) {
+                MakeAndPushMessage(pfrom, NetMsgType::VERACK);
+            } else {
+                LogInfo("[privatebroadcast] Disconnecting: does not support transactions relay (connected in vain), peer=%d%s",
+                        pfrom.GetId(), pfrom.LogIP(fLogIPs));
+                pfrom.fDisconnect = true;
+            }
+            return;
+        }
+
         if (greatest_common_version >= WTXID_RELAY_VERSION) {
             MakeAndPushMessage(pfrom, NetMsgType::WTXIDRELAY);
         }
@@ -3733,6 +3789,17 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
                        tx_relay->m_next_inv_send_time == 0s));
         }
 
+        if (pfrom.IsPrivateBroadcastConn()) {
+            pfrom.fSuccessfullyConnected = true;
+            // The peer may intend to later send us NetMsgType::FEEFILTER limiting
+            // cheap transactions, but we don't wait for that and thus we may send
+            // them a transaction below their threshold. This is ok because this
+            // relay logic is designed to work even in cases when the peer drops
+            // the transaction (due to it being too cheap, or for other reasons).
+            PushPrivateBroadcastTx(pfrom);
+            return;
+        }
+
         if (pfrom.GetCommonVersion() >= SHORT_IDS_BLOCKS_VERSION) {
             // Tell our peer we are willing to provide version 2 cmpctblocks.
             // However, we do not request new block announcements using
@@ -3884,6 +3951,13 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
         return;
     }
 
+    if (pfrom.IsPrivateBroadcastConn()) {
+        if (msg_type != NetMsgType::PONG && msg_type != NetMsgType::GETDATA) {
+            LogDebug(BCLog::PRIVBROADCAST, "Ignoring incoming message '%s', peer=%d%s", msg_type, pfrom.GetId(), pfrom.LogIP(fLogIPs));
+            return;
+        }
+    }
+
     if (msg_type == NetMsgType::ADDR || msg_type == NetMsgType::ADDRV2) {
         const auto ser_params{
             msg_type == NetMsgType::ADDRV2 ?
@@ -4087,6 +4161,33 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
             LogDebug(BCLog::NET, "received getdata for: %s peer=%d\n", vInv[0].ToString(), pfrom.GetId());
         }
 
+        if (pfrom.IsPrivateBroadcastConn()) {
+            const auto pushed_tx_opt{m_tx_for_private_broadcast.GetTxForNode(pfrom.GetId())};
+            if (!pushed_tx_opt) {
+                LogInfo("[privatebroadcast] Disconnecting: got GETDATA without sending an INV, peer=%d%s",
+                        pfrom.GetId(), fLogIPs ? strprintf(", peeraddr=%s", pfrom.addr.ToStringAddrPort()) : "");
+                pfrom.fDisconnect = true;
+                return;
+            }
+
+            const CTransactionRef& pushed_tx{*pushed_tx_opt};
+
+            // The GETDATA request must contain exactly one inv and it must be for the transaction
+            // that we INVed to the peer earlier.
+            if (vInv.size() == 1 && vInv[0].IsMsgTx() && vInv[0].hash == pushed_tx->GetHash().ToUint256()) {
+
+                MakeAndPushMessage(pfrom, NetMsgType::TX, TX_WITH_WITNESS(*pushed_tx));
+
+                peer->m_ping_queued = true; // Ensure a ping will be sent: mimic a request via RPC.
+                MaybeSendPing(pfrom, *peer, GetTime<std::chrono::microseconds>());
+            } else {
+                LogInfo("[privatebroadcast] Disconnecting: got an unexpected GETDATA message, peer=%d%s",
+                        pfrom.GetId(), fLogIPs ? strprintf(", peeraddr=%s", pfrom.addr.ToStringAddrPort()) : "");
+                pfrom.fDisconnect = true;
+            }
+            return;
+        }
+
         {
             LOCK(peer->m_getdata_requests_mutex);
             peer->m_getdata_requests.insert(peer->m_getdata_requests.end(), vInv.begin(), vInv.end());
@@ -4828,6 +4929,12 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
                     if (ping_time.count() >= 0) {
                         // Let connman know about this successful ping-pong
                         pfrom.PongReceived(ping_time);
+                        if (pfrom.IsPrivateBroadcastConn()) {
+                            m_tx_for_private_broadcast.NodeConfirmedReception(pfrom.GetId());
+                            LogInfo("[privatebroadcast] Got a PONG (the transaction will probably reach the network), marking for disconnect, peer=%d%s",
+                                    pfrom.GetId(), pfrom.LogIP(fLogIPs));
+                            pfrom.fDisconnect = true;
+                        }
                     } else {
                         // This should never happen
                         sProblem = "Timing mishap";
@@ -5535,6 +5642,18 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
 
     const auto current_time{GetTime<std::chrono::microseconds>()};
 
+    // The logic below does not apply to private broadcast peers, so skip it.
+    // Also in CConnman::PushMessage() we make sure that unwanted messages are
+    // not sent. This here is just an optimization.
+    if (pto->IsPrivateBroadcastConn()) {
+        if (pto->m_connected + PRIVATE_BROADCAST_MAX_CONNECTION_LIFETIME < current_time) {
+            LogInfo("[privatebroadcast] Disconnecting: did not complete the transaction send within %d seconds, peer=%d%s",
+                    count_seconds(PRIVATE_BROADCAST_MAX_CONNECTION_LIFETIME), pto->GetId(), pto->LogIP(fLogIPs));
+            pto->fDisconnect = true;
+        }
+        return true;
+    }
+
     if (pto->IsAddrFetchConn() && current_time - pto->m_connected > 10 * AVG_ADDRESS_BROADCAST_INTERVAL) {
         LogDebug(BCLog::NET, "addrfetch connection timeout, %s\n", pto->DisconnectMsg(fLogIPs));
         pto->fDisconnect = true;
diff --git a/src/private_broadcast.cpp b/src/private_broadcast.cpp
index c14cea84..6d312ac9 100644
--- a/src/private_broadcast.cpp
+++ b/src/private_broadcast.cpp
@@ -3,6 +3,9 @@
 // file COPYING or https://opensource.org/license/mit/.
 
 #include <private_broadcast.h>
+#include <util/check.h>
+
+#include <algorithm>
 
 bool PrivateBroadcast::Add(const CTransactionRef& tx)
     EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
@@ -11,3 +14,89 @@ bool PrivateBroadcast::Add(const CTransactionRef& tx)
     const bool inserted{m_transactions.try_emplace(tx).second};
     return inserted;
 }
+
+std::optional<CTransactionRef> PrivateBroadcast::PickTxForSend(const NodeId& will_send_to_nodeid)
+    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+{
+    LOCK(m_mutex);
+
+    const auto it{std::ranges::max_element(
+            m_transactions,
+            [](const auto& a, const auto& b) { return a < b; },
+            [](const auto& el) { return DerivePriority(el.second); })};
+
+    if (it != m_transactions.end()) {
+        auto& [tx, sent_to]{*it};
+        sent_to.emplace_back(will_send_to_nodeid, NodeClock::now());
+        return tx;
+    }
+
+    return std::nullopt;
+}
+
+std::optional<CTransactionRef> PrivateBroadcast::GetTxForNode(const NodeId& nodeid)
+    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+{
+    LOCK(m_mutex);
+    const auto tx_and_status{GetSendStatusByNode(nodeid)};
+    if (tx_and_status.has_value()) {
+        return tx_and_status.value().tx;
+    }
+    return std::nullopt;
+}
+
+void PrivateBroadcast::NodeConfirmedReception(const NodeId& nodeid)
+    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+{
+    LOCK(m_mutex);
+    const auto tx_and_status{GetSendStatusByNode(nodeid)};
+    if (tx_and_status.has_value()) {
+        tx_and_status.value().send_status.confirmed = NodeClock::now();
+    }
+}
+
+bool PrivateBroadcast::DidNodeConfirmReception(const NodeId& nodeid)
+    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+{
+    LOCK(m_mutex);
+    const auto tx_and_status{GetSendStatusByNode(nodeid)};
+    if (tx_and_status.has_value()) {
+        return tx_and_status.value().send_status.confirmed.has_value();
+    }
+    return false;
+}
+
+bool PrivateBroadcast::HavePendingTransactions()
+    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+{
+    LOCK(m_mutex);
+    return !m_transactions.empty();
+}
+
+PrivateBroadcast::Priority PrivateBroadcast::DerivePriority(const std::vector<SendStatus>& sent_to)
+{
+    Priority p;
+    p.num_picked = sent_to.size();
+    for (const auto& send_status : sent_to) {
+        p.last_picked = std::max(p.last_picked, send_status.picked);
+        if (send_status.confirmed.has_value()) {
+            ++p.num_confirmed;
+            p.last_confirmed = std::max(p.last_confirmed, send_status.confirmed.value());
+        }
+    }
+    return p;
+}
+
+std::optional<PrivateBroadcast::TxAndSendStatusForNode> PrivateBroadcast::GetSendStatusByNode(const NodeId& nodeid)
+    EXCLUSIVE_LOCKS_REQUIRED(m_mutex)
+{
+    AssertLockHeld(m_mutex);
+    for (auto& [tx, sent_to] : m_transactions) {
+        for (auto& send_status : sent_to) {
+            if (send_status.nodeid == nodeid) {
+                return TxAndSendStatusForNode{.tx = tx, .send_status = send_status};
+            }
+        }
+    }
+    return std::nullopt;
+}
diff --git a/src/private_broadcast.h b/src/private_broadcast.h
index 8e0b5ffb..5b8634a0 100644
--- a/src/private_broadcast.h
+++ b/src/private_broadcast.h
@@ -7,17 +7,24 @@
 
 #include <net.h>
 #include <primitives/transaction.h>
+#include <primitives/transaction_identifier.h>
 #include <sync.h>
 #include <threadsafety.h>
 #include <util/time.h>
 
 #include <optional>
+#include <tuple>
 #include <unordered_map>
 #include <vector>
 
 /**
  * Store a list of transactions to be broadcast privately. Supports the following operations:
  * - Add a new transaction
+ * - Pick a transaction for sending to one recipient
+ * - Query which transaction has been picked for sending to a given recipient node
+ * - Mark that a given recipient node has confirmed receipt of a transaction
+ * - Query whether a given recipient node has confirmed reception
+ * - Query whether any transactions that need sending are currently on the list
  */
 class PrivateBroadcast
 {
@@ -31,6 +38,46 @@ public:
     bool Add(const CTransactionRef& tx)
         EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
 
+    /**
+     * Pick the transaction with the fewest send attempts, and confirmations,
+     * and oldest send/confirm times.
+     * @param[in] will_send_to_nodeid Will remember that the returned transaction
+     * was picked for sending to this node.
+     * @return Most urgent transaction or nullopt if there are no transactions.
+     */
+    std::optional<CTransactionRef> PickTxForSend(const NodeId& will_send_to_nodeid)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
+    /**
+     * Get the transaction that was picked for sending to a given node by PickTxForSend().
+     * @param[in] nodeid Node to which a transaction is being (or was) sent.
+     * @return Transaction or nullopt if the nodeid is unknown.
+     */
+    std::optional<CTransactionRef> GetTxForNode(const NodeId& nodeid)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
+    /**
+     * Mark that the node has confirmed reception of the transaction we sent it by
+     * responding with `PONG` to our `PING` message.
+     * @param[in] nodeid Node that we sent a transaction to.
+     */
+    void NodeConfirmedReception(const NodeId& nodeid)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
+    /**
+     * Check if the node has confirmed reception of the transaction.
+     * @retval true Node has confirmed, `NodeConfirmedReception()` has been called.
+     * @retval false Node has not confirmed, `NodeConfirmedReception()` has not been called.
+     */
+    bool DidNodeConfirmReception(const NodeId& nodeid)
+        EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
+    /**
+     * Check if there are transactions that need to be broadcast.
+     */
+    bool HavePendingTransactions()
+        EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+
 private:
     /// Status of a transaction sent to a given node.
     struct SendStatus {
@@ -41,6 +88,29 @@ private:
         SendStatus(const NodeId& nodeid, const NodeClock::time_point& picked) : nodeid{nodeid}, picked{picked} {}
     };
 
+    /// Cumulative stats from all the send attempts for a transaction. Used to prioritize transactions.
+    struct Priority {
+        size_t num_picked{0}; ///< Number of times the transaction was picked for sending.
+        NodeClock::time_point last_picked{}; ///< The most recent time when the transaction was picked for sending.
+        size_t num_confirmed{0}; ///< Number of nodes that have confirmed reception of a transaction (by PONG).
+        NodeClock::time_point last_confirmed{}; ///< The most recent time when the transaction was confirmed.
+
+        auto operator<=>(const Priority& other) const
+        {
+            // Invert `other` and `this` in the comparison because smaller num_picked, num_confirmed or
+            // earlier times mean greater priority. In other words, if this.num_picked < other.num_picked
+            // then this > other.
+            return std::tie(other.num_picked, other.num_confirmed, other.last_picked, other.last_confirmed) <=>
+                   std::tie(num_picked, num_confirmed, last_picked, last_confirmed);
+        }
+    };
+
+    /// A pair of a transaction and a sent status for a given node. Convenience return type of GetSendStatusByNode().
+    struct TxAndSendStatusForNode {
+        const CTransactionRef& tx;
+        SendStatus& send_status;
+    };
+
     // No need for salted hasher because we are going to store just a bunch of locally originating transactions.
 
     struct CTransactionRefHash {
@@ -57,6 +127,20 @@ private:
         }
     };
 
+    /**
+     * Derive the sending priority of a transaction.
+     * @param[in] sent_to List of nodes that the transaction has been sent to.
+     */
+    static Priority DerivePriority(const std::vector<SendStatus>& sent_to);
+
+    /**
+     * Find which transaction we sent to a given node (marked by PickTxForSend()).
+     * @return That transaction together with the send status or nullopt if we did not
+     * send any transaction to the given node.
+     */
+    std::optional<TxAndSendStatusForNode> GetSendStatusByNode(const NodeId& nodeid)
+        EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
     mutable Mutex m_mutex;
     std::unordered_map<CTransactionRef, std::vector<SendStatus>, CTransactionRefHash, CTransactionRefComp>
         m_transactions GUARDED_BY(m_mutex);