[mempool] Track "unbroadcast" transactions

Summary:
- Mempool tracks locally submitted transactions (wallet or rpc)
- Transactions are removed from set when the node receives a GETDATA request
  from a peer, or if the transaction is removed from the mempool.

PR description:
> This PR introduces mempool tracking of unbroadcast transactions and periodic reattempts at initial broadcast. This is a part of the rebroadcast project, and a standalone privacy win.
>
> The current rebroadcast logic is terrible for privacy because 1. only the source wallet rebroadcasts transactions and 2. it does so quite frequently. In the current system, if a user submits a transaction that does not immediately get broadcast to the network (eg. they are offline), this "rebroadcast" behavior is the safety net that can actually serve as the initial broadcast. So, keeping the attempts frequent is important for initial delivery within a reasonable timespan.
>
> This PR aims to improve # 2 by reducing the wallet rebroadcast frequency to ~1/day from ~1/15 min. It achieves this by separating the notion of initial broadcast from rebroadcasts. With these changes, the mempool tracks locally submitted transactions & periodically reattempts initial broadcast. Transactions submitted via the wallet or RPC are added to an "unbroadcast" set & are removed when a peer sends a getdata request, or the transaction is removed from the mempool. Every 10-15 minutes, the node reattempts an initial broadcast. This enables reducing the wallet rebroadcast frequency while ensuring the transactions will be propagated to the network.
>
> For privacy improvements around # 1, please see [[bitcoin/bitcoin#16698 | PR16698]].

This is a backport of Core [[bitcoin/bitcoin#18038 | PR18038]] [1/7]
bitcoin/bitcoin@89eeb4a

Test Plan: ninja all check-all

Reviewers: #bitcoin_abc, majcosta

Reviewed By: #bitcoin_abc, majcosta

Subscribers: majcosta

Differential Revision: https://reviews.bitcoinabc.org/D9006

src/net_processing.cpp

@@ -1974,7 +1974,7 @@ static void ProcessGetBlockData(const Config &config, CNode &pfrom,
 }
 
 static void ProcessGetData(const Config &config, CNode &pfrom,
-                           CConnman &connman, const CTxMemPool &mempool,
+                           CConnman &connman, CTxMemPool &mempool,
                            const std::atomic<bool> &interruptMsgProc)
     LOCKS_EXCLUDED(cs_main) {
     AssertLockNotHeld(cs_main);
@@ -2032,7 +2032,13 @@ static void ProcessGetData(const Config &config, CNode &pfrom,
                     push = true;
                 }
             }
-            if (!push) {
+
+            if (push) {
+                // We interpret fulfilling a GETDATA for a transaction as a
+                // successful initial broadcast and remove it from our
+                // unbroadcast set.
+                mempool.RemoveUnbroadcastTx(TxId(inv.hash));
+            } else {
                 vNotFound.push_back(inv);
             }
         }

src/node/transaction.cpp

@@ -88,6 +88,10 @@ TransactionError BroadcastTransaction(NodeContext &node, const Config &config,
     }
 
     if (relay) {
+        // the mempool tracks locally submitted transactions to make a
+        // best-effort of initial broadcast
+        node.mempool->AddUnbroadcastTx(txid);
+
         RelayTransaction(txid, *node.connman);
     }
 

src/txmempool.cpp

@@ -476,6 +476,9 @@ void CTxMemPool::removeUnchecked(txiter it, MemPoolRemovalReason reason) {
         mapNextTx.erase(txin.prevout);
     }
 
+    /* add logging because unchecked */
+    RemoveUnbroadcastTx(it->GetTx().GetId(), true);
+
     if (vTxHashes.size() > 1) {
         vTxHashes[it->vTxHashesIdx] = std::move(vTxHashes.back());
         vTxHashes[it->vTxHashesIdx].second->vTxHashesIdx = it->vTxHashesIdx;
@@ -1065,6 +1068,17 @@ size_t CTxMemPool::DynamicMemoryUsage() const {
            memusage::DynamicUsage(vTxHashes) + cachedInnerUsage;
 }
 
+void CTxMemPool::RemoveUnbroadcastTx(const TxId &txid, const bool unchecked) {
+    LOCK(cs);
+
+    if (m_unbroadcast_txids.erase(txid)) {
+        LogPrint(
+            BCLog::MEMPOOL, "Removed %i from set of unbroadcast txns%s\n",
+            txid.GetHex(),
+            (unchecked ? " before confirmation that txn was sent out" : ""));
+    }
+}
+
 void CTxMemPool::RemoveStaged(setEntries &stage, bool updateDescendants,
                               MemPoolRemovalReason reason) {
     AssertLockHeld(cs);

src/txmempool.h

@@ -580,6 +580,12 @@ class CTxMemPool {
     std::vector<indexed_transaction_set::const_iterator>
     GetSortedDepthAndScore() const EXCLUSIVE_LOCKS_REQUIRED(cs);
 
+    /**
+     * Track locally submitted transactions to periodically retry initial
+     * broadcast
+     */
+    std::set<TxId> m_unbroadcast_txids GUARDED_BY(cs);
+
 public:
     indirectmap<COutPoint, const CTransaction *> mapNextTx GUARDED_BY(cs);
     std::map<TxId, Amount> mapDeltas;
@@ -777,6 +783,21 @@ class CTxMemPool {
 
     size_t DynamicMemoryUsage() const;
 
+    /** Adds a transaction to the unbroadcast set */
+    void AddUnbroadcastTx(const TxId &txid) {
+        LOCK(cs);
+        m_unbroadcast_txids.insert(txid);
+    }
+
+    /** Removes a transaction from the unbroadcast set */
+    void RemoveUnbroadcastTx(const TxId &txid, const bool unchecked = false);
+
+    /** Returns transactions in unbroadcast set */
+    const std::set<TxId> GetUnbroadcastTxs() const {
+        LOCK(cs);
+        return m_unbroadcast_txids;
+    }
+
 private:
     /**
      * UpdateForDescendants is used by UpdateTransactionsFromBlock to update the