Do not abort Dag run creation when a deadline alert fails

airflow-core/newsfragments/68917.bugfix.rst

@@ -0,0 +1 @@
+A deadline alert that cannot be decoded, or whose interval or reference fails to resolve, no longer aborts creation of the Dag run it belongs to. The failing alert is logged and skipped, the remaining alerts still get their deadlines, and a new ``deadline_alerts.deadline_creation_failed`` counter is emitted for it.

airflow-core/src/airflow/serialization/definitions/dag.py

@@ -61,6 +61,7 @@
 from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction
 from airflow.utils.helpers import prune_dict
 from airflow.utils.session import NEW_SESSION, provide_session
+from airflow.utils.sqlalchemy import PROHIBIT_COMMIT_ERROR_MESSAGE
 from airflow.utils.state import DagRunState, TaskInstanceState
 from airflow.utils.types import DagRunType
 
@@ -747,67 +748,93 @@ def _process_dagrun_deadline_alerts(
             select(DeadlineAlertModel).where(DeadlineAlertModel.serialized_dag_id == serialized_dag_id)
         ).all()
 
-        for deadline_alert in deadline_alert_records:
-            if not deadline_alert:
-                continue
-
-            deserialized_deadline_alert = decode_deadline_alert(
-                {
-                    Encoding.TYPE: DAT.DEADLINE_ALERT,
-                    Encoding.VAR: {
-                        DeadlineAlertFields.REFERENCE: deadline_alert.reference,
-                        DeadlineAlertFields.INTERVAL: deadline_alert.interval,
-                        DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
-                    },
-                }
-            )
+        if not deadline_alert_records:
+            return
 
-            interval = deserialized_deadline_alert.interval
+        team_name = (
+            DagModel.get_team_name(self.dag_id, session=session)
+            if airflow_conf.getboolean("core", "multi_team")
+            else None
+        )
+        metrics_tags = prune_dict({"dag_id": self.dag_id, "team_name": team_name})
+
+        # Read every value off the ORM up front. A failing alert can leave the session rolled back
+        # and its instances detached, so nothing after this point may touch ORM state -- otherwise
+        # the skip path raises DetachedInstanceError and takes the DagRun down anyway.
+        run_id = orm_dagrun.run_id
+        dagrun_id = orm_dagrun.id
+        dagrun_dag_id = orm_dagrun.dag_id
+        bundle_name = orm_dagrun.dag_model.bundle_name
+        stored_alerts = [
+            (alert.id, alert.reference, alert.interval, alert.callback_def)
+            for alert in deadline_alert_records
+            if alert
+        ]
+
+        for alert_id, alert_reference, alert_interval, alert_callback_def in stored_alerts:
+            # Deadline creation is best-effort. Catch per alert so one bad alert cannot starve the rest or abort the DagRun.
+            try:
+                deserialized_deadline_alert = decode_deadline_alert(
+                    {
+                        Encoding.TYPE: DAT.DEADLINE_ALERT,
+                        Encoding.VAR: {
+                            DeadlineAlertFields.REFERENCE: alert_reference,
+                            DeadlineAlertFields.INTERVAL: alert_interval,
+                            DeadlineAlertFields.CALLBACK: alert_callback_def,
+                        },
+                    }
+                )
 
-            if isinstance(interval, SerializedVariableInterval):
-                interval = interval.resolve(session=session)
+                interval = deserialized_deadline_alert.interval
 
-            if isinstance(deserialized_deadline_alert.reference, SerializedReferenceModels.TYPES.DAGRUN):
-                deadline_time = deserialized_deadline_alert.reference.evaluate_with(
-                    session=session,
-                    interval=interval,
-                    # TODO : Pretty sure we can drop these last two; verify after testing is complete
-                    dag_id=self.dag_id,
-                    run_id=orm_dagrun.run_id,
-                )
+                if isinstance(interval, SerializedVariableInterval):
+                    interval = interval.resolve(session=session)
 
-                if deadline_time is not None:
-                    session.add(
-                        Deadline(
-                            deadline_time=deadline_time,
-                            callback=deserialized_deadline_alert.callback,
-                            dagrun_id=orm_dagrun.id,
-                            deadline_alert_id=deadline_alert.id,
-                            dag_id=orm_dagrun.dag_id,
-                            bundle_name=orm_dagrun.dag_model.bundle_name,
-                        )
-                    )
-                    team_name = (
-                        DagModel.get_team_name(self.dag_id, session=session)
-                        if airflow_conf.getboolean("core", "multi_team")
-                        else None
-                    )
-                    stats.incr(
-                        "deadline_alerts.deadline_created",
-                        tags=prune_dict({"dag_id": self.dag_id, "team_name": team_name}),
-                    )
-                elif required_dagrun_column := _DAGRUN_REFERENCE_REQUIRED_COLUMNS.get(
-                    type(deserialized_deadline_alert.reference)
-                ):
-                    log.warning(
-                        "skipping deadline alert because the deadline reference evaluated to None",
+                if isinstance(deserialized_deadline_alert.reference, SerializedReferenceModels.TYPES.DAGRUN):
+                    deadline_time = deserialized_deadline_alert.reference.evaluate_with(
+                        session=session,
+                        interval=interval,
+                        # TODO : Pretty sure we can drop these last two; verify after testing is complete
                         dag_id=self.dag_id,
-                        run_id=orm_dagrun.run_id,
-                        deadline_alert_id=deadline_alert.id,
-                        reference_type=deserialized_deadline_alert.reference.reference_name,
-                        required_dagrun_column=required_dagrun_column,
+                        run_id=run_id,
                     )
 
+                    if deadline_time is not None:
+                        session.add(
+                            Deadline(
+                                deadline_time=deadline_time,
+                                callback=deserialized_deadline_alert.callback,
+                                dagrun_id=dagrun_id,
+                                deadline_alert_id=alert_id,
+                                dag_id=dagrun_dag_id,
+                                bundle_name=bundle_name,
+                            )
+                        )
+                        stats.incr("deadline_alerts.deadline_created", tags=metrics_tags)
+                    elif required_dagrun_column := _DAGRUN_REFERENCE_REQUIRED_COLUMNS.get(
+                        type(deserialized_deadline_alert.reference)
+                    ):
+                        log.warning(
+                            "skipping deadline alert because the deadline reference evaluated to None",
+                            dag_id=self.dag_id,
+                            run_id=run_id,
+                            deadline_alert_id=alert_id,
+                            reference_type=deserialized_deadline_alert.reference.reference_name,
+                            required_dagrun_column=required_dagrun_column,
+                        )
+            except Exception as exc:
+                if isinstance(exc, RuntimeError) and PROHIBIT_COMMIT_ERROR_MESSAGE in str(exc):
+                    # A rejected commit means the scheduler's HA locking is already broken. Recording
+                    # that as a skipped alert would hide it, so let it escape instead.
+                    raise
+                log.exception(
+                    "skipping deadline alert because creating its deadline failed",
+                    dag_id=self.dag_id,
+                    run_id=run_id,
+                    deadline_alert_id=alert_id,
+                )
+                stats.incr("deadline_alerts.deadline_creation_failed", tags=metrics_tags)
+
     @provide_session
     def set_task_instance_state(
         self,

airflow-core/src/airflow/serialization/definitions/deadline.py

@@ -406,7 +406,14 @@ def resolve(self, *, session: Session | None = None) -> timedelta:
                 f"VariableInterval '{self.key}' must be an integer (seconds), got: {value!r}"
             ) from e
 
-        return timedelta(seconds=seconds)
+        try:
+            return timedelta(seconds=seconds)
+        except OverflowError as e:
+            # ``timedelta`` raises OverflowError, which is not a ValueError, so callers
+            # inspecting for a bad interval would otherwise miss it.
+            raise ValueError(
+                f"VariableInterval '{self.key}' is too large to be a valid interval: {seconds} seconds"
+            ) from e
 
 
 @attrs.define

airflow-core/src/airflow/utils/sqlalchemy.py

@@ -627,6 +627,9 @@ def with_db_lock_timeout(session: Session, lock_timeout: int = 30) -> Generator[
             session.execute(text(f"SET SESSION innodb_lock_wait_timeout = {old_mysql_timeout}"))
 
 
+PROHIBIT_COMMIT_ERROR_MESSAGE = "UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!"
+
+
 class CommitProhibitorGuard:
     """Context manager class that powers prohibit_commit."""
 
@@ -639,7 +642,7 @@ def _validate_commit(self, _):
         if self.expected_commit:
             self.expected_commit = False
             return
-        raise RuntimeError("UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!")
+        raise RuntimeError(PROHIBIT_COMMIT_ERROR_MESSAGE)
 
     def __enter__(self) -> Self:
         event.listen(self.session, "before_commit", self._validate_commit)

airflow-core/tests/unit/models/test_dagrun.py

@@ -20,7 +20,7 @@
 import datetime
 from collections import defaultdict
 from collections.abc import Mapping
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
 from functools import partial, reduce
 from typing import TYPE_CHECKING
 from unittest import mock
@@ -1654,30 +1654,150 @@ def test_dagrun_deadline_does_not_warn_for_average_runtime_without_history(
             "log_level": "warning",
         } not in caplog
 
+    @pytest.mark.parametrize(
+        ("interval", "failure"),
+        [
+            pytest.param(
+                VariableInterval("missing_key"),
+                mock.patch.object(Variable, "get", side_effect=KeyError),
+                id="unresolvable_interval",
+            ),
+            pytest.param(
+                datetime.timedelta(hours=1),
+                mock.patch(
+                    "airflow.serialization.definitions.dag.decode_deadline_alert",
+                    autospec=True,
+                    side_effect=ValueError("corrupt deadline alert blob"),
+                ),
+                id="undecodable_alert",
+            ),
+            pytest.param(
+                datetime.timedelta(hours=1),
+                mock.patch.object(
+                    SerializedReferenceModels.FixedDatetimeDeadline,
+                    "evaluate_with",
+                    autospec=True,
+                    side_effect=RuntimeError("evaluate_with failed"),
+                ),
+                id="unevaluable_reference",
+            ),
+        ],
+    )
+    @mock.patch("airflow._shared.observability.metrics.stats.incr")
     @mock.patch.object(Deadline, "prune_deadlines")
-    def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _, session, deadline_test_dag):
+    def test_dagrun_deadline_failure_does_not_abort_dagrun(
+        self, _, mock_stats_incr, interval, failure, session, deadline_test_dag
+    ):
+        future_date = datetime.datetime(2037, 1, 1, tzinfo=datetime.timezone.utc)
 
-        with mock.patch.object(
-            Variable,
-            "get",
-            side_effect=KeyError,
+        scheduler_dag = deadline_test_dag(
+            deadline=DeadlineAlert(
+                reference=DeadlineReference.FIXED_DATETIME(future_date),
+                interval=interval,
+                callback=AsyncCallback(empty_callback_for_deadline),
+            ),
+        )
+
+        with (
+            conf_vars({("core", "multi_team"): "true"}),
+            mock.patch("airflow.models.dag.DagModel.get_team_name", return_value="team_alpha"),
+            failure,
         ):
-            future_date = datetime.datetime.now() + datetime.timedelta(days=365)
+            dag_run = self.create_dag_run(
+                dag=scheduler_dag,
+                task_states={"task_1": TaskInstanceState.SUCCESS},
+                session=session,
+            )
+
+        assert dag_run is not None
+        assert session.execute(select(Deadline)).scalars().one_or_none() is None
+        mock_stats_incr.assert_any_call(
+            "deadline_alerts.deadline_creation_failed",
+            tags={"dag_id": scheduler_dag.dag_id, "team_name": "team_alpha"},
+        )
 
-            scheduler_dag = deadline_test_dag(
-                deadline=DeadlineAlert(
-                    reference=DeadlineReference.FIXED_DATETIME(future_date),
-                    interval=VariableInterval("missing_key"),
-                    callback=AsyncCallback(empty_callback_for_deadline),
+    @mock.patch("airflow._shared.observability.metrics.stats.incr")
+    @mock.patch.object(Deadline, "prune_deadlines")
+    @mock.patch.object(Variable, "get")
+    def test_dagrun_deadline_failure_that_detaches_orm_objects_still_skips(
+        self, mock_variable_get, _, mock_stats_incr, session, deadline_test_dag
+    ):
+        """A failing interval resolution can roll back and close the caller's session underneath
+        the handler (``create_session`` reusing the scoped session does exactly that), detaching
+        every ORM instance. The skip path must still log and count without touching ORM state."""
+        scheduler_dag = deadline_test_dag(
+            deadline=DeadlineAlert(
+                reference=DeadlineReference.FIXED_DATETIME(
+                    datetime.datetime(2037, 1, 1, tzinfo=datetime.timezone.utc)
                 ),
-            )
+                interval=VariableInterval("missing_key"),
+                callback=AsyncCallback(empty_callback_for_deadline),
+            ),
+        )
+        # The alert rows must be persistent from an earlier transaction (as in the live
+        # scheduler, where the dag processor committed them) so the rollback below expires
+        # them and the close detaches them instead of returning them to transient.
+        session.commit()
 
-            with pytest.raises(ValueError, match="not found"):
-                self.create_dag_run(
-                    dag=scheduler_dag,
-                    task_states={"task_1": TaskInstanceState.SUCCESS},
-                    session=session,
-                )
+        def teardown_session_and_raise(*args, **kwargs):
+            session.rollback()
+            session.close()
+            raise KeyError("missing_key")
+
+        mock_variable_get.side_effect = teardown_session_and_raise
+
+        dag_run = self.create_dag_run(dag=scheduler_dag, session=session)
+
+        assert dag_run is not None
+        mock_stats_incr.assert_any_call(
+            "deadline_alerts.deadline_creation_failed",
+            tags={"dag_id": scheduler_dag.dag_id},
+        )
+
+    @pytest.mark.parametrize(
+        ("interval", "expect_deadline", "variable_value"),
+        [
+            pytest.param(datetime.timedelta(seconds=5), True, None, id="deadline_created"),
+            pytest.param(VariableInterval("missing_key"), False, None, id="deadline_skipped"),
+            pytest.param(VariableInterval("resolvable_key"), True, "4200", id="variable_interval_resolved"),
+        ],
+    )
+    @mock.patch.object(Deadline, "prune_deadlines")
+    def test_dagrun_deadline_handling_does_not_commit(
+        self, _, interval, expect_deadline, variable_value, session, deadline_test_dag
+    ):
+        """DagRuns are created under ``prohibit_commit``, so creating a deadline, skipping one whose
+        interval will not resolve, and resolving a ``VariableInterval`` out of the metadata database
+        all have to leave the scheduler's session uncommitted."""
+        future_date = datetime.datetime(2037, 1, 1, tzinfo=datetime.timezone.utc)
+        scheduler_dag = deadline_test_dag(
+            deadline=DeadlineAlert(
+                reference=DeadlineReference.FIXED_DATETIME(future_date),
+                interval=interval,
+                callback=AsyncCallback(empty_callback_for_deadline),
+            ),
+        )
+
+        if variable_value is None:
+            variable_lookup = mock.patch.object(Variable, "get", side_effect=KeyError)
+        else:
+            # A real row committed before the guard opens, so resolve() has to go through
+            # MetastoreBackend. Drop the session forwarding and the backend opens the same scoped
+            # session and commits it, which the guard rejects and the lookup then reports as missing.
+            Variable.set(key=interval.key, value=variable_value, session=session)
+            session.commit()
+            variable_lookup = nullcontext()
+
+        # No task_states: the helper's get_task_instance() opens its own session and would trip the guard.
+        with prohibit_commit(session), variable_lookup:
+            dag_run = self.create_dag_run(dag=scheduler_dag, session=session)
+        session.flush()
+
+        assert dag_run is not None
+        deadline = session.execute(select(Deadline)).scalars().one_or_none()
+        assert (deadline is not None) is expect_deadline
+        if variable_value is not None:
+            assert deadline.deadline_time == future_date + datetime.timedelta(seconds=int(variable_value))
 
 
 @pytest.mark.parametrize(

airflow-core/tests/unit/serialization/definitions/test_deadline.py

@@ -61,6 +61,7 @@ def test_resolve_forwards_session(self, mocker):
             (None, True, "not found"),
             ("abc", False, "must be an integer"),
             ("", False, "must be an integer"),
+            ("99999999999999", False, "too large to be a valid interval"),
         ],
     )
     def test_resolve_invalid(self, mocker, value, raise_missing, match):

shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml

@@ -326,6 +326,12 @@ metrics:
     legacy_name: "-"
     name_variables: []
 
+  - name: "deadline_alerts.deadline_creation_failed"
+    description: "Number of deadline alerts skipped because their deadline could not be created"
+    type: "counter"
+    legacy_name: "-"
+    name_variables: []
+
   - name: "deadline_alerts.deadline_missed"
     description: "Number of deadline alerts that fired because a Dag run missed its deadline"
     type: "counter"