Merge pull request #1845 from lbryio/stuck_peers_fixup

fix active peers not being removed after closing + parsing of data stream

lbrynet/blob_exchange/downloader.py

@@ -33,10 +33,14 @@ class BlobDownloader:
         async def _request_blob():
             if blob.get_is_verified():
                 return
-            success, keep_connection = await request_blob(
-                self.loop, blob, peer.address, peer.tcp_port, self.config.peer_connect_timeout,
-                self.config.blob_download_timeout
-            )
+            try:
+                success, keep_connection = await request_blob(
+                    self.loop, blob, peer.address, peer.tcp_port, self.config.peer_connect_timeout,
+                    self.config.blob_download_timeout
+                )
+            finally:
+                if peer in self.active_connections:
+                    self.active_connections.pop(peer)
             if not keep_connection and peer not in self.ignored:
                 self.ignored.add(peer)
                 log.debug("drop peer %s:%i", peer.address, peer.tcp_port)

lbrynet/blob_exchange/serialization.py

@@ -146,32 +146,26 @@ def _parse_blob_response(response_msg: bytes) -> typing.Tuple[typing.Optional[ty
     #   <blob bytes>
     #   <json><blob bytes>
 
-    extra_data = b''
-    response = None
     curr_pos = 0
     while True:
         next_close_paren = response_msg.find(b'}', curr_pos)
         if next_close_paren == -1:
-            break
+            return None, response_msg
         curr_pos = next_close_paren + 1
         try:
             response = json.loads(response_msg[:curr_pos])
-            if not isinstance(response, dict):
-                raise ValueError()
-            for key in response.keys():
-                if key not in [
-                        BlobPaymentAddressResponse.key,
-                        BlobAvailabilityResponse.key,
-                        BlobPriceResponse.key,
-                        BlobDownloadResponse.key]:
-                    raise ValueError()
-            extra_data = response_msg[curr_pos:]
-            break
         except ValueError:
-            response = None
-    if response is None:
-        extra_data = response_msg
-    return response, extra_data
+            continue
+        possible_response_keys = {
+                    BlobPaymentAddressResponse.key,
+                    BlobAvailabilityResponse.key,
+                    BlobPriceResponse.key,
+                    BlobDownloadResponse.key
+        }
+        if isinstance(response, dict) and response.keys():
+            if set(response.keys()).issubset(possible_response_keys):
+                return response, response_msg[curr_pos:]
+        return None, response_msg
 
 
 class BlobRequest: