Concurrent/retried Execution API XCom writes fail with 409 due to duplicate primary key

[prayaagmahajan]

Under which category would you file this issue?

Task SDK

Apache Airflow version

3.2.2

What happened and how to reproduce it?

An Airflow task runner attempted to store an XCom through the Airflow 3
Execution API:

POST /execution/xcoms/{dag_id}/{run_id}/{task_id}/{key}

The API server attempted the following statement:

INSERT INTO xcom
(dag_run_id, task_id, map_index, key, dag_id, run_id, value, timestamp)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)

The insert failed with an IntegrityError and the Execution API returned:

HTTP 409 Conflict

Relevant access log:

[2026-07-16T07:14:19.285110Z] request finished
method=POST
path=/execution/xcoms/{Run_ID}/runOnce__2026-07-16T07:01:26.419019+00:00/{Task_SENSOR}/{Task_SENSOR}
status_code=409

Three XCom INSERT failures occurred at approximately the same time.

The Airflow API exception handler classified the MySQL IntegrityError as:

"Unique constraint violation"

The XCom primary key is:

(dag_run_id, task_id, map_index, key)

In Airflow 3.2.2, XComModel.set() first deletes an existing XCom and
then inserts the replacement:

1. DELETE matching XCom
2. INSERT new XCom

With overlapping requests, both transactions can perform the DELETE
before either INSERT commits. One INSERT succeeds and another fails
with a duplicate primary-key error.

The approximately 81-second request duration may indicate that one
INSERT waited for another database transaction before the duplicate
conflict was returned.

What you think should happen instead?

Repeated or concurrent writes of the same XCom identity should be
handled idempotently.

Possible expected behavior:

• Use a database-appropriate atomic upsert.
• Serialize writes for the same XCom identity.
• If a repeated request contains the same value, treat it as already
  completed.
• Retry the write safely after an IntegrityError caused by the
  delete-then-insert race.

A transient/repeated Execution API request should not fail task
execution merely because the same XCom was already written.

Operating System

linux

Deployment

Docker-Compose

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

• Airflow: 3.2.2
• Python: 3.11
• Metadata database: MySQL 9.7.0

### Relevant log output

[2026-07-16T07:14:19.283188Z] ERROR - Error with id tuw1pVaf,
statement: INSERT INTO xcom
(dag_run_id, task_id, map_index, `key`, dag_id, run_id, value, timestamp)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)

File "airflow/api_fastapi/execution_api/routes/xcoms.py", line 392,
in set_xcom
    XComModel.set(...)

File "airflow/models/xcom.py", line 246, in set
    session.flush()

File "MySQLdb/cursors.py", line 179, in execute
    res = self._query(mogrified_query)

[2026-07-16T07:14:19.285110Z] request finished
method=POST
path=/execution/xcoms/.../.../.../...
status_code=409
duration_us=81032908

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.