Function _dispatch_callbacks_and_chain

celery/app/trace.py:426–473

Dispatch callbacks and chain for a completed task. Dispatches link callbacks and then the next chain step. Does NOT fire task lifecycle signals (on_success, task_postrun) or call mark_as_done — callers handle those separately. Note: dispatch is not atomic. If callb

(
        retval, callbacks, chain, parent_id, root_id, priority,
    )

        return I, R, I.state, I.retval

    def _dispatch_callbacks_and_chain(
        retval, callbacks, chain, parent_id, root_id, priority,
    ):
        """Dispatch callbacks and chain for a completed task.

        Dispatches link callbacks and then the next chain step.
        Does NOT fire task lifecycle signals (on_success, task_postrun)
        or call mark_as_done — callers handle those separately.

        Note: dispatch is not atomic.  If callbacks succeed but the
        chain step fails (or vice-versa), a Reject + redeliver may
        re-dispatch the already-sent callbacks.  This is acceptable
        under Celery's at-least-once delivery model.
        """
        if callbacks:
            if len(callbacks) > 1:
                sigs, groups = [], []
                for sig in callbacks:
                    sig = signature(sig, app=app)
                    if isinstance(sig, group):
                        groups.append(sig)
                    else:
                        sigs.append(sig)
                for group_ in groups:
                    group_.apply_async(
                        (retval,),
                        parent_id=parent_id, root_id=root_id,
                        priority=priority,
                    )
                if sigs:
                    group(sigs, app=app).apply_async(
                        (retval,),
                        parent_id=parent_id, root_id=root_id,
                        priority=priority,
                    )
            else:
                signature(callbacks[0], app=app).apply_async(
                    (retval,),
                    parent_id=parent_id, root_id=root_id,
                    priority=priority,
                )
        if chain:
            _chsig = signature(chain[-1], app=app)
            _chsig.apply_async(
                (retval,), chain=chain[:-1],
                parent_id=parent_id, root_id=root_id,
                priority=priority,
            )

    def trace_task(uuid, args, kwargs, request=None):
        # R      - is the possibly prepared return value.