Better chord unlock error handling for kvstores

marcusmartins · Mar 6, 2013 · 9193ca3 · 9193ca3
1 parent f9f2081
commit 9193ca3
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 6 deletions.
diff --git a/celery/app/builtins.py b/celery/app/builtins.py
@@ -112,7 +112,8 @@ def unlock_chord(group_id, callback, interval=None, propagate=None,
                 except Exception, exc:
                     app._tasks[callback.task].backend.fail_from_current_stack(
                         callback.id,
-                        exc=ChordError('Call callback error: %r' % (exc, )))
+                        exc=ChordError('Callback error: %r' % (exc, )),
+                    )
         else:
             return unlock_chord.retry(countdown=interval,
                                       max_retries=max_retries)

diff --git a/celery/backends/base.py b/celery/backends/base.py
@@ -485,6 +485,7 @@ def on_chord_part_return(self, task, propagate=None):
             return
         from celery import subtask
         from celery.result import GroupResult
+        app = self.app
         if propagate is None:
             propagate = self.app.conf.CELERY_CHORD_PROPAGATES
         gid = task.request.group
@@ -499,13 +500,22 @@ def on_chord_part_return(self, task, propagate=None):
             try:
                 ret = j(propagate=propagate)
             except Exception, exc:
-                culprit = deps._failed_join_report().next()
-                self.app._tasks[callback.task].backend.fail_from_current_stack(
-                    callback.id, exc=ChordError('Dependency %s raised %r' % (
-                        culprit.id, exc))
+                try:
+                    culprit = deps._failed_join_report().next()
+                    reason = 'Dependency %s raised %r' % (culprit.id, exc)
+                except StopIteration:
+                    reason = repr(exc)
+                app._tasks[callback.task].backend.fail_from_current_stack(
+                    callback.id, exc=ChordError(reason),
                 )
             else:
-                callback.delay(ret)
+                try:
+                    callback.delay(ret)
+                except Exception, exc:
+                    app._tasks[callback.task].backend.fail_from_current_stack(
+                        callback.id,
+                        exc=ChordError('Callback error: %r' % (exc, )),
+                    )
             finally:
                 deps.delete()
                 self.client.delete(key)