Bug [lbryio#2070] where blob_get RPC timed out.

lbry/blob_exchange/downloader.py

@@ -3,6 +3,7 @@
 import logging
 from lbry.utils import cache_concurrent
 from lbry.blob_exchange.client import request_blob
+from lbry.dht.node import prime_node_with_peer_addresses
 if typing.TYPE_CHECKING:
     from lbry.conf import Config
     from lbry.dht.node import Node
@@ -133,11 +134,12 @@ def close(self):
             protocol.close()
 
 
-async def download_blob(loop, config: 'Config', blob_manager: 'BlobManager', node: 'Node',
+async def download_blob(loop, config: 'Config', blob_manager: 'BlobManager', dht_node: 'Node',
                         blob_hash: str) -> 'AbstractBlob':
     search_queue = asyncio.Queue(loop=loop, maxsize=config.max_connections_per_download)
     search_queue.put_nowait(blob_hash)
-    peer_queue, accumulate_task = node.accumulate_peers(search_queue)
+    peer_queue, accumulate_task = dht_node.accumulate_peers(search_queue)
+    await prime_node_with_peer_addresses(config.fixed_peers, peer_queue)
     downloader = BlobDownloader(loop, config, blob_manager, peer_queue)
     try:
         return await downloader.download_blob(blob_hash)

lbry/dht/node.py

@@ -258,3 +258,13 @@ def accumulate_peers(self, search_queue: asyncio.Queue,
                          ) -> typing.Tuple[asyncio.Queue, asyncio.Task]:
         queue = peer_queue or asyncio.Queue(loop=self.loop)
         return queue, self.loop.create_task(self._accumulate_peers_for_value(search_queue, queue))
+
+
+async def prime_node_with_peer_addresses(peer_list: typing.List[typing.Tuple[str, int]], peer_queue: asyncio.Queue):
+    try:
+        peer_address_list = [(await resolve_host(url, port, proto='tcp'), port) for url, port in peer_list]
+        kademlia_peer_list = [make_kademlia_peer(None, address, None, tcp_port=port, allow_localhost=True)
+                              for address, port in peer_address_list]
+        peer_queue.put_nowait(kademlia_peer_list)
+    except asyncio.CancelledError:
+        pass

lbry/stream/downloader.py

@@ -3,9 +3,9 @@
 import logging
 import binascii
 
-from lbry.dht.peer import make_kademlia_peer
+from lbry.dht.node import prime_node_with_peer_addresses
 from lbry.error import DownloadSDTimeoutError
-from lbry.utils import resolve_host, lru_cache_concurrent
+from lbry.utils import lru_cache_concurrent
 from lbry.stream.descriptor import StreamDescriptor
 from lbry.blob_exchange.downloader import BlobDownloader
 if typing.TYPE_CHECKING:
@@ -48,26 +48,19 @@ async def cached_read_blob(blob_info: 'BlobInfo') -> bytes:
         self.cached_read_blob = cached_read_blob
 
     async def add_fixed_peers(self):
-        def _delayed_add_fixed_peers():
+        async def _delayed_add_fixed_peers(fixed_peers_delay):
+            await asyncio.sleep(fixed_peers_delay)
+            await prime_node_with_peer_addresses(self.config.fixed_peers, self.peer_queue)
             self.added_fixed_peers = True
-            self.peer_queue.put_nowait([
-                make_kademlia_peer(None, address, None, tcp_port=port, allow_localhost=True)
-                for address, port in addresses
-            ])
 
         if not self.config.fixed_peers:
             return
-        addresses = [
-            (await resolve_host(url, port, proto='tcp'), port)
-            for url, port in self.config.fixed_peers
-        ]
         if 'dht' in self.config.components_to_skip or not self.node or not \
                 len(self.node.protocol.routing_table.get_peers()) > 0:
             self.fixed_peers_delay = 0.0
         else:
             self.fixed_peers_delay = self.config.fixed_peer_delay
-
-        self.fixed_peers_handle = self.loop.call_later(self.fixed_peers_delay, _delayed_add_fixed_peers)
+        self.fixed_peers_handle = asyncio.create_task(_delayed_add_fixed_peers(self.fixed_peers_delay))
 
     async def load_descriptor(self, connection_id: int = 0):
         # download or get the sd blob