Merge #18038: P2P: Mempool tracks locally submitted transactions to improve wallet privacy

50fc4df [mempool] Persist unbroadcast set to mempool.dat (Amiti Uttarwar)
297a178 [test] Integration tests for unbroadcast functionality (Amiti Uttarwar)
6851502 [refactor/test] Extract P2PTxInvStore into test framework (Amiti Uttarwar)
dc1da48 [wallet] Update the rebroadcast frequency to be ~1/day. (Amiti Uttarwar)
e25e42f [p2p] Reattempt initial send of unbroadcast transactions (Amiti Uttarwar)
7e93eec [util] Add method that returns random time in milliseconds (Amiti Uttarwar)
89eeb4a [mempool] Track "unbroadcast" transactions (Amiti Uttarwar)

Pull request 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 #16698.
  Thank you to gmaxwell for the idea of how to break out this subset of functionality (bitcoin/bitcoin#16698 (comment))

ACKs for top commit:
  fjahr:
    Code review ACK 50fc4df
  MarcoFalke:
    ACK 50fc4df, I think this is ready for merge now 👻
  amitiuttarwar:
    The current tip `50fc4df` currently has 6 ACKs on it, so I've opened #18807 to address the last bits.
  jnewbery:
    utACK 50fc4df.
  ariard:
    Code Review ACK 50fc4df (minor points no need to invalid other ACKs)
  robot-visions:
    ACK 50fc4df
  sipa:
    utACK 50fc4df
  naumenkogs:
    utACK 50fc4df

Tree-SHA512: 2dd935d645d5e209f8abf87bfaa3ef0e4492705ce7e89ea64279cb27ffd37f4727fa94ad62d41be331177332f8edbebf3c7f4972f8cda10dd951b80a28ab3c0f

Author: fanquake

src/net_processing.cpp

@@ -810,6 +810,19 @@ void PeerLogicValidation::InitializeNode(CNode *pnode) {
         PushNodeVersion(pnode, connman, GetTime());
 }
 
+void PeerLogicValidation::ReattemptInitialBroadcast(CScheduler& scheduler) const
+{
+    std::set<uint256> unbroadcast_txids = m_mempool.GetUnbroadcastTxs();
+
+    for (const uint256& txid : unbroadcast_txids) {
+        RelayTransaction(txid, *connman);
+    }
+
+    // schedule next run for 10-15 minutes in the future
+    const std::chrono::milliseconds delta = std::chrono::minutes{10} + GetRandMillis(std::chrono::minutes{5});
+    scheduler.scheduleFromNow([&] { ReattemptInitialBroadcast(scheduler); }, delta);
+}
+
 void PeerLogicValidation::FinalizeNode(NodeId nodeid, bool& fUpdateConnectionTime) {
     fUpdateConnectionTime = false;
     LOCK(cs_main);
@@ -1159,6 +1172,10 @@ PeerLogicValidation::PeerLogicValidation(CConnman* connmanIn, BanMan* banman, CS
     // timer.
     static_assert(EXTRA_PEER_CHECK_INTERVAL < STALE_CHECK_INTERVAL, "peer eviction timer should be less than stale tip check timer");
     scheduler.scheduleEvery([this, consensusParams] { this->CheckForStaleTipAndEvictPeers(consensusParams); }, std::chrono::seconds{EXTRA_PEER_CHECK_INTERVAL});
+
+    // schedule next run for 10-15 minutes in the future
+    const std::chrono::milliseconds delta = std::chrono::minutes{10} + GetRandMillis(std::chrono::minutes{5});
+    scheduler.scheduleFromNow([&] { ReattemptInitialBroadcast(scheduler); }, delta);
 }
 
 /**
@@ -1587,7 +1604,7 @@ void static ProcessGetBlockData(CNode* pfrom, const CChainParams& chainparams, c
     }
 }
 
-void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnman* connman, const CTxMemPool& mempool, const std::atomic<bool>& interruptMsgProc) LOCKS_EXCLUDED(cs_main)
+void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnman* connman, CTxMemPool& mempool, const std::atomic<bool>& interruptMsgProc) LOCKS_EXCLUDED(cs_main)
 {
     AssertLockNotHeld(cs_main);
 
@@ -1636,7 +1653,13 @@ void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnm
                     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(inv.hash);
+            } else {
                 vNotFound.push_back(inv);
             }
         }

src/net_processing.h

@@ -76,6 +76,8 @@ class PeerLogicValidation final : public CValidationInterface, public NetEventsI
     void CheckForStaleTipAndEvictPeers(const Consensus::Params &consensusParams);
     /** If we have extra outbound peers, try to disconnect the one with the oldest block announcement */
     void EvictExtraOutboundPeers(int64_t time_in_seconds) EXCLUSIVE_LOCKS_REQUIRED(cs_main);
+    /** Retrieve unbroadcast transactions from the mempool and reattempt sending to peers */
+    void ReattemptInitialBroadcast(CScheduler& scheduler) const;
 
 private:
     int64_t m_stale_tip_check_time; //!< Next time to check for stale tip

src/node/transaction.cpp

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

src/random.cpp

@@ -592,6 +592,11 @@ std::chrono::microseconds GetRandMicros(std::chrono::microseconds duration_max)
     return std::chrono::microseconds{GetRand(duration_max.count())};
 }
 
+std::chrono::milliseconds GetRandMillis(std::chrono::milliseconds duration_max) noexcept
+{
+    return std::chrono::milliseconds{GetRand(duration_max.count())};
+}
+
 int GetRandInt(int nMax) noexcept
 {
     return GetRand(nMax);

src/random.h

@@ -69,6 +69,7 @@
 void GetRandBytes(unsigned char* buf, int num) noexcept;
 uint64_t GetRand(uint64_t nMax) noexcept;
 std::chrono::microseconds GetRandMicros(std::chrono::microseconds duration_max) noexcept;
+std::chrono::milliseconds GetRandMillis(std::chrono::milliseconds duration_max) noexcept;
 int GetRandInt(int nMax) noexcept;
 uint256 GetRandHash() noexcept;
 

src/txmempool.cpp

@@ -417,6 +417,8 @@ void CTxMemPool::removeUnchecked(txiter it, MemPoolRemovalReason reason)
     for (const CTxIn& txin : it->GetTx().vin)
         mapNextTx.erase(txin.prevout);
 
+    RemoveUnbroadcastTx(hash, true /* add logging because unchecked */ );
+
     if (vTxHashes.size() > 1) {
         vTxHashes[it->vTxHashesIdx] = std::move(vTxHashes.back());
         vTxHashes[it->vTxHashesIdx].second->vTxHashesIdx = it->vTxHashesIdx;
@@ -919,6 +921,15 @@ size_t CTxMemPool::DynamicMemoryUsage() const {
     return memusage::MallocUsage(sizeof(CTxMemPoolEntry) + 12 * sizeof(void*)) * mapTx.size() + memusage::DynamicUsage(mapNextTx) + memusage::DynamicUsage(mapDeltas) + memusage::DynamicUsage(mapLinks) + memusage::DynamicUsage(vTxHashes) + cachedInnerUsage;
 }
 
+void CTxMemPool::RemoveUnbroadcastTx(const uint256& 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);
     UpdateForRemoveFromMempool(stage, updateDescendants);

src/txmempool.h

@@ -549,6 +549,9 @@ 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<uint256> m_unbroadcast_txids GUARDED_BY(cs);
+
 public:
     indirectmap<COutPoint, const CTransaction*> mapNextTx GUARDED_BY(cs);
     std::map<uint256, CAmount> mapDeltas;
@@ -698,6 +701,21 @@ class CTxMemPool
 
     size_t DynamicMemoryUsage() const;
 
+    /** Adds a transaction to the unbroadcast set */
+    void AddUnbroadcastTx(const uint256& txid) {
+        LOCK(cs);
+        m_unbroadcast_txids.insert(txid);
+    }
+
+    /** Removes a transaction from the unbroadcast set */
+    void RemoveUnbroadcastTx(const uint256& txid, const bool unchecked = false);
+
+    /** Returns transactions in unbroadcast set */
+    const std::set<uint256> GetUnbroadcastTxs() const {
+        LOCK(cs);
+        return m_unbroadcast_txids;
+    }
+
 private:
     /** UpdateForDescendants is used by UpdateTransactionsFromBlock to update
      *  the descendants for a single transaction that has been added to the

src/validation.cpp

@@ -4998,6 +4998,7 @@ bool LoadMempool(CTxMemPool& pool)
     int64_t expired = 0;
     int64_t failed = 0;
     int64_t already_there = 0;
+    int64_t unbroadcast = 0;
     int64_t nNow = GetTime();
 
     try {
@@ -5051,12 +5052,21 @@ bool LoadMempool(CTxMemPool& pool)
         for (const auto& i : mapDeltas) {
             pool.PrioritiseTransaction(i.first, i.second);
         }
+
+        std::set<uint256> unbroadcast_txids;
+        file >> unbroadcast_txids;
+        unbroadcast = unbroadcast_txids.size();
+
+        for (const auto& txid : unbroadcast_txids) {
+            pool.AddUnbroadcastTx(txid);
+        }
+
     } catch (const std::exception& e) {
         LogPrintf("Failed to deserialize mempool data on disk: %s. Continuing anyway.\n", e.what());
         return false;
     }
 
-    LogPrintf("Imported mempool transactions from disk: %i succeeded, %i failed, %i expired, %i already there\n", count, failed, expired, already_there);
+    LogPrintf("Imported mempool transactions from disk: %i succeeded, %i failed, %i expired, %i already there, %i waiting for initial broadcast\n", count, failed, expired, already_there, unbroadcast);
     return true;
 }
 
@@ -5066,6 +5076,7 @@ bool DumpMempool(const CTxMemPool& pool)
 
     std::map<uint256, CAmount> mapDeltas;
     std::vector<TxMempoolInfo> vinfo;
+    std::set<uint256> unbroadcast_txids;
 
     static Mutex dump_mutex;
     LOCK(dump_mutex);
@@ -5076,6 +5087,7 @@ bool DumpMempool(const CTxMemPool& pool)
             mapDeltas[i.first] = i.second;
         }
         vinfo = pool.infoAll();
+        unbroadcast_txids = pool.GetUnbroadcastTxs();
     }
 
     int64_t mid = GetTimeMicros();
@@ -5100,6 +5112,10 @@ bool DumpMempool(const CTxMemPool& pool)
         }
 
         file << mapDeltas;
+
+        LogPrintf("Writing %d unbroadcast transactions to disk.\n", unbroadcast_txids.size());
+        file << unbroadcast_txids;
+
         if (!FileCommit(file.Get()))
             throw std::runtime_error("FileCommit failed");
         file.fclose();

src/wallet/wallet.cpp

@@ -1993,7 +1993,8 @@ void CWallet::ResendWalletTransactions()
     // that these are our transactions.
     if (GetTime() < nNextResend || !fBroadcastTransactions) return;
     bool fFirst = (nNextResend == 0);
-    nNextResend = GetTime() + GetRand(30 * 60);
+    // resend 12-36 hours from now, ~1 day on average.
+    nNextResend = GetTime() + (12 * 60 * 60) + GetRand(24 * 60 * 60);
     if (fFirst) return;
 
     // Only do it if there's been a new block since last time

test/functional/mempool_persist.py

@@ -40,10 +40,13 @@
 import time
 
 from test_framework.test_framework import BitcoinTestFramework
+from test_framework.mininode import P2PTxInvStore
 from test_framework.util import (
     assert_equal,
     assert_greater_than_or_equal,
     assert_raises_rpc_error,
+    connect_nodes,
+    disconnect_nodes,
     wait_until,
 )
 
@@ -80,6 +83,11 @@ def run_test(self):
         assert_greater_than_or_equal(tx_creation_time, tx_creation_time_lower)
         assert_greater_than_or_equal(tx_creation_time_higher, tx_creation_time)
 
+        # disconnect nodes & make a txn that remains in the unbroadcast set.
+        disconnect_nodes(self.nodes[0], 2)
+        self.nodes[0].sendtoaddress(self.nodes[2].getnewaddress(), Decimal("12"))
+        connect_nodes(self.nodes[0], 2)
+
         self.log.debug("Stop-start the nodes. Verify that node0 has the transactions in its mempool and node1 does not. Verify that node2 calculates its balance correctly after loading wallet transactions.")
         self.stop_nodes()
         # Give this node a head-start, so we can be "extra-sure" that it didn't load anything later
@@ -89,7 +97,7 @@ def run_test(self):
         self.start_node(2)
         wait_until(lambda: self.nodes[0].getmempoolinfo()["loaded"], timeout=1)
         wait_until(lambda: self.nodes[2].getmempoolinfo()["loaded"], timeout=1)
-        assert_equal(len(self.nodes[0].getrawmempool()), 5)
+        assert_equal(len(self.nodes[0].getrawmempool()), 6)
         assert_equal(len(self.nodes[2].getrawmempool()), 5)
         # The others have loaded their mempool. If node_1 loaded anything, we'd probably notice by now:
         assert_equal(len(self.nodes[1].getrawmempool()), 0)
@@ -105,17 +113,18 @@ def run_test(self):
         self.nodes[2].syncwithvalidationinterfacequeue()  # Flush mempool to wallet
         assert_equal(node2_balance, self.nodes[2].getbalance())
 
+        # start node0 with wallet disabled so wallet transactions don't get resubmitted
         self.log.debug("Stop-start node0 with -persistmempool=0. Verify that it doesn't load its mempool.dat file.")
         self.stop_nodes()
-        self.start_node(0, extra_args=["-persistmempool=0"])
+        self.start_node(0, extra_args=["-persistmempool=0", "-disablewallet"])
         wait_until(lambda: self.nodes[0].getmempoolinfo()["loaded"])
         assert_equal(len(self.nodes[0].getrawmempool()), 0)
 
         self.log.debug("Stop-start node0. Verify that it has the transactions in its mempool.")
         self.stop_nodes()
         self.start_node(0)
         wait_until(lambda: self.nodes[0].getmempoolinfo()["loaded"])
-        assert_equal(len(self.nodes[0].getrawmempool()), 5)
+        assert_equal(len(self.nodes[0].getrawmempool()), 6)
 
         mempooldat0 = os.path.join(self.nodes[0].datadir, self.chain, 'mempool.dat')
         mempooldat1 = os.path.join(self.nodes[1].datadir, self.chain, 'mempool.dat')
@@ -124,12 +133,12 @@ def run_test(self):
         self.nodes[0].savemempool()
         assert os.path.isfile(mempooldat0)
 
-        self.log.debug("Stop nodes, make node1 use mempool.dat from node0. Verify it has 5 transactions")
+        self.log.debug("Stop nodes, make node1 use mempool.dat from node0. Verify it has 6 transactions")
         os.rename(mempooldat0, mempooldat1)
         self.stop_nodes()
         self.start_node(1, extra_args=[])
         wait_until(lambda: self.nodes[1].getmempoolinfo()["loaded"])
-        assert_equal(len(self.nodes[1].getrawmempool()), 5)
+        assert_equal(len(self.nodes[1].getrawmempool()), 6)
 
         self.log.debug("Prevent bitcoind from writing mempool.dat to disk. Verify that `savemempool` fails")
         # to test the exception we are creating a tmp folder called mempool.dat.new
@@ -139,6 +148,27 @@ def run_test(self):
         assert_raises_rpc_error(-1, "Unable to dump mempool to disk", self.nodes[1].savemempool)
         os.rmdir(mempooldotnew1)
 
+        self.test_persist_unbroadcast()
+
+    def test_persist_unbroadcast(self):
+        node0 = self.nodes[0]
+        self.start_node(0)
+
+        # clear out mempool
+        node0.generate(1)
+
+        # disconnect nodes to make a txn that remains in the unbroadcast set.
+        disconnect_nodes(node0, 1)
+        node0.sendtoaddress(self.nodes[1].getnewaddress(), Decimal("12"))
+
+        # shutdown, then startup with wallet disabled
+        self.stop_nodes()
+        self.start_node(0, extra_args=["-disablewallet"])
+
+        # check that txn gets broadcast due to unbroadcast logic
+        conn = node0.add_p2p_connection(P2PTxInvStore())
+        node0.mockscheduler(16*60) # 15 min + 1 for buffer
+        wait_until(lambda: len(conn.get_invs()) == 1)
 
 if __name__ == '__main__':
     MempoolPersistTest().main()