Retry transient deadlocks in Trigger.submit_event/submit_failure

[uplsh580]

Retry transient MySQL/InnoDB deadlocks on the triggerer's per-event task_instance UPDATE so a single (1213, 'Deadlock found ...') no longer takes down the triggerer process.

Problem

When the triggerer fires an event, it resumes (or fails) the dependent deferred task instances through a single-row UPDATE:

handle_events -> Trigger.submit_event -> handle_event_submit -> session.flush()

UPDATE task_instance
SET state='scheduled', scheduled_dttm=..., updated_at=..., trigger_id=NULL, next_kwargs=...
WHERE task_instance.id = '...'

That UPDATE contends for task_instance row locks with the scheduler's bulk writes (e.g. SchedulerJobRunner.check_trigger_timeouts, Trigger.clean_unused) and, with more than one triggerer replica, with the other triggerer(s). On MySQL/InnoDB the lock-acquisition order differs between the set-based and row-by-row writers, so InnoDB occasionally aborts one side with (1213, 'Deadlock found when trying to get lock; try restarting transaction').

Neither submit_event/submit_failure nor anything above them in the triggerer call chain retries or catches this, so the DBAPIError propagates up to TriggerRunnerSupervisor.run and the triggerer process exits. Deferred tasks are picked up by the other replica, so no task fails, but the restart is noisy at the alerting level and (as reported) the process sometimes has to be SIGKILLed.

Reproduced on 3.1.7/3.1.8 and still reproducing on 3.2.2. This is the single-row path, which is not covered by the existing bulk-UPDATE PRs (#65836, #65920) or by the triggerer comms-channel fix (#66412).

Fix

Decorate Trigger.submit_event and Trigger.submit_failure with @retry_db_transaction, stacked under @provide_session exactly like the existing model-side usages (DagWarning.purge_inactive_dag_warnings, RenderedTaskInstanceFields). This is the same retry-on-deadlock treatment the bulk paths already get via run_with_db_retries(); it just extends it to the per-event single-row path.

This is safe and idempotent here:

• Both methods are only ever called from the triggerer without an outer session, so @provide_session owns the whole transaction — a rollback-and-retry cannot corrupt a caller's transaction.
• On retry the method re-runs its SELECT ... WHERE state == DEFERRED and re-applies the state transition, so re-processing the still-deferred rows is idempotent.

This is an incremental mitigation, not a redesign: it does not remove the concurrent-writer contention described in the issue, but it stops a single transient deadlock from killing the triggerer.

Tests

Added two regression tests in airflow-core/tests/unit/models/test_trigger.py that inject a 1213-style OperationalError on the first attempt of each path and assert the transaction is retried and the task instance still ends up SCHEDULED.

related: #65818

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code following the guidelines