When is TaskResult created?

[Mapiarz]

At what moment of a task life cycle is the TaskResult object created? TaskResult.status field supports all types of status but most of them would be unused if TaskResult was created only after a task has finished. So, what is it?

[Mapiarz]

Furthermore, I'm running into some race conditions. When I chain tasks using link, when I pass the task id to the linked task, the TaskResult object will not yet be available. I would have to wait a bit before trying to retrieve. On the other hand, when I use link_error, the TaskResult is always there already. Is that just by coincidence?

[auvipy]

can you please help to reproduce?

[Mapiarz]

Sure!

So the behavior I'm seeing is that a chained task, say A -> B, the latter task can execute before A is finished. If that's the case, it makes sense that TaskResult might not yet be available at the start of task B. I was caught off guard by this as one would expect a chain to work so that A always finishes before B is started. The link argument seems to be using chain under the hood and thus the behavior that I initially reported. This weirdness goes further as it seems that on_error or link_error always execute after a task has truly finished and thus TaskResult is always available. But that might be a coincidence and might not be guaranteed, I can't tell.

As a workaround I can use chord but that is optimized for reducing group results and makes my tasks have weird signature. Just meh.

Lastly, I am not sure when TaskResult is created but I have realized that I can use CELERY_TASK_TRACK_STARTED to force earlier creation of TaskResult. This will lead to TaskResult always being available in the chained task but possibly outdated?

Anyhow, using django-celery-results 2.4.0 with celery 5.2.7, backed by django with postgres and with rabbitmq as the broker. All settings default (except broker url, results backend set to django-db)

As for the code, there isn't much. The chain will throw unexpected exception and the error callback will work correctly.

@shared_task(bind=True)
def foo_task(self: Task) -> str:
    return self.request.id

@shared_task()
def bar_task() -> None:
    raise Exception("hello world")

@shared_task()
def failure_callback(task_id: str) -> None:
    TaskResult.objects.get(task_id=task_id)

@shared_task()
def success_callback(task_id: str) -> None:
    TaskResult.objects.get(task_id=task_id)

(foo_task.s() | success_callback.s()).delay()
bar_task.s().on_error(failure_callback.s()).delay()

[Mapiarz]

@auvipy any news on this? :)