rebuild changelog and redownload from peers.

oml/nodes.py

@@ -33,7 +33,6 @@ ENCODING='base64'
 
 class Node(Thread):
     _running = True
-    _pulling = False
     host = None
     local = None
     _online = None
@@ -49,10 +48,7 @@ class Node(Thread):
         Thread.__init__(self)
         self.daemon = True
         self.start()
-        self._pull = PeriodicCallback(self.pull, 60000)
-        self._pull.start()
         self.ping()
-        self.pull()
 
     def run(self):
         while self._running:
@@ -63,12 +59,6 @@ class Node(Thread):
                 self._send_response()
             elif action == 'ping':
                 self.online = self.can_connect()
-            elif action == 'pull':
-                self._pulling = True
-                self.online = self.can_connect()
-                if self.online:
-                    self.pullChanges()
-                self._pulling = False
             else:
                 logger.debug('unknown action %s', action)
 
@@ -77,9 +67,6 @@ class Node(Thread):
         self._q.put('')
         #return Thread.join(self)
 
-    def pull(self):
-        if state.online and not self._pulling:
-            self._q.put('pull')
 
     def ping(self):
         if state.online:
@@ -262,8 +249,6 @@ class Node(Thread):
             })
 
     def pullChanges(self):
-        if state.activity and state.activity.get('activity') == 'import':
-            return
         with db.session():
             u = user.models.User.get_or_create(self.user_id)
             if not self.online or not u.peered:
@@ -278,7 +263,7 @@ class Node(Thread):
                 return False
             if not changes:
                 return False
-            r = Changelog.apply_changes(u, changes)
+            r = Changelog.apply_changes(u, changes, first=from_revision == 0)
             return r
 
     def peering(self, action):
@@ -416,6 +401,7 @@ class Node(Thread):
 class Nodes(Thread):
     _nodes = {}
     _local = None
+    _pulling = False
 
     def __init__(self):
         self._q = Queue()
@@ -432,14 +418,21 @@ class Nodes(Thread):
         self._local = LocalNodes()
         self._cleanup = PeriodicCallback(lambda: self.queue('cleanup'), 120000)
         self._cleanup.start()
+        self._pullcb = PeriodicCallback(self.pull, 60000)
+        self._pullcb.start()
         Thread.__init__(self)
         self.daemon = True
         self.start()
+        self.pull()
 
     def cleanup(self):
         if self._running and self._local:
             self._local.cleanup()
 
+    def pull(self):
+        if state.online and not self._pulling:
+            self.queue('pull')
+
     def queue(self, *args):
         self._q.put(list(args))
 
@@ -480,6 +473,16 @@ class Nodes(Thread):
         if send_response:
             self._nodes[user_id].send_response()
 
+    def _pull(self):
+        if state.activity and state.activity.get('activity') == 'import':
+            return
+        self._pulling = True
+        for node in list(self._nodes.values()):
+            node.online = node.can_connect()
+            if node.online:
+                node.pullChanges()
+        self._pulling = False
+
     def run(self):
         while self._running:
             args = self._q.get()
@@ -488,6 +491,8 @@ class Nodes(Thread):
                     self.cleanup()
                 elif args[0] == 'add':
                     self._add(*args[1:])
+                elif args[0] == 'pull':
+                    self._pull()
                 else:
                     self._call(*args)