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_message(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          for id in range(NUM_PEERS):
 77              peers.append(node.add_outbound_p2p_connection(P2PStaller(stall_block), p2p_idx=id, connection_type="outbound-full-relay"))
 78              peers[-1].block_store = block_dict
 79              peers[-1].send_message(headers_message)
 80  
 81          # Need to wait until 1023 blocks are received - the magic total bytes number is a workaround in lack of an rpc
 82          # returning the number of downloaded (but not connected) blocks.
 83          bytes_recv = 172761 if not self.options.v2transport else 169692
 84          self.wait_until(lambda: self.total_bytes_recv_for_blocks() == bytes_recv)
 85  
 86          self.all_sync_send_with_ping(peers)
 87          # If there was a peer marked for stalling, it would get disconnected
 88          self.mocktime = int(time.time()) + 3
 89          node.setmocktime(self.mocktime)
 90          self.all_sync_send_with_ping(peers)
 91          assert_equal(node.num_test_p2p_connections(), NUM_PEERS)
 92  
 93          self.log.info("Check that increasing the window beyond 1024 blocks triggers stalling logic")
 94          headers_message.headers = [CBlockHeader(b) for b in blocks]
 95          with node.assert_debug_log(expected_msgs=['Stall started']):
 96              for p in peers:
 97                  p.send_message(headers_message)
 98              self.all_sync_send_with_ping(peers)
 99  
100          self.log.info("Check that the stalling peer is disconnected after 2 seconds")
101          self.mocktime += 3
102          node.setmocktime(self.mocktime)
103          peers[0].wait_for_disconnect()
104          assert_equal(node.num_test_p2p_connections(), NUM_PEERS - 1)
105          self.wait_until(lambda: self.is_block_requested(peers, stall_block))
106          # Make sure that SendMessages() is invoked, which assigns the missing block
107          # to another peer and starts the stalling logic for them
108          self.all_sync_send_with_ping(peers)
109  
110          self.log.info("Check that the stalling timeout gets doubled to 4 seconds for the next staller")
111          # No disconnect after just 3 seconds
112          self.mocktime += 3
113          node.setmocktime(self.mocktime)
114          self.all_sync_send_with_ping(peers)
115          assert_equal(node.num_test_p2p_connections(), NUM_PEERS - 1)
116  
117          self.mocktime += 2
118          node.setmocktime(self.mocktime)
119          self.wait_until(lambda: sum(x.is_connected for x in node.p2ps) == NUM_PEERS - 2)
120          self.wait_until(lambda: self.is_block_requested(peers, stall_block))
121          self.all_sync_send_with_ping(peers)
122  
123          self.log.info("Check that the stalling timeout gets doubled to 8 seconds for the next staller")
124          # No disconnect after just 7 seconds
125          self.mocktime += 7
126          node.setmocktime(self.mocktime)
127          self.all_sync_send_with_ping(peers)
128          assert_equal(node.num_test_p2p_connections(), NUM_PEERS - 2)
129  
130          self.mocktime += 2
131          node.setmocktime(self.mocktime)
132          self.wait_until(lambda: sum(x.is_connected for x in node.p2ps) == NUM_PEERS - 3)
133          self.wait_until(lambda: self.is_block_requested(peers, stall_block))
134          self.all_sync_send_with_ping(peers)
135  
136          self.log.info("Provide the withheld block and check that stalling timeout gets reduced back to 2 seconds")
137          with node.assert_debug_log(expected_msgs=['Decreased stalling timeout to 2 seconds']):
138              for p in peers:
139                  if p.is_connected and (stall_block in p.getdata_requests):
140                      p.send_message(msg_block(block_dict[stall_block]))
141  
142          self.log.info("Check that all outstanding blocks get connected")
143          self.wait_until(lambda: node.getblockcount() == NUM_BLOCKS)
144  
145      def total_bytes_recv_for_blocks(self):
146          total = 0
147          for info in self.nodes[0].getpeerinfo():
148              if ("block" in info["bytesrecv_per_msg"].keys()):
149                  total += info["bytesrecv_per_msg"]["block"]
150          return total
151  
152      def all_sync_send_with_ping(self, peers):
153          for p in peers:
154              if p.is_connected:
155                  p.sync_with_ping()
156  
157      def is_block_requested(self, peers, hash):
158          for p in peers:
159              if p.is_connected and (hash in p.getdata_requests):
160                  return True
161          return False
162  
163  
164  if __name__ == '__main__':
165      P2PIBDStallingTest().main()