Make backfill creation robust against scheduler race condition #64534

Description

The _create_backfill() method in airflow-core/src/airflow/models/backfill.py creates a backfill in two steps:

1. It commits the Backfill row to the database (session.commit())
2. Then it creates all the DagRun and BackfillDagRun rows linked to that backfill

The scheduler runs _mark_backfills_complete() every ~30 seconds, looking for backfills where all associated DagRuns have finished. If the scheduler runs between Step 1 and Step 2, it sees a backfill with zero DagRuns, concludes "all zero runs are done", and marks the backfill as complete. When Step 2 then creates the DagRuns, they're orphaned.

PR #62561 fixed this with a pragmatic guard: require at least one BackfillDagRun row before marking complete, with a 2-minute timeout to clean up orphaned backfills that failed during initialization. This works, but relies on a time-based heuristic rather than addressing the root cause.

During PR #62561 review, Daniel Standish suggested two more robust approaches as follow-ups:

Approach 1: Atomic transaction — Wrap the entire backfill creation in a single transaction so Backfill, DagRun, and BackfillDagRun rows all appear atomically. One way: change session.commit() to session.flush() in _create_backfill() (backfill.py L605). flush() assigns the ID without committing; create_session() commits everything at the end. Tradeoff: the AlreadyRunningBackfill check relies on the early commit to block concurrent duplicates, so this would need database-level locking. (Also suggested by Kaxil)

Approach 2: Add a state field or datetime — Add an explicit signal to distinguish "initializing" from "running" backfills. Either a nullable started_at/initialized_at datetime (set after DagRuns are created), or a state enum (QUEUED/RUNNING/COMPLETED/FAILED). Replaces the 2-minute heuristic with an explicit check. Requires a DB migration.

These approaches are not mutually exclusive. Either one alone improves on the current heuristic.

Key code locations:

• airflow-core/src/airflow/models/backfill.py — Backfill model and _create_backfill()
• airflow-core/src/airflow/jobs/scheduler_job_runner.py — _mark_backfills_complete()

Use case/motivation

The current fix in PR #62561 uses a 2-minute time-based heuristic to distinguish "still initializing" from "orphaned" backfills. This works but is fragile — it assumes backfill creation always completes within 2 minutes, and doesn't provide a clean way to detect or recover from failed initializations. A more robust solution would either eliminate the race window entirely (atomic transaction) or make the initialization state explicit (state field).

Related issues

• Backfill job marked as completed immediately, though backfill dag runs still are executing #61375 — Original bug report (backfill marked complete before DagRuns are created)
• PR Fix backfill marked complete before DagRuns are created #62561 — Current fix with the EXISTS guard + time-based cutoff
• Daniel's review comment suggesting these follow-up approaches
• Kaxil's review comment suggesting flush() instead of commit()

Code of Conduct

• I agree to follow this project's Code of Conduct