Asset-triggered DAG queueing can deadlock on MySQL and is not retried #69938

Apache Airflow version

main (3.4.0.dev), and all 2.x/3.x versions with data-aware scheduling

What happened

When an asset (dataset) event fans out to multiple consumer DAGs, the scheduler/task
path inserts one row per consumer into asset_dag_run_queue (dataset_dag_run_queue
in 2.x). On MySQL/InnoDB, concurrent fan-outs contend on hose inserts and can raise:

(1213, 'Deadlock found when trying to get lock; try restarting transaction')

sometimes followed by a secondary error:

SAVEPOINT sa_savepoint_1 does not exist

The deadlock is not retried, so the enqueue of downstream runs for that event
fails. The SAVEPOINT ... does not exist line is a secondary symptom: InnoDB
discards the savepoint when it rolls back on deadlock, so SQLAlchemy's nested
transaction cleanup then fails against a savepoint that no longer exists.

Root cause

AssetManager._queue_dagruns (airflow-core/src/airflow/assets/manager.py:487)
has two branches:

• Postgres — _queue_dagruns_nonpartitioned_postgres (line 811) does a single
  bulk insert(...).on_conflict_do_nothing().
• Everything else (MySQL) — _queue_dagruns_nonpartitioned_slow_path (line 792)
  loops per row inside session.begin_nested() (SAVEPOINT) and catches only
  exc.IntegrityError (line 802).

Two problems with the non-Postgres path:

1. A deadlock is an OperationalError (errno 1213), not an IntegrityError, so
   it is not swallowed by the except at line 802 — it propagates.
2. Nothing on this call path (register_asset_change → _queue_dagruns →
   _queue_dagruns_nonpartitioned_slow_path) is wrapped in @retry_db_transaction,
   so there is no deadlock retry.

The per-row SAVEPOINT loop was introduced in #26103 (fix for #25210) specifically to
tolerate duplicate-key IntegrityError from concurrent producers of the same asset.
Deadlock handling was never in scope, and the row-by-row approach — many
lock-holding round-trips per event — actively increases InnoDB deadlock likelihood
compared to the single-statement Postgres path.

What should happen

Fanning out an asset event to consumers should be resilient to InnoDB deadlocks:
the enqueue should either avoid the deadlock-prone construct or retry transparently,
so downstream runs are reliably queued.

Proposed fix

1. Give MySQL a bulk path, mirroring Postgres, using
   sqlalchemy.dialects.mysql.insert(...).on_duplicate_key_update(...) (or
   INSERT IGNORE). This collapses N lock-holding round-trips into one statement and
   removes the per-row savepoint churn, sharply reducing the deadlock window. (This is
   not version-gated — the syntax has been available far below Airflow's MySQL 8.0+
   floor.)
2. Add deadlock retry to the queueing call (@retry_db_transaction or catch
   OperationalError/1213 and retry) so any residual InnoDB deadlock self-heals
   instead of surfacing as a failure. A single-statement bulk upsert still can
   deadlock on InnoDB gap/next-key locks, so (2) is the robust backstop and (1) is
   the probability reduction — both are worth doing.

The partitioned path should get the same treatment where applicable.

Notes

• Postgres is unaffected (it already uses the bulk ON CONFLICT path).
• This is a robustness/enhancement issue, not a regression; behavior is unchanged on
  Postgres and functionally correct on MySQL except under concurrent fan-out.

Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting