Fixing error handling such that processing is aborted if there is a s…

…ingle failure.

* Required switching sentinel values to integers -- since the objects
  are pickled when being pushed onto the queue, the value popped off
  of the queue will *not* be the same object.  Sad.
* When any single service fails, abort processing for all to avoid
  marking tasks handled by failed services as 'done'.

bugwarrior/db.py

@@ -1,3 +1,4 @@
+from ConfigParser import NoOptionError
 import copy
 import os
 import re
@@ -10,7 +11,7 @@
 from taskw import TaskWarriorShellout
 from taskw.exceptions import TaskwarriorError
 
-from bugwarrior.config import asbool, NoOptionError
+from bugwarrior.config import asbool
 from bugwarrior.notifications import send_notification
 
 
@@ -26,6 +27,10 @@
 )
 
 
+# Sentinel value used for aborting processing of tasks
+ABORT_PROCESSING = 2
+
+
 class URLShortener(object):
     _instance = None
 
@@ -231,6 +236,8 @@ def _bool_option(section, option, default):
         'closed': expected_task_ids,
     }
     for issue in issue_generator:
+        if isinstance(issue, tuple) and issue[0] == ABORT_PROCESSING:
+            raise RuntimeError(issue[1])
         try:
             existing_uuid = find_local_uuid(
                 tw, key_list, issue, legacy_matching=legacy_matching

bugwarrior/services/__init__.py

@@ -6,12 +6,12 @@
 import six
 from twiggy import log
 
-from bugwarrior.db import MARKUP, URLShortener
+from bugwarrior.db import MARKUP, URLShortener, ABORT_PROCESSING
 
 
 # Sentinels for process completion status
-SERVICE_FINISHED_OK = object()
-SERVICE_FINISHED_ERROR = object()
+SERVICE_FINISHED_OK = 0
+SERVICE_FINISHED_ERROR = 1
 
 
 class IssueService(object):
@@ -406,10 +406,11 @@ def aggregate_issues(conf):
         issue = queue.get(True)
         if isinstance(issue, tuple):
             completion_type, args = issue
-            if completion_type is SERVICE_FINISHED_ERROR:
+            if completion_type == SERVICE_FINISHED_ERROR:
                 target, e = args
-                message = "Worker for [%s] failed." % (target, e)
-                log.name('bugwarrior').critical(message)
+                for process in processes:
+                    process.terminate()
+                yield ABORT_PROCESSING, e
             currently_running -= 1
             continue
         yield issue