Merge #18895: p2p: unbroadcast followups: rpcs, nLastResend, mempool sanity check

651f1d8 [test] wait for inital broadcast before comparing mempool entries (gzhao408)
9d3f7eb [mempool] sanity check that all unbroadcast txns are in mempool (gzhao408)
a7ebe48 [rpc] add unbroadcast info to mempool entries and getmempoolinfo (gzhao408)
d160069 [wallet] remove nLastResend logic (gzhao408)

Pull request description:

  Followup to #18038 by amitiuttarwar which introduces the unbroadcast set: "a mechanism for the mempool to track locally submitted transactions" and decreases the frequency of rebroadcast from 10-15 minutes to 12-36 hours.

  This PR addresses some of the outstanding TODOs building on top of it:
  - remove `nLastResend` logic, which is used to ensure rebroadcast doesn't happen again if no new block has been mined (makes sense in 10-15 min period, but not necessary for 12-36 hour period). (#18038 [comment](bitcoin/bitcoin#18038 (comment)))
  - expose unbroadcast info via RPCs, for more informative queries and testing (#18038 [comment](bitcoin/bitcoin#18038 (comment)))
  - add sanity check to verify unbroadcast transactions still exist in mempool before rebroadcasting (#18038 [comment](bitcoin/bitcoin#18038 (comment)))

ACKs for top commit:
  naumenkogs:
    Code review ACK 651f1d8
  amitiuttarwar:
    ACK 651f1d8 🎉
  MarcoFalke:
    Review ACK 651f1d8

Tree-SHA512: d5327e95ef39d44152b48df5c610502ae11c168f43dbbfb2885340c93d1ba9426eb3a5794573f5fc843502109cb3ffb63efa3f2db4f8f112efcde8f76d9a8845

Author: fanquake

src/net_processing.cpp

@@ -817,7 +817,12 @@ void PeerLogicValidation::ReattemptInitialBroadcast(CScheduler& scheduler) const
     std::set<uint256> unbroadcast_txids = m_mempool.GetUnbroadcastTxs();
 
     for (const uint256& txid : unbroadcast_txids) {
-        RelayTransaction(txid, *connman);
+        // Sanity check: all unbroadcast txns should exist in the mempool
+        if (m_mempool.exists(txid)) {
+            RelayTransaction(txid, *connman);
+        } else {
+            m_mempool.RemoveUnbroadcastTx(txid, true);
+        }
     }
 
     // schedule next run for 10-15 minutes in the future

src/rpc/blockchain.cpp

@@ -408,6 +408,7 @@ static std::vector<RPCResult> MempoolEntryDescription() { return {
     RPCResult{RPCResult::Type::ARR, "spentby", "unconfirmed transactions spending outputs from this transaction",
         {RPCResult{RPCResult::Type::STR_HEX, "transactionid", "child transaction id"}}},
     RPCResult{RPCResult::Type::BOOL, "bip125-replaceable", "Whether this transaction could be replaced due to BIP125 (replace-by-fee)"},
+    RPCResult{RPCResult::Type::BOOL, "unbroadcast", "Whether this transaction is currently unbroadcast (initial broadcast not yet confirmed)"},
 };}
 
 static void entryToJSON(const CTxMemPool& pool, UniValue& info, const CTxMemPoolEntry& e) EXCLUSIVE_LOCKS_REQUIRED(pool.cs)
@@ -469,6 +470,7 @@ static void entryToJSON(const CTxMemPool& pool, UniValue& info, const CTxMemPool
     }
 
     info.pushKV("bip125-replaceable", rbfStatus);
+    info.pushKV("unbroadcast", pool.IsUnbroadcastTx(tx.GetHash()));
 }
 
 UniValue MempoolToJSON(const CTxMemPool& pool, bool verbose)
@@ -1398,7 +1400,7 @@ UniValue MempoolInfoToJSON(const CTxMemPool& pool)
     ret.pushKV("maxmempool", (int64_t) maxmempool);
     ret.pushKV("mempoolminfee", ValueFromAmount(std::max(pool.GetMinFee(maxmempool), ::minRelayTxFee).GetFeePerK()));
     ret.pushKV("minrelaytxfee", ValueFromAmount(::minRelayTxFee.GetFeePerK()));
-
+    ret.pushKV("unbroadcastcount", uint64_t{pool.GetUnbroadcastTxs().size()});
     return ret;
 }
 
@@ -1417,6 +1419,7 @@ static UniValue getmempoolinfo(const JSONRPCRequest& request)
                         {RPCResult::Type::NUM, "maxmempool", "Maximum memory usage for the mempool"},
                         {RPCResult::Type::STR_AMOUNT, "mempoolminfee", "Minimum fee rate in " + CURRENCY_UNIT + "/kB for tx to be accepted. Is the maximum of minrelaytxfee and minimum mempool fee"},
                         {RPCResult::Type::STR_AMOUNT, "minrelaytxfee", "Current minimum relay fee for transactions"},
+                        {RPCResult::Type::NUM, "unbroadcastcount", "Current number of transactions that haven't passed initial broadcast yet"}
                     }},
                 RPCExamples{
                     HelpExampleCli("getmempoolinfo", "")

src/txmempool.h

@@ -704,7 +704,10 @@ class CTxMemPool
     /** Adds a transaction to the unbroadcast set */
     void AddUnbroadcastTx(const uint256& txid) {
         LOCK(cs);
-        m_unbroadcast_txids.insert(txid);
+        /** Sanity Check: the transaction should also be in the mempool */
+        if (exists(txid)) {
+            m_unbroadcast_txids.insert(txid);
+        }
     }
 
     /** Removes a transaction from the unbroadcast set */
@@ -716,6 +719,12 @@ class CTxMemPool
         return m_unbroadcast_txids;
     }
 
+    // Returns if a txid is in the unbroadcast set
+    bool IsUnbroadcastTx(const uint256& txid) const {
+        LOCK(cs);
+        return (m_unbroadcast_txids.count(txid) != 0);
+    }
+
 private:
     /** UpdateForDescendants is used by UpdateTransactionsFromBlock to update
      *  the descendants for a single transaction that has been added to the

src/wallet/wallet.cpp

@@ -1982,10 +1982,6 @@ void CWallet::ResendWalletTransactions()
     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
-    if (m_best_block_time < nLastResend) return;
-    nLastResend = GetTime();
-
     int submitted_tx_count = 0;
 
     { // cs_wallet scope

src/wallet/wallet.h

@@ -641,7 +641,6 @@ class CWallet final : public WalletStorage, public interfaces::Chain::Notificati
     int nWalletMaxVersion GUARDED_BY(cs_wallet) = FEATURE_BASE;
 
     int64_t nNextResend = 0;
-    int64_t nLastResend = 0;
     bool fBroadcastTransactions = false;
     // Local time that the tip block was received. Used to schedule wallet rebroadcasts.
     std::atomic<int64_t> m_best_block_time {0};

test/functional/mempool_packages.py

@@ -7,6 +7,7 @@
 from decimal import Decimal
 
 from test_framework.messages import COIN
+from test_framework.mininode import P2PTxInvStore
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
@@ -58,6 +59,7 @@ def chain_transaction(self, node, parent_txid, vout, value, fee, num_outputs):
 
     def run_test(self):
         # Mine some blocks and have them mature.
+        self.nodes[0].add_p2p_connection(P2PTxInvStore()) # keep track of invs
         self.nodes[0].generate(101)
         utxo = self.nodes[0].listunspent(10)
         txid = utxo[0]['txid']
@@ -72,6 +74,10 @@ def run_test(self):
             value = sent_value
             chain.append(txid)
 
+        # Wait until mempool transactions have passed initial broadcast (sent inv and received getdata)
+        # Otherwise, getrawmempool may be inconsistent with getmempoolentry if unbroadcast changes in between
+        self.nodes[0].p2p.wait_for_broadcast(chain)
+
         # Check mempool has MAX_ANCESTORS transactions in it, and descendant and ancestor
         # count and fees should look correct
         mempool = self.nodes[0].getrawmempool(True)
@@ -212,6 +218,10 @@ def run_test(self):
         for tx in chain[:MAX_ANCESTORS_CUSTOM]:
             assert tx in mempool1
         # TODO: more detailed check of node1's mempool (fees etc.)
+        # check transaction unbroadcast info (should be false if in both mempools)
+        mempool = self.nodes[0].getrawmempool(True)
+        for tx in mempool:
+            assert_equal(mempool[tx]['unbroadcast'], False)
 
         # TODO: test ancestor size limits
 

test/functional/mempool_unbroadcast.py

@@ -53,6 +53,13 @@ def test_broadcast(self):
         txFS = node.signrawtransactionwithwallet(txF["hex"])
         rpc_tx_hsh = node.sendrawtransaction(txFS["hex"])
 
+        # check transactions are in unbroadcast using rpc
+        mempoolinfo = self.nodes[0].getmempoolinfo()
+        assert_equal(mempoolinfo['unbroadcastcount'], 2)
+        mempool = self.nodes[0].getrawmempool(True)
+        for tx in mempool:
+            assert_equal(mempool[tx]['unbroadcast'], True)
+
         # check that second node doesn't have these two txns
         mempool = self.nodes[1].getrawmempool()
         assert rpc_tx_hsh not in mempool
@@ -71,6 +78,11 @@ def test_broadcast(self):
         assert rpc_tx_hsh in mempool
         assert wallet_tx_hsh in mempool
 
+        # check that transactions are no longer in first node's unbroadcast set
+        mempool = self.nodes[0].getrawmempool(True)
+        for tx in mempool:
+            assert_equal(mempool[tx]['unbroadcast'], False)
+
         self.log.info("Add another connection & ensure transactions aren't broadcast again")
 
         conn = node.add_p2p_connection(P2PTxInvStore())

test/functional/test_framework/mininode.py

@@ -645,6 +645,7 @@ def __init__(self):
         self.tx_invs_received = defaultdict(int)
 
     def on_inv(self, message):
+        super().on_inv(message) # Send getdata in response.
         # Store how many times invs have been received for each tx.
         for i in message.inv:
             if i.type == MSG_TX:
@@ -654,3 +655,12 @@ def on_inv(self, message):
     def get_invs(self):
         with mininode_lock:
             return list(self.tx_invs_received.keys())
+
+    def wait_for_broadcast(self, txns, timeout=60):
+        """Waits for the txns (list of txids) to complete initial broadcast.
+        The mempool should mark unbroadcast=False for these transactions.
+        """
+        # Wait until invs have been received (and getdatas sent) for each txid.
+        self.wait_until(lambda: set(self.get_invs()) == set([int(tx, 16) for tx in txns]), timeout)
+        # Flush messages and wait for the getdatas to be processed
+        self.sync_with_ping()