Fix manual runs getting a zero-length data interval across DST

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.

airflow-core/src/airflow/timetables/_cron.py

@@ -166,14 +166,21 @@ def _get_next(self, current: DateTime) -> DateTime:
         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)
 

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
 
 
+@pytest.mark.parametrize(
+    ("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
+
+
+@pytest.mark.parametrize(
+    ("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
+
+
+@pytest.mark.parametrize(
+    ("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.

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)
+@pytest.mark.parametrize(
+    "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)