refactor: peers_to_request/incomplete_peers -> available_peers

Author: jansegre

hathor/p2p/sync_v2/checkpoints.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from itertools import chain
 from typing import TYPE_CHECKING, Dict, List, NamedTuple, Optional
 
 from structlog import get_logger
@@ -45,11 +44,8 @@ class SyncCheckpoint:
     LC_INTERVAL: int = 5
 
     def __init__(self, manager: 'HathorManager'):
-        # All peers that have all the checkpoints to download
-        self.peers_to_request: List['NodeBlockSync'] = []
-
-        # All peers that we are connected but don't have all the checkpoints
-        self.incomplete_peers: List['NodeBlockSync'] = []
+        # List of peers and latest known height
+        self.available_peers: dict['NodeBlockSync', int] = {}
 
         # Indicate whether the checkpoint sync has been started.
         self._started: bool = False
@@ -122,8 +118,6 @@ def stop(self) -> bool:
             self.log.warn('already stopped')
             return False
         if self.peer_syncing is not None:
-            if self.peer_syncing in self.peers_to_request:
-                self.peers_to_request.remove(self.peer_syncing)
             self.peer_syncing = None
         self._started = False
         self._lc_run.stop()
@@ -192,15 +186,25 @@ def run_sync(self):
         finally:
             self._is_running = False
 
-    def get_peer_to_request(self) -> Optional['NodeBlockSync']:
+    def get_heighest_checkpoint_height(self) -> int:
+        """ Return the height of the heighest checkpoint.
         """
+        return self.manager.checkpoints[-1].height
+
+    def get_peer_to_request(self) -> Optional['NodeBlockSync']:
+        """ Search through the available peers and choose one to sync checkpoints from, if any
+
+        This method will return None if there are no suitable peers to download checkpoints from, which can happen if
+        none of the peers have the checkpoint height that we want or if their connection is not ready yet (or anymore).
         """
         # XXX: we could use a better peer selecting strategy here
-        for peer in self.peers_to_request[:]:
+        cutoff_height = self.get_heighest_checkpoint_height()
+        for peer, height in self.available_peers.items():
             if peer.protocol.state is None:
-                self.peers_to_request.remove(peer)
-            else:
-                return peer
+                continue
+            if height < cutoff_height:
+                continue
+            return peer
         return None
 
     def _run_sync(self):
@@ -242,7 +246,7 @@ def on_sync_error(self):
         """Called by NodeBlockSync when an error occurs (e.g. receive invalid blocks or receive too many blocks)."""
         self.log.debug('sync error')
         # TODO: emit an event and let the networking handle it
-        self.peers_to_request.remove(self.peer_syncing)
+        self.available_peers.remove(self.peer_syncing)
         self.peer_syncing = None
 
     def on_stream_ends(self):
@@ -283,20 +287,10 @@ def on_stream_ends(self):
         # All blocks are downloaded until the last checkpoint.
         # So, we stop the checkpoint sync and mark all connections as checkpoint finished.
         self.log.debug('stop all sync-checkpoints')
-        for peer_sync in chain(self.peers_to_request, self.incomplete_peers):
+        for peer_sync in self.available_peers.keys():
             peer_sync.sync_checkpoints_finished()
         self.stop()
 
     def update_peer_height(self, peer: 'NodeBlockSync', height: int) -> None:
         """Called by NodeBlockSync when we have updated information about a peers height."""
-        if height >= self.manager.checkpoints[-1].height:
-            if peer in self.incomplete_peers:
-                self.incomplete_peers.remove(peer)
-            # This peer has all checkpoints
-            self.peers_to_request.append(peer)
-        else:
-            # XXX: Maybe this isn't possible, but just in case
-            if peer in self.peers_to_request:
-                self.peers_to_request.remove(peer)
-            # This peer does not have all checkpoints
-            self.incomplete_peers.append(peer)
+        self.available_peers[peer] = height

hathor/p2p/sync_v2/manager.py

@@ -265,7 +265,7 @@ def handle_error(self, payload: str) -> None:
         if self.sync_checkpoints.is_started() and self.sync_checkpoints.peer_syncing == self:
             # Oops, we're syncing and we received an error, remove ourselves and let it recover
             self.sync_checkpoints.peer_syncing = None
-            self.sync_checkpoints.peers_to_request.remove(self)
+            self.sync_checkpoints.available_peers.pop(self)
         # forward message to overloaded handle_error:
         self.protocol.handle_error(payload)