fix: Calling of errbacks when chords fail

We had a special case for calling errbacks when a chord failed which
assumed they were old style. This change ensures that we call the proper
errback dispatch method which understands new and old style errbacks,
and adds test to confirm that things behave as one might expect now.

Author: maybe-sybr

celery/backends/base.py

@@ -278,19 +278,24 @@ def mark_as_retry(self, task_id, exc, traceback=None,
                                  traceback=traceback, request=request)
 
     def chord_error_from_stack(self, callback, exc=None):
-        # need below import for test for some crazy reason
-        from celery import group  # pylint: disable
         app = self.app
         try:
             backend = app._tasks[callback.task].backend
         except KeyError:
             backend = self
+        # We have to make a fake request since either the callback failed or
+        # we're pretending it did since we don't have information about the
+        # chord part(s) which failed. This request is constructed as a best
+        # effort for new style errbacks and may be slightly misleading about
+        # what really went wrong, but at least we call them!
+        fake_request = Context({
+            "id": callback.options.get("task_id"),
+            "errbacks": callback.options.get("link_error", []),
+            "delivery_info": dict(),
+            **callback
+        })
         try:
-            group(
-                [app.signature(errback)
-                 for errback in callback.options.get('link_error') or []],
-                app=app,
-            ).apply_async((callback.id,))
+            self._call_task_errbacks(fake_request, exc, None)
         except Exception as eb_exc:  # pylint: disable=broad-except
             return backend.fail_from_current_stack(callback.id, exc=eb_exc)
         else:

celery/canvas.py

@@ -1123,18 +1123,17 @@ def set_immutable(self, immutable):
             task.set_immutable(immutable)
 
     def link(self, sig):
-        # Simply link to first task
+        # Simply link to first task. Doing this is slightly misleading because
+        # the callback may be executed before all children in the group are
+        # completed and also if any children other than the first one fail.
+        #
+        # The callback signature is cloned and made immutable since it the
+        # first task isn't actually capable of passing the return values of its
+        # siblings to the callback task.
         sig = sig.clone().set(immutable=True)
         return self.tasks[0].link(sig)
 
     def link_error(self, sig):
-        try:
-            sig = sig.clone().set(immutable=True)
-        except AttributeError:
-            # See issue #5265.  I don't use isinstance because current tests
-            # pass a Mock object as argument.
-            sig['immutable'] = True
-            sig = Signature.from_dict(sig)
         # Any child task might error so we need to ensure that they are all
         # capable of calling the linked error signature. This opens the
         # possibility that the task is called more than once but that's better

t/integration/tasks.py

@@ -301,11 +301,6 @@ def fail_replaced(self, *args):
     raise self.replace(fail.si(*args))
 
 
-@shared_task
-def chord_error(*args):
-    return args
-
-
 @shared_task(bind=True)
 def return_priority(self, *_args):
     return "Priority: %s" % self.request.delivery_info['priority']
@@ -385,3 +380,15 @@ def _recurse(sig):
         if isinstance(sig, chord):
             _recurse(sig.body)
     _recurse(sig_obj)
+
+
+@shared_task
+def errback_old_style(request_id):
+    redis_count(request_id)
+    return request_id
+
+
+@shared_task
+def errback_new_style(request, exc, tb):
+    redis_count(request.id)
+    return request.id