Improvement of the idempotency of the celery workers

[pboers1988]

Contact Details

peter.boers@surf.nl

What happened?

When running the orchestrator in celery mode, the start stopping of tasks is not idempotent. There are some cases where tasks can be stuck in a "resumed" state and never picked up. Furthermore the running_processes counter in the database often does not reflect the accurate state of affairs.

Steps to reproduce

1. submit a large amount of workflows onto a queue
2. kill celery
3. A large number of tasks will remain in the created state.

It is also possible to do this when retrying a large number of workflows, they will then become stuck in resumed state. Or killing celery whilst executing long running tasks and start/stopping the engine

Sugested solution

As we heve implemented #832, we can now make the pick-up of tasks idempotent. Please design and implement a method so celery, at startup re-fills it's queue correctly.

Version

3.1.2rc4 and lower

What python version are you seeing the problem on?

All

Relevant log output

[hanstrompert]

Time-box finding the error to 3 points.

Suggested steps:

• reproduce both failure scenarios
• pinpoint what goes wrong
• design solution(s)
• implement solution(s)

[Igoranze]

A branch with the work containing how to start the workers and the server:
https://github.com/workfloworchestrator/orchestrator-core/tree/898-celery-workers

1. git checkout https://github.com/workfloworchestrator/orchestrator-core/tree/898-celery-workers
2. run uvicorn --reload --host 127.0.0.1 --port 8080 main:app
3. run worker celery -A orchestrator.tasks worker --loglevel=info -Q new_tasks,resume_tasks,new_workflows,resume_workflows

In the tasks.py there is a task validator, check the file out first before you try to reproduce the error

[Igoranze]

Forgot to mention, but you will also need a redis instance up and running, i'm using this docker compose file:

version: '3'

services:
  redis:
    image: redis
    ports:
      - "127.0.0.1:6379:6379"

volumes:
  redis-ssl-data:

[kvklink]

We are experiencing the issue with workflows getting stuck in a 'created' state as well when we fire off our nightly subscription validation. Funny enough, it's always exactly four workflows that are stuck.

Trace of how it fails, if relevant.

2025-06-19 08:50:35 [error    ] Unknown workflow failure       [orchestrator.services.processes] process_id=UUID('7a485fdd-5aa5-426c-bde4-f0eac9145b82') environment=prod
Traceback (most recent call last): environment=prod
  File "/app/orchestrator/services/processes.py", line 302, in safe_logstep environment=prod
    return _db_log_step(stat, step, process_state, broadcast_func=broadcast_func) environment=prod
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/services/processes.py", line 269, in _db_log_step environment=prod
    p = _update_process(stat.process_id, step, process_state) environment=prod
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/services/processes.py", line 130, in _update_process environment=prod
    raise ValueError(f"Failed to write failure step to process: process with PID {process_id} not found") environment=prod
ValueError: Failed to write failure step to process: process with PID 7a485fdd-5aa5-426c-bde4-f0eac9145b82 not found environment=prod
During handling of the above exception, another exception occurred: environment=prod
Traceback (most recent call last): environment=prod
  File "/app/orchestrator/services/processes.py", line 386, in run environment=prod
    result = f() environment=prod
             ^^^ environment=prod
  File "/app/orchestrator/services/processes.py", line 467, in <lambda> environment=prod
    return _run_process_async(pstat.process_id, lambda: runwf(pstat, _safe_logstep_with_func)) environment=prod
                                                        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/workflow.py", line 1438, in runwf environment=prod
    executed_steps = _exec_steps(steps, next_state, _logstep) environment=prod
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/workflow.py", line 1405, in _exec_steps environment=prod
    process = dblogstep(step, result_to_log) environment=prod
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/workflow.py", line 1421, in _logstep environment=prod
    return logstep(pstat, *x) environment=prod
           ^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/services/processes.py", line 309, in safe_logstep environment=prod
    return _db_log_step(stat, step, failure, broadcast_func=broadcast_func) environment=prod
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/services/processes.py", line 269, in _db_log_step environment=prod
    p = _update_process(stat.process_id, step, process_state) environment=prod
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ environment=prod
  File "/app/orchestrator/services/processes.py", line 130, in _update_process environment=prod
    raise ValueError(f"Failed to write failure step to process: process with PID {process_id} not found") environment=prod
ValueError: Failed to write failure step to process: process with PID 7a485fdd-5aa5-426c-bde4-f0eac9145b82 not found environment=prod

[Mark90]

We are experiencing the issue with workflows getting stuck in a 'created' state as well when we fire off our nightly subscription validation. Funny enough, it's always exactly four workflows that are stuck.

Trace of how it fails, if relevant.

Thanks for reporting this. Not sure how the trace relates to the problem yet but it's at least some indication.

Possibly related: this same errors occurred in this PR when adding a unittest for a workflow with an intentionally failing step #939 (comment)

[Mark90]

There is this in the _celery_start_process() function: when submitting a task to Celery fails, the process is deleted. IIRC that was added for a good reason, but it might be worth investigating the logs for that error message

https://github.com/workfloworchestrator/orchestrator-core/blob/main/orchestrator/services/celery.py#L73

Connection error when submitting task to Celery. Delete newly created process from database.