[test] wait for inital broadcast to complete before comparing mempool…

… entries

- mempool entry 'unbroadcast' field changes when initial broadcast is complete,
so anytime you want to compare mempool entries, wait for initial broadcast to complete
('unbroadcast' = False) otherwise it may change in between calls
- P2PTxInvStore can help track tx broadcast but needs to send msg_getdata for invs
- mempool_packages.py needs to wait especially because it compares entries as a whole
with getrawmempool and getmempoolentry

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(len(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)

test/functional/test_framework/mininode.py

@@ -638,6 +638,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:
@@ -647,3 +648,11 @@ def on_inv(self, message):
     def get_invs(self):
         with mininode_lock:
             return list(self.tx_invs_received.keys())
+
+    def wait_for_broadcast(self, num_txns):
+        # Waits for num_txns transactions to complete initial broadcast.
+        # Wait until num_txns invs have been received (and getdatas sent).
+        wait_until(lambda: len(self.tx_invs_received.keys()) == num_txns)
+        # Flush messages and wait for the getdatas to get processed
+        # The mempool should mark unbroadcast=False for these transactions.
+        self.sync_with_ping()