Fix TypeError clearing a DAG run when a queued-at deadline interval is stored as JSON

airflow-core/src/airflow/models/taskinstance.py

@@ -263,10 +263,30 @@ def _recalculate_dagrun_queued_at_deadlines(
     if not results:
         return
 
+    # Local import to avoid a circular import between models and serialization.
+    from airflow.serialization.decoders import decode_deadline_alert_model, resolve_deadline_alert_interval
+
     for deadline, deadline_alert in results:
-        # We can't use evaluate_with() since the new queued_at is not written to the DB yet.
-        deadline_interval = timedelta(seconds=deadline_alert.interval)
-        new_deadline_time = new_queued_at + deadline_interval
+        # We can't use evaluate_with() since the new queued_at is not written to the DB yet, and
+        # interval is stored as JSON, so it has to be decoded rather than passed to timedelta().
+        try:
+            interval = resolve_deadline_alert_interval(
+                decode_deadline_alert_model(deadline_alert), session=session
+            )
+        except (ValueError, TypeError):
+            # Either step can fail: the alert may hold a payload the decoder refuses, or a
+            # variable-backed interval may point at a Variable that is missing or non-numeric.
+            # Leave this deadline alone rather than failing the whole clear.
+            log.warning(
+                "Error while recalculating deadline %s for DagRun %s.%s, leaving it unchanged",
+                deadline.id,
+                dagrun.dag_id,
+                dagrun.run_id,
+                exc_info=True,
+            )
+            continue
+
+        new_deadline_time = new_queued_at + interval
 
         log.debug(
             "Recalculating deadline %s for DagRun %s.%s: old=%s, new=%s",

airflow-core/src/airflow/serialization/decoders.py

@@ -53,6 +53,9 @@
 )
 
 if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
+    from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
     from airflow.partition_mappers.base import PartitionMapper
     from airflow.partition_mappers.wait_policy import WaitPolicy
     from airflow.partition_mappers.window import Window
@@ -308,7 +311,7 @@ def _decode_deadline_callback(raw_callback: Any) -> AsyncCallback | SyncCallback
     return callback_cls(callback_callable=_SerializedCallbackPath(path), **fields)
 
 
-def decode_deadline_alert(encoded_data: dict):
+def decode_deadline_alert(encoded_data: dict) -> SerializedDeadlineAlert:
     """
     Decode a previously serialized deadline alert.
 
@@ -337,6 +340,45 @@ def decode_deadline_alert(encoded_data: dict):
     )
 
 
+def decode_deadline_alert_model(deadline_alert: DeadlineAlertModel) -> SerializedDeadlineAlert:
+    """
+    Decode a ``DeadlineAlert`` ORM row into its serialized representation.
+
+    :meta private:
+    """
+    return decode_deadline_alert(
+        {
+            DeadlineAlertFields.NAME: deadline_alert.name,
+            DeadlineAlertFields.REFERENCE: deadline_alert.reference,
+            DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+            DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
+        }
+    )
+
+
+def resolve_deadline_alert_interval(
+    alert: SerializedDeadlineAlert, *, session: Session
+) -> datetime.timedelta:
+    """
+    Resolve a decoded alert's interval to a ``timedelta``.
+
+    A ``SerializedVariableInterval`` reads its Airflow Variable here, so this is only called at
+    the point a deadline is actually calculated. It raises ``ValueError`` if the Variable is
+    missing or is not an integer number of seconds.
+
+    :param alert: The decoded alert whose interval should be resolved.
+    :param session: Existing SQLAlchemy Session. Required rather than defaulted: both callers run
+        under the scheduler's ``prohibit_commit`` guard, so the open session has to reach
+        ``Variable.get`` instead of ``provide_session`` handing back the same scoped session and
+        rolling it back on exit.
+
+    :meta private:
+    """
+    if isinstance(alert.interval, SerializedVariableInterval):
+        return alert.interval.resolve(session=session)
+    return alert.interval
+
+
 def decode_timetable(var: dict[str, Any]) -> CoreTimetable:
     """
     Decode a previously serialized timetable.

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

@@ -50,14 +50,9 @@
 from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.models.tasklog import LogTemplate
-from airflow.serialization.decoders import decode_deadline_alert
-from airflow.serialization.definitions.deadline import (
-    DeadlineAlertFields,
-    SerializedReferenceModels,
-    SerializedVariableInterval,
-)
+from airflow.serialization.decoders import decode_deadline_alert_model, resolve_deadline_alert_interval
+from airflow.serialization.definitions.deadline import SerializedReferenceModels
 from airflow.serialization.definitions.param import SerializedParamsDict
-from airflow.serialization.enums import DagAttributeTypes as DAT, Encoding
 from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction
 from airflow.utils.helpers import prune_dict
 from airflow.utils.session import NEW_SESSION, provide_session
@@ -751,21 +746,8 @@ def _process_dagrun_deadline_alerts(
             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,
-                    },
-                }
-            )
-
-            interval = deserialized_deadline_alert.interval
-
-            if isinstance(interval, SerializedVariableInterval):
-                interval = interval.resolve(session=session)
+            deserialized_deadline_alert = decode_deadline_alert_model(deadline_alert)
+            interval = resolve_deadline_alert_interval(deserialized_deadline_alert, session=session)
 
             if isinstance(deserialized_deadline_alert.reference, SerializedReferenceModels.TYPES.DAGRUN):
                 deadline_time = deserialized_deadline_alert.reference.evaluate_with(

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

@@ -1596,6 +1596,24 @@ def test_dagrun_deadline_variable_interval_stable(self, _, mock_get, session, de
         deadline = session.execute(select(Deadline)).scalars().one_or_none()
         assert deadline.deadline_time == first_deadline_time
 
+    @mock.patch.object(Deadline, "prune_deadlines")
+    def test_dagrun_deadline_variable_interval_uses_callers_session(self, _, session, deadline_test_dag):
+        """Caller's session must reach Variable.get."""
+        future_date = datetime.datetime.now() + datetime.timedelta(days=365)
+
+        scheduler_dag = deadline_test_dag(
+            deadline=DeadlineAlert(
+                reference=DeadlineReference.FIXED_DATETIME(future_date),
+                interval=VariableInterval("my_key"),
+                callback=AsyncCallback(empty_callback_for_deadline),
+            ),
+        )
+
+        with mock.patch.object(Variable, "get", return_value="60") as mock_get:
+            self.create_dag_run(dag=scheduler_dag, session=session)
+
+        mock_get.assert_called_once_with("my_key", session=session)
+
     def test_dagrun_deadline_logs_when_reference_column_is_null(self, session, deadline_test_dag, caplog):
         scheduler_dag = deadline_test_dag(
             deadline=DeadlineAlert(

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

@@ -73,6 +73,7 @@
     find_relevant_relatives,
 )
 from airflow.models.taskreschedule import TaskReschedule
+from airflow.models.variable import Variable
 from airflow.models.xcom import XComModel, XComModelV2, build_xcom_read_query
 from airflow.providers.standard.operators.bash import BashOperator
 from airflow.providers.standard.operators.empty import EmptyOperator
@@ -92,10 +93,11 @@
 )
 from airflow.sdk.api.datamodels._generated import AssetEventResponse, AssetResponse
 from airflow.sdk.definitions.callback import AsyncCallback
-from airflow.sdk.definitions.deadline import DeadlineReference
+from airflow.sdk.definitions.deadline import DeadlineReference, VariableInterval
 from airflow.sdk.definitions.param import process_params
 from airflow.sdk.definitions.taskgroup import TaskGroup
 from airflow.sdk.execution_time.comms import AssetEventsResult
+from airflow.sdk.serde import serialize
 from airflow.serialization.definitions.assets import SerializedAsset
 from airflow.serialization.definitions.baseoperator import SerializedBaseOperator
 from airflow.serialization.definitions.dag import SerializedDAG
@@ -4597,8 +4599,24 @@ async def empty_callback_for_deadline():
     pass
 
 
-def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, session):
-    """Test that clearing tasks recalculates all (and only) DAGRUN_QUEUED_AT deadlines."""
+@pytest.mark.parametrize(
+    "use_variable_interval",
+    [
+        pytest.param(False, id="fixed_timedelta_interval"),
+        pytest.param(True, id="variable_interval"),
+    ],
+)
+def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, session, use_variable_interval):
+    """Test that clearing tasks recalculates all (and only) DAGRUN_QUEUED_AT deadlines.
+
+    Since Airflow 3.3 the ``deadline_alert.interval`` column is JSON (a serialized ``timedelta``
+    or ``VariableInterval``), so the recalculation must decode it instead of passing the raw value
+    to ``timedelta()``. Storing the interval via ``serialize`` here mirrors production and covers
+    both interval kinds.
+    """
+    variable_key = "deadline_interval_key"
+    variable_seconds = 3600
+
     with dag_maker(
         dag_id="test_recalculate_deadlines",
         schedule=datetime.timedelta(days=1),
@@ -4618,29 +4636,52 @@ def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, se
         select(SerializedDagModel.id).where(SerializedDagModel.dag_id == dag.dag_id)
     )
 
-    deadline_configs = [
-        (DeadlineReference.DAGRUN_QUEUED_AT, datetime.timedelta(hours=1)),
-        (DeadlineReference.DAGRUN_QUEUED_AT, datetime.timedelta(hours=2)),
-        (DeadlineReference.FIXED_DATETIME, datetime.timedelta(hours=1)),
+    if use_variable_interval:
+        queued_interval = VariableInterval(variable_key)
+        queued_resolved = datetime.timedelta(seconds=variable_seconds)
+        # Both DAGRUN_QUEUED_AT deadlines resolve to the same interval from the Variable.
+        queued_configs = [
+            (DeadlineReference.DAGRUN_QUEUED_AT, queued_interval, queued_resolved),
+            (DeadlineReference.DAGRUN_QUEUED_AT, queued_interval, queued_resolved),
+        ]
+    else:
+        queued_configs = [
+            (
+                DeadlineReference.DAGRUN_QUEUED_AT,
+                datetime.timedelta(hours=1),
+                datetime.timedelta(hours=1),
+            ),
+            (
+                DeadlineReference.DAGRUN_QUEUED_AT,
+                datetime.timedelta(hours=2),
+                datetime.timedelta(hours=2),
+            ),
+        ]
+
+    # (reference type, interval stored on the alert, timedelta it resolves to)
+    deadline_configs = queued_configs + [
+        (DeadlineReference.FIXED_DATETIME, datetime.timedelta(hours=1), datetime.timedelta(hours=1)),
     ]
 
-    for deadline_type, interval in deadline_configs:
+    expected_resolved_by_alert = {}
+    for deadline_type, stored_interval, resolved_interval in deadline_configs:
         if deadline_type == DeadlineReference.DAGRUN_QUEUED_AT:
             reference = DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference()
-            deadline_time = dag_run.queued_at + interval
+            deadline_time = dag_run.queued_at + resolved_interval
         else:  # FIXED_DATETIME
             future_date = timezone.utcnow() + datetime.timedelta(days=7)
             reference = DeadlineReference.FIXED_DATETIME(future_date).serialize_reference()
-            deadline_time = future_date + interval
+            deadline_time = future_date + resolved_interval
 
         deadline_alert = DeadlineAlertModel(
             serialized_dag_id=serialized_dag_id,
             reference=reference,
-            interval=interval.total_seconds(),
-            callback_def={"path": f"{__name__}.empty_callback_for_deadline", "kwargs": {}},
+            interval=serialize(stored_interval),
+            callback_def=serialize(AsyncCallback(empty_callback_for_deadline)),
         )
         session.add(deadline_alert)
         session.flush()
+        expected_resolved_by_alert[deadline_alert.id] = resolved_interval
 
         deadline = Deadline(
             dagrun_id=dag_run.id,
@@ -4659,7 +4700,21 @@ def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, se
     }
 
     tis = session.scalars(select(TI).where(TI.dag_id == dag.dag_id, TI.run_id == dag_run.run_id)).all()
-    clear_task_instances(tis, session)
+
+    # VariableInterval.resolve() reads the Airflow Variable during recalculation.
+    with (
+        mock.patch.object(Variable, "get", return_value=str(variable_seconds))
+        if use_variable_interval
+        else contextlib.nullcontext()
+    ) as mock_variable_get:
+        clear_task_instances(tis, session)
+
+    if use_variable_interval:
+        # See resolve_deadline_alert_interval() for why the caller's session has to be forwarded.
+        assert mock_variable_get.call_args_list == [
+            mock.call(variable_key, session=session),
+            mock.call(variable_key, session=session),
+        ]
 
     dag_run = session.scalar(select(DagRun).where(DagRun.id == dag_run.id))
     assert dag_run.queued_at > original_queued_at
@@ -4672,13 +4727,84 @@ def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, se
     for deadline in deadlines_after:
         if deadline.deadline_time != deadline_times_by_alert[deadline.deadline_alert_id]:
             recalculated_count += 1
-            deadline_alert = session.get(DeadlineAlertModel, deadline.deadline_alert_id)
-            expected_time = dag_run.queued_at + datetime.timedelta(seconds=deadline_alert.interval)
+            expected_time = dag_run.queued_at + expected_resolved_by_alert[deadline.deadline_alert_id]
             assert deadline.deadline_time == expected_time
 
     assert recalculated_count == 2
 
 
+def test_clear_task_instances_skips_deadline_with_unresolvable_interval(dag_maker, session):
+    """A variable-backed interval that cannot be resolved must not abort the clear.
+
+    ``SerializedVariableInterval.resolve()`` raises ``ValueError`` when the Airflow Variable is
+    missing or is not an integer, and that happens while the DAG run is being cleared. The clear
+    should still go through, leaving the unresolvable deadline at its old time.
+    """
+    with dag_maker(
+        dag_id="test_recalculate_deadlines_unresolvable",
+        schedule=datetime.timedelta(days=1),
+    ) as dag:
+        EmptyOperator(task_id="task_1")
+
+    dag_run = dag_maker.create_dagrun()
+    ti = dag_run.get_task_instance("task_1", session=session)
+    ti.set_state(TaskInstanceState.SUCCESS, session=session)
+
+    original_queued_at = timezone.utcnow() - datetime.timedelta(hours=2)
+    dag_run.queued_at = original_queued_at
+    session.flush()
+
+    serialized_dag_id = session.scalar(
+        select(SerializedDagModel.id).where(SerializedDagModel.dag_id == dag.dag_id)
+    )
+
+    deadline_alert = DeadlineAlertModel(
+        serialized_dag_id=serialized_dag_id,
+        reference=DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference(),
+        interval=serialize(VariableInterval("missing_deadline_interval_key")),
+        callback_def=serialize(AsyncCallback(empty_callback_for_deadline)),
+    )
+    session.add(deadline_alert)
+    session.flush()
+
+    original_deadline_time = original_queued_at + datetime.timedelta(hours=1)
+    session.add(
+        Deadline(
+            dagrun_id=dag_run.id,
+            deadline_alert_id=deadline_alert.id,
+            deadline_time=original_deadline_time,
+            callback=AsyncCallback(empty_callback_for_deadline),
+            dag_id=dag_run.dag_id,
+        )
+    )
+    session.flush()
+
+    tis = session.scalars(select(TI).where(TI.dag_id == dag.dag_id, TI.run_id == dag_run.run_id)).all()
+
+    with (
+        mock.patch.object(
+            Variable, "get", side_effect=KeyError("missing_deadline_interval_key")
+        ) as mock_variable_get,
+        mock.patch("airflow.models.taskinstance.log") as mock_log,
+    ):
+        clear_task_instances(tis, session)
+
+    dag_run = session.scalar(select(DagRun).where(DagRun.id == dag_run.id))
+    assert dag_run.queued_at > original_queued_at
+
+    deadline = session.scalar(select(Deadline).where(Deadline.dagrun_id == dag_run.id))
+    assert deadline.deadline_time == original_deadline_time
+
+    # The alert decoded fine and the Variable lookup is what raised, so this is still covering
+    # an unresolvable interval rather than an earlier failure that lands in the same except.
+    mock_variable_get.assert_called_once_with("missing_deadline_interval_key", session=session)
+
+    assert mock_log.warning.call_count == 1
+    msg, *args = mock_log.warning.call_args.args
+    assert "Error while recalculating deadline" in msg
+    assert args[0] == deadline.id
+
+
 def test_get_dagrun_loaded_but_none_returns_dagrun(dag_maker, session):
     """
     Test that `get_dagrun()` fetches `DagRun` from DB when the `dag_run`