Return immediately from download when zero connected peers (#1733)

frontend/src/pages/squeak/Squeak.js

@@ -97,6 +97,10 @@ export default function SqueakPage() {
     setWaitingForDownloadAncestors(true);
     downloadSqueakRequest(hash, (response) => {
       setWaitingForDownloadAncestors(false);
+      if (response.getDownloadResult().getNumberPeers() === 0) {
+        alert("Unable to download because zero connected peers.");
+        return;
+      }
       setAncestorSqueaks(null); // Temporary fix until component unmounts correcyly
       getAncestorSqueaks(hash);
     });
@@ -107,6 +111,10 @@ export default function SqueakPage() {
     setWaitingForDownloadReplies(true);
     downloadRepliesRequest(hash, (response) => {
       setWaitingForDownloadReplies(false);
+      if (response.getDownloadResult().getNumberPeers() === 0) {
+        alert("Unable to download because zero connected peers.");
+        return;
+      }
       setReplySqueaks(null); // Temporary fix until component unmounts correcyly
       getReplySqueaks(hash, SQUEAKS_PER_PAGE, null);
     });

frontend/src/pages/squeakaddress/SqueakAddress.js

@@ -88,6 +88,10 @@ export default function SqueakAddressPage() {
     setWaitingForDownload(true);
     downloadAddressSqueaksRequest(address, (response) => {
       setWaitingForDownload(false);
+      if (response.getDownloadResult().getNumberPeers() === 0) {
+        alert("Unable to download because zero connected peers.");
+        return;
+      }
       setSqueaks([]);
       getSqueaks(address, SQUEAKS_PER_PAGE, null);
     });

proto/squeak_admin.proto

@@ -843,6 +843,12 @@ message DownloadResult {
 
     /// Number of squeaks requested
     int32 number_requested = 2;
+
+    /// Number of peers downloaded from
+    int32 number_peers = 3;
+
+    /// Download time in milliseconds.
+    int32 elapsed_time_ms = 4;
 }
 
 message DownloadSqueaksReply {

squeaknode/admin/messages.py

@@ -263,6 +263,8 @@ def download_result_to_message(download_result: DownloadResult) -> squeak_admin_
     return squeak_admin_pb2.DownloadResult(
         number_downloaded=download_result.number_downloaded,
         number_requested=download_result.number_requested,
+        number_peers=download_result.number_peers,
+        elapsed_time_ms=download_result.elapsed_time_ms,
     )
 
 

squeaknode/admin/squeak_admin_server_handler.py

@@ -648,8 +648,12 @@ class SqueakAdminServerHandler(object):
         squeak_hash = bytes.fromhex(squeak_hash_str)
         logger.info(
             "Handle download replies for hash: {}".format(squeak_hash_str))
-        self.squeak_controller.download_replies(squeak_hash)
-        return squeak_admin_pb2.DownloadRepliesReply()
+        download_result = self.squeak_controller.download_replies(squeak_hash)
+        logger.info("Download result: {}".format(download_result))
+        download_result_msg = download_result_to_message(download_result)
+        return squeak_admin_pb2.DownloadRepliesReply(
+            download_result=download_result_msg,
+        )
 
     def handle_download_address_squeaks(self, request):
         squeak_address = request.address

squeaknode/core/download_result.py

@@ -26,4 +26,5 @@ class DownloadResult(NamedTuple):
     """Represents a payment made by a buyer."""
     number_downloaded: int
     number_requested: int
-    request_time_s: int
+    elapsed_time_ms: int
+    number_peers: int

squeaknode/network/network_manager.py

@@ -100,14 +100,22 @@ class NetworkManager(object):
     def get_connected_peers(self) -> List[Peer]:
         return self.connection_manager.peers
 
-    def broadcast_msg(self, msg: MsgSerializable) -> None:
+    def broadcast_msg(self, msg: MsgSerializable) -> int:
+        """Send a message to all connected peers.
+
+        Returns:
+            int: the number of peers message was sent to.
+        """
+        count = 0
         for peer in self.connection_manager.peers:
             try:
                 peer.send_msg(msg)
+                count += 1
             except Exception:
                 logger.exception("Failed to send msg to peer: {}".format(
                     peer,
                 ))
+        return count
 
     def update_local_subscriptions(self, locator: CSqueakLocator) -> None:
         for peer in self.connection_manager.peers:

squeaknode/node/active_download_manager.py

@@ -21,6 +21,7 @@
 # SOFTWARE.
 import logging
 import threading
+import time
 import uuid
 from abc import ABC
 from abc import abstractmethod
@@ -32,6 +33,7 @@ from typing import Optional
 from squeak.core import CSqueak
 from squeak.messages import msg_getdata
 from squeak.messages import msg_getsqueaks
+from squeak.messages import MsgSerializable
 from squeak.net import CInterested
 from squeak.net import CInv
 from squeak.net import CSqueakLocator
@@ -53,14 +55,23 @@ class ActiveDownload(ABC):
         self.count = 0
         self._lock = threading.Lock()
         self.stopped = threading.Event()
+        self.num_peers = 0
+        self.start_time_ms: Optional[int] = None
 
     @abstractmethod
     def is_interested(self, squeak: CSqueak) -> bool:
         """Return True if the given squeak matches the download interest."""
 
     @abstractmethod
+    def get_download_msg(self) -> MsgSerializable:
+        """Get the message to send to peers to get download response."""
+
     def initiate_download(self, broadcast_fn) -> None:
-        """Broadcast a message to peers to get data."""
+        self.start_time_ms = int(time.time() * 1000)
+        msg = self.get_download_msg()
+        self.num_peers = broadcast_fn(msg)
+        if self.num_peers == 0:
+            self.mark_complete()
 
     def increment(self) -> None:
         with self._lock:
@@ -74,6 +85,12 @@ class ActiveDownload(ABC):
     def cancel(self):
         self.stopped.set()
 
+    def get_elapsed_time_ms(self):
+        if self.start_time_ms is None:
+            return 0
+        end_time_ms = int(time.time() * 1000)
+        return end_time_ms - self.start_time_ms
+
     def wait_for_complete(self, timeout_s: int) -> None:
         self.stopped.wait(timeout=timeout_s)
 
@@ -81,7 +98,8 @@ class ActiveDownload(ABC):
         return DownloadResult(
             number_downloaded=self.count,
             number_requested=self.limit,
-            request_time_s=-1,
+            elapsed_time_ms=self.get_elapsed_time_ms(),
+            number_peers=self.num_peers,
         )
 
 
@@ -94,14 +112,13 @@ class InterestDownload(ActiveDownload):
     def is_interested(self, squeak: CSqueak) -> bool:
         return squeak_matches_interest(squeak, self.interest)
 
-    def initiate_download(self, broadcast_fn) -> None:
+    def get_download_msg(self) -> MsgSerializable:
         locator = CSqueakLocator(
             vInterested=[self.interest],
         )
-        getsqueaks_msg = msg_getsqueaks(
+        return msg_getsqueaks(
             locator=locator,
         )
-        broadcast_fn(getsqueaks_msg)
 
 
 class HashDownload(ActiveDownload):
@@ -113,14 +130,13 @@ class HashDownload(ActiveDownload):
     def is_interested(self, squeak: CSqueak) -> bool:
         return self.squeak_hash == get_hash(squeak)
 
-    def initiate_download(self, broadcast_fn) -> None:
+    def get_download_msg(self) -> MsgSerializable:
         invs = [
             CInv(type=1, hash=self.squeak_hash)
         ]
-        getdata_msg = msg_getdata(
+        return msg_getdata(
             inv=invs,
         )
-        broadcast_fn(getdata_msg)
 
 
 class ActiveDownloadManager:

squeaknode/node/squeak_controller.py

@@ -737,8 +737,8 @@ class SqueakController:
         )
         return self.active_download_manager.download_interest(10, interest)
 
-    def broadcast_msg(self, msg: MsgSerializable) -> None:
-        self.network_manager.broadcast_msg(msg)
+    def broadcast_msg(self, msg: MsgSerializable) -> int:
+        return self.network_manager.broadcast_msg(msg)
 
     def disconnect_peer(self, peer_address: PeerAddress) -> None:
         logger.info("Disconnect to peer: {}".format(

tests/node/test_active_download.py

@@ -126,5 +126,6 @@ def test_download_hash_get_result(download_hash, squeak):
     assert download_result == DownloadResult(
         number_downloaded=1,
         number_requested=1,
-        request_time_s=-1,
+        elapsed_time_ms=0,
+        number_peers=0,
     )