Retry chain: how to get if signature of previously run task was immutable?

[WilliamDEdwards]

I need to be able to retry Celery chains. That is: all tasks in a given chain with a non-success state should be run again.

As far as I can see, that is not natively supported. So, I came up with the following:

• From the previous chain, get tasks with the status 'failure', that is:

• • The task that failed.
• • All tasks after that, as Celery gives those the 'failure' status as well.

• Create new chain with those tasks.
• When constructing signatures, re-use UUIDs of the previous (failed) tasks. This ensures that the UUIDs stored in a local database refer to the retried tasks.

Using result_extended, I have access to almost all needed information to re-create signatures for previously run tasks:

• Task name (celery.AsyncResult.name)
• Args (celery.AsyncResult.args)
• Kwargs (celery.AsyncResult.kwargs)
• UUID (celery.AsyncResult.uuid)

But one piece of the puzzle is missing: how do I know if the signature was mutable (.s) or immutable (.si)?

Untested POC:

from app.engines.celery import celery  # App (instance of celery.Celery)
from celery import chain
from celery.canvas import Signature

new_chain_signatures: list[Signature]
task_uuid: list  # List of task UUIDs in previous chain, stored outside of Celery

for task_uuid in task_uuids:
    result = celery.AsyncResult(task_uuid)

    if result.status.lower() != 'failure':
        continue

    signature = celery.signature(result.name).si(*result.args, *result.kwargs).set(task_id=task_uuid)

    new_chain_signatures.append(signature)

    chain(*new_chain_signatures)()

[WilliamDEdwards]

Even with result_extended, whether a signature was mutable or immutable is not stored. This makes sense, because Celery stores results, not signatures (or even tasks in a more abstract sense).

Therefore, I have decided to store this piece of information in a local database, and reconstruct the signature as follows:

signature = celery.signature(
    result.name,
    immutable=task.internal_metadata["immutable"],
    args=result.args,
    kwargs=result.kwargs,
).set(task_id=task.uuid)

where result is an instance of celery.AsyncResult, and task is a local database object