fix#4834 (canvas): fix chord error handling when body is a chain (#10247)

aurangzaib048 · auvipy · web-flow · commit 72e9240aa1e2 · 2026-04-06T20:35:42.000+06:00
* fix(canvas): fix chord error handling when body is a chain (#4834) Three bugs caused chord_error_from_stack to crash with TypeError when a chord member failed and the chord body was a chain (created by chain(chain(group(...), task), task)): 1. prepare_steps set self.id = last_task_id (the input parameter, None) instead of the last task's actual UUID from results[0].id. The chord body chain had id=None, causing get_key_for_task(None) to crash in the error handler. 2. Errbacks set on the outer chain were applied to the chord task but not to the chord body. chord_error_from_stack reads errbacks from the body (request.chord), so link_error handlers never fired. 3. No defensive guard in chord_error_from_stack for callback.id being None. Fix: - canvas.py prepare_steps: set chain ID to results[0].id - canvas.py prepare_steps: propagate link_error to chord body - backends/base.py chord_error_from_stack: generate UUID if callback.id is None Fixes #4834 * style: shorten inline comments * test: add integration test for chord error in nested chain Verifies that chain(chain(group(ok, failing), task), task) does not crash with TypeError in chord_error_from_stack when a chord member fails. The test checks that no 'task_id must not be empty' error is logged during error propagation. * fix: narrow exception assertion in integration test Replace broad pytest.raises((ExpectedException, Exception)) with explicit checks that the exception is not TypeError or the internal ValueError from get_key_for_task. The chord failure itself is expected, but internal errors are not. --------- Co-authored-by: Asif Saif Uddin {"Auvi":"অভি"} <auvipy@gmail.com>
diff --git a/celery/backends/base.py b/celery/backends/base.py
@@ -317,6 +317,13 @@ def chord_error_from_stack(self, callback, exc=None):
         # Handle group callbacks specially to prevent hanging body tasks
         if isinstance(callback, group):
             return self._handle_group_chord_error(group_callback=callback, backend=backend, exc=exc)
+
+        # Generate an ID if missing so the error can be stored.
+        callback_id = callback.id
+        if not callback_id:
+            from kombu.utils.uuid import uuid
+            callback_id = callback.options['task_id'] = uuid()
+
         # 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
@@ -330,9 +337,9 @@ def chord_error_from_stack(self, callback, exc=None):
         try:
             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)
+            return backend.fail_from_current_stack(callback_id, exc=eb_exc)
         else:
-            return backend.fail_from_current_stack(callback.id, exc=exc)
+            return backend.fail_from_current_stack(callback_id, exc=exc)
 
     def _handle_group_chord_error(self, group_callback, backend, exc=None):
         """Handle chord errors when the callback is a group.
diff --git a/celery/canvas.py b/celery/canvas.py
@@ -1260,6 +1260,9 @@ def prepare_steps(self, args, kwargs, tasks,
             if link_error:
                 for errback in maybe_list(link_error):
                     task.link_error(errback)
+                    # Propagate to chord body for chord_error_from_stack.
+                    if isinstance(task, chord) and task.body:
+                        task.body.link_error(errback)
 
             tasks.append(task)
             results.append(res)
@@ -1277,7 +1280,8 @@ def prepare_steps(self, args, kwargs, tasks,
                 while node.parent:
                     node = node.parent
                 prev_res = node
-        self.id = last_task_id
+        # Use the last task's actual ID, not the input parameter.
+        self.id = results[0].id if results else last_task_id
         return tasks, results
 
     def apply(self, args=None, kwargs=None, **options):
diff --git a/t/integration/test_canvas.py b/t/integration/test_canvas.py
@@ -3269,6 +3269,47 @@ def test_chord_error_propagation_with_different_body_types(
         error_found = check_for_logs(caplog=caplog, message="ValueError: task_id must not be empty")
         assert not error_found, "The 'task_id must not be empty' error was found in the logs"
 
+    @flaky
+    def test_chord_error_in_nested_chain_does_not_crash(self, manager, caplog):
+        """Chord error with chain body must not raise internal TypeError.
+
+        Regression test for https://github.com/celery/celery/issues/4834
+        chain(chain(group(ok, failing), task), task) crashed with
+        TypeError in chord_error_from_stack because the chain body
+        had id=None.
+        """
+        try:
+            manager.app.backend.ensure_chords_allowed()
+        except NotImplementedError as e:
+            raise pytest.skip(e.args[0])
+
+        c = chain(
+            chain(
+                group(add.si(1, 1), fail.si()),
+                identity.s(),
+            ),
+            identity.s(),
+        )
+        result = c.apply_async()
+
+        try:
+            result.get(timeout=TIMEOUT)
+        except Exception as exc:
+            assert not isinstance(exc, TypeError), (
+                f"Internal TypeError raised: {exc}"
+            )
+            assert "task_id must not be empty" not in str(exc), (
+                f"Internal ValueError raised: {exc}"
+            )
+
+        error_found = check_for_logs(
+            caplog=caplog,
+            message="task_id must not be empty",
+        )
+        assert not error_found, (
+            "chord_error_from_stack crashed with 'task_id must not be empty'"
+        )
+
 
 class test_signature_serialization:
     """
diff --git a/t/unit/backends/test_base.py b/t/unit/backends/test_base.py
@@ -910,6 +910,36 @@ def test_chord_error_from_stack_backend_fallback(self):
         # Verify self was used as fallback backend
         backend.fail_from_current_stack.assert_called_once()
 
+    def test_chord_error_from_stack_generates_id_when_callback_id_is_none(self):
+        """chord_error_from_stack must not crash when callback.id is None.
+
+        Regression test for https://github.com/celery/celery/issues/4834
+        When a chord body is a chain without an explicit task_id, the
+        error handler must generate an ID so the error result can be
+        stored and errback handlers can fire.
+        """
+        backend = self.b
+
+        callback = MagicMock(name='callback')
+        callback.id = None
+        callback.options = {'task_id': None, 'link_error': []}
+        callback.keys.return_value = []
+        callback.task = 'nonexistent.task'
+
+        backend._call_task_errbacks = Mock()
+        backend.fail_from_current_stack = Mock()
+
+        backend.chord_error_from_stack(callback, exc=ValueError('test'))
+
+        # fail_from_current_stack must be called with a non-None ID
+        call_args = backend.fail_from_current_stack.call_args
+        actual_id = call_args[0][0]
+        assert actual_id is not None, (
+            "chord_error_from_stack must generate an ID when callback.id is None"
+        )
+        # The generated ID must also be stored on the callback
+        assert callback.options['task_id'] is not None
+
     def _create_mock_frozen_group(self, group_id="group-id", task_ids=None, task_names=None):
         """Helper to create mock frozen group with results."""
         if task_ids is None:
diff --git a/t/unit/tasks/test_canvas.py b/t/unit/tasks/test_canvas.py
@@ -860,6 +860,106 @@ def test_upgrade_to_chord_on_chain(self):
         assert isinstance(final_task.tasks[0].body, chord)
         assert final_task.tasks[0].body.body == chain1
 
+    def test_chain_body_gets_id_when_used_as_chord_body(self):
+        """Chain used as chord body must have a non-None ID after freeze.
+
+        Regression test for https://github.com/celery/celery/issues/4834
+        When chain(chain(group(...), task), task) is frozen, the chord
+        body is a chain. Previously, the chain's ID stayed None because
+        prepare_steps set self.id = last_task_id (the input parameter,
+        None) instead of the last task's actual UUID.
+        """
+        # Build chord with a chain body: group becomes chord,
+        # the two following tasks become the body chain.
+        g = group(self.add.si(1, 1), self.add.si(2, 2))
+        body_chain = chord(g, self.add.s(10), app=self.app)
+        canvas = chain(body_chain, self.add.s(20), app=self.app)
+        canvas.freeze()
+
+        # Find the chord
+        chords = [t for t in canvas.tasks if isinstance(t, chord)]
+        assert chords, "Expected a chord in the frozen chain"
+        chord_task = chords[0]
+
+        # The chord body (a chain) must have a non-None ID
+        assert chord_task.body.id is not None, (
+            "Chord body chain must have an ID after freeze"
+        )
+
+    def test_chain_body_id_propagated_to_header_tasks(self):
+        """Header tasks must carry the chord body's ID in request.chord.
+
+        Regression test for https://github.com/celery/celery/issues/4834
+        The header tasks store a reference to the chord body. After
+        freeze, this reference must have a valid task_id so that
+        chord_error_from_stack can store the error result.
+        """
+        g = group(self.add.si(1, 1), self.add.si(2, 2))
+        body_chain = chord(g, self.add.s(10), app=self.app)
+        canvas = chain(body_chain, self.add.s(20), app=self.app)
+        canvas.freeze()
+
+        chords = [t for t in canvas.tasks if isinstance(t, chord)]
+        chord_task = chords[0]
+
+        # Every header task's chord option must have the body's ID
+        body_id = chord_task.body.id
+        for header_task in chord_task.tasks.tasks:
+            header_chord = header_task.options.get('chord')
+            assert header_chord is not None, "Header task must have chord option"
+            assert header_chord.id == body_id, (
+                f"Header chord ID {header_chord.id} != body ID {body_id}"
+            )
+
+    def test_chain_errbacks_propagated_to_chord_body(self):
+        """Errbacks on a chain must propagate to the chord body.
+
+        Regression test for https://github.com/celery/celery/issues/4834
+        chord_error_from_stack reads errbacks from the chord body
+        (request.chord), not from the chord task. Without propagation,
+        link_error handlers set on a chain containing a chord never fire.
+        """
+        @self.app.task(shared=False)
+        def on_error(*args):
+            pass
+
+        g = group(self.add.si(1, 1), self.add.si(2, 2))
+        body_chord = chord(g, self.add.s(10), app=self.app)
+        c = chain(body_chord, self.add.s(20), app=self.app)
+
+        # Simulate what apply_async does: pass link_error to prepare_steps
+        errback = on_error.s()
+        c.prepare_steps(
+            c.args, c.kwargs, c.tasks, app=self.app,
+            link_error=[errback], clone=False,
+        )
+
+        # The chord body must have the errback
+        chords = [t for t in c.tasks if isinstance(t, chord)]
+        assert chords
+        body_errbacks = chords[0].body.options.get('link_error', [])
+        assert len(body_errbacks) >= 1, (
+            "Errback must be propagated to chord body"
+        )
+
+    def test_chain_id_matches_result_after_freeze(self):
+        """Chain.id must equal the returned result ID after freeze.
+
+        After freeze(), a chain's ID should reflect the last task's
+        result (since that is what the chain's result represents).
+        """
+        c = self.add.s(1, 1) | self.add.s(2) | self.add.s(3)
+        result = c.freeze()
+        assert c.id is not None
+        assert c.id == result.id
+
+    def test_chain_id_with_explicit_id_preserved(self):
+        """Chain.freeze(explicit_id) must set chain.id to that value."""
+        c = self.add.s(1, 1) | self.add.s(2)
+        result = c.freeze('my-explicit-id')
+        assert c.id == 'my-explicit-id'
+        assert result.id == 'my-explicit-id'
+
 
 class test_group(CanvasCase):
     def test_repr(self):
