diff --git a/airflow-core/src/airflow/models/trigger.py b/airflow-core/src/airflow/models/trigger.py index 4915637c88d4b..439c0ca07ed4b 100644 --- a/airflow-core/src/airflow/models/trigger.py +++ b/airflow-core/src/airflow/models/trigger.py @@ -37,7 +37,7 @@ from airflow.models.taskinstance import TaskInstance from airflow.serialization.enums import stringify_encoding_keys from airflow.triggers.base import BaseTaskEndEvent -from airflow.utils.retries import run_with_db_retries +from airflow.utils.retries import retry_db_transaction, run_with_db_retries from airflow.utils.session import NEW_SESSION, provide_session from airflow.utils.sqlalchemy import UtcDateTime, get_dialect_name, with_row_locks from airflow.utils.state import TaskInstanceState @@ -270,12 +270,20 @@ def clean_unused(cls, *, session: Session = NEW_SESSION) -> None: @classmethod @provide_session + @retry_db_transaction def submit_event(cls, trigger_id, event: TriggerEvent, *, session: Session = NEW_SESSION) -> None: """ Fire an event. Resume all tasks that were in deferred state. Send an event to all assets associated to the trigger. + + The per-task-instance UPDATE issued here contends with the scheduler's bulk + ``task_instance`` writes (e.g. ``check_trigger_timeouts``) and with other triggerer + replicas, so a transient MySQL/InnoDB deadlock (error 1213) is possible. Retrying the + transaction keeps a single deadlock from propagating up to ``handle_events`` and killing + the triggerer process. The retried body re-reads the still-deferred rows, so it is + idempotent. See https://github.com/apache/airflow/issues/65818. """ # Resume deferred tasks for task_instance in session.scalars( @@ -305,6 +313,7 @@ def submit_event(cls, trigger_id, event: TriggerEvent, *, session: Session = NEW @classmethod @provide_session + @retry_db_transaction def submit_failure(cls, trigger_id, exc=None, *, session: Session = NEW_SESSION) -> None: """ When a trigger has failed unexpectedly, mark everything that depended on it as failed. @@ -316,6 +325,11 @@ def submit_failure(cls, trigger_id, exc=None, *, session: Session = NEW_SESSION) We use a special __fail__ value for next_method to achieve this that the runtime code understands as immediate-fail, and pack the error into next_kwargs. + + Like :meth:`submit_event`, this mutates ``task_instance`` rows that the scheduler and + other triggerer replicas write concurrently, so the transaction is retried on a transient + deadlock instead of letting it take down the triggerer. See + https://github.com/apache/airflow/issues/65818. """ for task_instance in session.scalars( select(TaskInstance).where( diff --git a/airflow-core/tests/unit/models/test_trigger.py b/airflow-core/tests/unit/models/test_trigger.py index b48f00622fe9b..03274e22dfbb1 100644 --- a/airflow-core/tests/unit/models/test_trigger.py +++ b/airflow-core/tests/unit/models/test_trigger.py @@ -27,6 +27,7 @@ import pytz from cryptography.fernet import Fernet from sqlalchemy import delete, func, select +from sqlalchemy.exc import OperationalError from airflow._shared.timezones import timezone from airflow.jobs.job import Job @@ -344,6 +345,81 @@ def test_submit_failure(session, create_task_instance): assert updated_task_instance.next_method == "__fail__" +def _deadlock_error() -> OperationalError: + """Build an OperationalError that mimics a MySQL/InnoDB deadlock (error 1213).""" + return OperationalError( + statement="UPDATE task_instance SET state='scheduled', trigger_id=NULL WHERE id = %s", + params=("some-ti-id",), + orig=Exception("(1213, 'Deadlock found when trying to get lock; try restarting transaction')"), + ) + + +def test_submit_event_retries_transient_deadlock(session, create_task_instance): + """ + A transient deadlock raised while resuming a deferred task instance is retried rather than + propagated. Before this, a single MySQL 1213 in the submit_event -> handle_event_submit path + escaped up to the triggerer and killed the process. Regression test for issue #65818. + """ + trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}) + session.add(trigger) + task_instance = create_task_instance( + session=session, logical_date=timezone.utcnow(), state=State.DEFERRED + ) + task_instance.trigger_id = trigger.id + task_instance.next_kwargs = {"cheesecake": True} + session.commit() + + real_handle_event_submit = handle_event_submit + calls = {"count": 0} + + def flaky_handle_event_submit(event, *, task_instance, session): + calls["count"] += 1 + if calls["count"] == 1: + # Deadlock surfaces on the flush inside handle_event_submit, as in the reported stack. + raise _deadlock_error() + return real_handle_event_submit(event, task_instance=task_instance, session=session) + + payload = "payload" + event = TriggerEvent(payload) + with patch("airflow.models.trigger.handle_event_submit", side_effect=flaky_handle_event_submit): + Trigger.submit_event(trigger.id, event, session=session) + + # The first attempt deadlocked, the retry resumed the task instance successfully. + assert calls["count"] == 2 + session.refresh(task_instance) + assert task_instance.state == State.SCHEDULED + assert task_instance.next_kwargs == {"event": payload, "cheesecake": True} + + +def test_submit_failure_retries_transient_deadlock(session, create_task_instance): + """ + submit_failure mutates the same ``task_instance`` rows as submit_event, so it is retried on a + transient deadlock too. Regression test for issue #65818. + """ + trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}) + session.add(trigger) + task_instance = create_task_instance(task_id="fake", logical_date=timezone.utcnow(), state=State.DEFERRED) + task_instance.trigger_id = trigger.id + session.commit() + + calls = {"count": 0} + + def flaky_format_exception(*args, **kwargs): + calls["count"] += 1 + if calls["count"] == 1: + # A deadlock while marking the dependent task instances failed must be retried. + raise _deadlock_error() + return ["RuntimeError: boom\n"] + + with patch("airflow.models.trigger.format_exception", side_effect=flaky_format_exception): + Trigger.submit_failure(trigger.id, exc=RuntimeError("boom"), session=session) + + assert calls["count"] == 2 + updated_task_instance = session.scalar(select(TaskInstance)) + assert updated_task_instance.state == State.SCHEDULED + assert updated_task_instance.next_method == "__fail__" + + @pytest.mark.parametrize( ("event_cls", "expected"), [