Merge #16197: net: Use mockable time for tx download

fab365835639a3da03f8ad9a58a0db6c6c4c2314 [qa] Test that getdata requests work as expected (Suhas Daftuar)
fa883ab35ad2d4328e35b1e855d0833740a6b910 net: Use mockable time for tx download (MarcoFalke)

Pull request description:

  Two commits:

  * First commit changes to mockable time for tx download (refactoring, should only have an effect on regtest)
  * Second commit adds a test that uses mocktime to test tx download

ACKs for top commit:
  laanwj:
    code review ACK 16197/commits/fab365835639a3da03f8ad9a58a0db6c6c4c2314
  jamesob:
    ACK https://github.com/bitcoin/bitcoin/pull/16197/commits/fab365835639a3da03f8ad9a58a0db6c6c4c2314

Tree-SHA512: 3a64a3e283ec4bab1f6e506404b11f0a564a5b61d2a7508ae738a61f035e57220484c66e0ae47d847fe9f7e3ff5cc834909d7b34a9bbcea6abe01f8742806908

2 files changed, 176 insertions, 0 deletions

diff --git a/test/functional/p2p_tx_download.py b/test/functional/p2p_tx_download.py
new file mode 100755
index 0000000000..19d78ff303
--- /dev/null
+++ b/test/functional/p2p_tx_download.py
@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+# Copyright (c) 2019 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""
+Test transaction download behavior
+"""
+
+from test_framework.messages import (
+    CInv,
+    CTransaction,
+    FromHex,
+    MSG_TX,
+    MSG_TYPE_MASK,
+    msg_inv,
+    msg_notfound,
+)
+from test_framework.mininode import (
+    P2PInterface,
+    mininode_lock,
+)
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import (
+    assert_equal,
+    wait_until,
+)
+from test_framework.address import ADDRESS_BCRT1_UNSPENDABLE
+
+import time
+
+
+class TestP2PConn(P2PInterface):
+    def __init__(self):
+        super().__init__()
+        self.tx_getdata_count = 0
+
+    def on_getdata(self, message):
+        for i in message.inv:
+            if i.type & MSG_TYPE_MASK == MSG_TX:
+                self.tx_getdata_count += 1
+
+
+# Constants from net_processing
+GETDATA_TX_INTERVAL = 60  # seconds
+MAX_GETDATA_RANDOM_DELAY = 2  # seconds
+INBOUND_PEER_TX_DELAY = 2  # seconds
+MAX_GETDATA_IN_FLIGHT = 100
+TX_EXPIRY_INTERVAL = GETDATA_TX_INTERVAL * 10
+
+# Python test constants
+NUM_INBOUND = 10
+MAX_GETDATA_INBOUND_WAIT = GETDATA_TX_INTERVAL + MAX_GETDATA_RANDOM_DELAY + INBOUND_PEER_TX_DELAY
+
+
+class TxDownloadTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.setup_clean_chain = False
+        self.num_nodes = 2
+
+    def test_tx_requests(self):
+        self.log.info("Test that we request transactions from all our peers, eventually")
+
+        txid = 0xdeadbeef
+
+        self.log.info("Announce the txid from each incoming peer to node 0")
+        msg = msg_inv([CInv(t=1, h=txid)])
+        for p in self.nodes[0].p2ps:
+            p.send_message(msg)
+            p.sync_with_ping()
+
+        outstanding_peer_index = [i for i in range(len(self.nodes[0].p2ps))]
+
+        def getdata_found(peer_index):
+            p = self.nodes[0].p2ps[peer_index]
+            with mininode_lock:
+                return p.last_message.get("getdata") and p.last_message["getdata"].inv[-1].hash == txid
+
+        node_0_mocktime = int(time.time())
+        while outstanding_peer_index:
+            node_0_mocktime += MAX_GETDATA_INBOUND_WAIT
+            self.nodes[0].setmocktime(node_0_mocktime)
+            wait_until(lambda: any(getdata_found(i) for i in outstanding_peer_index))
+            for i in outstanding_peer_index:
+                if getdata_found(i):
+                    outstanding_peer_index.remove(i)
+
+        self.nodes[0].setmocktime(0)
+        self.log.info("All outstanding peers received a getdata")
+
+    def test_inv_block(self):
+        self.log.info("Generate a transaction on node 0")
+        tx = self.nodes[0].createrawtransaction(
+            inputs=[{  # coinbase
+                "txid": self.nodes[0].getblock(self.nodes[0].getblockhash(1))['tx'][0],
+                "vout": 0
+            }],
+            outputs={ADDRESS_BCRT1_UNSPENDABLE: 50 - 0.00025},
+        )
+        tx = self.nodes[0].signrawtransactionwithkey(
+            hexstring=tx,
+            privkeys=[self.nodes[0].get_deterministic_priv_key().key],
+        )['hex']
+        ctx = FromHex(CTransaction(), tx)
+        txid = int(ctx.rehash(), 16)
+
+        self.log.info(
+            "Announce the transaction to all nodes from all {} incoming peers, but never send it".format(NUM_INBOUND))
+        msg = msg_inv([CInv(t=1, h=txid)])
+        for p in self.peers:
+            p.send_message(msg)
+            p.sync_with_ping()
+
+        self.log.info("Put the tx in node 0's mempool")
+        self.nodes[0].sendrawtransaction(tx)
+
+        # Since node 1 is connected outbound to an honest peer (node 0), it
+        # should get the tx within a timeout. (Assuming that node 0
+        # announced the tx within the timeout)
+        # The timeout is the sum of
+        # * the worst case until the tx is first requested from an inbound
+        #   peer, plus
+        # * the first time it is re-requested from the outbound peer, plus
+        # * 2 seconds to avoid races
+        timeout = 2 + (MAX_GETDATA_RANDOM_DELAY + INBOUND_PEER_TX_DELAY) + (
+            GETDATA_TX_INTERVAL + MAX_GETDATA_RANDOM_DELAY)
+        self.log.info("Tx should be received at node 1 after {} seconds".format(timeout))
+        self.sync_mempools(timeout=timeout)
+
+    def test_in_flight_max(self):
+        self.log.info("Test that we don't request more than {} transactions from any peer, every {} minutes".format(
+            MAX_GETDATA_IN_FLIGHT, TX_EXPIRY_INTERVAL / 60))
+        txids = [i for i in range(MAX_GETDATA_IN_FLIGHT + 2)]
+
+        p = self.nodes[0].p2ps[0]
+
+        with mininode_lock:
+            p.tx_getdata_count = 0
+
+        p.send_message(msg_inv([CInv(t=1, h=i) for i in txids]))
+        wait_until(lambda: p.tx_getdata_count >= MAX_GETDATA_IN_FLIGHT, lock=mininode_lock)
+        with mininode_lock:
+            assert_equal(p.tx_getdata_count, MAX_GETDATA_IN_FLIGHT)
+
+        self.log.info("Now check that if we send a NOTFOUND for a transaction, we'll get one more request")
+        p.send_message(msg_notfound(vec=[CInv(t=1, h=txids[0])]))
+        wait_until(lambda: p.tx_getdata_count >= MAX_GETDATA_IN_FLIGHT + 1, timeout=10, lock=mininode_lock)
+        with mininode_lock:
+            assert_equal(p.tx_getdata_count, MAX_GETDATA_IN_FLIGHT + 1)
+
+        WAIT_TIME = TX_EXPIRY_INTERVAL // 2 + TX_EXPIRY_INTERVAL
+        self.log.info("if we wait about {} minutes, we should eventually get more requests".format(WAIT_TIME / 60))
+        self.nodes[0].setmocktime(int(time.time() + WAIT_TIME))
+        wait_until(lambda: p.tx_getdata_count == MAX_GETDATA_IN_FLIGHT + 2)
+        self.nodes[0].setmocktime(0)
+
+    def run_test(self):
+        # Setup the p2p connections
+        self.peers = []
+        for node in self.nodes:
+            for i in range(NUM_INBOUND):
+                self.peers.append(node.add_p2p_connection(TestP2PConn()))
+
+        self.log.info("Nodes are setup with {} incoming connections each".format(NUM_INBOUND))
+
+        # Test the in-flight max first, because we want no transactions in
+        # flight ahead of this test.
+        self.test_in_flight_max()
+
+        self.test_inv_block()
+
+        self.test_tx_requests()
+
+
+if __name__ == '__main__':
+    TxDownloadTest().main()
diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py
index 7ad2b78d4e..0e0e6ed18e 100755
--- a/test/functional/test_runner.py
+++ b/test/functional/test_runner.py
@@ -91,6 +91,7 @@ BASE_SCRIPTS = [
     'wallet_labels.py',
     'p2p_segwit.py',
     'p2p_timeouts.py',
+    'p2p_tx_download.py',
     'wallet_dump.py',
     'wallet_listtransactions.py',
     # vv Tests less than 60s vv