p2p_ibd_stalling.py
1 #!/usr/bin/env python3 2 # Copyright (c) 2022- 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 stalling logic during IBD 7 """ 8 9 import time 10 11 from test_framework.blocktools import ( 12 create_block, 13 create_coinbase 14 ) 15 from test_framework.messages import ( 16 MSG_BLOCK, 17 MSG_TYPE_MASK, 18 ) 19 from test_framework.p2p import ( 20 CBlockHeader, 21 msg_block, 22 msg_headers, 23 P2PDataStore, 24 ) 25 from test_framework.test_framework import BitcoinTestFramework 26 from test_framework.util import ( 27 assert_equal, 28 ) 29 30 31 class P2PStaller(P2PDataStore): 32 def __init__(self, stall_block): 33 self.stall_block = stall_block 34 super().__init__() 35 36 def on_getdata(self, message): 37 for inv in message.inv: 38 self.getdata_requests.append(inv.hash) 39 if (inv.type & MSG_TYPE_MASK) == MSG_BLOCK: 40 if (inv.hash != self.stall_block): 41 self.send_without_ping(msg_block(self.block_store[inv.hash])) 42 43 def on_getheaders(self, message): 44 pass 45 46 47 class P2PIBDStallingTest(BitcoinTestFramework): 48 def set_test_params(self): 49 self.setup_clean_chain = True 50 self.num_nodes = 1 51 52 def run_test(self): 53 NUM_BLOCKS = 1025 54 NUM_PEERS = 4 55 node = self.nodes[0] 56 tip = int(node.getbestblockhash(), 16) 57 blocks = [] 58 height = 1 59 block_time = node.getblock(node.getbestblockhash())['time'] + 1 60 self.log.info("Prepare blocks without sending them to the node") 61 block_dict = {} 62 for _ in range(NUM_BLOCKS): 63 blocks.append(create_block(tip, create_coinbase(height), block_time)) 64 blocks[-1].solve() 65 tip = blocks[-1].sha256 66 block_time += 1 67 height += 1 68 block_dict[blocks[-1].sha256] = blocks[-1] 69 stall_block = blocks[0].sha256 70 71 headers_message = msg_headers() 72 headers_message.headers = [CBlockHeader(b) for b in blocks[:NUM_BLOCKS-1]] 73 peers = [] 74 75 self.log.info("Check that a staller does not get disconnected if the 1024 block lookahead buffer is filled") 76 self.mocktime = int(time.time()) + 1 77 node.setmocktime(self.mocktime) 78 for id in range(NUM_PEERS): 79 peers.append(node.add_outbound_p2p_connection(P2PStaller(stall_block), p2p_idx=id, connection_type="outbound-full-relay")) 80 peers[-1].block_store = block_dict 81 peers[-1].send_and_ping(headers_message) 82 83 # Need to wait until 1023 blocks are received - the magic total bytes number is a workaround in lack of an rpc 84 # returning the number of downloaded (but not connected) blocks. 85 bytes_recv = 172761 if not self.options.v2transport else 169692 86 self.wait_until(lambda: self.total_bytes_recv_for_blocks() == bytes_recv) 87 88 self.all_sync_send_with_ping(peers) 89 # If there was a peer marked for stalling, it would get disconnected 90 self.mocktime += 3 91 node.setmocktime(self.mocktime) 92 self.all_sync_send_with_ping(peers) 93 assert_equal(node.num_test_p2p_connections(), NUM_PEERS) 94 95 self.log.info("Check that increasing the window beyond 1024 blocks triggers stalling logic") 96 headers_message.headers = [CBlockHeader(b) for b in blocks] 97 with node.assert_debug_log(expected_msgs=['Stall started']): 98 for p in peers: 99 p.send_without_ping(headers_message) 100 self.all_sync_send_with_ping(peers) 101 102 self.log.info("Check that the stalling peer is disconnected after 2 seconds") 103 self.mocktime += 3 104 node.setmocktime(self.mocktime) 105 peers[0].wait_for_disconnect() 106 assert_equal(node.num_test_p2p_connections(), NUM_PEERS - 1) 107 self.wait_until(lambda: self.is_block_requested(peers, stall_block)) 108 # Make sure that SendMessages() is invoked, which assigns the missing block 109 # to another peer and starts the stalling logic for them 110 self.all_sync_send_with_ping(peers) 111 112 self.log.info("Check that the stalling timeout gets doubled to 4 seconds for the next staller") 113 # No disconnect after just 3 seconds 114 self.mocktime += 3 115 node.setmocktime(self.mocktime) 116 self.all_sync_send_with_ping(peers) 117 assert_equal(node.num_test_p2p_connections(), NUM_PEERS - 1) 118 119 self.mocktime += 2 120 node.setmocktime(self.mocktime) 121 self.wait_until(lambda: sum(x.is_connected for x in node.p2ps) == NUM_PEERS - 2) 122 self.wait_until(lambda: self.is_block_requested(peers, stall_block)) 123 self.all_sync_send_with_ping(peers) 124 125 self.log.info("Check that the stalling timeout gets doubled to 8 seconds for the next staller") 126 # No disconnect after just 7 seconds 127 self.mocktime += 7 128 node.setmocktime(self.mocktime) 129 self.all_sync_send_with_ping(peers) 130 assert_equal(node.num_test_p2p_connections(), NUM_PEERS - 2) 131 132 self.mocktime += 2 133 node.setmocktime(self.mocktime) 134 self.wait_until(lambda: sum(x.is_connected for x in node.p2ps) == NUM_PEERS - 3) 135 self.wait_until(lambda: self.is_block_requested(peers, stall_block)) 136 self.all_sync_send_with_ping(peers) 137 138 self.log.info("Provide the withheld block and check that stalling timeout gets reduced back to 2 seconds") 139 with node.assert_debug_log(expected_msgs=['Decreased stalling timeout to 2 seconds']): 140 for p in peers: 141 if p.is_connected and (stall_block in p.getdata_requests): 142 p.send_without_ping(msg_block(block_dict[stall_block])) 143 144 self.log.info("Check that all outstanding blocks get connected") 145 self.wait_until(lambda: node.getblockcount() == NUM_BLOCKS) 146 147 def total_bytes_recv_for_blocks(self): 148 total = 0 149 for info in self.nodes[0].getpeerinfo(): 150 if ("block" in info["bytesrecv_per_msg"].keys()): 151 total += info["bytesrecv_per_msg"]["block"] 152 return total 153 154 def all_sync_send_with_ping(self, peers): 155 for p in peers: 156 if p.is_connected: 157 p.sync_with_ping() 158 159 def is_block_requested(self, peers, hash): 160 for p in peers: 161 if p.is_connected and (hash in p.getdata_requests): 162 return True 163 return False 164 165 166 if __name__ == '__main__': 167 P2PIBDStallingTest(__file__).main()