light peer: get code recursion -> capped loop

Author: carver

p2p/constants.py

@@ -35,6 +35,7 @@
 
 # Timeout used when waiting for a reply from a remote node.
 REPLY_TIMEOUT = 3
+MAX_REQUEST_ATTEMPTS = 3
 
 # Timeout used when performing the check to ensure peers are on the same side of chain splits as
 # us.

p2p/exceptions.py

@@ -138,6 +138,9 @@ class BadAckMessage(BaseP2PError):
 class BadLESResponse(BaseP2PError):
     """
     Raised when the response to a LES request doesn't contain the data we asked for.
+
+    The peer can be treated as violating protocol. Often, the repurcussion should be
+    disconnection and blacklisting.
     """
     pass
 

p2p/lightchain.py

@@ -41,6 +41,7 @@
 from p2p import protocol
 from p2p.constants import (
     COMPLETION_TIMEOUT,
+    MAX_REQUEST_ATTEMPTS,
     REPLY_TIMEOUT,
 )
 from p2p.p2p_proto import (
@@ -159,16 +160,42 @@ async def get_contract_code(self, block_hash: Hash32, address: Address) -> bytes
         :param address: which contract to look up
 
         :return: bytecode of the contract, ``b''`` if no code is set
-        """
-        peer = cast(LESPeer, self.peer_pool.highest_td_peer)
 
+        :raise NoEligiblePeers: if no peers are available to fulfill the request
+        :raise TimeoutError: if an individual request or the overall process times out
+        """
         # get account for later verification, and
         # to confirm that our highest total difficulty peer has the info
         try:
             account = await self.get_account(block_hash, address)
         except HeaderNotFound as exc:
             raise NoEligiblePeers("Our best peer does not have header %s" % block_hash) from exc
 
+        code_hash = account.code_hash
+
+        for _ in range(MAX_REQUEST_ATTEMPTS):
+            peer = cast(LESPeer, self.peer_pool.highest_td_peer)
+            try:
+                return await self._get_contract_code_from_peer(block_hash, address, peer, code_hash)
+            except BadLESResponse as exc:
+                self.logger.warn("Disconnecting from peer, because: %s", exc)
+                await self.disconnect_peer(peer, DisconnectReason.subprotocol_error)
+                # reattempt after removing this peer from our pool
+
+        raise TimeoutError("Could not get contract code within %d attempts" % MAX_REQUEST_ATTEMPTS)
+
+    async def _get_contract_code_from_peer(
+            self,
+            block_hash: Hash32,
+            address: Address,
+            peer: LESPeer,
+            code_hash: Hash32) -> bytes:
+        """
+        A single attempt to get the contract code from the given peer
+
+        :raise BadLESResponse: if the peer replies with contract code that does not match the
+            account's code hash
+        """
         # request contract code
         request_id = gen_request_id()
         peer.sub_proto.send_get_contract_code(block_hash, keccak(address), request_id)
@@ -180,18 +207,20 @@ async def get_contract_code(self, block_hash: Hash32, address: Address) -> bytes
             bytecode = reply['codes'][0]
 
         # validate bytecode against a proven account
-        if account.code_hash == keccak(bytecode):
+        if code_hash == keccak(bytecode):
             return bytecode
         elif bytecode == b'':
             # TODO disambiguate failure types here, and raise the appropriate exception
             # An (incorrectly) empty bytecode might indicate a bad-acting peer, or it might not
             raise NoEligiblePeers("Our best peer incorrectly responded with an empty code value")
         else:
             # a bad-acting peer sent an invalid non-empty bytecode
-            # disconnect from the peer
-            await self.disconnect_peer(peer, DisconnectReason.subprotocol_error)
-            # try again with another peer
-            return await self.get_contract_code(block_hash, address)
+            raise BadLESResponse("Peer %s sent code %s that did not match hash %s in account %s" % (
+                peer,
+                encode_hex(bytecode),
+                encode_hex(code_hash),
+                encode_hex(address),
+            ))
 
     async def _get_block_header_by_hash(self, peer: LESPeer, block_hash: Hash32) -> BlockHeader:
         self.logger.debug("Fetching header %s from %s", encode_hex(block_hash), peer)