PR feedback

Author: pipermerriam

p2p/chain.py

@@ -305,6 +305,9 @@ async def _handle_msg(self, peer: HeaderRequestingPeer, cmd: protocol.Command,
         elif isinstance(cmd, les.GetBlockHeaders):
             msg = cast(Dict[str, Any], msg)
             await self._handle_get_block_headers(cast(LESPeer, peer), msg)
+        elif isinstance(cmd, les.BlockHeaders):
+            # `BlockHeaders` messages are handled at the peer level.
+            pass
         else:
             self.logger.debug("Ignoring %s message from %s", cmd, peer)
 
@@ -533,6 +536,9 @@ async def _handle_msg(self, peer: HeaderRequestingPeer, cmd: protocol.Command,
             await self._handle_new_block(peer, cast(Dict[str, Any], msg))
         elif isinstance(cmd, eth.GetBlockHeaders):
             await self._handle_get_block_headers(peer, cast(Dict[str, Any], msg))
+        elif isinstance(cmd, eth.BlockHeaders):
+            # `BlockHeaders` messages are handled at the peer level.
+            pass
         elif isinstance(cmd, eth.GetBlockBodies):
             # Only serve up to eth.MAX_BODIES_FETCH items in every request.
             block_hashes = cast(List[Hash32], msg)[:eth.MAX_BODIES_FETCH]

p2p/eth.py

@@ -74,7 +74,7 @@ class GetBlockHeaders(Command):
 
 
 class HeaderRequest(BaseHeaderRequest):
-    MAX_HEADERS_FETCH = MAX_HEADERS_FETCH
+    max_size = MAX_HEADERS_FETCH
 
     def __init__(self,
                  block_number_or_hash: BlockIdentifier,

p2p/les.py

@@ -170,7 +170,7 @@ class GetBlockHeadersQuery(rlp.Serializable):
 class HeaderRequest(BaseHeaderRequest):
     request_id: int
 
-    MAX_HEADERS_FETCH = MAX_HEADERS_FETCH
+    max_size = MAX_HEADERS_FETCH
 
     def __init__(self,
                  block_number_or_hash: BlockIdentifier,

p2p/peer.py

@@ -381,6 +381,24 @@ def handle_sub_proto_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgT
         else:
             self.logger.warn("Peer %s has no subscribers, discarding %s msg", self, cmd)
 
+        cmd_type = type(cmd)
+
+        if cmd_type in self.pending_requests:
+            request, future = self.pending_requests[cmd_type]
+            try:
+                request.validate_response(msg)
+            except ValidationError as err:
+                self.logger.debug(
+                    "Response validation failure for pending %s request from peer %s: %s",
+                    cmd_type.__name__,
+                    self,
+                    err,
+                )
+                pass
+            else:
+                future.set_result(msg)
+                self.pending_requests.pop(cmd_type)
+
     def process_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgType) -> None:
         if cmd.is_base_protocol:
             self.handle_p2p_msg(cmd, msg)
@@ -531,21 +549,11 @@ def max_headers_fetch(self) -> int:
         return les.MAX_HEADERS_FETCH
 
     def handle_sub_proto_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgType) -> None:
-        cmd_type = type(cmd)
-
         if isinstance(cmd, les.Announce):
             self.head_info = cmd.as_head_info(msg)
             self.head_td = self.head_info.total_difficulty
             self.head_hash = self.head_info.block_hash
-        elif cmd_type in self.pending_requests:
-            request, future = self.pending_requests[cmd_type]
-            try:
-                request.validate_response(msg)
-            except ValidationError:
-                pass
-            else:
-                future.set_result(msg)
-                self.pending_requests.pop(cmd_type)
+
         super().handle_sub_proto_msg(cmd, msg)
 
     async def send_sub_proto_handshake(self) -> None:
@@ -603,14 +611,26 @@ def request_block_headers(self,
 
     async def wait_for_block_headers(self, request: les.HeaderRequest) -> Tuple[BlockHeader, ...]:
         future: 'asyncio.Future[protocol._DecodedMsgType]' = asyncio.Future()
+        if les.BlockHeaders in self.pending_requests:
+            # the `finally` block below should prevent this from happening, but
+            # were two requests to the same peer to be fired off at the same
+            # time, this will prevent us from overwriting the first one.
+            raise ValueError(
+                "There is already a pending `BlockHeaders` request for peer {0}".format(self)
+            )
         self.pending_requests[les.BlockHeaders] = cast(
             Tuple[protocol.BaseRequest, 'asyncio.Future[protocol._DecodedMsgType]'],
             (request, future),
         )
-        response = cast(
-            Dict[str, Any],
-            await self.wait(future, timeout=self._response_timeout),
-        )
+        try:
+            response = cast(
+                Dict[str, Any],
+                await self.wait(future, timeout=self._response_timeout),
+            )
+        finally:
+            # We always want to be sure that this method cleans up the
+            # `pending_requests` so that we don't end up in a situation.
+            self.pending_requests.pop(les.BlockHeaders, None)
         return cast(Tuple[BlockHeader, ...], response['headers'])
 
     async def get_block_headers(self,
@@ -631,8 +651,6 @@ def max_headers_fetch(self) -> int:
         return eth.MAX_HEADERS_FETCH
 
     def handle_sub_proto_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgType) -> None:
-        cmd_type = type(cmd)
-
         if isinstance(cmd, eth.NewBlock):
             msg = cast(Dict[str, Any], msg)
             header, _, _ = msg['block']
@@ -642,16 +660,6 @@ def handle_sub_proto_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgT
                 self.head_hash = actual_head
                 self.head_td = actual_td
 
-        if cmd_type in self.pending_requests:
-            request, future = self.pending_requests[cmd_type]
-            try:
-                request.validate_response(msg)
-            except ValidationError:
-                pass
-            else:
-                future.set_result(msg)
-                self.pending_requests.pop(cmd_type)
-
         super().handle_sub_proto_msg(cmd, msg)
 
     async def send_sub_proto_handshake(self) -> None:

p2p/protocol.py

@@ -140,7 +140,7 @@ class BaseHeaderRequest(BaseRequest):
 
     @property
     @abstractmethod
-    def MAX_HEADERS_FETCH(self) -> int:
+    def max_size(self) -> int:
         pass
 
     def generate_block_numbers(self,
@@ -158,7 +158,7 @@ def generate_block_numbers(self,
         elif block_number is None:
             block_number = cast(BlockNumber, self.block_number_or_hash)
 
-        max_headers = min(self.MAX_HEADERS_FETCH, self.max_headers)
+        max_headers = min(self.max_size, self.max_headers)
 
         # inline import until this module is moved to `trinity`
         from trinity.utils.headers import sequence_builder

tests/p2p/test_header_request_object.py

@@ -15,7 +15,7 @@
 
 
 class HeaderRequest(BaseHeaderRequest):
-    MAX_HEADERS_FETCH = 192
+    max_size = 192
 
     def __init__(self,
                  block_number_or_hash,