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

[dstandish]

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

[dstandish]

Opened #69977 to address this.

Fix taken: route MySQL to a single-statement INSERT ... ON DUPLICATE KEY UPDATE
(a no-op update on conflict), mirroring the existing PostgreSQL ON CONFLICT DO NOTHING
bulk path. This removes the per-row SAVEPOINT loop (and the secondary
SAVEPOINT ... does not exist error) and collapses the fan-out into one statement, holding
locks far more briefly — the practical mitigation for the InnoDB deadlock.

On the deadlock-retry backstop originally proposed here: after tracing the call path on
main, a clean localized retry is not feasible, so it is intentionally not part of this fix.
Asset-event registration runs inside the execution API TI state-update request
(_create_ti_state_update_query_and_update_state → TaskInstance.register_asset_changes_in_db
→ AssetManager.register_asset_change), whose transaction owner is the whole FastAPI request
(SessionDep); there is no request-level deadlock retry in the execution API. Since a deadlock
aborts that entire transaction:

• a @retry_db_transaction on the queueing method cannot revive the aborted outer transaction;
• splitting the queueing into its own retried transaction would break the atomicity the design
  requires — the queue rows must commit in the same transaction as the ti.state change, or a
  task could be marked SUCCESS while its downstream queueing rolls back;
• retrying the entire execution-API request would be a cross-cutting change across all
  endpoints, out of scope here.

The bulk upsert is therefore the appropriate resolution, and #69977 closes this.

---

Drafted-by: Claude Code (Opus 4.8) (no human review before posting)