{{ message }}
Repository navigation
fix(canvas): fire task lifecycle hooks for map/starmap items - #10880
Open
Sanjays2402 wants to merge 1 commit into
Open
Sanjays2402 wants to merge 1 commit into
Sanjays2402 wants to merge 1 commit into
Conversation
The celery.map and celery.starmap builtin tasks executed each item by calling the task directly, bypassing the task trace entirely. Task lifecycle hooks (before_start/on_success/on_failure/after_return) and task signals never fired for mapped items, unlike single tasks. Route every item through Task.apply() so the trace runs per item. get(disable_sync_subtasks=False) is safe here because the EagerResult is already computed and never blocks. Fixes celery#7585
auvipy
self-requested a review
October 18, 2026 03:57
MehrazRumman
left a comment
MehrazRumman
left a comment
Contributor
There was a problem hiding this comment.
Checked at faaec5c.
What checks out
- Every mapped item now goes through the trace. The three new tests fail on
mainand pass here, andt/unit/app/test_builtins.pyplust/unit/tasks/test_canvas.pypass (378 passed, 1 xfailed). - Exceptions from an item still propagate out of the map task, now with
on_failure/after_returnfired first.
Behaviour changes beyond the hooks
Running items through apply() changes more than hook delivery. Compared with main:
main this PR
request inside an item (has id, is_eager,
called_directly) (False, False, True) (True, True, False)
INFO "Task ... succeeded" lines for a 100-item map 0 100
item raises Ignore (map over [1, 2, 3], x == 2) map task raises [1, None, 3]
- Log and signal volume. Inside a worker at INFO level, every item now logs its own "succeeded" line, and
task_prerun/task_postrun/task_successfire per item. Achunks()call over thousands of items produces thousands of log lines, and signal-based metrics (Prometheus exporters and similar) will count each item as a task. That is probably acceptable for hooks, but it should be called out. A quieter path, such as suppressing the per-item success log, may be worth considering. Ignore/Rejectfrom an item no longer stop the map. Onmainthe exception propagated and the whole map task ended as ignored. Nowapply()records the item asIGNORED,.get()returnsNone, and the map carries on withNonein the results. To keep the old semantics, check theEagerResultstate and re-raise forIGNORED/REJECTED.- The request context changes for the task. Items now see a fresh
request.id,is_eager=Trueandcalled_directly=False. Code that branches on those flags behaves differently. In particular,self.retry()/autoretry_forin eager mode re-runs the item inline and immediately, so the #7586 fix retries without honouringcountdown/retry_backoff, and it blocks the map task while doing so. Worth a sentence in the docs. - With
task_store_eager_result = True, every item's result is now written to the backend.
None of these is necessarily wrong, but together they are a user-visible change to map/starmap/chunks. They belong in the changelog, and point 2 probably needs a decision from the maintainers.
Member
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Fixes #7585
Fixes #7586
The
celery.mapandcelery.starmapbuiltin tasks ran each item by calling the task instance directly (task(item)/task(*item)), which skips the task trace completely. That meansbefore_start,on_success,on_failureandafter_returnnever fired for mapped items, and task signals were never sent, even though all of them fire for the same task executed on its own.chunksis affected too since it fans out throughxstarmap.The fix routes every item through
Task.apply()so each one gets a full trace..get(disable_sync_subtasks=False)is safe here: theEagerResultis already computed, so it never blocks, and passingFalsekeeps this working inside real workers where the join guard is on. Exceptions still propagate out of the map task exactly as before, and nowon_failurefires for them.Repro (before the fix,
map apply hookswas[]):Tests: added regression tests in
t/unit/app/test_builtins.pycovering hook order per mapped item formapandstarmap, pluson_failurefiring when an item raises. Rant/unit/app/test_builtins.py,t/unit/tasks/test_canvas.py,t/unit/app/test_trace.py,t/unit/tasks/test_result.py,t/unit/tasks/test_tasks.py: 638 passed, 1 skipped, 1 xfailed, all green.Also fixes #7586:
autoretry_forwas ignored for mapped items because the retry logic lives in the trace path. Verified: a task withautoretry_for=(ValueError,)andmax_retries=3raised after a single attempt throughmapbefore this change; now each item retries independently.