Guard 2 to 3+ migration against custom Dag bundle configs using startup-based initialization

[jason810496]

• related: make bundle_name not nullable #47592
• closes: Flaky “Requested bundle is not configured” error during 2 to 3.1+ upgrade with custom Dag bundle names #63323

Why

Some users encounter a flaky error during 2.x → 3.1+ upgrades. If the deployment configures any bundle other than the default dags-folder, the 0082_3_1_0_make_bundle_name_not_nullable migration writes a single hard-coded bundle name that no longer matches the user's runtime config, so triggering a DagRun raises:

airflow/airflow-core/src/airflow/dag_processing/bundles/manager.py

Lines 333 to 345 in 450a08e

	def get_bundle(self, name: str, version: str | None = None) -> BaseDagBundle:
	"""
	Get a DAG bundle by name.
	:param name: The name of the DAG bundle.
	:param version: The version of the DAG bundle you need (optional). If not provided, ``tracking_ref`` will be used instead.
	:return: The DAG bundle.
	"""
	cfg_bundle = self._bundle_config.get(name)
	if not cfg_bundle:
	raise ValueError(f"Requested bundle '{name}' is not configured.")
	return cfg_bundle.bundle_class(name=name, version=version, **cfg_bundle.kwargs)

Reproducible edge cases:

Case 1: Dag bundle name is not dags-folder

AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST='[{"name": "main", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {"path": "/files/dags", "refresh_interval": 1}}]'

Case 2: Multiple custom Dag bundles

AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST='[{"name": "main1", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {"path": "/files/dags", "refresh_interval": 1}}, {"name": "main2", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {"path": "/files/dags", "refresh_interval": 1}}]'

In both cases the migration inserts dags-folder rather than the configured bundle, leaving the Dag rows pointing at a bundle_name that the running deployment cannot resolve.

How

• At DagFileProcessorManager startup, after sync_bundles_to_db flushes the latest bundle state, scan DagModel for legacy-candidate rows (NULL relative_fileloc and no DagVersion for that dag_id) and route them to the most-specific configured bundle whose absolute path contains the Dag's fileloc. relative_fileloc is written at the same time so fileloc-based stale-detection works later.
• Rows whose fileloc does not lie under any configured bundle's path are left untouched rather than forced onto a default bundle. Writing bundle_name without a verified relative_fileloc would produce an active row that task workloads cannot execute (no dag_rel_path). Skipped rows then follow the normal staleness lifecycle: when the operator removed the bundle from config, sync_bundles_to_db flips it to active=False and the next deactivate_stale_dags scan marks the row is_stale=True. Once the operator restores a matching bundle, the next parser run resets is_stale and writes fresh bundle_name and relative_fileloc — no manual airflow dags reserialize required.
• Safe under core.multi_team: a bundle path belongs to at most one team, so routing by the most-specific containing path cannot cross a team boundary. The same routing applies in single- and multi-team deployments.
• Bundles that are configured but failed to persist (e.g. construction error during sync) are excluded from reassignment targets to avoid FK violations against dag_bundle.
• The repair is chunked (_REASSIGN_BATCH_SIZE=1000) and each chunk runs in its own internally-owned transaction (no caller-provided session is committed), bounding the row-lock window. Each UPDATE re-asserts the legacy-candidate predicate as a compare-and-swap guard so a concurrent parser write that lands between SELECT and UPDATE wins the race.
• Parser-loop stale-Dag scan now skips rows with relative_fileloc IS NULL so the standard fileloc-based deactivation does not crash on legacy rows the repair couldn't recover.

What

• airflow-core/src/airflow/dag_processing/bundles/manager.py
  • DagBundlesManager.reassign_dags_with_unconfigured_bundles — chunked, CAS-guarded fileloc-aware reassignment that owns its own transactions.
  • DagBundlesManager._resolve_active_bundle_paths — returns configured-and-active bundle names (in config order) and {name: path} ordered by path length descending for longest-prefix matching.
  • DagBundlesManager.bundle_names — list of configured bundle names in config order.
  • New module-level _best_bundle_for_fileloc — short-circuits on the first prefix match given descending-by-length input; rejects parent-traversal results.
  • Explicit session.flush() after sync_bundles_to_db so subsequent reads see the freshly synced bundle rows under Airflow's autoflush=False sessions.
• airflow-core/src/airflow/dag_processing/manager.py
  • Calls reassign_dags_with_unconfigured_bundles once at startup from sync_bundles (not on every bundle refresh).
  • Stale-Dag parser loop skips rows with NULL relative_fileloc to avoid Path(None) crashes on legacy rows.
• Tests
  • airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py — parametrized coverage for custom names, multiple bundles, overlapping paths (deepest wins), fileloc outside any bundle (row skipped, not forced to default), missing fileloc, multi-team routing, relative_fileloc backfill, FK-safety when a configured bundle is missing from dag_bundle, concurrent-DFP startup, batching boundary, and the full sync → repair → stale-scan → re-parse lifecycle.
  • airflow-core/tests/unit/dag_processing/test_manager.py — asserts reassignment runs once at startup and not on bundle refresh, and that legacy rows with NULL relative_fileloc are still picked up by deactivate_deleted_dags.

Alternative considered: fix it in the DB migration (#63185, reverted)

The earlier approach (PR #63185, commit 450a08e78, reverted in eb7e928d9) tried to fix this inside the 0082_3_1_0_make_bundle_name_not_nullable Alembic migration itself — reading the user's bundle config from inside the migration and rewriting bundle_name per Dag during schema upgrade.

That approach was abandoned because:

• Migrations should not import Airflow application code. Pulling DagBundlesManager / conf.getjson("dag_processor", "dag_bundle_config_list") into an Alembic revision couples schema upgrades to the live config and the bundle-loading code path, which can change shape between releases and break re-runs on older revisions.
• The migration runs before bundles are constructible. The Dag processor hasn't started, bundles haven't been instantiated, and their path isn't safely resolvable in every deployment (e.g. remote/git bundles). Inferring the right target during schema upgrade is fragile.
• It only handled the dags-folder case. The migration wrote a single hard-coded bundle name, so deployments with a custom bundle name or multiple bundles still ended up with the wrong bundle_name and the same runtime error (Flaky “Requested bundle is not configured” error during 2 to 3.1+ upgrade with custom Dag bundle names #63323).
• Multi-team safety has to live next to the runtime config. Routing across team boundaries is a policy decision that belongs in the component that knows about teams, not in a schema migration.

Doing the reassignment at DagFileProcessorManager startup keeps the migration purely schema-level, lets us use the real bundle objects (get_bundle(name).path) for fileloc matching, and gives a single place to enforce FK correctness against dag_bundle.

---

Was generative AI tooling used to co-author this PR?

• Yes, with help of Claude Code Opus 4.7 following the guidelines

[jason810496]

So the two steps are:

Migration: set bundle_name = 'dags-folder' where NULL (satisfies NOT NULL) remains unchanged
sync_bundles_to_db(): reassign DAGs with unconfigured bundle_name to the actual configured bundle

So based on the offline discussion and the context in this thread, I will make the migration file as is, and harden the startup operation for Dag-Processor.

Since running airflow dags reserialize could also workaround this issue, so approach for me will be consolidate some common utils for both dag_reserialize CLI and DagBundlesManager.sync_bundles_to_db (or DagFileProcessorManager.sync_bundles )

airflow/airflow-core/src/airflow/cli/commands/dag_command.py

Line 721 in 50e27b3

def dag_reserialize(args, session: Session = NEW_SESSION) -> None:

airflow/airflow-core/src/airflow/dag_processing/bundles/manager.py

Lines 211 to 213 in 076f519

	@provide_session
	def sync_bundles_to_db(self, *, session: Session = NEW_SESSION) -> None:
	self.log.debug("Syncing DAG bundles to the database")

airflow/airflow-core/src/airflow/dag_processing/manager.py

Lines 258 to 260 in 48d7ac5

	def sync_bundles(self) -> None:
	"""Sync configured DAG bundles to the metadata database."""
	DagBundlesManager().sync_bundles_to_db()

[Copilot]

Pull request overview

This PR aims to prevent a flaky “Requested bundle '{name}' is not configured.” error during 2.x → 3.1+ upgrades by reassigning DAGs that reference unconfigured bundle names to a configured fallback bundle at DagFileProcessorManager startup.

Changes:

• Add DagBundlesManager.reassign_dags_with_unconfigured_bundles() to bulk-update DagModel.bundle_name to a fallback configured bundle.
• Invoke the reassignment immediately after sync_bundles_to_db() during DagFileProcessorManager.sync_bundles() startup.
• Add unit tests covering reassignment behavior and ensuring reassignment is not performed during bundle refresh.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 8 comments.

File	Description
airflow-core/src/airflow/dag_processing/manager.py	Calls bundle DB sync + startup reassignment.
airflow-core/src/airflow/dag_processing/bundles/manager.py	Implements the DB-level reassignment helper.
airflow-core/tests/unit/dag_processing/test_manager.py	Verifies reassignment is called once at startup (not on refresh).
airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py	Adds DB tests for reassignment logic and sync+reassign flow.