Merge pull request #1249 from carver/reliability-messaging-cleanup

Improve fast-sync reliability & clean up messages

Author: carver

p2p/peer.py

@@ -497,7 +497,7 @@ def decrypt_body(self, data: bytes, body_size: int) -> bytes:
         self.ingress_mac.update(sxor(self.mac_enc(fmac_seed), fmac_seed))
         expected_frame_mac = self.ingress_mac.digest()[:MAC_LEN]
         if not bytes_eq(expected_frame_mac, frame_mac):
-            raise DecryptionError('Invalid frame mac: expected %s, got %s'.format(
+            raise DecryptionError('Invalid frame mac: expected {}, got {}'.format(
                 expected_frame_mac, frame_mac))
         return self.aes_dec.update(frame_ciphertext)[:body_size]
 

p2p/protocol.py

@@ -93,7 +93,7 @@ def decode_payload(self, rlp_data: bytes) -> PayloadType:
             data = rlp.decode(rlp_data, sedes=decoder, recursive_cache=True)
         except rlp.DecodingError as err:
             raise MalformedMessage(
-                "Malformed %s message: %r".format(type(self).__name__, err)
+                "Malformed {} message: {!r}".format(type(self).__name__, err)
             ) from err
 
         if isinstance(self.structure, sedes.CountableList):

p2p/service.py

@@ -207,8 +207,7 @@ async def cleanup(self) -> None:
 
         self.events.cleaned_up.set()
 
-    async def cancel(self) -> None:
-        """Trigger the CancelToken and wait for the cleaned_up event to be set."""
+    def cancel_nowait(self) -> None:
         if self.is_cancelled:
             self.logger.warning("Tried to cancel %s, but it was already cancelled", self)
             return
@@ -218,6 +217,11 @@ async def cancel(self) -> None:
         self.logger.debug("Cancelling %s", self)
         self.events.cancelled.set()
         self.cancel_token.trigger()
+
+    async def cancel(self) -> None:
+        """Trigger the CancelToken and wait for the cleaned_up event to be set."""
+        self.cancel_nowait()
+
         try:
             await asyncio.wait_for(
                 self.events.cleaned_up.wait(), timeout=self._wait_until_finished_timeout)

trinity/protocol/common/managers.py

@@ -51,7 +51,7 @@ def subscription_msg_types(self) -> Set[Type[Command]]:
 
     msg_queue_maxsize = 100
 
-    response_timout: float = ROUND_TRIP_TIMEOUT
+    response_timeout: float = ROUND_TRIP_TIMEOUT
 
     pending_request: Tuple[float, 'asyncio.Future[TResponsePayload]'] = None
 
@@ -80,7 +80,7 @@ async def payload_candidates(
         candidates will stop arriving.
         """
         if timeout is None:
-            timeout = self.response_timout
+            timeout = self.response_timeout
 
         start_at = time.perf_counter()
 
@@ -174,13 +174,13 @@ def _is_pending(self) -> bool:
 
     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")
+            self.logger.debug("Stream %r shutting down, cancelling the pending request", self)
             _, 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, raising an exception on the pending request")
+            self.logger.debug("Peer stream %r shutting down, cancelling the pending request", self)
             _, future = self.pending_request
             future.set_exception(PeerConnectionLost("Pending request can't complete: peer is gone"))
 
@@ -201,6 +201,9 @@ def __init__(
         self._response_command_type = listening_for
 
     async def launch_service(self) -> None:
+        if self._cancel_token.triggered:
+            raise PeerConnectionLost("Peer %s is gone. Ignoring new requests to it" % self._peer)
+
         self._response_stream = ResponseCandidateStream(
             self._peer,
             self._response_command_type,

trinity/protocol/eth/peer.py

@@ -19,13 +19,13 @@
     NewBlock,
     Status,
 )
-from trinity.protocol.eth import constants
+from .constants import MAX_HEADERS_FETCH
 from .proto import ETHProtocol
 from .handlers import ETHExchangeHandler
 
 
 class ETHPeer(BasePeer):
-    max_headers_fetch = constants.MAX_HEADERS_FETCH
+    max_headers_fetch = MAX_HEADERS_FETCH
 
     _supported_sub_protocols = [ETHProtocol]
     sub_proto: ETHProtocol = None

trinity/protocol/les/peer.py

@@ -23,7 +23,7 @@
     Status,
     StatusV2,
 )
-from trinity.protocol.les.constants import (
+from .constants import (
     MAX_HEADERS_FETCH,
 )
 from .proto import (

trinity/sync/common/chain.py

@@ -36,7 +36,6 @@
 from trinity.protocol.les.peer import LESPeer
 from trinity.utils.datastructures import TaskQueue
 
-
 HeaderRequestingPeer = Union[LESPeer, ETHPeer]
 
 
@@ -49,8 +48,6 @@ class BaseHeaderChainSyncer(BaseService, PeerSubscriber):
     """
     # We'll only sync if we are connected to at least min_peers_to_sync.
     min_peers_to_sync = 1
-    # Post-processing steps can exit out of sync (for example, fast sync) by triggering this token:
-    complete_token = None
     # TODO: Instead of a fixed timeout, we should use a variable one that gets adjusted based on
     # the round-trip times from our download requests.
     _reply_timeout = 60
@@ -64,12 +61,7 @@ def __init__(self,
                  db: AsyncHeaderDB,
                  peer_pool: PeerPool,
                  token: CancelToken = None) -> None:
-        self.complete_token = CancelToken('trinity.sync.common.BaseHeaderChainSyncer.SyncCompleted')
-        if token is None:
-            master_service_token = self.complete_token
-        else:
-            master_service_token = token.chain(self.complete_token)
-        super().__init__(master_service_token)
+        super().__init__(token)
         self.chain = chain
         self.db = db
         self.peer_pool = peer_pool
@@ -146,7 +138,7 @@ async def sync(self, peer: HeaderRequestingPeer) -> None:
         try:
             await self._sync(peer)
         except OperationCancelled as e:
-            self.logger.info("Sync with %s aborted: %s", peer, e)
+            self.logger.info("Sync with %s was shut down: %s", peer, e)
         finally:
             self._syncing = False
 
@@ -199,7 +191,23 @@ async def _sync(self, peer: HeaderRequestingPeer) -> None:
                 break
 
             if not headers:
-                self.logger.info("Got no new headers from %s, aborting sync", peer)
+                if last_received_header is None:
+                    request_parent = head
+                else:
+                    request_parent = last_received_header
+                if head_td < peer.head_td:
+                    # peer claims to have a better header, but didn't return it. Boot peer
+                    # TODO ... also blacklist, because it keeps trying to reconnect
+                    self.logger.warning(
+                        "%s announced difficulty %s, but didn't return any headers after %r@%s",
+                        peer,
+                        peer.head_td,
+                        request_parent,
+                        head_td,
+                    )
+                    await peer.disconnect(DisconnectReason.subprotocol_error)
+                else:
+                    self.logger.info("Got no new headers from %s, aborting sync", peer)
                 break
 
             first = headers[0]
@@ -235,24 +243,17 @@ async def _sync(self, peer: HeaderRequestingPeer) -> None:
                 await peer.disconnect(DisconnectReason.subprotocol_error)
                 break
 
+            for header in headers:
+                head_td += header.difficulty
+
             # Setting the latest header hash for the peer, before queuing header processing tasks
             self._target_header_hash = peer.head_hash
 
-            unrequested_headers = tuple(h for h in headers if h not in self.header_queue)
-            await self.header_queue.add(unrequested_headers)
+            new_headers = tuple(h for h in headers if h not in self.header_queue)
+            await self.header_queue.add(new_headers)
             last_received_header = headers[-1]
             start_at = last_received_header.block_number + 1
 
-        # erase any pending tasks, to restart on next _sync() run
-        try:
-            batch_id, pending_tasks = self.header_queue.get_nowait()
-        except asyncio.QueueFull:
-            # nothing pending, continue
-            pass
-        else:
-            # fully remove pending tasks from queue
-            self.header_queue.complete(batch_id, pending_tasks)
-
     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."""
@@ -273,12 +274,15 @@ async def get_missing_tail(self: 'BaseHeaderChainSyncer',
                                    ) -> AsyncGenerator[BlockHeader, None]:
             iter_headers = iter(headers)
             for header in iter_headers:
-                is_missing = not await self.wait(self.db.coro_header_exists(header.hash))
-                if is_missing:
+                if header in self.header_queue:
+                    self.logger.debug("Discarding header that is already queued: %s", header)
+                    continue
+                is_present = await self.wait(self.db.coro_header_exists(header.hash))
+                if is_present:
+                    self.logger.debug("Discarding header that we already have: %s", header)
+                else:
                     yield header
                     break
-                else:
-                    self.logger.debug("Discarding header that we already have: %s", header)
 
             for header in iter_headers:
                 yield header

trinity/sync/full/chain.py

@@ -47,7 +47,6 @@
 from trinity.sync.common.chain import BaseHeaderChainSyncer
 from trinity.utils.timer import Timer
 
-
 HeaderRequestingPeer = Union[LESPeer, ETHPeer]
 # (ReceiptBundle, (Receipt, (root_hash, receipt_trie_data))
 ReceiptBundle = Tuple[Tuple[Receipt, ...], Tuple[Hash32, Dict[Hash32, bytes]]]
@@ -67,6 +66,9 @@ class FastChainSyncer(BaseHeaderChainSyncer):
     highest TD, at which point we must run the StateDownloader to fetch the state for our chain
     head.
     """
+    NO_PEER_RETRY_PAUSE = 5
+    """If no peers are available for downloading the chain data, retry after this many seconds"""
+
     db: AsyncChainDB
 
     subscription_msg_types: Set[Type[Command]] = {
@@ -92,8 +94,16 @@ async def _load_and_process_headers(self) -> None:
             # TODO implement the maximum task size at each step instead of this magic number
             max_headers = min((MAX_BODIES_FETCH, MAX_RECEIPTS_FETCH)) * 4
             batch_id, headers = await self.header_queue.get(max_headers)
-            await self._process_headers(headers)
-            self.header_queue.complete(batch_id, headers)
+            try:
+                await self._process_headers(headers)
+            except NoEligiblePeers:
+                self.logger.info(
+                    f"No available peers to sync with, retrying in {self.NO_PEER_RETRY_PAUSE}s"
+                )
+                self.header_queue.complete(batch_id, tuple())
+                await self.sleep(self.NO_PEER_RETRY_PAUSE)
+            else:
+                self.header_queue.complete(batch_id, headers)
 
     async def _calculate_td(self, headers: Tuple[BlockHeader, ...]) -> int:
         """Return the score (total difficulty) of the last header in the given list.
@@ -148,7 +158,7 @@ async def _process_headers(self, headers: Tuple[BlockHeader, ...]) -> None:
         # order to see if the sync is completed. Instead we just check that we have the peer's
         # head_hash in our chain.
         if await self.wait(self.db.coro_header_exists(target_hash)):
-            self.complete_token.trigger()
+            self.cancel_nowait()
 
     async def _download_block_bodies(self,
                                      target_td: int,
@@ -295,6 +305,9 @@ async def _download_receipts(self,
             receipt_bundles = tuple(concat(all_receipt_bundles))
             headers = tuple(concat(all_missing_headers))
 
+            if len(receipt_bundles) == 0:
+                continue
+
             # process all of the returned receipts, storing their trie data
             # dicts in the database
             receipts, trie_roots_and_data_dicts = zip(*receipt_bundles)