This is an automated email from the ASF dual-hosted git repository.
Lee-W 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 6194df6d703 Fix manual runs getting a zero-length data interval across
DST (#70089)
6194df6d703 is described below
commit 6194df6d703073e37cbb3379cdbe627505c24f29
Author: Harjoth Khara <[email protected]>
AuthorDate: Wed Aug 12 01:01:04 2026 -0700
Fix manual runs getting a zero-length data interval across DST (#70089)
---
airflow-core/newsfragments/70089.bugfix.rst | 1 +
airflow-core/src/airflow/timetables/_cron.py | 11 +-
.../unit/timetables/test_interval_timetable.py | 167 +++++++++++++++++++++
.../unit/timetables/test_trigger_timetable.py | 47 ++++++
4 files changed, 224 insertions(+), 2 deletions(-)
diff --git a/airflow-core/newsfragments/70089.bugfix.rst
b/airflow-core/newsfragments/70089.bugfix.rst
new file mode 100644
index 00000000000..a8fa4b5e224
--- /dev/null
+++ b/airflow-core/newsfragments/70089.bugfix.rst
@@ -0,0 +1 @@
+Fix cron timetables that could schedule a Dag run for a period that has not
elapsed yet on a DST transition day.
diff --git a/airflow-core/src/airflow/timetables/_cron.py
b/airflow-core/src/airflow/timetables/_cron.py
index db9950afb33..ed0b3c76bb7 100644
--- a/airflow-core/src/airflow/timetables/_cron.py
+++ b/airflow-core/src/airflow/timetables/_cron.py
@@ -166,14 +166,21 @@ class CronMixin:
return convert_to_utc(current.in_timezone(self._timezone) + delta)
def _get_prev(self, current: DateTime) -> DateTime:
- """Get the first schedule before specified time, with DST fixed."""
+ """Get the first schedule strictly before specified time, with DST
fixed."""
naive = make_naive(current, self._timezone)
cron = croniter(self._expression, start_time=naive)
scheduled = cron.get_prev(datetime.datetime)
if TYPE_CHECKING:
assert isinstance(scheduled, datetime.datetime)
if not _covers_every_hour(cron):
- return convert_to_utc(make_aware(scheduled, self._timezone))
+ prev = convert_to_utc(make_aware(scheduled, self._timezone))
+ # croniter steps back on naive wall clock, but make_aware can map
a tick inside
+ # a DST transition forward onto current or later. Keep stepping
until strictly
+ # earlier; get_prev is strictly decreasing, so this terminates.
+ while prev >= current:
+ scheduled = cron.get_prev(datetime.datetime)
+ prev = convert_to_utc(make_aware(scheduled, self._timezone))
+ return prev
delta = naive - scheduled
return convert_to_utc(current.in_timezone(self._timezone) - delta)
diff --git a/airflow-core/tests/unit/timetables/test_interval_timetable.py
b/airflow-core/tests/unit/timetables/test_interval_timetable.py
index dbe27c6784f..df57d3bd5a3 100644
--- a/airflow-core/tests/unit/timetables/test_interval_timetable.py
+++ b/airflow-core/tests/unit/timetables/test_interval_timetable.py
@@ -383,6 +383,173 @@ def
test_cron_next_dagrun_info_alignment(last_data_interval: DataInterval, expec
assert info == expected_info
[email protected](
+ ("timetable", "run_after", "expected_interval"),
+ [
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("America/New_York")),
+ pendulum.DateTime(2024, 3, 10, 7, 30, tzinfo=utc),
+ DataInterval(
+ pendulum.DateTime(2024, 3, 9, 7, tzinfo=utc),
+ pendulum.DateTime(2024, 3, 10, 7, tzinfo=utc),
+ ),
+ id="new-york-dst-gap",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("America/New_York")),
+ pendulum.DateTime(2024, 3, 10, 7, tzinfo=utc),
+ DataInterval(
+ pendulum.DateTime(2024, 3, 9, 7, tzinfo=utc),
+ pendulum.DateTime(2024, 3, 10, 7, tzinfo=utc),
+ ),
+ id="new-york-dst-gap-exact",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("Europe/Zurich")),
+ pendulum.DateTime(2023, 3, 26, 1, 30, tzinfo=utc),
+ DataInterval(
+ pendulum.DateTime(2023, 3, 25, 1, tzinfo=utc),
+ pendulum.DateTime(2023, 3, 26, 1, tzinfo=utc),
+ ),
+ id="zurich-dst-gap",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("Australia/Lord_Howe")),
+ pendulum.DateTime(2024, 10, 5, 15, 30, tzinfo=utc),
+ DataInterval(
+ pendulum.DateTime(2024, 10, 4, 15, 30, tzinfo=utc),
+ pendulum.DateTime(2024, 10, 5, 15, 30, tzinfo=utc),
+ ),
+ id="lord-howe-half-hour-dst-gap",
+ ),
+ pytest.param(
+ # make_aware pins fold=1, so the 1am tick on the fold day is its
second
+ # occurrence (06:00Z) and has not happened yet at 05:30Z.
+ CronDataIntervalTimetable("0 1 * * *",
pendulum.timezone("America/New_York")),
+ pendulum.DateTime(2024, 11, 3, 5, 30, tzinfo=utc),
+ DataInterval(
+ pendulum.DateTime(2024, 11, 1, 5, tzinfo=utc),
+ pendulum.DateTime(2024, 11, 2, 5, tzinfo=utc),
+ ),
+ id="new-york-fold",
+ ),
+ ],
+)
+def test_cron_infer_manual_data_interval_dst(
+ timetable: CronDataIntervalTimetable,
+ run_after: pendulum.DateTime,
+ expected_interval: DataInterval,
+):
+ """The inferred interval must not be zero-length, nor end after
``run_after``."""
+ interval = timetable.infer_manual_data_interval(run_after=run_after)
+ assert interval == expected_interval
+ assert interval.start < interval.end
+ assert interval.end <= run_after
+
+
[email protected](
+ ("timetable", "run_after", "expected_info"),
+ [
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("America/New_York")),
+ pendulum.DateTime(2024, 3, 10, 7, 30, tzinfo=utc),
+ DagRunInfo.interval(
+ pendulum.DateTime(2024, 3, 10, 7, tzinfo=utc),
+ pendulum.DateTime(2024, 3, 11, 6, tzinfo=utc),
+ ),
+ id="new-york-dst-gap",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("Europe/Zurich")),
+ pendulum.DateTime(2023, 3, 26, 1, 30, tzinfo=utc),
+ DagRunInfo.interval(
+ pendulum.DateTime(2023, 3, 26, 1, tzinfo=utc),
+ pendulum.DateTime(2023, 3, 27, tzinfo=utc),
+ ),
+ id="zurich-dst-gap",
+ ),
+ ],
+)
+def test_cron_next_dagrun_info_after_dst_manual_run(
+ timetable: CronDataIntervalTimetable,
+ run_after: pendulum.DateTime,
+ expected_info: DagRunInfo,
+):
+ """A zero-length interval here would trip the point-in-time guard and skip
a period."""
+ manual_interval = timetable.infer_manual_data_interval(run_after=run_after)
+ info = timetable.next_dagrun_info(
+ last_automated_data_interval=manual_interval,
+ restriction=TimeRestriction(None, None, True),
+ )
+ assert info == expected_info
+
+
[email protected](
+ ("timetable", "current_time", "expected_info"),
+ [
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("America/New_York")),
+ pendulum.DateTime(2024, 3, 10, 7, 30, tzinfo=utc),
+ DagRunInfo.interval(
+ pendulum.DateTime(2024, 3, 9, 7, tzinfo=utc),
+ pendulum.DateTime(2024, 3, 10, 7, tzinfo=utc),
+ ),
+ id="new-york-dst-gap",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("America/New_York")),
+ pendulum.DateTime(2024, 3, 10, 7, tzinfo=utc),
+ DagRunInfo.interval(
+ pendulum.DateTime(2024, 3, 9, 7, tzinfo=utc),
+ pendulum.DateTime(2024, 3, 10, 7, tzinfo=utc),
+ ),
+ id="new-york-dst-gap-exact",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("Europe/Zurich")),
+ pendulum.DateTime(2023, 3, 26, 1, 30, tzinfo=utc),
+ DagRunInfo.interval(
+ pendulum.DateTime(2023, 3, 25, 1, tzinfo=utc),
+ pendulum.DateTime(2023, 3, 26, 1, tzinfo=utc),
+ ),
+ id="zurich-dst-gap",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 2 * * *",
pendulum.timezone("Australia/Lord_Howe")),
+ pendulum.DateTime(2024, 10, 5, 15, 30, tzinfo=utc),
+ DagRunInfo.interval(
+ pendulum.DateTime(2024, 10, 4, 15, 30, tzinfo=utc),
+ pendulum.DateTime(2024, 10, 5, 15, 30, tzinfo=utc),
+ ),
+ id="lord-howe-half-hour-dst-gap",
+ ),
+ pytest.param(
+ CronDataIntervalTimetable("0 1 * * *",
pendulum.timezone("America/New_York")),
+ pendulum.DateTime(2024, 11, 3, 5, 30, tzinfo=utc),
+ DagRunInfo.interval(
+ pendulum.DateTime(2024, 11, 1, 5, tzinfo=utc),
+ pendulum.DateTime(2024, 11, 2, 5, tzinfo=utc),
+ ),
+ id="new-york-fold",
+ ),
+ ],
+)
+def test_cron_next_dagrun_info_no_catchup_dst(
+ timetable: CronDataIntervalTimetable,
+ current_time: pendulum.DateTime,
+ expected_info: DagRunInfo,
+) -> None:
+ """``_skip_to_latest`` must not propose a run whose period has not
elapsed."""
+ with time_machine.travel(current_time, tick=False):
+ info = timetable.next_dagrun_info(
+ last_automated_data_interval=None,
+ restriction=TimeRestriction(earliest=None, latest=None,
catchup=False),
+ )
+ assert info is not None
+ assert info == expected_info
+ assert info.run_after <= current_time
+
+
class TestCronIntervalDst:
"""
Test cron interval timetable can correctly enter a DST boundary.
diff --git a/airflow-core/tests/unit/timetables/test_trigger_timetable.py
b/airflow-core/tests/unit/timetables/test_trigger_timetable.py
index eef68b0679f..f4718f8d04b 100644
--- a/airflow-core/tests/unit/timetables/test_trigger_timetable.py
+++ b/airflow-core/tests/unit/timetables/test_trigger_timetable.py
@@ -1149,3 +1149,50 @@ def
test_iter_partition_dagrun_infos_dst_america_new_york_spring_forward() -> No
# run_after == partition_date for both ticks.
for info in infos:
assert info.run_after == info.partition_date
+
+
+def _get_cron_trigger_run_immediately_info() -> DagRunInfo | None:
+ timetable = CronTriggerTimetable("0 1 * * *", timezone="America/New_York",
run_immediately=True)
+ return timetable.next_dagrun_info(
+ last_automated_data_interval=None,
+ restriction=TimeRestriction(earliest=None, latest=None, catchup=False),
+ )
+
+
+def _get_multi_cron_trigger_run_immediately_info() -> DagRunInfo | None:
+ timetable = MultipleCronTriggerTimetable("0 1 * * *",
timezone="America/New_York", run_immediately=True)
+ return timetable.next_dagrun_info(
+ last_automated_data_interval=None,
+ restriction=TimeRestriction(earliest=None, latest=None, catchup=False),
+ )
+
+
+def _get_cron_partition_run_immediately_info() -> DagRunInfo | None:
+ timetable = CoreCronPartitionTimetable("0 1 * * *",
timezone="America/New_York", run_immediately=True)
+ return timetable.next_dagrun_info_v2(
+ last_dagrun_info=None,
+ restriction=TimeRestriction(earliest=None, latest=None, catchup=False),
+ )
+
+
+@time_machine.travel(pendulum.DateTime(2024, 11, 3, 5, 30, tzinfo=utc),
tick=False)
[email protected](
+ "get_info",
+ [
+ pytest.param(_get_cron_trigger_run_immediately_info,
id="cron-trigger"),
+ pytest.param(_get_multi_cron_trigger_run_immediately_info,
id="multi-cron-trigger"),
+ pytest.param(_get_cron_partition_run_immediately_info,
id="cron-partition"),
+ ],
+)
+def test_run_immediately_does_not_pick_future_run(get_info:
typing.Callable[[], DagRunInfo | None]) -> None:
+ """``run_immediately`` must select a past tick, never one that has not
happened yet.
+
+ At 01:30 EDT on the fold day the 1am tick's second occurrence (06:00Z) is
still in
+ the future, so the previous day's tick is the one to run. Covers
``CronTriggerTimetable``,
+ ``MultipleCronTriggerTimetable``, and ``CronPartitionTimetable``, which
all share the same
+ ``CronMixin._get_prev``.
+ """
+ info = get_info()
+ assert info is not None
+ assert info.run_after <= pendulum.DateTime(2024, 11, 3, 5, 30, tzinfo=utc)
+ assert info.run_after == pendulum.DateTime(2024, 11, 2, 5, tzinfo=utc)
