jason810496 opened a new pull request, #72498: URL: https://github.com/apache/airflow/pull/72498
- closes: https://github.com/apache/airflow/issues/66754 - related: https://github.com/apache/airflow/pull/66132 -- is actual the same root case, but the previous patch only solved the `CronTriggerTimetable` to `CronDataIntervalTimetable` case, it didn't solved the hourly to daily case. - suppress: #66754 - related: #66754 ## Why Switching a Dag to a coarser cron (hourly to daily, for example) can make the realigned next run collide with the logical_date already on record, so the scheduler loops on "run already exists; skipping dagrun creation" forever instead of advancing. ## Notes for tracing the e2e behavior ``` DagModel(dag_id="schedule_change") next_dagrun = 2026-05-04T00:00 next_dagrun_data_interval = [2026-05-04, 2026-05-05) next_dagrun_create_after = 2026-05-05T00:00 | | read, every _do_scheduling tick once now() >= next_dagrun_create_after v _create_dag_runs() existing_dagruns.get((dag_id, next_dagrun)) -> found R1 "run already exists; skipping dagrun creation" | v calculate_dagrun_date_fields(reference_run=R1) = dag.next_dagrun_info( last_automated_run_info=R1.data_interval, # fixed: R1 never changes schedule=daily cron, # fixed: Dag is not edited again ) -> deterministically returns (2026-05-04, 2026-05-05) again | | write (no new DagRun created) v DagModel.next_dagrun* unchanged: still (2026-05-04, 2026-05-05) ``` `DagModel.next_dagrun*` can look like a stale cache, but it is not a failed invalidation: `calculate_dagrun_date_fields` reruns on every tick. The problem is that its inputs never change, `reference_run` stays `R1` forever since no new DagRun is ever created, so the recompute is a deterministic function of fixed inputs and keeps landing on the same answer. ### Call chain ``` SchedulerJobRunner._execute_helper (main loop) \- _do_scheduling |- [A] _create_dagruns_for_dags | \- DagModel.dags_needing_dagruns (reads next_dagrun_create_after) | \- _create_dag_runs | |- serdag.create_dagrun(state=QUEUED) new DagRun row only | \- DagModel.calculate_dagrun_date_fields | \- dag.next_dagrun_info() writes next_dagrun / next_dagrun_create_after | |- [B] _start_queued_dagruns (QUEUED DagRun -> RUNNING) | \- [C] _schedule_all_dag_runs TI scheduling lives here, not in [A] \- _schedule_dag_run \- DagRun.update_state \- task_instance_scheduling_decisions -> schedulable_tis \- DagRun.schedule_tis TI.state = SCHEDULED (the actual TI scheduling) (elsewhere, at parse time, the same sink function is reached via a different top) DagFileProcessorManager.persist_parsing_result \- update_dag_parsing_results_in_db \- SerializedDAG.bulk_write_to_db \- DagModelOperation.update_dags \- DagModel.calculate_dagrun_date_fields \- dag.next_dagrun_info() same sink as [A] ``` `next_dagrun_info` writes `DagModel.next_dagrun*` columns that gate `dags_needing_dagruns`. ### Why one step is enough for the `_DataIntervalTimetable` subclasses ``` no NEW-schedule instant can exist in this gap v v --------*===========*---------------------------------------*------> time last.start last.end next(align_end) = align_end (largest NEW-schedule instant <= last.end) ``` For both shipped subclasses `earliest` is itself always some instant of the same schedule, so if it beats `align_end` it cannot land in the gap either, it must be at or after `next(align_end)`. `DeltaDataIntervalTimetable` never reaches this branch at all, its `_align_to_prev` / `_align_to_next` are the identity. Neither invariant is enforced by `next_dagrun_info` itself though, so a custom `Timetable` could break it, which is why the guard retries instead of assuming one step always suffices. The retry is bounded and raises if exhausted rather than looping forever. --- ##### Was generative AI tooling used to co-author this PR? - [x] Yes, with help of Claude Code Sonnet 5 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
