This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 016dae42b05 [v3-3-test] Fix Dag scheduling stall after switching to a
coarser cron (#72498) (#72679)
016dae42b05 is described below
commit 016dae42b05a4eed24e7267eab4775738cb4753e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Sep 8 11:48:58 2026 +0530
[v3-3-test] Fix Dag scheduling stall after switching to a coarser cron
(#72498) (#72679)
* Fix Dag scheduling stall after switching to a coarser cron
Changing a Dag's schedule to a coarser cron (e.g. hourly to daily) could
make the realigned next run collide with the logical_date of the run
already on record. The scheduler would then loop on "run already exists;
skipping dagrun creation" forever instead of advancing, since every input
to the recompute (the DAG's static start_date, the stale reference run)
is a fixed constant that reproduces the identical collision every time.
A single-step guard for this class of collision already existed for the
narrower case of a CronTriggerTimetable-to-CronDataIntervalTimetable
switch. A single step turns out to always be enough for both shipped
_DataIntervalTimetable subclasses, but only because of invariants that
are specific to them and aren't enforced anywhere -- a custom Timetable
isn't guaranteed to uphold them. Retry a bounded number of times instead
of assuming one step suffices, and fail loudly if that budget is
exhausted rather than looping unbounded, which would otherwise hang the
scheduler's main loop for every Dag if a future _get_next implementation
ever stopped strictly advancing.
* Simplify the coarser-schedule-stall guard to a single bounded retry
Only one advance past the previous run's start is provably necessary
for both shipped _DataIntervalTimetable subclasses; a second retry was
unproven margin that only existed to satisfy a synthetic test case, not
any real or documented timetable behavior. A bounded while loop implied
that margin was load-bearing. Replacing it with a single retry followed
by a loud failure keeps the same fail-fast guarantee -- a custom
Timetable needing more than one step is still caught immediately rather
than silently retried -- while matching what the invariant actually
proves.
* Use ValueError for timetable alignment guard
* Avoid exposing timetable internals in error message
* Simplify schedule change test runner helper
* Clarify zero-length interval test wording
* Clarify coarser schedule regression test comment
(cherry picked from commit 31117014f3c5a8cd4d4acd4223309eb344e21981)
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
airflow-core/src/airflow/timetables/interval.py | 25 +--
.../tests/unit/jobs/test_schedule_change_stall.py | 173 +++++++++++++++++++++
.../unit/timetables/test_interval_timetable.py | 125 ++++++++++++++-
3 files changed, 304 insertions(+), 19 deletions(-)
diff --git a/airflow-core/src/airflow/timetables/interval.py
b/airflow-core/src/airflow/timetables/interval.py
index 30a7f5033fa..d2e7289b517 100644
--- a/airflow-core/src/airflow/timetables/interval.py
+++ b/airflow-core/src/airflow/timetables/interval.py
@@ -111,19 +111,20 @@ class _DataIntervalTimetable(Timetable):
# Data interval starts from the end of the previous interval.
start = align_last_data_interval_end
- # CronTriggerTimetable stores its runs as point-in-time intervals
- # (start == end == logical_date). After a switch to a
- # CronDataIntervalTimetable the aligned `start` lands back on that
- # same logical_date, so without this guard we'd propose a run
- # identical to the existing one — which collides with the
- # (dag_id, logical_date) unique constraint and leaves the scheduler
- # looping on "run already exists; skipping dagrun creation" until
- # the next period elapses. Advance one period to skip past it.
- if (
- last_automated_data_interval.start ==
last_automated_data_interval.end
- and start == last_automated_data_interval.start
- ):
+ # A schedule change (e.g. a coarser cron) can realign `start` onto
or
+ # before the previous run's start, colliding with the (dag_id,
+ # logical_date) unique constraint and stalling the scheduler on
"run
+ # already exists; skipping dagrun creation". One retry past it is
+ # provably enough for both shipped subclasses; fail loudly instead
of
+ # retrying indefinitely, which would hang the scheduler for every
Dag
+ # if `_get_next` ever stopped strictly advancing.
+ if start <= last_automated_data_interval.start:
start = self._get_next(start)
+ if start <= last_automated_data_interval.start:
+ raise ValueError(
+ f"{type(self).__name__} timetable did not advance past
"
+ f"{last_automated_data_interval.start} after one retry"
+ )
if restriction.latest is not None and start > restriction.latest:
return None
end = self._get_next(start)
diff --git a/airflow-core/tests/unit/jobs/test_schedule_change_stall.py
b/airflow-core/tests/unit/jobs/test_schedule_change_stall.py
new file mode 100644
index 00000000000..78c7c04191a
--- /dev/null
+++ b/airflow-core/tests/unit/jobs/test_schedule_change_stall.py
@@ -0,0 +1,173 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Regression test for https://github.com/apache/airflow/issues/66754: changing a
Dag's
+cron to a coarser one (e.g. hourly to daily) must not permanently stall DagRun
creation.
+Calls the DagRun-creation and TI-scheduling code paths directly
+(``SchedulerJobRunner._create_dag_runs``, ``DagRun.update_state``,
+``DagRun.schedule_tis``) against a real Dag re-sync
(``SerializedDAG.bulk_write_to_db``
+via ``dag_maker``) and real ``DagRun``/``TaskInstance`` rows, not just
+``CronDataIntervalTimetable`` in isolation, to prove the stall reproduces at
the
+database level and that the guard in ``timetables/interval.py`` resolves it.
+"""
+
+from __future__ import annotations
+
+import datetime
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow._shared.timezones import timezone
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.models import DagRun
+from airflow.providers.standard.operators.bash import BashOperator
+from airflow.utils.state import DagRunState, TaskInstanceState
+
+from tests_common.test_utils.db import (
+ clear_db_assets,
+ clear_db_backfills,
+ clear_db_callbacks,
+ clear_db_dags,
+ clear_db_deadline,
+ clear_db_import_errors,
+ clear_db_jobs,
+ clear_db_pools,
+ clear_db_runs,
+ clear_db_triggers,
+)
+from tests_common.test_utils.mock_executor import MockExecutor
+
+pytestmark = pytest.mark.db_test
+
+DAG_ID = "schedule_change_stall_coarser_cron"
+START_DATE = timezone.datetime(2026, 5, 4)
+
+
+def _clean_db():
+ clear_db_dags()
+ clear_db_runs()
+ clear_db_backfills()
+ clear_db_pools()
+ clear_db_import_errors()
+ clear_db_jobs()
+ clear_db_assets()
+ clear_db_deadline()
+ clear_db_callbacks()
+ clear_db_triggers()
+
+
[email protected](autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def _make_runner():
+ return SchedulerJobRunner(job=Job(), executors=[MockExecutor()])
+
+
+@time_machine.travel(START_DATE, tick=False)
+def test_coarser_schedule_change_does_not_stall_dagrun_creation(dag_maker,
session):
+ # 1. First Dag scheduling: hourly, catchup=True.
+ with dag_maker(
+ dag_id=DAG_ID,
+ schedule="0 * * * *",
+ start_date=START_DATE,
+ catchup=True,
+ max_active_runs=1,
+ session=session,
+ ):
+ BashOperator(task_id="do_something", bash_command="true")
+
+ dag_model = dag_maker.dag_model
+ assert dag_model.next_dagrun == START_DATE
+ assert dag_model.next_dagrun_create_after == START_DATE +
datetime.timedelta(hours=1)
+
+ runner = _make_runner()
+
+ # Tick once the hourly run becomes due.
+ with time_machine.travel(START_DATE + datetime.timedelta(hours=1),
tick=False):
+ runner._create_dag_runs([dag_model], session)
+ session.flush()
+
+ runs = session.scalars(select(DagRun).where(DagRun.dag_id == DAG_ID)).all()
+ assert len(runs) == 1, f"expected exactly one DagRun after the first tick,
got {runs}"
+ hourly_run = runs[0]
+ assert hourly_run.logical_date == START_DATE
+ assert hourly_run.data_interval_end == START_DATE +
datetime.timedelta(hours=1)
+
+ # 2. The first TI of the first Dag scheduling already ran.
+ ti = hourly_run.get_task_instances(session=session)[0]
+ ti.state = TaskInstanceState.SUCCESS
+ session.merge(ti)
+ session.flush()
+ hourly_run.dag = dag_maker.serialized_dag
+ hourly_run.update_state(session=session)
+ session.flush()
+ assert hourly_run.state == DagRunState.SUCCESS
+
+ # 3. Change the same Dag's schedule: hourly to daily, drop end_date.
+ with dag_maker(
+ dag_id=DAG_ID,
+ schedule="0 0 * * *",
+ start_date=START_DATE,
+ catchup=True,
+ max_active_runs=1,
+ session=session,
+ ):
+ BashOperator(task_id="do_something", bash_command="true")
+
+ session.expire_all()
+ dag_model = dag_maker.dag_model
+
+ # 4. Advance past the next expected (daily) boundary and tick again.
+ with time_machine.travel(START_DATE + datetime.timedelta(days=1,
minutes=5), tick=False):
+ runner._create_dag_runs([dag_model], session)
+ session.flush()
+
+ session.expire_all()
+ runs = session.scalars(select(DagRun).where(DagRun.dag_id ==
DAG_ID).order_by(DagRun.logical_date)).all()
+
+ expected_logical_date = START_DATE + datetime.timedelta(days=1)
+ new_runs = [r for r in runs if r.logical_date == expected_logical_date]
+ assert len(new_runs) == 1, (
+ f"expected a new DagRun at logical_date={expected_logical_date} (the
next day's slot); "
+ f"instead the scheduler produced these runs: "
+ f"{[(r.run_id, r.logical_date, r.state) for r in runs]}. "
+ "This is the 'run already exists; skipping dagrun creation' stall: the
scheduler is stuck "
+ "proposing the pre-existing hourly run's logical_date forever instead
of advancing."
+ )
+
+ new_run = new_runs[0]
+ new_ti = new_run.get_task_instances(session=session)[0]
+ assert new_ti.state != TaskInstanceState.REMOVED
+
+ # Drive scheduling decisions for the new run and confirm its TI reaches
SCHEDULED,
+ # i.e. it is not stuck in a not-yet-scheduled/None/queued limbo.
+ new_run.dag = dag_maker.serialized_dag
+ schedulable_tis, _ = new_run.update_state(session=session)
+ new_run.schedule_tis(schedulable_tis, session=session)
+ session.flush()
+ session.expire_all()
+ new_ti = new_run.get_task_instances(session=session)[0]
+ assert new_ti.state == TaskInstanceState.SCHEDULED, (
+ f"new run's TaskInstance should reach SCHEDULED, got {new_ti.state!r},
it is stuck."
+ )
diff --git a/airflow-core/tests/unit/timetables/test_interval_timetable.py
b/airflow-core/tests/unit/timetables/test_interval_timetable.py
index dbe27c6784f..eedd7d62dea 100644
--- a/airflow-core/tests/unit/timetables/test_interval_timetable.py
+++ b/airflow-core/tests/unit/timetables/test_interval_timetable.py
@@ -27,7 +27,11 @@ import time_machine
from airflow._shared.timezones.timezone import utc
from airflow.exceptions import AirflowTimetableInvalid
from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction,
Timetable
-from airflow.timetables.interval import CronDataIntervalTimetable,
DeltaDataIntervalTimetable
+from airflow.timetables.interval import (
+ CronDataIntervalTimetable,
+ DeltaDataIntervalTimetable,
+ _DataIntervalTimetable,
+)
START_DATE = pendulum.DateTime(2021, 9, 4, tzinfo=utc)
@@ -73,12 +77,12 @@ def test_no_catchup_first_starts_at_current_time(
)
@time_machine.travel(pendulum.DateTime(2021, 9, 7, 15, tzinfo=utc))
def test_zero_length_last_interval_does_not_re_emit_logical_date(catchup:
bool) -> None:
- """A zero-length ``data_interval`` (``start == end``) on the previous run
- must not cause ``next_dagrun_info`` to re-emit that run's logical_date.
-
- These appear when a DAG was scheduled by ``CronTriggerTimetable`` and later
- switched to ``CronDataIntervalTimetable``. Without the guard the scheduler
- loops on "run already exists; skipping dagrun creation".
+ """A zero-length ``data_interval`` (``start == end``) on the previous run
must not
+ cause ``next_dagrun_info`` to re-emit that run's logical_date. These
appear when a
+ Dag was scheduled by ``CronTriggerTimetable`` and later switched to
+ ``CronDataIntervalTimetable``; without the guard the scheduler loops on
"run
+ already exists; skipping dagrun creation". The guard's single retry always
+ resolves this for ``CronDataIntervalTimetable``.
"""
timetable = CronDataIntervalTimetable("0 17 * * *", utc)
last_run_at = pendulum.DateTime(2021, 9, 5, 17, tzinfo=utc)
@@ -92,6 +96,113 @@ def
test_zero_length_last_interval_does_not_re_emit_logical_date(catchup: bool)
assert next_info == DagRunInfo.interval(start=expected_start,
end=expected_end)
[email protected](
+ "catchup",
+ [pytest.param(True, id="catchup_true"), pytest.param(False,
id="catchup_false")],
+)
+@time_machine.travel(pendulum.DateTime(2021, 9, 7, 15, tzinfo=utc))
+def test_coarser_schedule_does_not_re_emit_logical_date(catchup: bool) -> None:
+ """Switching to a coarser cron must not re-emit the previous run's
logical_date.
+ An hourly run leaves the interval [00:00, 01:00). Aligning that end to a
daily
+ schedule lands back on 00:00, the logical_date the run already occupies,
which
+ would stall the scheduler on "run already exists; skipping dagrun
creation" if
+ the guard didn't advance past it. The guard's single retry resolves this
here
+ (see ``test_guard_raises_when_one_retry_is_not_enough`` for a case where
it isn't).
+ """
+ timetable = CronDataIntervalTimetable("0 0 * * *", utc)
+ last = DataInterval(
+ start=pendulum.DateTime(2021, 9, 6, 0, tzinfo=utc),
+ end=pendulum.DateTime(2021, 9, 6, 1, tzinfo=utc),
+ )
+ next_info = timetable.next_dagrun_info(
+ last_automated_data_interval=last,
+ restriction=TimeRestriction(earliest=None, latest=None,
catchup=catchup),
+ )
+ expected_start = pendulum.DateTime(2021, 9, 7, 0, tzinfo=utc)
+ expected_end = pendulum.DateTime(2021, 9, 8, 0, tzinfo=utc)
+ assert next_info == DagRunInfo.interval(start=expected_start,
end=expected_end)
+
+
+class _FixedStepTimetable(_DataIntervalTimetable):
+ """Test double for a custom Timetable whose alignment isn't
schedule-quantized.
+
+ The shipped subclasses guarantee (see the guard's comment in
``next_dagrun_info``)
+ that a single ``_get_next`` step past the realigned ``start`` clears
+ ``last_automated_data_interval.start``, but only because of invariants
specific to
+ them. This double doesn't uphold those invariants: ``_align_to_prev``
overshoots to
+ a fixed point well before the reference interval, and ``_get_next`` only
advances by
+ a small fixed step, so escaping the collision can need more steps than the
guard's
+ single retry budgets for -- in which case the guard raises instead of
retrying again.
+
+ Only the two methods this scenario reaches are implemented; the rest keep
the base
+ class's ``NotImplementedError``.
+ """
+
+ def __init__(self, align_point: pendulum.DateTime, step:
datetime.timedelta) -> None:
+ super().__init__()
+ self._align_point = align_point
+ self._step = step
+ self.get_next_call_count = 0
+
+ def _align_to_prev(self, current: pendulum.DateTime) -> pendulum.DateTime:
+ return self._align_point
+
+ def _get_next(self, current: pendulum.DateTime) -> pendulum.DateTime:
+ self.get_next_call_count += 1
+ return current + self._step
+
+ def infer_manual_data_interval(self, *, run_after: pendulum.DateTime) ->
DataInterval:
+ raise NotImplementedError()
+
+
+def test_guard_single_retry_resolves_collision() -> None:
+ """The collision guard's one retry is enough when a single step clears the
+ stale boundary. This uses a test-only Timetable double (not a shipped
+ subclass) to exercise the guard's mechanism directly, independent of any
+ real cron/timedelta alignment semantics.
+ """
+ last_start = pendulum.DateTime(2021, 9, 7, 15, tzinfo=utc)
+ last = DataInterval(start=last_start, end=last_start +
datetime.timedelta(minutes=1))
+ align_point = last_start - datetime.timedelta(minutes=1)
+ step = datetime.timedelta(minutes=2)
+
+ single_step = align_point + step
+ assert single_step > last.start, "test setup must reproduce the
single-retry success case"
+
+ timetable = _FixedStepTimetable(align_point=align_point, step=step)
+ next_info = timetable.next_dagrun_info(
+ last_automated_data_interval=last,
+ restriction=TimeRestriction(earliest=None, latest=None, catchup=True),
+ )
+
+ # 1 retry inside the guard, plus 1 more to compute the returned interval's
`end`.
+ assert timetable.get_next_call_count == 2
+ expected_start = last_start + datetime.timedelta(minutes=1)
+ expected_end = last_start + datetime.timedelta(minutes=3)
+ assert next_info == DagRunInfo.interval(start=expected_start,
end=expected_end)
+
+
+def test_guard_raises_when_one_retry_is_not_enough() -> None:
+ """The guard must fail loudly, not hang the scheduler, when its one retry
+ doesn't clear the collision. Retrying indefinitely on a ``_get_next`` that
+ never advances past the collision would spin an unbounded loop forever, on
+ a code path that runs for every Dag in every scheduler loop. Bounding the
+ guard to a single retry and raising instead turns that into a clear,
+ immediate error rather than a wedged scheduler.
+ """
+ last_start = pendulum.DateTime(2021, 9, 7, 15, tzinfo=utc)
+ last = DataInterval(start=last_start, end=last_start +
datetime.timedelta(minutes=1))
+ align_point = last_start - datetime.timedelta(minutes=6)
+ step = datetime.timedelta(minutes=2)
+
+ timetable = _FixedStepTimetable(align_point=align_point, step=step)
+ with pytest.raises(ValueError, match="timetable did not advance past"):
+ timetable.next_dagrun_info(
+ last_automated_data_interval=last,
+ restriction=TimeRestriction(earliest=None, latest=None,
catchup=True),
+ )
+
+
@pytest.mark.parametrize(
"earliest",
[pytest.param(None, id="none"), pytest.param(START_DATE, id="start_date")],
