Fix backfill marked complete before DagRuns are created (#61375)

The scheduler's _mark_backfills_complete() could mark a backfill as
completed during the window between the Backfill row commit and DagRun
creation. Add an EXISTS guard on BackfillDagRun so backfills still
being initialized are skipped.

Author: shivaam

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -84,7 +84,7 @@
     TaskInletAssetReference,
     TaskOutletAssetReference,
 )
-from airflow.models.backfill import Backfill
+from airflow.models.backfill import Backfill, BackfillDagRun
 from airflow.models.callback import Callback
 from airflow.models.dag import DagModel
 from airflow.models.dag_version import DagVersion
@@ -1895,6 +1895,9 @@ def _mark_backfills_complete(self, session: Session = NEW_SESSION) -> None:
         # todo: AIP-78 simplify this function to an update statement
         query = select(Backfill).where(
             Backfill.completed_at.is_(None),
+            # Guard: backfill must have at least one BackfillDagRun association,
+            # otherwise it is still being set up (see #61375).
+            exists(select(BackfillDagRun.id).where(BackfillDagRun.backfill_id == Backfill.id)),
             ~exists(
                 select(DagRun.id).where(
                     and_(DagRun.backfill_id == Backfill.id, DagRun.state.in_(unfinished_states))
@@ -1903,9 +1906,13 @@ def _mark_backfills_complete(self, session: Session = NEW_SESSION) -> None:
         )
         backfills = list(session.scalars(query))
         if not backfills:
+            self.log.debug("No backfills to mark as complete.")
             return
         self.log.info("marking %s backfills as complete", len(backfills))
         for b in backfills:
+            self.log.info(
+                "Marking backfill %s (dag_id=%s) as complete at %s", b.id, b.dag_id, now
+            )
             b.completed_at = now
 
     @add_debug_span

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -64,7 +64,7 @@
     AssetPartitionDagRun,
     PartitionedAssetKeyLog,
 )
-from airflow.models.backfill import Backfill, _create_backfill
+from airflow.models.backfill import Backfill, BackfillDagRun, ReprocessBehavior, _create_backfill
 from airflow.models.dag import DagModel, get_last_dagrun, infer_automated_data_interval
 from airflow.models.dag_version import DagVersion
 from airflow.models.dagbundle import DagBundleModel
@@ -8688,6 +8688,57 @@ def test_mark_backfills_completed(dag_maker, session):
     assert b.completed_at.timestamp() > 0
 
 
+def test_mark_backfills_complete_skips_initializing_backfill(dag_maker, session):
+    clear_db_backfills()
+    dag_id = "test_backfill_race_lifecycle"
+    with dag_maker(serialized=True, dag_id=dag_id, schedule="@daily"):
+        BashOperator(task_id="hi", bash_command="echo hi")
+    b = Backfill(
+        dag_id=dag_id,
+        from_date=pendulum.parse("2021-01-01"),
+        to_date=pendulum.parse("2021-01-03"),
+        max_active_runs=10,
+        dag_run_conf={},
+        reprocess_behavior=ReprocessBehavior.NONE,
+    )
+    session.add(b)
+    session.commit()
+    backfill_id = b.id
+    session.expunge_all()
+    runner = SchedulerJobRunner(
+        job=Job(job_type=SchedulerJobRunner.job_type), executors=[MockExecutor(do_update=False)]
+    )
+    runner._mark_backfills_complete()
+    b = session.get(Backfill, backfill_id)
+    assert b.completed_at is None
+    session.expunge_all()
+    dr = DagRun(
+        dag_id=dag_id,
+        run_id="backfill__2021-01-01T00:00:00+00:00",
+        run_type=DagRunType.BACKFILL_JOB,
+        logical_date=pendulum.parse("2021-01-01"),
+        data_interval=(pendulum.parse("2021-01-01"), pendulum.parse("2021-01-02")),
+        run_after=pendulum.parse("2021-01-02"),
+        state=DagRunState.SUCCESS,
+        backfill_id=backfill_id,
+    )
+    session.add(dr)
+    session.flush()
+    session.add(
+        BackfillDagRun(
+            backfill_id=backfill_id,
+            dag_run_id=dr.id,
+            logical_date=pendulum.parse("2021-01-01"),
+            sort_ordinal=1,
+        )
+    )
+    session.commit()
+    session.expunge_all()
+    runner._mark_backfills_complete()
+    b = session.get(Backfill, backfill_id)
+    assert b.completed_at is not None
+
+
 class Key1Mapper(CorePartitionMapper):
     """Partition Mapper that returns only key-1 as downstream key"""