Correctly handle disconnections from maxed out peers

A maxed out peer will disconnect during the P2P handshake by
sending us a disconnect msg with the TOO_MANY_PEERS reason.  We must
detect that and wait a while before attempting to connect again.

Closes #1167

p2p/abc.py

@@ -201,6 +201,7 @@ class TransportAPI(ABC):
     session: SessionAPI
     remote: NodeAPI
     read_state: TransportState
+    logger: ExtendedDebugLogger
 
     @property
     @abstractmethod

p2p/commands.py

@@ -94,6 +94,9 @@ class BaseCommand(CommandAPI[TCommandPayload]):
     def __init__(self, payload: TCommandPayload) -> None:
         self.payload = payload
 
+    def __repr__(self) -> str:
+        return f"{self.__class__}(payload={self.payload})"
+
     def encode(self, cmd_id: int, snappy_support: bool) -> MessageAPI:
         raw_payload_data = self.serialization_codec.encode(self.payload)
 

p2p/exceptions.py

@@ -46,9 +46,9 @@ class HandshakeFailure(BaseP2PError):
     pass
 
 
-class TooManyPeersFailure(HandshakeFailure):
+class HandshakeFailureTooManyPeers(HandshakeFailure):
     """
-    The remote disconnected from us because it has too many peers
+    The remote disconnected from us during a handshake because it has too many peers
     """
     pass
 

p2p/handshake.py

@@ -30,8 +30,10 @@
 )
 from p2p.connection import Connection
 from p2p.constants import DEVP2P_V5
+from p2p.disconnect import DisconnectReason
 from p2p.exceptions import (
     HandshakeFailure,
+    HandshakeFailureTooManyPeers,
     NoMatchingPeerCapabilities,
 )
 from p2p.multiplexer import (
@@ -40,6 +42,7 @@
 )
 from p2p.p2p_proto import (
     DevP2PReceipt,
+    Disconnect,
     Hello,
     HelloPayload,
     BaseP2PProtocol,
@@ -127,6 +130,10 @@ async def _do_p2p_handshake(transport: TransportAPI,
     # The base `p2p` protocol handshake directly streams the messages as it has
     # strict requirements about receiving the `Hello` message first.
     async for _, cmd in stream_transport_messages(transport, base_protocol, token=token):
+        if isinstance(cmd, Disconnect):
+            if cmd.payload == DisconnectReason.TOO_MANY_PEERS:
+                raise HandshakeFailureTooManyPeers(f"Peer disconnected because it is already full")
+
         if not isinstance(cmd, Hello):
             raise HandshakeFailure(
                 f"First message across the DevP2P connection must be a Hello "
@@ -304,6 +311,7 @@ async def dial_out(remote: NodeAPI,
         token,
     )
 
+    transport.logger.debug2("Initiating p2p handshake with %s", remote)
     try:
         multiplexer, devp2p_receipt, protocol_receipts = await negotiate_protocol_handshakes(
             transport=transport,
@@ -320,6 +328,7 @@ async def dial_out(remote: NodeAPI,
         await asyncio.sleep(0)
         raise
 
+    transport.logger.debug2("Completed p2p handshake with %s", remote)
     connection = Connection(
         multiplexer=multiplexer,
         devp2p_receipt=devp2p_receipt,

p2p/tracking/connection.py

@@ -9,7 +9,7 @@
 from p2p.exceptions import (
     BaseP2PError,
     HandshakeFailure,
-    TooManyPeersFailure,
+    HandshakeFailureTooManyPeers,
 )
 
 
@@ -23,7 +23,7 @@ def register_error(exception: Type[BaseP2PError], timeout_seconds: int) -> None:
 
 
 register_error(HandshakeFailure, 10)  # 10 seconds
-register_error(TooManyPeersFailure, 60)  # one minute
+register_error(HandshakeFailureTooManyPeers, 60)  # one minute
 
 
 def get_timeout_for_failure(failure: BaseP2PError) -> int:

p2p/transport.py

@@ -110,7 +110,7 @@ async def connect(cls,
                       remote: NodeAPI,
                       private_key: datatypes.PrivateKey,
                       token: CancelToken) -> TransportAPI:
-        """Perform the auth and P2P handshakes with the given remote.
+        """Perform the auth handshake with the given remote.
 
         Return an instance of the given peer_class (must be a subclass of
         BasePeer) connected to that remote in case both handshakes are
@@ -122,6 +122,7 @@ async def connect(cls,
         handshake or if none of the sub-protocols supported by us is also
         supported by the remote.
         """
+        cls.logger.debug2("Initiating auth handshake with %s", remote)
         try:
             (aes_secret,
              mac_secret,
@@ -133,6 +134,7 @@ async def connect(cls,
         except (ConnectionRefusedError, OSError) as e:
             raise UnreachablePeer(f"Can't reach {remote!r}") from e
 
+        cls.logger.debug2("Completed auth handshake with %s", remote)
         return cls(
             remote=remote,
             private_key=private_key,
@@ -203,7 +205,7 @@ async def receive_connection(cls,
         ip, socket, *_ = peername
         remote_address = Address(ip, socket)
 
-        cls.logger.debug("Receiving handshake from %s", remote_address)
+        cls.logger.debug("Receiving auth handshake from %s", remote_address)
 
         initiator_remote = Node(initiator_pubkey, remote_address)
 

scripts/peer.py

@@ -15,11 +15,12 @@
 )
 
 from eth_typing import BlockNumber
+from eth_utils import DEBUG2_LEVEL_NUM
+
 from eth.chains.mainnet import MainnetChain, MAINNET_GENESIS_HEADER, MAINNET_VM_CONFIGURATION
 from eth.chains.ropsten import RopstenChain, ROPSTEN_GENESIS_HEADER, ROPSTEN_VM_CONFIGURATION
 from eth.db.atomic import AtomicDB
 from eth.db.backends.memory import MemoryDB
-from eth_utils import DEBUG2_LEVEL_NUM
 
 from p2p import ecies
 from p2p.constants import DEVP2P_V5
@@ -35,14 +36,19 @@
 
 
 def _main() -> None:
-    logging.basicConfig(level=DEBUG2_LEVEL_NUM, format='%(asctime)s %(levelname)s: %(message)s')
-
     parser = argparse.ArgumentParser()
     parser.add_argument('-enode', type=str, help="The enode we should connect to", required=True)
     parser.add_argument('-mainnet', action='store_true')
     parser.add_argument('-light', action='store_true', help="Connect as a light node")
+    parser.add_argument('-debug', action="store_true")
     args = parser.parse_args()
 
+    log_level = logging.INFO
+    if args.debug:
+        log_level = DEBUG2_LEVEL_NUM
+    logging.basicConfig(
+        level=log_level, format='%(asctime)s %(levelname)s: %(message)s', datefmt='%H:%M:%S')
+
     peer_class: Union[Type[ETHPeer], Type[LESPeer]]
     pool_class: Union[Type[ETHPeerPool], Type[LESPeerPool]]