Enable file transfers + fix bug in get unl request expiry.

requirements.txt

@@ -8,6 +8,6 @@ ntplib >= 0.3.3
 twisted >= 15.4.0
 ipaddress >= 1.0.14
 requests >= 2.8.1
-pyp2p >= 0.7.7
+pyp2p >= 0.7.8
 jsonschema >= 2.5.1
 apigen >= 1.4.1

storjnode/network/monitor.py

@@ -22,7 +22,7 @@
 _log = storjnode.log.getLogger(__name__)
 
 
-SKIP_BANDWIDTH_TEST = True
+SKIP_BANDWIDTH_TEST = False
 if os.environ.get("STORJNODE_MONITOR_MAX_TRIES"):
     MAX_TRIES = int(os.environ.get("STORJNODE_MONITOR_MAX_TRIES"))
 else:
@@ -313,6 +313,8 @@ def _process_bandwidth_test(self):
             skip_bandwidth_test = SKIP_BANDWIDTH_TEST
             if self.node.sim_dht is not None:
                 if not self.node.sim_dht.has_mutex:
+                    skip_bandwidth_test = True
+                else:
                     if not self.node.sim_dht.can_test_knode(nodeid):
                         skip_bandwidth_test = True
             if skip_bandwidth_test:
@@ -385,7 +387,15 @@ def crawl(self):
 
         # add initial peers
         peers = self.node.get_neighbours()
+        x = []
+        y = []
         for peer in peers:
+            if peer.can_test:
+                x.append(peer)
+            else:
+                y.append(peer)
+
+        for peer in x + y:
             if peer.id == self.node.get_id():
                 continue
 

storjnode/network/node.py

@@ -355,13 +355,12 @@ def get_unl_by_node_id(self, nodeid):
         unl_req = sign(unl_req, self.get_key())
 
         # Record start time.
-        future_timeout = time.time() + 10
+        future_timeout = time.time() + 30
 
         # Handle responses for this request.
         def handler_builder(self, d, their_node_id, wif):
             def handler(node, msg):
                 # Is this a response to our request?
-                remove_handler = 0
                 try:
                     if type(msg) in [type(b"")]:
                         msg = literal_eval(zlib.decompress(msg))
@@ -387,17 +386,12 @@ def handler(node, msg):
                         _log.debug("unl response: their sig")
                         return
 
-                    remove_handler = 1
-
                     # Everything passed: fire callback.
+                    node.remove_message_handler(handler)
                     d.callback(msg[u"unl"])
                 except (ValueError, KeyError, zlib.error):
                     _log.debug("unl response:val or key er")
                     pass  # not a unl response
-                finally:
-                    if remove_handler:
-                        # Remove this callback.
-                        node.remove_message_handler(handler)
 
             return handler
 
@@ -406,16 +400,16 @@ def handler(node, msg):
         handler = handler_builder(self, d, nodeid, self.get_key())
 
         # Expire this request.
-        def expire_old_unl_request():
-            _log.debug("Get unl request timed out")
+        def expire_old_unl_request(node, msg):
             if time.time() >= future_timeout:
-                d.errback(Exception("Get unl request timed out"))
-
-            return defer.Deferred()
-
-        LoopingCall(expire_old_unl_request).start(30, now=False)
+                _log.debug("Get unl request timed out")
+                node.remove_message_handler(expire_old_unl_request)
+                if handler in self._message_handlers:
+                    node.remove_message_handler(handler)
+                    d.errback(Exception("Get unl request timed out"))
 
         # Register new handler for this UNL request.
+        self.add_message_handler(expire_old_unl_request)
         self.add_message_handler(handler)
 
         # Send our get UNL request to node.

storjnode/network/repeat_relay.py

@@ -71,7 +71,9 @@ def relay(self, node_id, msg, rebroadcast=True):
             "rebroadcast": ENABLE_REPEAT
         }
 
-        self.relaying.append(relay_info)
+        if ENABLE_REPEAT:
+            self.relaying.append(relay_info)
+
         self.node.relay_message(node_id, msg)
 
         return None

tests/network/node.py

@@ -390,7 +390,7 @@ def handler(node, message):
     ########################
 
     def test_network_monitor_service(self):
-        limit = 2
+        limit = 1
         interval = 60 * 15
         crawled_event = threading.Event()
         results = {}
@@ -401,10 +401,12 @@ def handler(key, shard):
             crawled_event.set()
 
         self.swarm.reverse()
+        node = self.swarm[-1]
         for n in self.swarm:
-            if not n.sim_dht.has_mutex:
-                node = n
-                break
+            if n.sim_dht.has_mutex:
+                if n.sim_dht.has_testable_neighbours():
+                    node = n
+                    break
 
         # Todo: figure out how to choose node that has mutex
         # and how to make it so it has more neighbours than other