Merge pull request #1063 from ethereum/piper/foundation-for-request-and-response-linking

Implement stronger linkage between requests for headers and their responses

Author: pipermerriam

p2p/chain.py

@@ -29,7 +29,10 @@
     BLANK_ROOT_HASH, EMPTY_UNCLE_HASH, GENESIS_BLOCK_NUMBER, GENESIS_PARENT_HASH)
 from eth.chains import AsyncChain
 from eth.db.trie import make_trie_root_and_nodes
-from eth.exceptions import HeaderNotFound, ValidationError
+from eth.exceptions import (
+    HeaderNotFound,
+    ValidationError as EthValidationError,
+)
 from eth.rlp.headers import BlockHeader
 from eth.rlp.receipts import Receipt
 from eth.rlp.transactions import BaseTransaction, BaseTransactionFields
@@ -42,11 +45,10 @@
 from p2p.constants import MAX_REORG_DEPTH, SEAL_CHECK_RANDOM_SAMPLE_RATE
 from p2p.exceptions import NoEligiblePeers, OperationCancelled
 from p2p.p2p_proto import DisconnectReason
-from p2p.peer import BasePeer, ETHPeer, LESPeer, PeerPool, PeerSubscriber
+from p2p.peer import BasePeer, ETHPeer, LESPeer, HeaderRequest, PeerPool, PeerSubscriber
 from p2p.rlp import BlockBody
 from p2p.service import BaseService
 from p2p.utils import (
-    get_block_numbers_for_request,
     get_asyncio_executor,
     Timer,
 )
@@ -205,6 +207,13 @@ async def _sync(self, peer: HeaderRequestingPeer) -> None:
                 self.logger.warn("Timeout waiting for header batch from %s, aborting sync", peer)
                 await peer.disconnect(DisconnectReason.timeout)
                 break
+            except ValueError as err:
+                self.logger.warn(
+                    "Invalid header response sent by peer %s disconnecting: %s",
+                    peer, err,
+                )
+                await peer.disconnect(DisconnectReason.useless_peer)
+                break
 
             if not headers:
                 self.logger.info("Got no new headers from %s, aborting sync", peer)
@@ -220,7 +229,7 @@ async def _sync(self, peer: HeaderRequestingPeer) -> None:
             self.logger.debug("Got new header chain starting at #%d", first.block_number)
             try:
                 await self.chain.coro_validate_chain(headers, self._seal_check_random_sample_rate)
-            except ValidationError as e:
+            except EthValidationError as e:
                 self.logger.warn("Received invalid headers from %s, aborting sync: %s", peer, e)
                 break
             try:
@@ -244,22 +253,39 @@ async def _fetch_missing_headers(
             self, peer: HeaderRequestingPeer, start_at: int) -> Tuple[BlockHeader, ...]:
         """Fetch a batch of headers starting at start_at and return the ones we're missing."""
         self.logger.debug("Fetching chain segment starting at #%d", start_at)
-        peer.request_block_headers(start_at, peer.max_headers_fetch, reverse=False)
+        request = peer.request_block_headers(
+            start_at,
+            peer.max_headers_fetch,
+            skip=0,
+            reverse=False,
+        )
+
         # Pass the peer's token to self.wait() because we want to abort if either we
         # or the peer terminates.
-        headers = list(await self.wait(
+        headers = tuple(await self.wait(
             self._new_headers.get(),
             token=peer.cancel_token,
             timeout=self._reply_timeout))
-        for header in headers.copy():
-            try:
-                await self.wait(self.db.coro_get_block_header_by_hash(header.hash))
-            except HeaderNotFound:
-                break
-            else:
-                self.logger.debug("Discarding %s as we already have it", header)
-                headers.remove(header)
-        return tuple(headers)
+
+        # check that the response headers are a valid match for our
+        # requested headers.
+        request.validate_headers(headers)
+
+        # the inner list comprehension is required to get python to evaluate
+        # the asynchronous comprehension
+        missing_headers = tuple([
+            header
+            for header
+            in headers
+            if not (await self.wait(self.db.coro_header_exists(header.hash)))
+        ])
+        if len(missing_headers) != len(headers):
+            self.logger.debug(
+                "Discarding %d / %d headers that we already have",
+                len(headers) - len(missing_headers),
+                len(headers),
+            )
+        return headers
 
     def _handle_block_headers(self, headers: Tuple[BlockHeader, ...]) -> None:
         if not headers:
@@ -298,9 +324,13 @@ async def _handle_msg(self, peer: HeaderRequestingPeer, cmd: protocol.Command,
 
     async def _handle_get_block_headers(self, peer: LESPeer, msg: Dict[str, Any]) -> None:
         self.logger.debug("Peer %s made header request: %s", peer, msg)
-        query = msg['query']
-        headers = await self._handler.lookup_headers(
-            query.block_number_or_hash, query.max_headers, query.skip, query.reverse)
+        request = HeaderRequest(
+            msg['query'].block_number_or_hash,
+            msg['query'].max_headers,
+            msg['query'].skip,
+            msg['query'].reverse,
+        )
+        headers = await self._handler.lookup_headers(request)
         self.logger.trace("Replying to %s with %d headers", peer, len(headers))
         peer.sub_proto.send_block_headers(headers, buffer_value=0, request_id=msg['request_id'])
 
@@ -581,12 +611,16 @@ async def _handle_block_bodies(self,
     async def _handle_get_block_headers(
             self,
             peer: ETHPeer,
-            header_request: Dict[str, Any]) -> None:
-        self.logger.debug("Peer %s made header request: %s", peer, header_request)
-
-        headers = await self._handler.lookup_headers(
-            header_request['block_number_or_hash'], header_request['max_headers'],
-            header_request['skip'], header_request['reverse'])
+            query: Dict[str, Any]) -> None:
+        self.logger.debug("Peer %s made header request: %s", peer, query)
+        request = HeaderRequest(
+            query['block_number_or_hash'],
+            query['max_headers'],
+            query['skip'],
+            query['reverse'],
+        )
+
+        headers = await self._handler.lookup_headers(request)
         self.logger.trace("Replying to %s with %d headers", peer, len(headers))
         peer.sub_proto.send_block_headers(headers)
 
@@ -697,34 +731,49 @@ async def handle_get_node_data(self, peer: ETHPeer, node_hashes: List[Hash32]) -
         self.logger.trace("Replying to %s with %d trie nodes", peer, len(nodes))
         peer.sub_proto.send_node_data(nodes)
 
-    async def lookup_headers(self, block_number_or_hash: Union[int, bytes], max_headers: int,
-                             skip: int, reverse: bool) -> List[BlockHeader]:
+    async def lookup_headers(self,
+                             request: HeaderRequest) -> Tuple[BlockHeader, ...]:
         """
         Lookup :max_headers: headers starting at :block_number_or_hash:, skipping :skip: items
         between each, in reverse order if :reverse: is True.
         """
-        if isinstance(block_number_or_hash, bytes):
-            try:
-                header = await self.wait(
-                    self.db.coro_get_block_header_by_hash(cast(Hash32, block_number_or_hash)))
-            except HeaderNotFound:
-                self.logger.debug(
-                    "Peer requested starting header %r that is unavailable, returning nothing",
-                    block_number_or_hash)
-                return []
-            block_number = header.block_number
-        elif isinstance(block_number_or_hash, int):
-            block_number = block_number_or_hash
+        try:
+            block_numbers = await self._get_block_numbers_for_request(request)
+        except HeaderNotFound:
+            self.logger.debug(
+                "Peer requested starting header %r that is unavailable, returning nothing",
+                request.block_number_or_hash)
+            block_numbers = tuple()  # type: ignore
+
+        headers: Tuple[BlockHeader, ...] = tuple([
+            header
+            async for header
+            in self._generate_available_headers(block_numbers)
+        ])
+        return headers
+
+    async def _get_block_numbers_for_request(self,
+                                             request: HeaderRequest) -> Tuple[BlockNumber, ...]:
+        """
+        Generate the block numbers for a given `HeaderRequest`.
+        """
+        if isinstance(request.block_number_or_hash, bytes):
+            header = await self.wait(
+                self.db.coro_get_block_header_by_hash(cast(Hash32, request.block_number_or_hash)))
+            return request.generate_block_numbers(header.block_number)
+        elif isinstance(request.block_number_or_hash, int):
+            # We don't need to pass in the block number to
+            # `generate_block_numbers` since the request is based on a numbered
+            # block identifier.
+            return request.generate_block_numbers()
         else:
             raise TypeError(
-                "Unexpected type for 'block_number_or_hash': %s", type(block_number_or_hash))
-
-        block_numbers = get_block_numbers_for_request(block_number, max_headers, skip, reverse)
-        headers = [header async for header in self._generate_available_headers(block_numbers)]
-        return headers
+                "Invariant: unexpected type for 'block_number_or_hash': %s",
+                type(request.block_number_or_hash),
+            )
 
     async def _generate_available_headers(
-            self, block_numbers: Tuple[BlockNumber]) -> AsyncGenerator[BlockHeader, None]:
+            self, block_numbers: Tuple[BlockNumber, ...]) -> AsyncGenerator[BlockHeader, None]:
         """
         Generates the headers requested, halting on the first header that is not locally available.
         """

p2p/eth.py

@@ -3,12 +3,15 @@
     cast,
     List,
     Tuple,
-    Union,
     TYPE_CHECKING
 )
 
 from rlp import sedes
 
+from eth_typing import (
+    BlockIdentifier,
+)
+
 from eth.rlp.headers import BlockHeader
 from eth.rlp.receipts import Receipt
 from eth.rlp.transactions import BaseTransactionFields
@@ -27,6 +30,7 @@
         ChainInfo
     )
 
+
 # Max number of items we can ask for in ETH requests. These are the values used in geth and if we
 # ask for more than this the peers will disconnect from us.
 MAX_STATE_FETCH = 384
@@ -145,8 +149,11 @@ def send_node_data(self, nodes: List[bytes]) -> None:
         header, body = cmd.encode(nodes)
         self.send(header, body)
 
-    def send_get_block_headers(self, block_number_or_hash: Union[int, bytes],
-                               max_headers: int, reverse: bool = True
+    def send_get_block_headers(self,
+                               block_number_or_hash: BlockIdentifier,
+                               max_headers: int,
+                               skip: int,
+                               reverse: bool,
                                ) -> None:
         """Send a GetBlockHeaders msg to the remote.
 
@@ -159,8 +166,6 @@ def send_get_block_headers(self, block_number_or_hash: Union[int, bytes],
                 "Cannot ask for more than {} block headers in a single request".format(
                     MAX_HEADERS_FETCH))
         cmd = GetBlockHeaders(self.cmd_id_offset)
-        # Number of block headers to skip between each item (i.e. step in python APIs).
-        skip = 0
         data = {
             'block_number_or_hash': block_number_or_hash,
             'max_headers': max_headers,
@@ -169,7 +174,7 @@ def send_get_block_headers(self, block_number_or_hash: Union[int, bytes],
         header, body = cmd.encode(data)
         self.send(header, body)
 
-    def send_block_headers(self, headers: List[BlockHeader]) -> None:
+    def send_block_headers(self, headers: Tuple[BlockHeader, ...]) -> None:
         cmd = BlockHeaders(self.cmd_id_offset)
         header, body = cmd.encode(headers)
         self.send(header, body)

p2p/les.py

@@ -9,6 +9,7 @@
 )
 
 from eth_typing import (
+    BlockIdentifier,
     Hash32
 )
 
@@ -314,8 +315,12 @@ def send_get_block_bodies(self, block_hashes: List[bytes], request_id: int) -> N
         header, body = GetBlockBodies(self.cmd_id_offset).encode(data)
         self.send(header, body)
 
-    def send_get_block_headers(self, block_number_or_hash: Union[int, bytes],
-                               max_headers: int, request_id: int, reverse: bool = True
+    def send_get_block_headers(self,
+                               block_number_or_hash: BlockIdentifier,
+                               max_headers: int,
+                               skip: int,
+                               reverse: bool,
+                               request_id: int,
                                ) -> None:
         """Send a GetBlockHeaders msg to the remote.
 
@@ -338,7 +343,7 @@ def send_get_block_headers(self, block_number_or_hash: Union[int, bytes],
         self.send(header, body)
 
     def send_block_headers(
-            self, headers: List[BlockHeader], buffer_value: int, request_id: int) -> None:
+            self, headers: Tuple[BlockHeader, ...], buffer_value: int, request_id: int) -> None:
         data = {
             'request_id': request_id,
             'headers': headers,

p2p/lightchain.py

@@ -324,7 +324,7 @@ async def _get_block_header_by_hash(self, block_hash: Hash32, peer: LESPeer) ->
         self.logger.debug("Fetching header %s from %s", encode_hex(block_hash), peer)
         request_id = gen_request_id()
         max_headers = 1
-        peer.sub_proto.send_get_block_headers(block_hash, max_headers, request_id)
+        peer.sub_proto.send_get_block_headers(block_hash, max_headers, 0, False, request_id)
         reply = await self._wait_for_reply(request_id)
         if not reply['headers']:
             raise HeaderNotFound("Peer {} has no block with hash {}".format(peer, block_hash))