fix: handle concurrent XCom writes with savepoint + UPDATE fallback to avoid 409

[nagasrisai]

Closes #69956

Summary

When two task runners (concurrent execution or task retry) write the same XCom key at roughly the same time, both call XComModel.set which does a DELETE followed by an INSERT. Both DELETEs succeed (they each see the record that was just written by the peer or see nothing), then both try to INSERT the same primary key. The second INSERT hits the unique constraint and the Execution API converts the IntegrityError to an HTTP 409, causing the task to fail.

Fix

Wrap the XComModel.set call in a savepoint (session.begin_nested()) so that a constraint violation only rolls back the XCom write and not any earlier work in the same request (e.g. the task_map merge). The IntegrityError is caught and handled with an explicit UPDATE that overwrites the existing entry with the latest value, which is the correct semantic for a retry.

try:
    with session.begin_nested():
        XComModel.set(...)
except IntegrityError:
    # concurrent write already committed; overwrite with latest value
    session.execute(update(XComModel).where(...).values(value=value, dag_result=dag_result))

Changes

• airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
  • Add update to the sqlalchemy import
  • Add IntegrityError import from sqlalchemy.exc
  • Wrap XComModel.set in session.begin_nested() and catch IntegrityError with an UPDATE fallback