p2p_tx_download.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2019-2021 The Limenka 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 LimenkaTestFramework
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(LimenkaTestFramework):
69 def set_test_params(self):
70 self.num_nodes = 2
71 self.extra_args= [['-datacarriersize=100000', '-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. (Assuming that node 0
120 # announced the tx within the timeout)
121 # The timeout is the sum of
122 # * the worst case until the tx is first requested from an inbound
123 # peer, plus
124 # * the first time it is re-requested from the outbound peer, plus
125 # * 2 seconds to avoid races
126 assert self.nodes[1].getpeerinfo()[0]['inbound'] == False
127 timeout = 2 + NONPREF_PEER_TX_DELAY + GETDATA_TX_INTERVAL
128 self.log.info("Tx should be received at node 1 after {} seconds".format(timeout))
129 self.nodes[0].bumpmocktime(timeout)
130 self.sync_mempools()
131
132 self.nodes[0].setmocktime(0)
133
134 def test_in_flight_max(self):
135 self.log.info("Test that we don't load peers with more than {} transaction requests immediately".format(MAX_PEER_TX_REQUEST_IN_FLIGHT))
136 txids = [i for i in range(MAX_PEER_TX_REQUEST_IN_FLIGHT + 2)]
137
138 p = self.nodes[0].p2ps[0]
139
140 with p2p_lock:
141 p.tx_getdata_count = 0
142
143 mock_time = int(time.time() + 1)
144 self.nodes[0].setmocktime(mock_time)
145 for i in range(MAX_PEER_TX_REQUEST_IN_FLIGHT):
146 p.send_message(msg_inv([CInv(t=MSG_WTX, h=txids[i])]))
147 p.sync_with_ping()
148 mock_time += NONPREF_PEER_TX_DELAY
149 self.nodes[0].setmocktime(mock_time)
150 p.wait_until(lambda: p.tx_getdata_count >= MAX_PEER_TX_REQUEST_IN_FLIGHT)
151 for i in range(MAX_PEER_TX_REQUEST_IN_FLIGHT, len(txids)):
152 p.send_message(msg_inv([CInv(t=MSG_WTX, h=txids[i])]))
153 p.sync_with_ping()
154 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))
155 self.nodes[0].setmocktime(mock_time + NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_TX_DELAY - 1)
156 p.sync_with_ping()
157 with p2p_lock:
158 assert_equal(p.tx_getdata_count, MAX_PEER_TX_REQUEST_IN_FLIGHT)
159 self.log.info("If we wait {} seconds after announcement, we should eventually get more requests".format(NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_TX_DELAY))
160 self.nodes[0].setmocktime(mock_time + NONPREF_PEER_TX_DELAY + OVERLOADED_PEER_TX_DELAY)
161 p.wait_until(lambda: p.tx_getdata_count == len(txids))
162
163 def test_expiry_fallback(self):
164 self.log.info('Check that expiry will select another peer for download')
165 WTXID = 0xffaa
166 peer1 = self.nodes[0].add_p2p_connection(TestP2PConn())
167 peer2 = self.nodes[0].add_p2p_connection(TestP2PConn())
168 for p in [peer1, peer2]:
169 p.send_message(msg_inv([CInv(t=MSG_WTX, h=WTXID)]))
170 # One of the peers is asked for the tx
171 peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1)
172 with p2p_lock:
173 _peer_expiry, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1)
174 assert_equal(peer_fallback.tx_getdata_count, 0)
175 self.nodes[0].setmocktime(int(time.time()) + GETDATA_TX_INTERVAL + 1) # Wait for request to _peer_expiry to expire
176 peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1)
177 self.restart_node(0) # reset mocktime
178
179 def test_disconnect_fallback(self):
180 self.log.info('Check that disconnect will select another peer for download')
181 WTXID = 0xffbb
182 peer1 = self.nodes[0].add_p2p_connection(TestP2PConn())
183 peer2 = self.nodes[0].add_p2p_connection(TestP2PConn())
184 for p in [peer1, peer2]:
185 p.send_message(msg_inv([CInv(t=MSG_WTX, h=WTXID)]))
186 # One of the peers is asked for the tx
187 peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1)
188 with p2p_lock:
189 peer_disconnect, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1)
190 assert_equal(peer_fallback.tx_getdata_count, 0)
191 peer_disconnect.peer_disconnect()
192 peer_disconnect.wait_for_disconnect()
193 peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1)
194
195 def test_notfound_fallback(self):
196 self.log.info('Check that notfounds will select another peer for download immediately')
197 WTXID = 0xffdd
198 peer1 = self.nodes[0].add_p2p_connection(TestP2PConn())
199 peer2 = self.nodes[0].add_p2p_connection(TestP2PConn())
200 for p in [peer1, peer2]:
201 p.send_message(msg_inv([CInv(t=MSG_WTX, h=WTXID)]))
202 # One of the peers is asked for the tx
203 peer2.wait_until(lambda: sum(p.tx_getdata_count for p in [peer1, peer2]) == 1)
204 with p2p_lock:
205 peer_notfound, peer_fallback = (peer1, peer2) if peer1.tx_getdata_count == 1 else (peer2, peer1)
206 assert_equal(peer_fallback.tx_getdata_count, 0)
207 peer_notfound.send_and_ping(msg_notfound(vec=[CInv(MSG_WTX, WTXID)])) # Send notfound, so that fallback peer is selected
208 peer_fallback.wait_until(lambda: peer_fallback.tx_getdata_count >= 1, timeout=1)
209
210 def test_preferred_inv(self, connection_type: ConnectionType):
211 if connection_type == ConnectionType.WHITELIST:
212 self.log.info('Check invs from preferred (whitelisted) peers are downloaded immediately')
213 self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1'])
214 elif connection_type == ConnectionType.OUTBOUND:
215 self.log.info('Check invs from preferred (outbound) peers are downloaded immediately')
216 self.restart_node(0)
217 elif connection_type == ConnectionType.INBOUND:
218 self.log.info('Check invs from non-preferred peers are downloaded after {} s'.format(NONPREF_PEER_TX_DELAY))
219 self.restart_node(0)
220 else:
221 raise Exception("invalid connection_type")
222
223 mock_time = int(time.time() + 1)
224 self.nodes[0].setmocktime(mock_time)
225
226 if connection_type == ConnectionType.OUTBOUND:
227 peer = self.nodes[0].add_outbound_p2p_connection(
228 TestP2PConn(), wait_for_verack=True, p2p_idx=1, connection_type="outbound-full-relay")
229 else:
230 peer = self.nodes[0].add_p2p_connection(TestP2PConn())
231
232 peer.send_message(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
233 peer.sync_with_ping()
234 if connection_type != ConnectionType.INBOUND:
235 peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1)
236 else:
237 with p2p_lock:
238 assert_equal(peer.tx_getdata_count, 0)
239 self.nodes[0].setmocktime(mock_time + NONPREF_PEER_TX_DELAY)
240 peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1)
241
242 def test_preferred_tiebreaker_inv(self):
243 self.log.info("Test that preferred peers are always selected over non-preferred when ready")
244
245 self.restart_node(0)
246 self.nodes[0].setmocktime(int(time.time()))
247
248 # Peer that is immediately asked, but never responds.
249 # This will set us up to have two ready requests, one
250 # of which is preferred and one which is not
251 unresponsive_peer = self.nodes[0].add_outbound_p2p_connection(
252 TestP2PConn(), wait_for_verack=True, p2p_idx=0, connection_type="outbound-full-relay")
253 unresponsive_peer.send_message(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
254 unresponsive_peer.sync_with_ping()
255 unresponsive_peer.wait_until(lambda: unresponsive_peer.tx_getdata_count >= 1, timeout=1)
256
257 # A bunch of incoming (non-preferred) connections that advertise the same tx
258 non_pref_peers = []
259 NUM_INBOUND = 10
260 for _ in range(NUM_INBOUND):
261 non_pref_peers.append(self.nodes[0].add_p2p_connection(TestP2PConn()))
262 non_pref_peers[-1].send_message(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
263 non_pref_peers[-1].sync_with_ping()
264
265 # Check that no request made due to in-flight
266 self.nodes[0].bumpmocktime(NONPREF_PEER_TX_DELAY)
267 with p2p_lock:
268 for peer in non_pref_peers:
269 assert_equal(peer.tx_getdata_count, 0)
270
271 # Now add another outbound (preferred) which is immediately ready for consideration
272 # upon advertisement
273 pref_peer = self.nodes[0].add_outbound_p2p_connection(
274 TestP2PConn(), wait_for_verack=True, p2p_idx=1, connection_type="outbound-full-relay")
275 pref_peer.send_message(msg_inv([CInv(t=MSG_WTX, h=0xff00ff00)]))
276 pref_peer.sync_with_ping()
277
278 assert_equal(len(self.nodes[0].getpeerinfo()), NUM_INBOUND + 2)
279
280 # Still have to wait for in-flight to timeout
281 with p2p_lock:
282 assert_equal(pref_peer.tx_getdata_count, 0)
283
284 # Timeout in-flight
285 self.nodes[0].bumpmocktime(GETDATA_TX_INTERVAL - NONPREF_PEER_TX_DELAY)
286
287 # Preferred peers are *always* selected next if ready
288 pref_peer.wait_until(lambda: pref_peer.tx_getdata_count >= 1, timeout=10)
289
290 # And none for non-preferred
291 for non_pref_peer in non_pref_peers:
292 with p2p_lock:
293 assert_equal(non_pref_peer.tx_getdata_count, 0)
294
295 def test_txid_inv_delay(self, glob_wtxid=False):
296 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))
297 self.restart_node(0, extra_args=['-whitelist=noban@127.0.0.1'])
298 mock_time = int(time.time() + 1)
299 self.nodes[0].setmocktime(mock_time)
300 peer = self.nodes[0].add_p2p_connection(TestP2PConn(wtxidrelay=False))
301 if glob_wtxid:
302 # Add a second wtxid-relay connection otherwise TXID_RELAY_DELAY is waived in
303 # lack of wtxid-relay peers
304 self.nodes[0].add_p2p_connection(TestP2PConn(wtxidrelay=True))
305 peer.send_message(msg_inv([CInv(t=MSG_TX, h=0xff11ff11)]))
306 peer.sync_with_ping()
307 with p2p_lock:
308 assert_equal(peer.tx_getdata_count, 0 if glob_wtxid else 1)
309 self.nodes[0].setmocktime(mock_time + TXID_RELAY_DELAY)
310 peer.wait_until(lambda: peer.tx_getdata_count >= 1, timeout=1)
311
312 def test_large_inv_batch(self):
313 self.log.info('Test how large inv batches are handled with relay permission')
314 self.restart_node(0, extra_args=['-whitelist=relay@127.0.0.1'])
315 peer = self.nodes[0].add_p2p_connection(TestP2PConn())
316 peer.send_message(msg_inv([CInv(t=MSG_WTX, h=wtxid) for wtxid in range(MAX_PEER_TX_ANNOUNCEMENTS + 1)]))
317 peer.wait_until(lambda: peer.tx_getdata_count == MAX_PEER_TX_ANNOUNCEMENTS + 1)
318
319 self.log.info('Test how large inv batches are handled without relay permission')
320 self.restart_node(0)
321 peer = self.nodes[0].add_p2p_connection(TestP2PConn())
322 peer.send_message(msg_inv([CInv(t=MSG_WTX, h=wtxid) for wtxid in range(MAX_PEER_TX_ANNOUNCEMENTS + 1)]))
323 peer.wait_until(lambda: peer.tx_getdata_count == MAX_PEER_TX_ANNOUNCEMENTS)
324 peer.sync_with_ping()
325
326 def test_spurious_notfound(self):
327 self.log.info('Check that spurious notfound is ignored')
328 self.nodes[0].p2ps[0].send_message(msg_notfound(vec=[CInv(MSG_TX, 1)]))
329
330 def test_rejects_filter_reset(self):
331 self.log.info('Check that rejected tx is not requested again')
332 node = self.nodes[0]
333 fill_mempool(self, node, tx_sync_fun=self.no_op)
334 self.wallet.rescan_utxos()
335 mempoolminfee = node.getmempoolinfo()['mempoolminfee']
336 peer = node.add_p2p_connection(TestP2PConn())
337 low_fee_tx = self.wallet.create_self_transfer(fee_rate=Decimal("0.9")*mempoolminfee)
338 assert_equal(node.testmempoolaccept([low_fee_tx['hex']])[0]["reject-reason"], "mempool min fee not met")
339 peer.send_and_ping(msg_tx(low_fee_tx['tx']))
340 peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(low_fee_tx['wtxid'], 16))]))
341 node.setmocktime(int(time.time()))
342 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
343 peer.sync_with_ping()
344 assert_equal(peer.tx_getdata_count, 0)
345
346 self.log.info('Check that rejection filter is cleared after new block comes in')
347 self.generate(self.wallet, 1, sync_fun=self.no_op)
348 peer.sync_with_ping()
349 peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(low_fee_tx['wtxid'], 16))]))
350 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
351 peer.wait_for_getdata([int(low_fee_tx['wtxid'], 16)])
352
353 def test_inv_wtxidrelay_mismatch(self):
354 self.log.info("Check that INV messages that don't match the wtxidrelay setting are ignored")
355 node = self.nodes[0]
356 wtxidrelay_on_peer = node.add_p2p_connection(TestP2PConn(wtxidrelay=True))
357 wtxidrelay_off_peer = node.add_p2p_connection(TestP2PConn(wtxidrelay=False))
358 random_tx = self.wallet.create_self_transfer()
359
360 # MSG_TX INV from wtxidrelay=True peer -> mismatch, ignored
361 wtxidrelay_on_peer.send_and_ping(msg_inv([CInv(t=MSG_TX, h=int(random_tx['txid'], 16))]))
362 node.setmocktime(int(time.time()))
363 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
364 wtxidrelay_on_peer.sync_with_ping()
365 assert_equal(wtxidrelay_on_peer.tx_getdata_count, 0)
366
367 # MSG_WTX INV from wtxidrelay=False peer -> mismatch, ignored
368 wtxidrelay_off_peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(random_tx['wtxid'], 16))]))
369 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
370 wtxidrelay_off_peer.sync_with_ping()
371 assert_equal(wtxidrelay_off_peer.tx_getdata_count, 0)
372
373 # MSG_TX INV from wtxidrelay=False peer works
374 wtxidrelay_off_peer.send_and_ping(msg_inv([CInv(t=MSG_TX, h=int(random_tx['txid'], 16))]))
375 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
376 wtxidrelay_off_peer.wait_for_getdata([int(random_tx['txid'], 16)])
377
378 # MSG_WTX INV from wtxidrelay=True peer works
379 wtxidrelay_on_peer.send_and_ping(msg_inv([CInv(t=MSG_WTX, h=int(random_tx['wtxid'], 16))]))
380 node.bumpmocktime(MAX_GETDATA_INBOUND_WAIT)
381 wtxidrelay_on_peer.wait_for_getdata([int(random_tx['wtxid'], 16)])
382
383 def run_test(self):
384 self.wallet = MiniWallet(self.nodes[0])
385
386 # Run tests without mocktime that only need one peer-connection first, to avoid restarting the nodes
387 self.test_expiry_fallback()
388 self.test_disconnect_fallback()
389 self.test_notfound_fallback()
390 self.test_preferred_tiebreaker_inv()
391 self.test_preferred_inv(ConnectionType.INBOUND)
392 self.test_preferred_inv(ConnectionType.OUTBOUND)
393 self.test_preferred_inv(ConnectionType.WHITELIST)
394 self.test_txid_inv_delay()
395 self.test_txid_inv_delay(True)
396 self.test_large_inv_batch()
397 self.test_spurious_notfound()
398
399 # Run each test against new limenkad instances, as setting mocktimes has long-term effects on when
400 # the next trickle relay event happens.
401 for test, with_inbounds in [
402 (self.test_in_flight_max, True),
403 (self.test_inv_block, True),
404 (self.test_tx_requests, True),
405 (self.test_rejects_filter_reset, False),
406 (self.test_inv_wtxidrelay_mismatch, False),
407 ]:
408 self.stop_nodes()
409 self.start_nodes()
410 self.connect_nodes(1, 0)
411 # Setup the p2p connections
412 self.peers = []
413 if with_inbounds:
414 for node in self.nodes:
415 for _ in range(NUM_INBOUND):
416 self.peers.append(node.add_p2p_connection(TestP2PConn()))
417 self.log.info("Nodes are setup with {} incoming connections each".format(NUM_INBOUND))
418 test()
419
420 if __name__ == '__main__':
421 TxDownloadTest(__file__).main()
422