Merge pull request #1068 from sumanjeet0012/fix/kademlia-dht-bootstrap-fallback

Implement fallback mechanism in Kademlia DHT for peer lookup.

Author: seetadev

libp2p/kad_dht/peer_routing.py

@@ -43,6 +43,7 @@
 logger = logging.getLogger("kademlia-example.peer_routing")
 
 MAX_PEER_LOOKUP_ROUNDS = 20  # Maximum number of rounds in peer lookup
+MIN_PEERS_THRESHOLD = 5  # Minimum peers threshold for fallback to connected peers
 
 
 class PeerRouting(IPeerRouting):
@@ -158,6 +159,9 @@ async def find_closest_peers_network(
         Find the closest peers to a target key in the entire network.
 
         Performs an iterative lookup by querying peers for their closest peers.
+        If the routing table has fewer peers than MIN_PEERS_THRESHOLD, it falls
+        back to using connected peers first, then peers from the peerstore if
+        needed, to gather up to 'count' initial query targets.
 
         Returns
         -------
@@ -168,6 +172,47 @@ async def find_closest_peers_network(
         # Start with closest peers from our routing table
         closest_peers = self.routing_table.find_local_closest_peers(target_key, count)
         logger.debug("Local closest peers: %d found", len(closest_peers))
+
+        # Fallback to connected peers and peerstore if routing table has
+        # insufficient peers
+        if len(closest_peers) < MIN_PEERS_THRESHOLD:
+            # First, try connected peers
+            connected_peers = self.host.get_connected_peers()
+            if connected_peers:
+                logger.debug(
+                    "Routing table has insufficient peers (%d < %d), "
+                    "adding %d connected peers",
+                    len(closest_peers),
+                    MIN_PEERS_THRESHOLD,
+                    len(connected_peers),
+                )
+                closest_peers.extend(connected_peers)
+
+            # If still not enough, get peers from peerstore
+            if len(closest_peers) < count:
+                try:
+                    peerstore_peers = self.host.get_peerstore().peer_ids()
+                    # Filter out our own ID and already included peers
+                    local_id = self.host.get_id()
+                    existing_peers = set(closest_peers)
+                    new_peerstore_peers = [
+                        p
+                        for p in peerstore_peers
+                        if p != local_id and p not in existing_peers
+                    ]
+                    if new_peerstore_peers:
+                        logger.debug(
+                            "Adding %d peers from peerstore", len(new_peerstore_peers)
+                        )
+                        closest_peers.extend(new_peerstore_peers)
+                except Exception as e:
+                    logger.debug(f"Failed to get peers from peerstore: {e}")
+
+            # Deduplicate and sort by distance, keeping closest peers
+            closest_peers = sort_peer_ids_by_distance(
+                target_key, list(dict.fromkeys(closest_peers))
+            )[:count]
+
         queried_peers: set[ID] = set()
         rounds = 0
 

newsfragments/905.feature.rst

@@ -0,0 +1 @@
+Added fallback mechanism in Kademlia DHT to use connected peers and peerstore when routing table has insufficient peers.

tests/core/kad_dht/test_unit_peer_routing.py

@@ -30,6 +30,7 @@
 from libp2p.kad_dht.peer_routing import (
     ALPHA,
     MAX_PEER_LOOKUP_ROUNDS,
+    MIN_PEERS_THRESHOLD,
     PROTOCOL_ID,
     PeerRouting,
 )
@@ -145,18 +146,134 @@ async def test_find_peer_not_found(self, peer_routing, mock_host):
             assert result is None
 
     @pytest.mark.trio
-    async def test_find_closest_peers_network_empty_start(self, peer_routing):
-        """Test network search with no local peers."""
+    async def test_find_closest_peers_network_empty_start(
+        self, peer_routing, mock_host
+    ):
+        """Test network search with no local peers and no fallback peers."""
         target_key = b"target_key"
 
         # Mock routing table to return empty list
         with patch.object(
             peer_routing.routing_table, "find_local_closest_peers", return_value=[]
         ):
+            # Mock no connected peers and empty peerstore
+            mock_host.get_connected_peers.return_value = []
+            mock_host.get_peerstore().peer_ids.return_value = []
             result = await peer_routing.find_closest_peers_network(target_key)
 
             assert result == []
 
+    @pytest.mark.trio
+    async def test_find_closest_peers_network_fallback_to_connected_peers(
+        self, peer_routing, mock_host
+    ):
+        """
+        Test that network search falls back to connected peers when routing
+        table has insufficient peers.
+        """
+        target_key = b"target_key"
+
+        # Create few local peers (less than MIN_PEERS_THRESHOLD)
+        local_peers = [create_valid_peer_id(f"local{i}") for i in range(2)]
+
+        # Create connected peers
+        connected_peers = [create_valid_peer_id(f"connected{i}") for i in range(3)]
+
+        # Mock routing table to return insufficient peers
+        with patch.object(
+            peer_routing.routing_table,
+            "find_local_closest_peers",
+            return_value=local_peers,
+        ):
+            # Mock host to return connected peers
+            mock_host.get_connected_peers.return_value = connected_peers
+            # Mock peerstore to return empty (connected peers should be enough)
+            mock_host.get_peerstore().peer_ids.return_value = []
+
+            # Mock _query_peer_for_closest to return empty results
+            with patch.object(peer_routing, "_query_peer_for_closest", return_value=[]):
+                result = await peer_routing.find_closest_peers_network(
+                    target_key, count=10
+                )
+
+                # Should include both local and connected peers
+                assert len(result) > 0
+                # Verify host.get_connected_peers was called
+                mock_host.get_connected_peers.assert_called()
+
+    @pytest.mark.trio
+    async def test_find_closest_peers_network_fallback_to_peerstore(
+        self, peer_routing, mock_host
+    ):
+        """
+        Test that network search falls back to peerstore when connected peers
+        are insufficient.
+        """
+        target_key = b"target_key"
+
+        # Create few local peers (less than MIN_PEERS_THRESHOLD)
+        local_peers = [create_valid_peer_id(f"local{i}") for i in range(2)]
+
+        # Create few connected peers (not enough to reach count)
+        connected_peers = [create_valid_peer_id(f"connected{i}") for i in range(2)]
+
+        # Create peerstore peers
+        peerstore_peers = [create_valid_peer_id(f"peerstore{i}") for i in range(5)]
+
+        # Mock routing table to return insufficient peers
+        with patch.object(
+            peer_routing.routing_table,
+            "find_local_closest_peers",
+            return_value=local_peers,
+        ):
+            # Mock host to return connected peers
+            mock_host.get_connected_peers.return_value = connected_peers
+            # Mock peerstore to return additional peers
+            mock_host.get_peerstore().peer_ids.return_value = peerstore_peers
+
+            # Mock _query_peer_for_closest to return empty results
+            with patch.object(peer_routing, "_query_peer_for_closest", return_value=[]):
+                result = await peer_routing.find_closest_peers_network(
+                    target_key, count=20
+                )
+
+                # Should include peers from all sources
+                assert len(result) > 0
+                # Verify peerstore.peer_ids was called
+                mock_host.get_peerstore().peer_ids.assert_called()
+
+    @pytest.mark.trio
+    async def test_find_closest_peers_network_no_fallback_when_sufficient_peers(
+        self, peer_routing, mock_host
+    ):
+        """
+        Test that fallback is not triggered when routing table has
+        sufficient peers.
+        """
+        target_key = b"target_key"
+
+        # Create enough local peers (>= MIN_PEERS_THRESHOLD)
+        local_peers = [
+            create_valid_peer_id(f"local{i}") for i in range(MIN_PEERS_THRESHOLD + 1)
+        ]
+
+        # Mock routing table to return sufficient peers
+        with patch.object(
+            peer_routing.routing_table,
+            "find_local_closest_peers",
+            return_value=local_peers,
+        ):
+            # Mock _query_peer_for_closest to return empty results
+            with patch.object(peer_routing, "_query_peer_for_closest", return_value=[]):
+                result = await peer_routing.find_closest_peers_network(
+                    target_key, count=10
+                )
+
+                # Should have peers from routing table
+                assert len(result) > 0
+                # Verify host.get_connected_peers was NOT called
+                mock_host.get_connected_peers.assert_not_called()
+
     @pytest.mark.trio
     async def test_find_closest_peers_network_with_peers(self, peer_routing, mock_host):
         """Test network search with some initial peers."""
@@ -171,7 +288,10 @@ async def test_find_closest_peers_network_with_peers(self, peer_routing, mock_ho
             "find_local_closest_peers",
             return_value=initial_peers,
         ):
-            # Mock _query_peer_for_closest to return empty results (no new peers found)
+            # Mock get_connected_peers and peerstore to return empty
+            mock_host.get_connected_peers.return_value = []
+            mock_host.get_peerstore().peer_ids.return_value = []
+            # Mock _query_peer_for_closest to return empty results
             with patch.object(peer_routing, "_query_peer_for_closest", return_value=[]):
                 result = await peer_routing.find_closest_peers_network(
                     target_key, count=5
@@ -182,7 +302,7 @@ async def test_find_closest_peers_network_with_peers(self, peer_routing, mock_ho
                 assert all(peer in initial_peers for peer in result)
 
     @pytest.mark.trio
-    async def test_find_closest_peers_convergence(self, peer_routing):
+    async def test_find_closest_peers_convergence(self, peer_routing, mock_host):
         """Test that network search converges properly."""
         target_key = b"target_key"
 
@@ -195,6 +315,9 @@ async def test_find_closest_peers_convergence(self, peer_routing):
             "find_local_closest_peers",
             return_value=initial_peers,
         ):
+            # Mock get_connected_peers and peerstore to return empty
+            mock_host.get_connected_peers.return_value = []
+            mock_host.get_peerstore().peer_ids.return_value = []
             with patch.object(peer_routing, "_query_peer_for_closest", return_value=[]):
                 with patch(
                     "libp2p.kad_dht.peer_routing.sort_peer_ids_by_distance",
@@ -430,7 +553,7 @@ def test_constants(self):
         assert PROTOCOL_ID == "/ipfs/kad/1.0.0"
 
     @pytest.mark.trio
-    async def test_edge_case_max_rounds_reached(self, peer_routing):
+    async def test_edge_case_max_rounds_reached(self, peer_routing, mock_host):
         """Test that lookup stops after maximum rounds."""
         target_key = b"target_key"
         initial_peers = [create_valid_peer_id("peer1")]
@@ -444,6 +567,9 @@ def mock_query_side_effect(peer, key):
             "find_local_closest_peers",
             return_value=initial_peers,
         ):
+            # Mock get_connected_peers and peerstore to return empty
+            mock_host.get_connected_peers.return_value = []
+            mock_host.get_peerstore().peer_ids.return_value = []
             with patch.object(
                 peer_routing,
                 "_query_peer_for_closest",