This is an automated email from the ASF dual-hosted git repository.
o-nikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 00ce83c6c44 Fix deadline never firing after non-deadline Dag edit
(#68734)
00ce83c6c44 is described below
commit 00ce83c6c443f75020ef8764718e585e0048efe9
Author: Ehab Abdalla <[email protected]>
AuthorDate: Wed Jul 15 15:52:41 2026 -0400
Fix deadline never firing after non-deadline Dag edit (#68734)
The structural issue was that cls(dag) was constructed before knowing which
branch (skip / UPDATE / INSERT) we'd take. By deferring construction past the
branch decision, each branch can prepare dag.data correctly before the
constructor runs — eliminating the need for post-hoc patching of internal
attributes.
---
airflow-core/src/airflow/models/serialized_dag.py | 18 ++--
.../tests/unit/models/test_serialized_dag.py | 113 +++++++++++++++++++++
2 files changed, 125 insertions(+), 6 deletions(-)
diff --git a/airflow-core/src/airflow/models/serialized_dag.py
b/airflow-core/src/airflow/models/serialized_dag.py
index fd773e8ed4f..fbff450a993 100644
--- a/airflow-core/src/airflow/models/serialized_dag.py
+++ b/airflow-core/src/airflow/models/serialized_dag.py
@@ -643,6 +643,7 @@ class SerializedDagModel(Base):
dag_version = _prefetched.dag_version
name_updated = False
+ reused_deadline_data: dict[str, dict] | None = None
if dag.data.get("dag", {}).get("deadline"):
# Try to reuse existing deadline UUIDs if the deadline definitions
haven't changed.
# This preserves the hash and avoids unnecessary
SerializedDagModel recreations.
@@ -674,6 +675,7 @@ class SerializedDagModel(Base):
)
name_updated = bool(name_updates)
dag.data["dag"]["deadline"] = existing_deadline_uuids
+ reused_deadline_data = deadline_uuid_mapping
deadline_uuid_mapping = {}
else:
# At least one deadline has changed, generate new UUIDs
and update the hash.
@@ -684,13 +686,9 @@ class SerializedDagModel(Base):
else:
deadline_uuid_mapping = {}
- new_serialized_dag = cls(dag)
+ new_dag_hash = cls.hash(dag.data)
- if (
- serialized_dag_hash == new_serialized_dag.dag_hash
- and dag_version
- and dag_version.bundle_name == bundle_name
- ):
+ if serialized_dag_hash == new_dag_hash and dag_version and
dag_version.bundle_name == bundle_name:
# Serialized content is unchanged, so we don't create a new
DagVersion.
# But if the bundle advanced, refresh the latest version's pointer
in place — tasks resolve
# their code from ``ti.dag_version.bundle_version`` at run time,
so a stale
@@ -728,6 +726,7 @@ class SerializedDagModel(Base):
# This is for dynamic DAGs that the hashes changes often. We
should update
# the serialized dag, the dag_version and the dag_code instead of
a new version
# if the dag_version is not associated with any task instances
+ new_serialized_dag = cls(dag)
# Use direct UPDATE to avoid loading the full serialized DAG
result = session.execute(
@@ -773,8 +772,15 @@ class SerializedDagModel(Base):
session=session,
)
log.debug("Writing Serialized DAG: %s to the DB", dag.dag_id)
+
+ if reused_deadline_data:
+ deadline_uuid_mapping = {str(uuid6.uuid7()): data for data in
reused_deadline_data.values()}
+ dag.data["dag"]["deadline"] = list(deadline_uuid_mapping.keys())
+
+ new_serialized_dag = cls(dag)
new_serialized_dag.dag_version = dagv
session.add(new_serialized_dag)
+
cls._create_deadline_alert_records(new_serialized_dag,
deadline_uuid_mapping)
log.debug("DAG: %s written to the DB", dag.dag_id)
DagCode.write_code(dagv, dag.fileloc, session=session)
diff --git a/airflow-core/tests/unit/models/test_serialized_dag.py
b/airflow-core/tests/unit/models/test_serialized_dag.py
index 03a4159eb2b..fcb49805b4c 100644
--- a/airflow-core/tests/unit/models/test_serialized_dag.py
+++ b/airflow-core/tests/unit/models/test_serialized_dag.py
@@ -1042,3 +1042,116 @@ class TestSerializedDagModel:
# The name must have been updated in the DB.
assert updated_alert.name == "updated name"
+
+ def test_non_deadline_edit_creates_deadline_alert_for_new_serdag(self,
testing_dag_bundle, session):
+ """Non-deadline Dag edit must create deadline_alert rows for the new
serialized_dag row."""
+ dag_id = "test_deadline_orphan"
+
+ dag = DAG(
+ dag_id=dag_id,
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(minutes=5),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+ EmptyOperator(task_id="task1", dag=dag)
+ scheduler_dag = sync_dag_to_db(dag, session=session)
+
+ # Create a dagrun so the existing dag_version has task instances,
+ # forcing write_dag into the INSERT branch (new serialized_dag row).
+ scheduler_dag.create_dagrun(
+ run_id="test1",
+ run_after=DEFAULT_DATE,
+ state=DagRunState.QUEUED,
+ logical_date=DEFAULT_DATE,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ triggered_by=DagRunTriggeredByType.TEST,
+ run_type=DagRunType.MANUAL,
+ )
+ session.commit()
+
+ orig_serdag = session.scalar(select(SDM).where(SDM.dag_id ==
dag_id).order_by(SDM.created_at.desc()))
+ orig_alert = session.scalar(select(DAM).where(DAM.serialized_dag_id ==
orig_serdag.id))
+ assert orig_alert is not None
+
+ # Add a second task (non-deadline change) — deadline definition is
untouched.
+ EmptyOperator(task_id="task2", dag=dag)
+ SDM.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="testing", session=session)
+ session.commit()
+
+ new_serdag = session.scalar(select(SDM).where(SDM.dag_id ==
dag_id).order_by(SDM.created_at.desc()))
+ assert new_serdag.id != orig_serdag.id, "A new serialized_dag row
should have been created"
+
+ new_alert = session.scalar(select(DAM).where(DAM.serialized_dag_id ==
new_serdag.id))
+ assert new_alert is not None, (
+ f"serialized_dag {new_serdag.id} has deadline UUIDs in JSON but no
deadline_alert row links to it"
+ )
+
+ # The old alert should still be intact for the old serialized_dag.
+ old_alert = session.scalar(select(DAM).where(DAM.serialized_dag_id ==
orig_serdag.id))
+ assert old_alert is not None
+
+ def test_non_deadline_edit_preserves_alert_in_update_branch(self,
testing_dag_bundle, session):
+ """UPDATE branch (no task instances): existing deadline_alert stays
linked after non-deadline edit."""
+ dag_id = "test_deadline_update_branch"
+
+ dag = DAG(
+ dag_id=dag_id,
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(minutes=5),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+ EmptyOperator(task_id="task1", dag=dag)
+ sync_dag_to_db(dag, session=session)
+ session.commit()
+
+ # No dagrun created — dag_version has no task instances, so write_dag
+ # takes the UPDATE branch (in-place update of the existing
serialized_dag row).
+ orig_serdag = session.scalar(select(SDM).where(SDM.dag_id == dag_id))
+ orig_alert = session.scalar(select(DAM).where(DAM.serialized_dag_id ==
orig_serdag.id))
+ assert orig_alert is not None
+
+ EmptyOperator(task_id="task2", dag=dag)
+ SDM.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="testing", session=session)
+ session.commit()
+
+ # Same row was updated in place — only one serialized_dag should exist.
+ serdag_count =
session.scalar(select(func.count()).select_from(SDM).where(SDM.dag_id ==
dag_id))
+ assert serdag_count == 1
+
+ updated_serdag = session.scalar(select(SDM).where(SDM.dag_id ==
dag_id))
+ alert = session.scalar(select(DAM).where(DAM.serialized_dag_id ==
updated_serdag.id))
+ assert alert is not None, "deadline_alert must still be linked after
UPDATE-branch re-serialization"
+
+ def test_deadline_reuse_skips_write_when_hash_matches(self,
testing_dag_bundle, session):
+ """When nothing changed, write_dag short-circuits and existing alerts
remain valid."""
+ dag_id = "test_deadline_noop"
+
+ dag = DAG(
+ dag_id=dag_id,
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(minutes=5),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+ EmptyOperator(task_id="task1", dag=dag)
+ sync_dag_to_db(dag, session=session)
+ session.commit()
+
+ orig_serdag = session.scalar(select(SDM).where(SDM.dag_id == dag_id))
+ orig_alert = session.scalar(select(DAM).where(DAM.serialized_dag_id ==
orig_serdag.id))
+ assert orig_alert is not None
+
+ # Re-serialize the exact same Dag — nothing changed.
+ did_write = SDM.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="testing", session=session)
+ session.commit()
+
+ assert did_write is False
+
+ alert = session.scalar(select(DAM).where(DAM.serialized_dag_id ==
orig_serdag.id))
+ assert alert is not None
+ assert alert.id == orig_alert.id
