Fix MySQL deadlock when queueing asset-triggered Dags

[FrankYang0529]

Why

• When an asset event fans out to multiple consumer Dags, the non-Postgres path inserts asset_dag_run_queue rows one by one (SAVEPOINT + merge() per row).
• Concurrent producers of the same asset insert the same rows in nondeterministic set order, so on MySQL it acquires index locks in opposite order and gets deadlock (errno 1213).

How

• Update MySQL as the same single-statement bulk path Postgres already has: one multi-row upsert INSERT ... ON DUPLICATE KEY UPDATE.
• Insert queue rows sorted by dag_id on all backends, so concurrent fan-outs acquire row locks in a consistent order and cannot ABBA-deadlock each other.
• Add @retry_db_transaction to Trigger.submit_event.

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

---

Was generative AI tooling used to co-author this PR?

• Yes - Claude Code

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[viiccwen]

Hello PoAn,

@retry_db_transaction retries the entire submit_event() method, but AssetManager.register_asset_change() performs non-transactional side effects before the queue insert that may deadlock.

In particular, it invokes on_asset_changed, on_asset_event_emitted, and incre asset.updates before calling _queue_dagruns(). If the queue insert raises a deadlock:

1. the listeners and metric have already run.
2. the database transaction is rolled back.
3. submit_event() is replayed.
4. the same listeners and metric run again for the eventually successful event.

Listener plugins may perform external, non-idempotent operations such as sending webhooks or publishing messages, so those effects cannot be rolled back with the database transaction. This can also replay listeners for earlier assets if a trigger is associated with multiple assets and a later asset fails.

The new retry test mocks AssetManager.register_asset_change() entirely, so it only verifies that the decorator retries and does not cover these duplicated side effects.

IMO, the retry boundary needs to exclude listener/metric emission, or those side effects need to be deferred until the database transaction has successfully committed. : D

[FrankYang0529]

@viiccwen Thanks for the review. You're right. The listener hooks and the asset.updates metric run before _queue_dagruns() are not transactional and would run again on a replay.

Another path (task success → Execution API ti_update_state → register_asset_changes_in_db) retries when response is 5xx.

airflow/task-sdk/src/airflow/sdk/api/client.py

Lines 1208 to 1215 in 6ba529e

	@retry(
	retry=retry_if_exception(_should_retry_api_request),
	stop=stop_after_attempt(API_RETRIES),
	wait=wait_random_exponential(min=API_RETRY_WAIT_MIN, max=API_RETRY_WAIT_MAX),
	before_sleep=_log_and_trace_retry,
	reraise=True,
	)
	def request(self, *args, **kwargs):

However, the deadlock error doesn't return 5xx. It only marks task as failed.

airflow/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

Lines 467 to 478 in 6ba529e

	except Exception:
	# Set a task to failed in case any unexpected exception happened during task state update
	log.exception(
	"Error updating Task Instance state. Setting the task to failed.",
	payload=ti_patch_payload,
	)
	ti = session.get(TI, task_instance_id, with_for_update={"of": TI})
	if session.bind is not None:
	query = TI.duration_expression_update(timezone.utcnow(), query, session.bind)
	query = query.values(state=(updated_state := TaskInstanceState.FAILED))
	if ti is not None:
	_handle_fail_fast_for_dag(ti=ti, dag_id=dag_id, session=session, dag_bag=dag_bag)

I agree that we should remove @retry_db_transaction change to align the behavior.

Since issue author already created another PR #69977, I will close this one.