fixed start/end when already run once

airflow/timetables/simple.py

@@ -141,11 +141,14 @@ def next_dagrun_info(
         current_time = timezone.coerce_datetime(timezone.utcnow())
 
         if last_automated_data_interval is not None:  # has already run once
-            start = last_automated_data_interval.end
-            end = current_time
+            if last_automated_data_interval.end > current_time:  # start date is future
+                start = restriction.earliest
+                elapsed = last_automated_data_interval.end - last_automated_data_interval.start
 
-            if start > end:  # Skip scheduling if the last run ended in the future
-                return None
+                end = start + elapsed.as_timedelta()
+            else:
+                start = last_automated_data_interval.end
+                end = current_time
         else:  # first run
             start = restriction.earliest
             end = max(restriction.earliest, current_time)

tests/timetables/test_continuous_timetable.py

@@ -91,13 +91,12 @@ def test_no_runs_after_end_date(timetable, restriction):
     assert next_info is None
 
 
-@time_machine.travel(DURING_DATE)
+@time_machine.travel(BEFORE_DATE)
 def test_no_false_triggering_with_future_start_date_after_run(timetable, restriction):
-    FUTURE_DATE = DURING_DATE.add(days=1)
-
     next_info = timetable.next_dagrun_info(
-        last_automated_data_interval=DataInterval(START_DATE, FUTURE_DATE),
+        last_automated_data_interval=DataInterval(BEFORE_DATE, BEFORE_DATE.add(hours=1)),
         restriction=restriction,
     )
 
-    assert next_info is None
+    assert next_info is not None
+    assert next_info.data_interval.start == START_DATE