langgenius/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-20 19:50:06 -04:00
Code Issues 775 Packages Projects Releases 10 Wiki Activity

[Bug]: Race conditions in Celery tasks cause IntegrityError and state inconsistency in Workflow/Node execution storage #21760

New Issue
Open
opened 2026-02-21 20:14:11 -05:00 by yindo · 3 comments
yindo commented 2026-02-21 20:14:11 -05:00
Owner
Copy Link
Copy Source

Originally created by @NieRonghua on GitHub (Jan 19, 2026).

Originally assigned to: @fatelei on GitHub.

Self Checks

  • I have read the Contributing Guide and Language Policy.
  • This is only for bug report, if you would like to ask a question, please head to Discussions.
  • I have searched for existing issues search for existing issues, including closed ones.
  • I confirm that I am using English to submit this report, otherwise it will be closed.
  • 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
  • Please do not modify this template :) and fill in all the required fields.

Dify version

1.11.1

Cloud or Self Hosted

Self Hosted (Docker)

Steps to reproduce

Description

In high-concurrency scenarios or when workflow nodes execute very rapidly, the asynchronous tasks responsible for saving execution states (save_workflow_node_execution_task and save_workflow_execution_task) encounter race conditions. This leads to two critical problems:

  1. Database Integrity Errors: Frequent Duplicate entry errors in Celery worker logs.
  2. State Inconsistency (Regression): A completed workflow/node execution can be reverted to RUNNING state if messages are processed out of order.

Root Cause Analysis

The current implementation uses a "Check-Then-Act" pattern which is not atomic:

# Current Logic (Simplified)
existing = session.scalar(select(Model).where(Model.id == id)) # Time T1
if existing:
    update(existing)
else:
    # Time T2 - Another worker might insert here!
    session.add(new_record)
    session.commit() # Crash! IntegrityError

Race Condition Scenario (Integrity Error)

Time Worker A (Task: Node Start) Worker B (Task: Node Finish) DB State
T0 Starts processing Starts processing Empty
T1 SELECT (Not Found) SELECT (Not Found) Empty
T2 session.add(new) session.add(new) Empty
T3 session.commit() -> Success Inserted by A
T4 session.commit() -> Fail IntegrityError 
# error log
The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/app/deploy/tasks/workflow_node_execution_tasks.py", line 84, in save_workflow_node_execution_task
    session.commit()
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 2028, in commit
    trans.commit(_to_root=True)
  File "<string>", line 2, in commit
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/state_changes.py", line 139, in _go
    ret_value = fn(self, *arg, **kw)
                ^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1313, in commit
    self._prepare_impl()
  File "<string>", line 2, in _prepare_impl
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/state_changes.py", line 139, in _go
    ret_value = fn(self, *arg, **kw)
                ^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1288, in _prepare_impl
    self.session.flush()
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4352, in flush
    self._flush(objects)
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4487, in _flush
    with util.safe_reraise():
         ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4448, in _flush
    flush_context.execute()
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/unitofwork.py", line 466, in execute
    rec.execute(self)
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/unitofwork.py", line 642, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/persistence.py", line 93, in save_obj
    _emit_insert_statements(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/persistence.py", line 1048, in _emit_insert_statements
    result = connection.execute(
             ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1418, in execute
    return meth(
           ^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/sql/elements.py", line 515, in _execute_on_connection
    return connection._execute_clauseelement(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1640, in _execute_clauseelement
    ret = self._execute_context(
          ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1846, in _execute_context
    return self._exec_single_context(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1986, in _exec_single_context
    self._handle_dbapi_exception(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 2355, in _handle_dbapi_exception
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 941, in do_execute
    cursor.execute(statement, parameters)
  File "/usr/local/lib/python3.12/site-packages/pymysql/cursors.py", line 153, in execute
    result = self._query(query)
             ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pymysql/cursors.py", line 322, in _query
    conn.query(q)
  File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 563, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 825, in _read_query_result
    result.read()
  File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 1199, in read
    first_packet = self.connection._read_packet()
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 775, in _read_packet
    packet.raise_for_error()
  File "/usr/local/lib/python3.12/site-packages/pymysql/protocol.py", line 219, in raise_for_error
    err.raise_mysql_exception(self._data)
  File "/usr/local/lib/python3.12/site-packages/pymysql/err.py", line 150, in raise_mysql_exception
    raise errorclass(errno, errval)
sqlalchemy.exc.IntegrityError: (pymysql.err.IntegrityError) (1062, "Duplicate entry '158b3f02-7498-4d6e-bbb2-ba5b9e3bcaf5-2026-01-19 14:32:38' for key 'workflow_node_executions.PRIMARY'")
[SQL: INSERT INTO workflow_node_executions (id, tenant_id, app_id, workflow_id, triggered_from, workflow_run_id, `index`, predecessor_node_id, node_execution_id, node_id, node_type, title, inputs, process_data, outputs, status, error, elapsed_time, execution_metadata, created_at, created_by_role, created_by, finished_at) VALUES (%(id)s, %(tenant_id)s, %(app_id)s, %(workflow_id)s, %(triggered_from)s, %(workflow_run_id)s, %(index)s, %(predecessor_node_id)s, %(node_execution_id)s, %(node_id)s, %(node_type)s, %(title)s, %(inputs)s, %(process_data)s, %(outputs)s, %(status)s, %(error)s, %(elapsed_time)s, %(execution_metadata)s, %(created_at)s, %(created_by_role)s, %(created_by)s, %(finished_at)s)]
[parameters: {'id': '158b3f02-7498-4d6e-bbb2-ba5b9e3bcaf5', 'tenant_id': 'f2276ff4-8044-4af1-a4d0-9b971d24d6d2', 'app_id': '9bafcf4a-2d3d-4518-af5b-ff47b050b0a4', 'workflow_id': '7fb5db11-d627-4ba2-a7a8-a2b627bce4b5', 'triggered_from': 'workflow-run', 'workflow_run_id': 'd632b56b-e48e-496e-a92c-e36ec6946dc8', 'index': 21, 'predecessor_node_id': None, 'node_execution_id': '158b3f02-7498-4d6e-bbb2-ba5b9e3bcaf5', 'node_id': '1752546413547', 'node_type': 'answer', 'title': '【联网】最终出话拼接', 'inputs': '{}', 'process_data': '{}', 'outputs': '{"answer": "<think>\\n\\u9996\\u5148\\uff0c\\u7528\\u6237\\u95ee\\u9898\\u662f\\uff1a\\u201c\\u987a\\u4e30\\u548c\\u6781\\u5154\\u5408\\u4f5c\\u4e3b\ ... (13664 characters truncated) ... 5\\u8865\\u672b\\u7aef\\u77ed\\u677f\\uff0c\\u4f18\\u5316\\u5168\\u94fe\\u6761\\u670d\\u52a1 |", "files": {"value_type": "array[file]", "value": []}}', 'status': 'succeeded', 'error': None, 'elapsed_time': 0.006605, 'execution_metadata': '{}', 'created_at': datetime.datetime(2026, 1, 19, 14, 32, 38, 468084), 'created_by_role': 'account', 'created_by': 'daf6d8aa-4d06-4416-ab71-c16d64499c6c', 'finished_at': datetime.datetime(2026, 1, 19, 14, 32, 38, 474689)}]

Race Condition Scenario (State Regression)

Even worse, if retries are involved or queues are delayed:

  1. Worker A processes "SUCCEEDED" event: Inserts record with status SUCCEEDED.
  2. Worker B processes a delayed "RUNNING" event (originally generated earlier but processed later):
    • Finds existing record (Status: SUCCEEDED).
    • Updates status to RUNNING (Overwrite!).
  3. Result: The workflow appears stuck in RUNNING forever in the UI, even though it actually finished.

Impact

  • Operational Noise: High volume of sqlalchemy.exc.IntegrityError in logs masks real issues.
  • Data Corruption: Valid final states are lost, causing workflows to "hang".
  • Resource Waste: Failed tasks trigger Celery retries (exponential backoff), unnecessarily loading the system.

Affected Modules

  1. Workflow Node Storage: api/tasks/workflow_node_execution_tasks.py
    • Function: save_workflow_node_execution_task
    • Table: workflow_node_executions
  2. Workflow Run Storage: api/tasks/workflow_execution_tasks.py
    • Function: save_workflow_execution_task
    • Table: workflow_runs

✔️ Expected Behavior

  1. Concurrency Resilience: The system should strictly handle concurrent INSERT operations for the same Primary Key. If a duplicate key is detected during insertion, it must automatically failover to an UPDATE operation without raising an unhandled exception or crashing the worker task.
  2. State Monotonicity: Workflow and Node execution statuses should follow a strictly monotonic transition from "Non-Terminal" (e.g., RUNNING) to "Terminal" (e.g., SUCCEEDED, FAILED).
    • Once a record reaches a Terminal state in the database, subsequent update requests carrying a Non-Terminal state (e.g., delayed packets) must be ignored or partially applied (updating only metadata) without reverting the status.
  3. Zero Integrity Errors: Under normal load or high concurrency, the Celery worker logs should remain free of sqlalchemy.exc.IntegrityError related to duplicate primary keys for execution tables.

❌ Actual Behavior

No response

Originally created by @NieRonghua on GitHub (Jan 19, 2026). Originally assigned to: @fatelei on GitHub. ### Self Checks - [x] I have read the [Contributing Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) and [Language Policy](https://github.com/langgenius/dify/issues/1542). - [x] This is only for bug report, if you would like to ask a question, please head to [Discussions](https://github.com/langgenius/dify/discussions/categories/general). - [x] I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones. - [x] I confirm that I am using English to submit this report, otherwise it will be closed. - [x] 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：） - [x] Please do not modify this template :) and fill in all the required fields. ### Dify version 1.11.1 ### Cloud or Self Hosted Self Hosted (Docker) ### Steps to reproduce ### Description In high-concurrency scenarios or when workflow nodes execute very rapidly, the asynchronous tasks responsible for saving execution states (`save_workflow_node_execution_task` and `save_workflow_execution_task`) encounter race conditions. This leads to two critical problems: 1. **Database Integrity Errors**: Frequent `Duplicate entry` errors in Celery worker logs. 2. **State Inconsistency (Regression)**: A completed workflow/node execution can be reverted to `RUNNING` state if messages are processed out of order. ### Root Cause Analysis The current implementation uses a "Check-Then-Act" pattern which is not atomic: ```python # Current Logic (Simplified) existing = session.scalar(select(Model).where(Model.id == id)) # Time T1 if existing: update(existing) else: # Time T2 - Another worker might insert here! session.add(new_record) session.commit() # Crash! IntegrityError ``` #### Race Condition Scenario (Integrity Error) | Time | Worker A (Task: Node Start) | Worker B (Task: Node Finish) | DB State | | :--- | :--- | :--- | :--- | | T0 | Starts processing | Starts processing | Empty | | T1 | SELECT (Not Found) | SELECT (Not Found) | Empty | | T2 | `session.add(new)` | `session.add(new)` | Empty | | T3 | `session.commit()` -> **Success** | | Inserted by A | | T4 | | `session.commit()` -> **Fail** | **IntegrityError** | ``` # error log The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/app/deploy/tasks/workflow_node_execution_tasks.py", line 84, in save_workflow_node_execution_task session.commit() File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 2028, in commit trans.commit(_to_root=True) File "<string>", line 2, in commit File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/state_changes.py", line 139, in _go ret_value = fn(self, *arg, **kw) ^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1313, in commit self._prepare_impl() File "<string>", line 2, in _prepare_impl File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/state_changes.py", line 139, in _go ret_value = fn(self, *arg, **kw) ^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1288, in _prepare_impl self.session.flush() File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4352, in flush self._flush(objects) File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4487, in _flush with util.safe_reraise(): ^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__ raise exc_value.with_traceback(exc_tb) File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4448, in _flush flush_context.execute() File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/unitofwork.py", line 466, in execute rec.execute(self) File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/unitofwork.py", line 642, in execute util.preloaded.orm_persistence.save_obj( File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/persistence.py", line 93, in save_obj _emit_insert_statements( File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/persistence.py", line 1048, in _emit_insert_statements result = connection.execute( ^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1418, in execute return meth( ^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/sql/elements.py", line 515, in _execute_on_connection return connection._execute_clauseelement( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1640, in _execute_clauseelement ret = self._execute_context( ^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1846, in _execute_context return self._exec_single_context( ^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1986, in _exec_single_context self._handle_dbapi_exception( File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 2355, in _handle_dbapi_exception raise sqlalchemy_exception.with_traceback(exc_info[2]) from e File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context self.dialect.do_execute( File "/usr/local/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 941, in do_execute cursor.execute(statement, parameters) File "/usr/local/lib/python3.12/site-packages/pymysql/cursors.py", line 153, in execute result = self._query(query) ^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/pymysql/cursors.py", line 322, in _query conn.query(q) File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 563, in query self._affected_rows = self._read_query_result(unbuffered=unbuffered) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 825, in _read_query_result result.read() File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 1199, in read first_packet = self.connection._read_packet() ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/pymysql/connections.py", line 775, in _read_packet packet.raise_for_error() File "/usr/local/lib/python3.12/site-packages/pymysql/protocol.py", line 219, in raise_for_error err.raise_mysql_exception(self._data) File "/usr/local/lib/python3.12/site-packages/pymysql/err.py", line 150, in raise_mysql_exception raise errorclass(errno, errval) sqlalchemy.exc.IntegrityError: (pymysql.err.IntegrityError) (1062, "Duplicate entry '158b3f02-7498-4d6e-bbb2-ba5b9e3bcaf5-2026-01-19 14:32:38' for key 'workflow_node_executions.PRIMARY'") [SQL: INSERT INTO workflow_node_executions (id, tenant_id, app_id, workflow_id, triggered_from, workflow_run_id, `index`, predecessor_node_id, node_execution_id, node_id, node_type, title, inputs, process_data, outputs, status, error, elapsed_time, execution_metadata, created_at, created_by_role, created_by, finished_at) VALUES (%(id)s, %(tenant_id)s, %(app_id)s, %(workflow_id)s, %(triggered_from)s, %(workflow_run_id)s, %(index)s, %(predecessor_node_id)s, %(node_execution_id)s, %(node_id)s, %(node_type)s, %(title)s, %(inputs)s, %(process_data)s, %(outputs)s, %(status)s, %(error)s, %(elapsed_time)s, %(execution_metadata)s, %(created_at)s, %(created_by_role)s, %(created_by)s, %(finished_at)s)] [parameters: {'id': '158b3f02-7498-4d6e-bbb2-ba5b9e3bcaf5', 'tenant_id': 'f2276ff4-8044-4af1-a4d0-9b971d24d6d2', 'app_id': '9bafcf4a-2d3d-4518-af5b-ff47b050b0a4', 'workflow_id': '7fb5db11-d627-4ba2-a7a8-a2b627bce4b5', 'triggered_from': 'workflow-run', 'workflow_run_id': 'd632b56b-e48e-496e-a92c-e36ec6946dc8', 'index': 21, 'predecessor_node_id': None, 'node_execution_id': '158b3f02-7498-4d6e-bbb2-ba5b9e3bcaf5', 'node_id': '1752546413547', 'node_type': 'answer', 'title': '【联网】最终出话拼接', 'inputs': '{}', 'process_data': '{}', 'outputs': '{"answer": "<think>\\n\\u9996\\u5148\\uff0c\\u7528\\u6237\\u95ee\\u9898\\u662f\\uff1a\\u201c\\u987a\\u4e30\\u548c\\u6781\\u5154\\u5408\\u4f5c\\u4e3b\ ... (13664 characters truncated) ... 5\\u8865\\u672b\\u7aef\\u77ed\\u677f\\uff0c\\u4f18\\u5316\\u5168\\u94fe\\u6761\\u670d\\u52a1 |", "files": {"value_type": "array[file]", "value": []}}', 'status': 'succeeded', 'error': None, 'elapsed_time': 0.006605, 'execution_metadata': '{}', 'created_at': datetime.datetime(2026, 1, 19, 14, 32, 38, 468084), 'created_by_role': 'account', 'created_by': 'daf6d8aa-4d06-4416-ab71-c16d64499c6c', 'finished_at': datetime.datetime(2026, 1, 19, 14, 32, 38, 474689)}] ``` #### Race Condition Scenario (State Regression) Even worse, if retries are involved or queues are delayed: 1. **Worker A** processes "SUCCEEDED" event: Inserts record with status `SUCCEEDED`. 2. **Worker B** processes a delayed "RUNNING" event (originally generated earlier but processed later): * Finds existing record (Status: `SUCCEEDED`). * Updates status to `RUNNING` (Overwrite!). 3. **Result**: The workflow appears stuck in `RUNNING` forever in the UI, even though it actually finished. ### Impact * **Operational Noise**: High volume of `sqlalchemy.exc.IntegrityError` in logs masks real issues. * **Data Corruption**: Valid final states are lost, causing workflows to "hang". * **Resource Waste**: Failed tasks trigger Celery retries (exponential backoff), unnecessarily loading the system. ### Affected Modules 1. **Workflow Node Storage**: `api/tasks/workflow_node_execution_tasks.py` * Function: `save_workflow_node_execution_task` * Table: `workflow_node_executions` 2. **Workflow Run Storage**: `api/tasks/workflow_execution_tasks.py` * Function: `save_workflow_execution_task` * Table: `workflow_runs` ### ✔️ Expected Behavior 1. **Concurrency Resilience**: The system should strictly handle concurrent `INSERT` operations for the same Primary Key. If a duplicate key is detected during insertion, it must automatically failover to an `UPDATE` operation without raising an unhandled exception or crashing the worker task. 2. **State Monotonicity**: Workflow and Node execution statuses should follow a strictly monotonic transition from "Non-Terminal" (e.g., `RUNNING`) to "Terminal" (e.g., `SUCCEEDED`, `FAILED`). * Once a record reaches a **Terminal** state in the database, subsequent update requests carrying a **Non-Terminal** state (e.g., delayed packets) must be ignored or partially applied (updating only metadata) without reverting the status. 3. **Zero Integrity Errors**: Under normal load or high concurrency, the Celery worker logs should remain free of `sqlalchemy.exc.IntegrityError` related to duplicate primary keys for execution tables. ### ❌ Actual Behavior _No response_
yindo added the 🐞 bugstale labels 2026-02-21 20:14:11 -05:00
yindo commented 2026-02-21 20:14:11 -05:00
Author
Owner
Copy Link
Copy Source

@dosubot[bot] commented on GitHub (Jan 19, 2026):

You're correct—Dify's Celery-based workflow/node execution storage currently uses a non-atomic check-then-act pattern, which is vulnerable to race conditions under high concurrency. This leads to IntegrityErrors (duplicate primary keys) and state inconsistencies, such as terminal states being overwritten by non-terminal ones if events are processed out of order. The affected code is in save_workflow_node_execution_task and save_workflow_execution_task, which do not implement atomic upsert logic or enforce monotonic state transitions, so these issues persist in async/Celery mode source.

A merged PR (#24643) fixed this for synchronous storage by switching to UUID v7 and adding a retry mechanism, but this fix has not been applied to Celery/asynchronous tasks. As of v2.0.0, there is still no official fix for async storage—Celery writes are only safe with a single worker unless you implement distributed locking, idempotency, or robust upsert logic source.

Workarounds until a fix is released include: running only one Celery worker for workflow storage tasks, manually recovering stuck nodes by updating their status in the database, and adding robust exception handling and monitoring for duplicate key errors. Some users have temporarily disabled Celery async writing to avoid the issue source. For MySQL, you can also set the session isolation level to READ COMMITTED to avoid visibility issues, but this does not solve the race condition itself source.

The best long-term solution is to implement atomic upsert logic (e.g., INSERT ... ON DUPLICATE KEY UPDATE or equivalent) and enforce monotonic state transitions so terminal states can't be overwritten by earlier events. Keep an eye on future releases and PRs for a comprehensive fix.

To reply, just mention @dosu.

How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other

Chat with Dosu Open in Cursor Join Discord Share on X

@dosubot[bot] commented on GitHub (Jan 19, 2026): <!-- Answer --> You're correct—Dify's Celery-based workflow/node execution storage currently uses a non-atomic check-then-act pattern, which is vulnerable to race conditions under high concurrency. This leads to IntegrityErrors (duplicate primary keys) and state inconsistencies, such as terminal states being overwritten by non-terminal ones if events are processed out of order. The affected code is in `save_workflow_node_execution_task` and `save_workflow_execution_task`, which do not implement atomic upsert logic or enforce monotonic state transitions, so these issues persist in async/Celery mode [source](https://github.com/langgenius/dify/blob/0f1db88dcb5952a3f299ca8ace37c77708045531/api/tasks/workflow_node_execution_tasks.py#L59-L170). A merged PR ([#24643](https://github.com/langgenius/dify/pull/24643)) fixed this for synchronous storage by switching to UUID v7 and adding a retry mechanism, but this fix has not been applied to Celery/asynchronous tasks. As of v2.0.0, there is still no official fix for async storage—Celery writes are only safe with a single worker unless you implement distributed locking, idempotency, or robust upsert logic [source](https://github.com/langgenius/dify/issues/25135). Workarounds until a fix is released include: running only one Celery worker for workflow storage tasks, manually recovering stuck nodes by updating their status in the database, and adding robust exception handling and monitoring for duplicate key errors. Some users have temporarily disabled Celery async writing to avoid the issue [source](https://github.com/langgenius/dify/issues/25135). For MySQL, you can also set the session isolation level to READ COMMITTED to avoid visibility issues, but this does not solve the race condition itself [source](https://github.com/langgenius/dify/issues/28999). The best long-term solution is to implement atomic upsert logic (e.g., `INSERT ... ON DUPLICATE KEY UPDATE` or equivalent) and enforce monotonic state transitions so terminal states can't be overwritten by earlier events. Keep an eye on future releases and PRs for a comprehensive fix. <!-- Dosu Comment Footer --> *To reply, just mention [@dosu](https://go.dosu.dev/dosubot).* --- <sup>How did I do? [Good](https://app.dosu.dev/response-feedback/aa9b67cd-8671-4e30-82a6-169f958b8e78?feedback_type=great_response) | [Irrelevant](https://app.dosu.dev/response-feedback/aa9b67cd-8671-4e30-82a6-169f958b8e78?feedback_type=irrelevant_answer) | [Incorrect](https://app.dosu.dev/response-feedback/aa9b67cd-8671-4e30-82a6-169f958b8e78?feedback_type=incorrect_sources) | [Verbose](https://app.dosu.dev/response-feedback/aa9b67cd-8671-4e30-82a6-169f958b8e78?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/aa9b67cd-8671-4e30-82a6-169f958b8e78?feedback_type=hallucination) | [Report 🐛](https://app.dosu.dev/response-feedback/aa9b67cd-8671-4e30-82a6-169f958b8e78?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/aa9b67cd-8671-4e30-82a6-169f958b8e78?feedback_type=other)</sup> [![Chat with Dosu](https://dosu.dev/dosu-chat-badge.svg)](https://app.dosu.dev/6088a4f4-073a-43f3-b3d4-4f06fb0c05a0/ask?utm_source=github)&nbsp;[![Open in Cursor](https://dosu.dev/dosu-in-cursor.svg)](https://cursor.com/link/prompt?text=You%27re%20correct%E2%80%94Dify%27s%20Celery-based%20workflow/node%20execution%20storage%20currently%20uses%20a%20non-atomic%20check-then-act%20pattern%2C%20which%20is%20vulnerable%20to%20race%20conditions%20under%20high%20concurrency.%20This%20leads%20to%20IntegrityErrors%20%28duplicate%20primary%20keys%29%20and%20state%20inconsistencies%2C%20such%20as%20terminal%20states%20being%20overwritten%20by%20non-terminal%20ones%20if%20events%20are%20processed%20out%20of%20order.%20The%20affected%20code%20is%20in%20%60save_workflow_node_execution_task%60%20and%20%60save_workflow_execution_task%60%2C%20which%20do%20not%20implement%20atomic%20upsert%20logic%20or%20enforce%20monotonic%20state%20transitions%2C%20so%20these%20issues%20persist%20in%20async/Celery%20mode%20%5Bsource%5D%28https%3A//github.com/langgenius/dify/blob/0f1db88dcb5952a3f299ca8ace37c77708045531/api/tasks/workflow_node_execution_tasks.py%23L59-L170%29.%0A%0AA%20merged%20PR%20%28%5B%2324643%5D%28https%3A//github.com/langgenius/dify/pull/24643%29%29%20fixed%20this%20for%20synchronous%20storage%20by%20switching%20to%20UUID%20v7%20and%20adding%20a%20retry%20mechanism%2C%20but%20this%20fix%20has%20not%20been%20applied%20to%20Celery/asynchronous%20tasks.%20As%20of%20v2.0.0%2C%20there%20is%20still%20no%20official%20fix%20for%20async%20storage%E2%80%94Celery%20writes%20are%20only%20safe%20with%20a%20single%20worker%20unless%20you%20implement%20distributed%20locking%2C%20idempotency%2C%20or%20robust%20upsert%20logic%20%5Bsource%5D%28https%3A//github.com/langgenius/dify/issues/25135%29.%0A%0AWorkarounds%20until%20a%20fix%20is%20released%20include%3A%20running%20only%20one%20Celery%20worker%20for%20workflow%20storage%20tasks%2C%20manually%20recovering%20stuck%20nodes%20by%20updating%20their%20status%20in%20the%20database%2C%20and%20adding%20robust%20exception%20handling%20and%20monitoring%20for%20duplicate%20key%20errors.%20Some%20users%20have%20temporarily%20disabled%20Celery%20async%20writing%20to%20avoid%20the%20issue%20%5Bsource%5D%28https%3A//github.com/langgenius/dify/issues/25135%29.%20For%20MySQL%2C%20you%20can%20also%20set%20the%20session%20isolation%20level%20to%20READ%20COMMITTED%20to%20avoid%20visibility%20issues%2C%20but%20this%20does%20not%20solve%20the%20race%20condition%20itself%20%5Bsource%5D%28https%3A//github.com/langgenius/dify/issues/28999%29.%0A%0AThe%20best%20long-term%20solution%20is%20to%20implement%20atomic%20upsert%20logic%20%28e.g.%2C%20%60INSERT%20...%20ON%20DUPLICATE%20KEY%20UPDATE%60%20or%20equivalent%29%20and%20enforce%20monotonic%20state%20transitions%20so%20terminal%20states%20can%27t%20be%20overwritten%20by%20earlier%20events.%20Keep%20an%20eye%20on%20future%20releases%20and%20PRs%20for%20a%20comprehensive%20fix.)&nbsp;[![Join Discord](https://img.shields.io/badge/join-5865F2?logo=discord&logoColor=white&label=)](https://go.dosu.dev/discord-bot)&nbsp;[![Share on X](https://img.shields.io/badge/X-share-black)](https://twitter.com/intent/tweet?text=%40dosu_ai%20helped%20me%20solve%20this%20issue!&url=https%3A//github.com/langgenius/dify/issues/31222)
yindo commented 2026-02-21 20:14:11 -05:00
Author
Owner
Copy Link
Copy Source

@NieRonghua commented on GitHub (Jan 19, 2026):

@fatelei
I've submitted a PR to fix this issue. Please review it. The modified code is already running in my production environment.

@NieRonghua commented on GitHub (Jan 19, 2026): @fatelei I've submitted a PR to fix this issue. Please review it. The modified code is already running in my production environment.
yindo commented 2026-02-21 20:14:12 -05:00
Author
Owner
Copy Link
Copy Source

@dosubot[bot] commented on GitHub (Feb 19, 2026):

Hi, @NieRonghua. I'm Dosu, and I'm helping the Dify team manage their backlog and am marking this issue as stale.

Issue Summary:

  • You reported race conditions in Dify's Celery-based workflow/node execution storage causing IntegrityErrors and state inconsistencies.
  • The problem stems from a non-atomic check-then-act pattern, with fixes so far only applied to synchronous storage.
  • Workarounds include using a single-worker Celery setup and manual recovery processes.
  • You submitted a PR addressing the issue, which is deployed in your production environment but still pending review.
  • The long-term fix involves implementing atomic upsert logic and enforcing monotonic state transitions.

Next Steps:

  • Please let me know if this issue is still relevant with the latest version of Dify by commenting here to keep the discussion open.
  • Otherwise, this issue will be automatically closed in 15 days.

Thanks for your understanding and contribution!

@dosubot[bot] commented on GitHub (Feb 19, 2026): Hi, @NieRonghua. I'm [Dosu](https://dosu.dev), and I'm helping the Dify team manage their backlog and am marking this issue as stale. **Issue Summary:** - You reported race conditions in Dify's Celery-based workflow/node execution storage causing IntegrityErrors and state inconsistencies. - The problem stems from a non-atomic check-then-act pattern, with fixes so far only applied to synchronous storage. - Workarounds include using a single-worker Celery setup and manual recovery processes. - You submitted a PR addressing the issue, which is deployed in your production environment but still pending review. - The long-term fix involves implementing atomic upsert logic and enforcing monotonic state transitions. **Next Steps:** - Please let me know if this issue is still relevant with the latest version of Dify by commenting here to keep the discussion open. - Otherwise, this issue will be automatically closed in 15 days. Thanks for your understanding and contribution!
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
Knowledge Pipeline
Proposal
WIP
cloud
dependencies
feat:webapp
good first issue
outdated
performance
pull-request
Mirrored from GitHub Pull Request
 python:uv
refactor
size:L
stale
status: accepting prs
status: blocked
status: in discussion
stt & tts
vulnerablity
web
wont-fix
🌊 feat:workflow
🌚 invalid
🐞 bug
👻 feat:rag
💪 enhancement
📚 documentation
🔨 feat:tools
🙋‍♂️ question
🙏 help wanted
🤔 cant-reproduce
🤖 feat:agent
No Label stale 🐞 bug
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langgenius/dify#21760
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?