feat(p2p): Run periodic bootstrap discovery

Author: msbrogli

hathor/builder/cli_builder.py

@@ -258,10 +258,10 @@ def create_manager(self, reactor: Reactor) -> HathorManager:
             dns_hosts.extend(self._args.dns)
 
         if dns_hosts:
-            self.manager.add_peer_discovery(DNSPeerDiscovery(dns_hosts))
+            p2p_manager.add_peer_discovery(DNSPeerDiscovery(dns_hosts))
 
         if self._args.bootstrap:
-            self.manager.add_peer_discovery(BootstrapPeerDiscovery(self._args.bootstrap))
+            p2p_manager.add_peer_discovery(BootstrapPeerDiscovery(self._args.bootstrap))
 
         if self._args.test_mode_tx_weight:
             _set_test_mode(TestMode.TEST_TX_WEIGHT)
@@ -277,7 +277,7 @@ def create_manager(self, reactor: Reactor) -> HathorManager:
             self.log.warn('--memory-indexes is implied for memory storage or JSON storage')
 
         for description in self._args.listen:
-            self.manager.add_listen_address(description)
+            p2p_manager.add_listen_address(description)
 
         if self._args.peer_id_blacklist:
             self.log.info('with peer id blacklist', blacklist=self._args.peer_id_blacklist)

hathor/manager.py

@@ -44,7 +44,6 @@
 from hathor.feature_activation.feature_service import FeatureService
 from hathor.mining import BlockTemplate, BlockTemplates
 from hathor.p2p.manager import ConnectionsManager
-from hathor.p2p.peer_discovery import PeerDiscovery
 from hathor.p2p.peer_id import PeerId
 from hathor.p2p.protocol import HathorProtocol
 from hathor.profiler import get_cpu_profiler
@@ -180,8 +179,6 @@ def __init__(self,
 
         self.consensus_algorithm = consensus_algorithm
 
-        self.peer_discoveries: list[PeerDiscovery] = []
-
         self.connections = p2p_manager
 
         self.metrics = Metrics(
@@ -206,9 +203,6 @@ def __init__(self,
         # Thread pool used to resolve pow when sending tokens
         self.pow_thread_pool = ThreadPool(minthreads=0, maxthreads=settings.MAX_POW_THREADS, name='Pow thread pool')
 
-        # List of addresses to listen for new connections (eg: [tcp:8000])
-        self.listen_addresses: list[str] = []
-
         # Full verification execute all validations for transactions and blocks when initializing the node
         # Can be activated on the command line with --full-verification
         self._full_verification = full_verification
@@ -274,7 +268,6 @@ def start(self) -> None:
         self.state = self.NodeState.INITIALIZING
         self.pubsub.publish(HathorEvents.MANAGER_ON_START)
         self._event_manager.load_started()
-        self.connections.start()
         self.pow_thread_pool.start()
 
         # Disable get transaction lock when initializing components
@@ -297,10 +290,7 @@ def start(self) -> None:
         # Metric starts to capture data
         self.metrics.start()
 
-        for description in self.listen_addresses:
-            self.listen(description)
-
-        self.do_discovery()
+        self.connections.start()
 
         self.start_time = time.time()
 
@@ -350,13 +340,6 @@ def stop(self) -> Deferred:
 
         return defer.DeferredList(waits)
 
-    def do_discovery(self) -> None:
-        """
-        Do a discovery and connect on all discovery strategies.
-        """
-        for peer_discovery in self.peer_discoveries:
-            peer_discovery.discover_and_connect(self.connections.connect_to)
-
     def start_profiler(self, *, reset: bool = False) -> None:
         """
         Start profiler. It can be activated from a web resource, as well.
@@ -686,12 +669,6 @@ def _sync_v2_resume_validations(self) -> None:
             self.sync_v2_step_validations(depended_final_txs, quiet=True)
             self.log.debug('pending validations finished')
 
-    def add_listen_address(self, addr: str) -> None:
-        self.listen_addresses.append(addr)
-
-    def add_peer_discovery(self, peer_discovery: PeerDiscovery) -> None:
-        self.peer_discoveries.append(peer_discovery)
-
     def get_new_tx_parents(self, timestamp: Optional[float] = None) -> list[VertexId]:
         """Select which transactions will be confirmed by a new transaction.
 
@@ -1128,17 +1105,6 @@ def _log_if_feature_is_active(self, block: Block, feature: Feature) -> None:
         if self._feature_service.is_feature_active(block=block, feature=feature):
             self.log.info('Feature is ACTIVE for block', feature=feature.value, block_height=block.get_height())
 
-    def listen(self, description: str, use_ssl: Optional[bool] = None) -> None:
-        endpoint = self.connections.listen(description, use_ssl)
-        # XXX: endpoint: IStreamServerEndpoint does not intrinsically have a port, but in practice all concrete cases
-        #      that we have will have a _port attribute
-        port = getattr(endpoint, '_port', None)
-
-        if self.hostname:
-            proto, _, _ = description.partition(':')
-            address = '{}://{}:{}'.format(proto, self.hostname, port)
-            self.my_peer.entrypoints.append(address)
-
     def has_sync_version_capability(self) -> bool:
         return settings.CAPABILITY_SYNC_VERSION in self.capabilities
 

hathor/p2p/manager.py

@@ -24,6 +24,7 @@
 
 from hathor.conf import HathorSettings
 from hathor.p2p.netfilter.factory import NetfilterFactory
+from hathor.p2p.peer_discovery import PeerDiscovery
 from hathor.p2p.peer_id import PeerId
 from hathor.p2p.peer_storage import PeerStorage
 from hathor.p2p.protocol import HathorProtocol
@@ -114,6 +115,12 @@ def __init__(self,
 
         self.network = network
 
+        # List of addresses to listen for new connections (eg: [tcp:8000])
+        self.listen_addresses: list[str] = []
+
+        # List of peer discovery methods.
+        self.peer_discoveries: list[PeerDiscovery] = []
+
         # Options
         self.localhost_only = False
 
@@ -202,6 +209,21 @@ def set_manager(self, manager: 'HathorManager') -> None:
             indexes.enable_deps_index()
             indexes.enable_mempool_index()
 
+    def add_listen_address(self, addr: str) -> None:
+        """Add address to listen for incoming connections."""
+        self.listen_addresses.append(addr)
+
+    def add_peer_discovery(self, peer_discovery: PeerDiscovery) -> None:
+        """Add a peer discovery method."""
+        self.peer_discoveries.append(peer_discovery)
+
+    def do_discovery(self) -> None:
+        """
+        Do a discovery and connect on all discovery strategies.
+        """
+        for peer_discovery in self.peer_discoveries:
+            peer_discovery.discover_and_connect(self.connect_to)
+
     def disable_rate_limiter(self) -> None:
         """Disable global rate limiter."""
         self.rate_limiter.unset_limit(self.GlobalRateLimiter.SEND_TIPS)
@@ -217,9 +239,15 @@ def enable_rate_limiter(self, max_hits: int = 16, window_seconds: float = 1) ->
     def start(self) -> None:
         self.lc_reconnect.start(5, now=False)
         self.lc_sync_update.start(self.lc_sync_update_interval, now=False)
+
         if settings.ENABLE_PEER_WHITELIST:
             self._start_whitelist_reconnect()
 
+        for description in self.listen_addresses:
+            self.listen(description)
+
+        self.do_discovery()
+
     def _start_whitelist_reconnect(self) -> None:
         # The deferred returned by the LoopingCall start method
         # executes when the looping call stops running
@@ -442,7 +470,7 @@ def on_receive_peer(self, peer: PeerId, origin: Optional[ReadyState] = None) ->
         if peer.id == self.my_peer.id:
             return
         peer = self.received_peer_storage.add_or_merge(peer)
-        self.connect_to_if_not_connected(peer, 0)
+        self.connect_to_if_not_connected(peer, int(self.reactor.seconds()))
 
     def peers_cleanup(self) -> None:
         """Clean up aged peers."""
@@ -469,7 +497,7 @@ def reconnect_to_all(self) -> None:
         # when we have no connected peers left, run the discovery process again
         assert self.manager is not None
         if len(self.connected_peers) < 1:
-            self.manager.do_discovery()
+            self.do_discovery()
         now = int(self.reactor.seconds())
         # We need to use list() here because the dict might change inside connect_to_if_not_connected
         # when the peer is disconnected and without entrypoint
@@ -614,7 +642,7 @@ def connect_to(self, description: str, peer: Optional[PeerId] = None, use_ssl: O
         )
 
     def listen(self, description: str, use_ssl: Optional[bool] = None) -> IStreamServerEndpoint:
-        """ Start to listen to new connection according to the description.
+        """ Start to listen for new connection according to the description.
 
         If `ssl` is True, then the connection will be wraped by a TLS.
 
@@ -641,6 +669,17 @@ def listen(self, description: str, use_ssl: Optional[bool] = None) -> IStreamSer
 
         self.log.info('listen on', endpoint=description)
         endpoint.listen(factory)
+
+        # XXX: endpoint: IStreamServerEndpoint does not intrinsically have a port, but in practice all concrete cases
+        #      that we have will have a _port attribute
+        port = getattr(endpoint, '_port', None)
+        assert self.manager is not None
+        if self.manager.hostname and port is not None:
+            proto, _, _ = description.partition(':')
+            address = '{}://{}:{}'.format(proto, self.manager.hostname, port)
+            assert self.manager.my_peer is not None
+            self.manager.my_peer.entrypoints.append(address)
+
         return endpoint
 
     def get_connection_to_drop(self, protocol: HathorProtocol) -> HathorProtocol: