p2p_tx_download.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2019-present The Bitcoin Core developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """
6 Test transaction download behavior
7 """
8 from decimal import Decimal
9 from enum import Enum
10 import time
11
12 from test_framework.mempool_util import (
13 fill_mempool,
14 )
15 from test_framework.messages import (
16 CInv,
17 MSG_TX,
18 MSG_TYPE_MASK,
19 MSG_WTX,
20 msg_inv,
21 msg_notfound,
22 msg_tx,
23 )
24 from test_framework.p2p import (
25 P2PInterface,
26 p2p_lock,
27 NONPREF_PEER_TX_DELAY,
28 GETDATA_TX_INTERVAL,
29 TXID_RELAY_DELAY,
30 OVERLOADED_PEER_TX_DELAY
31 )
32 from test_framework.test_framework import BitcoinTestFramework
33 from test_framework.util import (
34 assert_equal,
35 )
36 from test_framework.wallet import MiniWallet
37
38
39 class TestP2PConn(P2PInterface):
40 def __init__(self, wtxidrelay=True):
41 super().__init__(wtxidrelay=wtxidrelay)
42 self.tx_getdata_count = 0
43
44 def on_getdata(self, message):
45 for i in message.inv:
46 if i.type & MSG_TYPE_MASK == MSG_TX or i.type & MSG_TYPE_MASK == MSG_WTX:
47 self.tx_getdata_count += 1
48
49
50 # Constants from txdownloadman
51 MAX_PEER_TX_REQUEST_IN_FLIGHT = 100
52 MAX_PEER_TX_ANNOUNCEMENTS = 5000
53
54 # Python test constants
55 NUM_INBOUND = 10
56 MAX_GETDATA_INBOUND_WAIT = GETDATA_TX_INTERVAL + NONPREF_PEER_TX_DELAY + TXID_RELAY_DELAY
57
58 class ConnectionType(Enum):
59 """ Different connection types
60 1. INBOUND: Incoming connection, not whitelisted
61 2. OUTBOUND: Outgoing connection
62 3. WHITELIST: Incoming connection, but whitelisted
63 """
64 INBOUND = 0
65 OUTBOUND = 1
66 WHITELIST = 2
67
68 class TxDownloadTest(BitcoinTestFramework):
69 def set_test_params(self):
70 self.num_nodes = 2
71 self.extra_args= [['-maxmempool=5', '-persistmempool=0']] * self.num_nodes
72
73 def test_tx_requests(self):
74 self.log.info("Test that we request transactions from all our peers, eventually")
75
76 txid = 0xdeadbeef
77
78 self.log.info("Announce the txid from each incoming peer to node 0")
79 msg = msg_inv([CInv(t=MSG_WTX, h=txid)])
80 for p in self.nodes[0].p2ps:
81 p.send_and_ping(msg)
82
83 outstanding_peer_index = [i for i in range(len(self.nodes[0].p2ps))]
84
85 def getdata_found(peer_index):
86 p = self.nodes[0].p2ps[peer_index]
87 with p2p_lock:
88 return p.last_message.get("getdata") and p.last_message["getdata"].inv[-1].hash == txid
89
90 node_0_mocktime = int(time.time())
91 while outstanding_peer_index:
92 node_0_mocktime += MAX_GETDATA_INBOUND_WAIT
93 self.nodes[0].setmocktime(node_0_mocktime)
94 self.wait_until(lambda: any(getdata_found(i) for i in outstanding_peer_index))
95 for i in outstanding_peer_index:
96 if getdata_found(i):
97 outstanding_peer_index.remove(i)
98
99 self.nodes[0].setmocktime(0)
100 self.log.info("All outstanding peers received a getdata")
101
102 def test_inv_block(self):
103 self.log.info("Generate a transaction on node 0")
104 tx = self.wallet.create_self_transfer()
105 wtxid = int(tx['wtxid'], 16)
106
107 self.nodes[0].setmocktime(int(time.time()))
108
109 self.log.info(
110 "Announce the transaction to all nodes from all {} incoming peers, but never send it".format(NUM_INBOUND))
111 msg = msg_inv([CInv(t=MSG_WTX, h=wtxid)])
112 for p in self.peers:
113 p.send_and_ping(msg)
114
115 self.log.info("Put the tx in node 0's mempool")
116 self.nodes[0].sendrawtransaction(tx['hex'])
117
118 # Since node 1 is connected outbound to an honest peer (node 0), it
119 # should get the tx within a timeout.
120 # However, node 0 sees the same connection as inbound, so it may not
121 # announce the tx within the timeout.
122 # The timeout is the sum of
123 # * the worst case until the tx is first requested from an inbound
124 # peer, plus
125 # * the first time it is re-requested from the outbound peer, plus
126 # * 2 seconds to avoid races
127 assert_equal(self.nodes[0].getpeerinfo()[0]["inbound"], True)
128 assert_equal(self.nodes[0].getpeerinfo()[0]["inv_to_send"], 1)
129 assert_equal(self.nodes[1].getpeerinfo()[0]['inbound'], False)
130 timeout = 2 + NONPREF_PEER_TX_DELAY + GETDATA_TX_INTERVAL
131 self.log.info("Tx should be received at node 1 after {} seconds".format(timeout))
132 self.nodes[0].bumpmocktime(timeout)
133 # Use an unrelated peer to ensure node 0 calls `SendMessages` at least once
134 self.nodes[0].p2ps[0].sync_with_ping()
135 assert_equal(self.nodes[0].getpeerinfo()[0]['inv_to_send'], 0) # May fail rarely, retry the test
136 self.sync_mempools()
137
138 self.nodes[0].setmocktime(0)
139
140 def test_in_flight_max(self):
141 self.log.info("Test that we don't load peers with more than {} transaction requests immediately".format(MAX_PEER_TX_REQUEST_IN_FLIGHT))
142 txids = [i for i in range(MAX_PEER_TX_REQUEST_IN_FLIGHT + 2)]
143
144 p = self.nodes[0].p2ps[0]
145
146 with p2p_lock:
147 p.tx_getdata_count = 0
148
149 mock_time = int(time.time() + 1)
150 self.nodes[0].setmocktime(mock_time)
151 for i in range(MAX_PEER_TX_REQUEST_IN_FLIGHT):
152 p.send_without_ping(msg_inv([CInv(t=MSG_WTX, h=txids[i])]))
153 p.sync_with_ping()
154 mock_time += NONPREF_PEER_TX_DELAY
155 self.nodes[0].setmocktime(mock_time)
156 p.wait_until(lambda: p.tx_getdata_count >= MAX_PEER_TX_REQUEST_IN_FLIGHT)
157 for i in range(MAX_PEER_TX_REQUEST_IN_FLIGHT, len(txids)):
158 p.send_without_ping(msg_inv([CInv(t=MSG_WTX, h=txids[i])]))
159 p.sync_with_ping()
160 self.log.info("No more than {} requests should be seen within {} seconds after announcement".format(MAX_PEER_TX_REQUEST_IN_FLIGHT, NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_TX_DELAY - 1))
161 self.nodes[0].setmocktime(mock_time + NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_TX_DELAY - 1)
162 p.sync_with_ping()
163 with p2p_lock:
164 assert_equal(p.tx_getdata_count, MAX_PEER_TX_REQUEST_IN_FLIGHT)
165 self.log.info("If we wait {} seconds after announcement, we should eventually get more requests".format(NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_TX_DELAY))
166 self.nodes[0].setmocktime(mock_time + NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_TX_DELAY)
167 p.wait_until(lambda: p.tx_getdata_count == len(txids))
168
169 def test_expiry_fallback(self):
170 self.log.info('Check that expiry will select another peer for download')
171 WTXID = 0xffaa
172 peer1 = self.nodes[0].add_p2p_connection(TestP2PConn())
173 peer2 = self.nodes[0].add_p2p_connection(TestP2PConn())
174 for p in [peer1, peer2]:
175 p.send_without_ping(msg_inv([CInv(t=MSG_WTX, h=WTXID)]))
176 # One of the peers is asked for the tx
177 peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1)
178 with p2p_lock:
179 _peer_expiry, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1)
180 assert_equal(peer_fallback.tx_getdata_count, 0)
181 self.nodes[0].setmocktime(int(time.time()) + GETDATA_TX_INTERVAL + 1) # Wait for request to _peer_expiry to expire
182 peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1)
183 self.restart_node(0) # reset mocktime
184
185 def test_disconnect_fallback(self):
186 self.log.info('Check that disconnect will select another peer for download')
187 WTXID = 0xffbb
188 peer1 = self.nodes[0].add_p2p_connection(TestP2PConn())
189 peer2 = self.nodes[0].add_p2p_connection(TestP2PConn())
190 for p in [peer1, peer2]:
191 p.send_without_ping(msg_inv([CInv(t=MSG_WTX, h=WTXID)]))
192 # One of the peers is asked for the tx
193 peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1)
194 with p2p_lock:
195 peer_disconnect, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1)
196 assert_equal(peer_fallback.tx_getdata_count, 0)
197 peer_disconnect.peer_disconnect()
198 peer_disconnect.wait_for_disconnect()
199 peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1)
200
201 def test_notfound_fallback(self):
202 self.log.info('Check that notfounds will select another peer for download immediately')
203 WTXID = 0xffdd
204 peer1 = self.nodes[0].add_p2p_connection(TestP2PConn())
205 peer2 = self.nodes[0].add_p2p_connection(TestP2PConn())
206 for p in [peer1, peer2]:
207 p.send_without_ping(msg_inv([CInv(t=MSG_WTX, h=WTXID)]))
208 # One of the peers is asked for the tx
209 peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1)
210 with p2p_lock:
211 peer_notfound, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1)
212 assert_equal(peer_fallback.tx_getdata_count, 0)
213 peer_notfound.send_and_ping(msg_notfound(vec=[CInv(MSG_WTX, WTXID)])) # Send notfound, so that fallback peer is selected
214 peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1)
215
216 def test_preferred_inv(self, connection_type: ConnectionType):
217 if connection_type == ConnectionType.WHITELIST:
218 self.log.info('Check invs from preferred (whitelisted) peers are downloaded immediately')
219 self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1'])
220 elif connection_type == ConnectionType.OUTBOUND:
221 self.log.info('Check invs from preferred (outbound) peers are downloaded immediately')
222 self.restart_node(0)
223 elif connection_type == ConnectionType.INBOUND:
224 self.log.info('Check invs from non-preferred peers are downloaded after {} s'.format(NONPREF_PEER_TX_DELAY))
225 self.restart_node(0)
226 else:
227 raise Exception("invalid connection_type")
228
229 mock_time = int(time.time() + 1)
230 self.nodes[0].setmocktime(mock_time)
231
232 if connection_type == ConnectionType.OUTBOUND:
233 peer = self.nodes[0].add_outbound_p2p_connection(
234 TestP2PConn(), wait_for_verack=True, p2p_idx=1, connection_type="outbound-full-relay")
235 else:
236 peer = self.nodes[0].add_p2p_connection(TestP2PConn())
237
238 peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
239 if connection_type != ConnectionType.INBOUND:
240 peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1)
241 else:
242 with p2p_lock:
243 assert_equal(peer.tx_getdata_count, 0)
244 self.nodes[0].setmocktime(mock_time + NONPREF_PEER_TX_DELAY)
245 peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1)
246
247 def test_preferred_tiebreaker_inv(self):
248 self.log.info("Test that preferred peers are always selected over non-preferred when ready")
249
250 self.restart_node(0)
251 self.nodes[0].setmocktime(int(time.time()))
252
253 # Peer that is immediately asked, but never responds.
254 # This will set us up to have two ready requests, one
255 # of which is preferred and one which is not
256 unresponsive_peer = self.nodes[0].add_outbound_p2p_connection(
257 TestP2PConn(), wait_for_verack=True, p2p_idx=0, connection_type="outbound-full-relay")
258 unresponsive_peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
259 unresponsive_peer.wait_until(lambda: unresponsive_peer.tx_getdata_count >= 1, timeout=1)
260
261 # A bunch of incoming (non-preferred) connections that advertise the same tx
262 non_pref_peers = []
263 NUM_INBOUND = 10
264 for _ in range(NUM_INBOUND):
265 non_pref_peers.append(self.nodes[0].add_p2p_connection(TestP2PConn()))
266 non_pref_peers[-1].send_and_ping(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
267
268 # Check that no request made due to in-flight
269 self.nodes[0].bumpmocktime(NONPREF_PEER_TX_DELAY)
270 with p2p_lock:
271 for peer in non_pref_peers:
272 assert_equal(peer.tx_getdata_count, 0)
273
274 # Now add another outbound (preferred) which is immediately ready for consideration
275 # upon advertisement
276 pref_peer = self.nodes[0].add_outbound_p2p_connection(
277 TestP2PConn(), wait_for_verack=True, p2p_idx=1, connection_type="outbound-full-relay")
278 pref_peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
279
280 assert_equal(len(self.nodes[0].getpeerinfo()), NUM_INBOUND + 2)
281
282 # Still have to wait for in-flight to timeout
283 with p2p_lock:
284 assert_equal(pref_peer.tx_getdata_count, 0)
285
286 # Timeout in-flight
287 self.nodes[0].bumpmocktime(GETDATA_TX_INTERVAL - NONPREF_PEER_TX_DELAY)
288
289 # Preferred peers are *always* selected next if ready
290 pref_peer.wait_until(lambda: pref_peer.tx_getdata_count >= 1, timeout=10)
291
292 # And none for non-preferred
293 for non_pref_peer in non_pref_peers:
294 with p2p_lock:
295 assert_equal(non_pref_peer.tx_getdata_count, 0)
296
297 def test_txid_inv_delay(self, glob_wtxid=False):
298 self.log.info('Check that inv from a txid-relay peers are delayed by {} s, with a wtxid peer {}'.format(TXID_RELAY_DELAY, glob_wtxid))
299 self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1'])
300 mock_time = int(time.time() + 1)
301 self.nodes[0].setmocktime(mock_time)
302 peer = self.nodes[0].add_p2p_connection(TestP2PConn(wtxidrelay=False))
303 if glob_wtxid:
304 # Add a second wtxid-relay connection otherwise TXID_RELAY_DELAY is waived in
305 # lack of wtxid-relay peers
306 self.nodes[0].add_p2p_connection(TestP2PConn(wtxidrelay=True))
307 peer.send_and_ping(msg_inv([CInv(t=MSG_TX, h=0xff11ff11)]))
308 with p2p_lock:
309 assert_equal(peer.tx_getdata_count, 0 if glob_wtxid else 1)
310 self.nodes[0].setmocktime(mock_time + TXID_RELAY_DELAY)
311 peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1)
312
313 def test_large_inv_batch(self):
314 self.log.info('Test how large inv batches are handled with relay permission')
315 self.restart_node(0, extra_args=['-whitelist=relay@127.0.0.1'])
316 peer = self.nodes[0].add_p2p_connection(TestP2PConn())
317 peer.send_without_ping(msg_inv([CInv(t=MSG_WTX, h=wtxid) for wtxid in range(MAX_PEER_TX_ANNOUNCEMENTS + 1)]))
318 peer.wait_until(lambda: peer.tx_getdata_count == MAX_PEER_TX_ANNOUNCEMENTS + 1)
319
320 self.log.info('Test how large inv batches are handled without relay permission')
321 self.restart_node(0)
322 peer = self.nodes[0].add_p2p_connection(TestP2PConn())
323 peer.send_without_ping(msg_inv([CInv(t=MSG_WTX, h=wtxid) for wtxid in range(MAX_PEER_TX_ANNOUNCEMENTS + 1)]))
324 peer.wait_until(lambda: peer.tx_getdata_count == MAX_PEER_TX_ANNOUNCEMENTS)
325 peer.sync_with_ping()
326
327 def test_spurious_notfound(self):
328 self.log.info('Check that spurious notfound is ignored')
329 self.nodes[0].p2ps[0].send_without_ping(msg_notfound(vec=[CInv(MSG_TX, 1)]))
330
331 def test_rejects_filter_reset(self):
332 self.log.info('Check that rejected tx is not requested again')
333 node = self.nodes[0]
334 fill_mempool(self, node, tx_sync_fun=self.no_op)
335 self.wallet.rescan_utxos()
336 mempoolminfee = node.getmempoolinfo()['mempoolminfee']
337 peer = node.add_p2p_connection(TestP2PConn())
338 low_fee_tx = self.wallet.create_self_transfer(fee_rate=Decimal("0.9")*mempoolminfee)
339 assert_equal(node.testmempoolaccept([low_fee_tx['hex']])[0]["reject-reason"], "mempool min fee not met")
340 peer.send_and_ping(msg_tx(low_fee_tx['tx']))
341 peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(low_fee_tx['wtxid'], 16))]))
342 node.setmocktime(int(time.time()))
343 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
344 peer.sync_with_ping()
345 assert_equal(peer.tx_getdata_count, 0)
346
347 self.log.info('Check that rejection filter is cleared after new block comes in')
348 self.generate(self.wallet, 1, sync_fun=self.no_op)
349 peer.sync_with_ping()
350 peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(low_fee_tx['wtxid'], 16))]))
351 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
352 peer.wait_for_getdata([int(low_fee_tx['wtxid'], 16)])
353
354 def test_inv_wtxidrelay_mismatch(self):
355 self.log.info("Check that INV messages that don't match the wtxidrelay setting are ignored")
356 node = self.nodes[0]
357 wtxidrelay_on_peer = node.add_p2p_connection(TestP2PConn(wtxidrelay=True))
358 wtxidrelay_off_peer = node.add_p2p_connection(TestP2PConn(wtxidrelay=False))
359 random_tx = self.wallet.create_self_transfer()
360
361 # MSG_TX INV from wtxidrelay=True peer -> mismatch, ignored
362 wtxidrelay_on_peer.send_and_ping(msg_inv([CInv(t=MSG_TX, h=int(random_tx['txid'], 16))]))
363 node.setmocktime(int(time.time()))
364 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
365 wtxidrelay_on_peer.sync_with_ping()
366 assert_equal(wtxidrelay_on_peer.tx_getdata_count, 0)
367
368 # MSG_WTX INV from wtxidrelay=False peer -> mismatch, ignored
369 wtxidrelay_off_peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(random_tx['wtxid'], 16))]))
370 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
371 wtxidrelay_off_peer.sync_with_ping()
372 assert_equal(wtxidrelay_off_peer.tx_getdata_count, 0)
373
374 # MSG_TX INV from wtxidrelay=False peer works
375 wtxidrelay_off_peer.send_and_ping(msg_inv([CInv(t=MSG_TX, h=int(random_tx['txid'], 16))]))
376 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
377 wtxidrelay_off_peer.wait_for_getdata([int(random_tx['txid'], 16)])
378
379 # MSG_WTX INV from wtxidrelay=True peer works
380 wtxidrelay_on_peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(random_tx['wtxid'], 16))]))
381 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
382 wtxidrelay_on_peer.wait_for_getdata([int(random_tx['wtxid'], 16)])
383
384 def run_test(self):
385 self.wallet = MiniWallet(self.nodes[0])
386
387 # Run tests without mocktime that only need one peer-connection first, to avoid restarting the nodes
388 self.test_expiry_fallback()
389 self.test_disconnect_fallback()
390 self.test_notfound_fallback()
391 self.test_preferred_tiebreaker_inv()
392 self.test_preferred_inv(ConnectionType.INBOUND)
393 self.test_preferred_inv(ConnectionType.OUTBOUND)
394 self.test_preferred_inv(ConnectionType.WHITELIST)
395 self.test_txid_inv_delay()
396 self.test_txid_inv_delay(True)
397 self.test_large_inv_batch()
398 self.test_spurious_notfound()
399
400 # Run each test against new bitcoind instances, as setting mocktimes has long-term effects on when
401 # the next trickle relay event happens.
402 for test, with_inbounds in [
403 (self.test_in_flight_max, True),
404 (self.test_inv_block, True),
405 (self.test_tx_requests, True),
406 (self.test_rejects_filter_reset, False),
407 (self.test_inv_wtxidrelay_mismatch, False),
408 ]:
409 self.stop_nodes()
410 self.start_nodes()
411 self.connect_nodes(1, 0)
412 # Setup the p2p connections
413 self.peers = []
414 if with_inbounds:
415 for node in self.nodes:
416 for _ in range(NUM_INBOUND):
417 self.peers.append(node.add_p2p_connection(TestP2PConn()))
418 self.log.info("Nodes are setup with {} incoming connections each".format(NUM_INBOUND))
419 test()
420
421 if __name__ == '__main__':
422 TxDownloadTest(__file__).main()
423