raise PeerConnectionLost when peer dies in request

Author: carver

trinity/protocol/common/managers.py

@@ -17,6 +17,7 @@
     ValidationError,
 )
 
+from p2p.exceptions import PeerConnectionLost
 from p2p.peer import BasePeer, PeerSubscriber
 from p2p.protocol import (
     BaseRequest,
@@ -134,7 +135,7 @@ async def _run(self) -> None:
     async def _handle_msg(self, msg: TResponsePayload) -> None:
         if self.pending_request is None:
             self.logger.debug(
-                "Got unexpected %s payload from %", self.response_msg_name, self._peer
+                "Got unexpected %s payload from %s", self.response_msg_name, self._peer
             )
             return
 
@@ -179,15 +180,24 @@ def _request(self, request: BaseRequest[TRequestPayload]) -> None:
     def _is_pending(self) -> bool:
         return self.pending_request is not None
 
+    async def _cleanup(self) -> None:
+        if self.pending_request is not None:
+            self.logger.debug("Stream shutting down, raising an exception on the pending request")
+            _, future = self.pending_request
+            future.set_exception(PeerConnectionLost("Pending request can't complete: peer is gone"))
+
     def deregister_peer(self, peer: BasePeer) -> None:
         if self.pending_request is not None:
-            self.logger.debug("Peer disconnected, trigger a timeout on the pending request")
+            self.logger.debug("Peer disconnected, raising an exception on the pending request")
             _, future = self.pending_request
-            future.set_exception(TimeoutError("Peer disconnected, simulating inevitable timeout"))
+            future.set_exception(PeerConnectionLost("Pending request can't complete: peer is gone"))
 
     def get_stats(self) -> Tuple[str, str]:
         return (self.response_msg_name, self.response_times.get_stats())
 
+    def __repr__(self) -> str:
+        return f'<ResponseCandidateStream({self._peer!s}, {self.response_msg_type!r})>'
+
 
 class ExchangeManager(Generic[TRequestPayload, TResponsePayload, TResult]):
     _response_stream: ResponseCandidateStream[TRequestPayload, TResponsePayload] = None

trinity/sync/full/chain.py

@@ -12,6 +12,7 @@
     cast,
 )
 
+from cancel_token import OperationCancelled
 from cytoolz import (
     concat,
     merge,
@@ -28,7 +29,7 @@
 from eth.rlp.transactions import BaseTransaction
 
 from p2p import protocol
-from p2p.exceptions import NoEligiblePeers
+from p2p.exceptions import NoEligiblePeers, PeerConnectionLost
 from p2p.protocol import Command
 
 from trinity.db.chain import AsyncChainDB
@@ -204,7 +205,7 @@ async def _get_block_bodies(self,
         """
         Requests the batch of block bodies from the given peer, returning the
         returned block bodies data and the headers for which block bodies were not
-        returned for.
+        returned.
         """
         self.logger.debug("Requesting block bodies for %d headers from %s", len(batch), peer)
         try:
@@ -214,6 +215,16 @@ async def _get_block_bodies(self,
                 "Timed out requesting block bodies for %d headers from %s", len(batch), peer,
             )
             return tuple(), batch
+        except OperationCancelled:
+            self.logger.trace("Pending block bodies call to %r cancelled", peer)
+            return tuple(), batch
+        except PeerConnectionLost:
+            self.logger.debug("Peer went away, cancelling the block body request and moving on...")
+            return tuple(), batch
+        except Exception:
+            self.logger.exception("Unknown error when getting block bodies")
+            return tuple(), batch
+        else:
             self.logger.debug(
                 "Got block bodies for %d headers from %s", len(block_body_bundles), peer,
             )
@@ -306,6 +317,15 @@ async def _get_receipts(self,
                 "Timed out requesting receipts for %d headers from %s", len(batch), peer,
             )
             return tuple(), batch
+        except OperationCancelled:
+            self.logger.trace("Pending receipts call to %r cancelled", peer)
+            return tuple(), batch
+        except PeerConnectionLost:
+            self.logger.debug("Peer went away, cancelling the receipts request and moving on...")
+            return tuple(), batch
+        except Exception:
+            self.logger.exception("Unknown error when getting receipts")
+            return tuple(), batch
         else:
             self.logger.debug(
                 "Got receipts for %d headers from %s", len(receipt_bundles), peer,