Fix queue sticking problem when an error occurs on the client

backtrac/client/client.py

@@ -58,17 +58,22 @@ def get_index(self, path):
             for f in files:
                 path = os.path.join(root, f)
                 filepath = FilePath(path)
-                index[path] = {
-                    'mtime': filepath.getModificationTime(),
-                    'size': filepath.getsize(),
-                }
+                try:
+                    index[path] = {
+                        'mtime': filepath.getModificationTime(),
+                        'size': filepath.getsize(),
+                    }
+                except OSError:
+                    # file could be a broken symlink, or deleted mid-scan
+                    continue
         return index
 
     @defer.inlineCallbacks
     def check_index(self, path, index):
         backup = yield self.broker.check_index(path, index)
         for path in backup:
-            self.backup_queue.add(BackupJob(path))
+            #self.backup_queue.add(BackupJob(path))
+            self.transfer_queue.add(FilePath(path))
 
     def remote_put_file(self, path):
         self.monitor.add_exclusion(path)
@@ -83,8 +88,6 @@ def remote_put_file(self, path):
     @defer.inlineCallbacks
     def start(self):
         broker = yield self.broker.connect(client=self)
-        self.backup_queue.start()
-        self.transfer_queue.start()
         paths = yield self.broker.get_paths()
         for path in paths:
             path = normpath(path)
@@ -93,6 +96,8 @@ def start(self):
             self.check_present_state(path)
             self.check_index(path, index)
             #self.walk_path(path)
+        self.backup_queue.start()
+        self.transfer_queue.start()
 
 def get_server_status():
     broker = BackupBroker(server='localhost', secret_key=settings.SECRET_KEY,

backtrac/client/queue.py

@@ -53,6 +53,9 @@ def consume_create(self, filepath):
             return self.client.broker.create_item(filepath.path, 'd')
 
     def consume_update(self, filepath):
+        if filepath.isdir():
+            # we can't backup a directory
+            return
         try:
             attrs = {
                 'mtime': filepath.getModificationTime(),
@@ -85,10 +88,14 @@ def _check_result(self, backup_required, path):
 
 class TransferQueue(BackupQueue):
     def consume(self, filepath):
+        if filepath.isdir():
+            # we can't transfer a directory
+            return
         try:
             mtime = filepath.getModificationTime()
             size = filepath.getsize()
         except (OSError, IOError):
+            # returning will allow the queue to move on
             return
         d = Deferred()
         self.client.broker.put_file(filepath.path, mtime, size).addCallback(
@@ -106,8 +113,11 @@ def _transfer(self, collector, filepath, d=None):
             if d is not None:
                 pager.wait().chainDeferred(d)
             print '%s, %d bytes' % (filepath.path, filepath.getsize())
-        except (OSError, IOError):
-            pass
+        except (OSError, IOError) as e:
+            print "Error:", e
+            if d is not None:
+                # make sure we callback, otherwise the queue will stick
+                d.errback(e)
 
 class QueueManager(object):
     def __init__(self, queues):