net_processing: store transactions for private broadcast in PeerManager

Extend `PeerManager` with a transaction storage and a new method
`InitiateTxBroadcastPrivate()` which:
* adds a transaction to that storage and
* calls `CConnman::PrivateBroadcast::NumToOpenAdd()` to open dedicated
  privacy connections that will pick an entry from the transaction
  storage and broadcast it.

src/CMakeLists.txt

@@ -244,6 +244,7 @@ add_library(bitcoin_node STATIC EXCLUDE_FROM_ALL
   policy/rbf.cpp
   policy/settings.cpp
   policy/truc_policy.cpp
+  private_broadcast.cpp
   rest.cpp
   rpc/blockchain.cpp
   rpc/external_signer.cpp

src/net_processing.cpp

@@ -44,6 +44,7 @@
 #include <policy/policy.h>
 #include <primitives/block.h>
 #include <primitives/transaction.h>
+#include <private_broadcast.h>
 #include <protocol.h>
 #include <random.h>
 #include <scheduler.h>
@@ -196,6 +197,8 @@ static constexpr double MAX_ADDR_RATE_PER_SECOND{0.1};
 static constexpr size_t MAX_ADDR_PROCESSING_TOKEN_BUCKET{MAX_ADDR_TO_SEND};
 /** The compactblocks version we support. See BIP 152. */
 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};
 
 // Internal stuff
 namespace {
@@ -538,6 +541,7 @@ public:
     PeerManagerInfo GetInfo() const 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);
     void SetBestBlock(int height, std::chrono::seconds time) override
     {
         m_best_height = height;
@@ -1070,6 +1074,9 @@ private:
     void PushAddress(Peer& peer, const CAddress& addr) EXCLUSIVE_LOCKS_REQUIRED(g_msgproc_mutex);
 
     void LogBlockHeader(const CBlockIndex& index, const CNode& peer, bool via_compact_block);
+
+    /// The transactions to be broadcast privately.
+    PrivateBroadcast m_tx_for_private_broadcast;
 };
 
 const CNodeState* PeerManagerImpl::State(NodeId pnode) const
@@ -2147,6 +2154,17 @@ void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wt
     }
 }
 
+void 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)) {
+        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 {
+        LogDebug(BCLog::PRIVBROADCAST, "Ignoring unnecessary request to schedule an already scheduled transaction: %s", txstr);
+    }
+}
+
 void PeerManagerImpl::RelayAddress(NodeId originator,
                                    const CAddress& addr,
                                    bool fReachable)

src/net_processing.h

@@ -124,6 +124,12 @@ public:
      */
     virtual void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) = 0;
 
+    /**
+     * Initiate a private transaction broadcast. This is done
+     * asynchronously via short-lived connections to peers on privacy networks.
+     */
+    virtual void InitiateTxBroadcastPrivate(const CTransactionRef& tx) = 0;
+
     /** Send ping message to all peers */
     virtual void SendPings() = 0;
 

src/node/transaction.cpp

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

src/private_broadcast.cpp

@@ -0,0 +1,13 @@
+// Copyright (c) 2023-present The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or https://opensource.org/license/mit/.
+
+#include <private_broadcast.h>
+
+bool PrivateBroadcast::Add(const CTransactionRef& tx)
+    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+{
+    LOCK(m_mutex);
+    const bool inserted{m_transactions.try_emplace(tx).second};
+    return inserted;
+}

src/private_broadcast.h

@@ -0,0 +1,65 @@
+// Copyright (c) 2023-present The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or https://opensource.org/license/mit/.
+
+#ifndef BITCOIN_PRIVATE_BROADCAST_H
+#define BITCOIN_PRIVATE_BROADCAST_H
+
+#include <net.h>
+#include <primitives/transaction.h>
+#include <sync.h>
+#include <threadsafety.h>
+#include <util/time.h>
+
+#include <optional>
+#include <unordered_map>
+#include <vector>
+
+/**
+ * Store a list of transactions to be broadcast privately. Supports the following operations:
+ * - Add a new transaction
+ */
+class PrivateBroadcast
+{
+public:
+    /**
+     * 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.
+     */
+    bool Add(const CTransactionRef& tx)
+        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 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} {}
+    };
+
+    // No need for salted hasher because we are going to store just a bunch of locally originating transactions.
+
+    struct CTransactionRefHash {
+        size_t operator()(const CTransactionRef& tx) const
+        {
+            return static_cast<size_t>(tx->GetWitnessHash().ToUint256().GetUint64(0));
+        }
+    };
+
+    struct CTransactionRefComp {
+        bool operator()(const CTransactionRef& a, const CTransactionRef& b) const
+        {
+            return a->GetWitnessHash() == b->GetWitnessHash(); // If wtxid equals, then txid also equals.
+        }
+    };
+
+    mutable Mutex m_mutex;
+    std::unordered_map<CTransactionRef, std::vector<SendStatus>, CTransactionRefHash, CTransactionRefComp>
+        m_transactions GUARDED_BY(m_mutex);
+};
+
+#endif // BITCOIN_PRIVATE_BROADCAST_H