reload dag bundle configuration without restarts

[samraj2k]

Summary

PR 2/2 of stack:
#73209
#73217

Dag processors currently read the bundle list only at startup. A deployment must restart them to add, update, or remove a bundle. In a large deployment, these restarts arent feasible if the bundles are being added frequently.

This PR makes adds the capability of hot reload. That means each Dag processor request the active bundle list at the existing bundle_refresh_check_interval. The configured class returns the complete list of active bundles. A missing bundle is inactive.

Note

The approach here allows eventual convergence.
Consider the following case. At time t the following bundles are active:

finance
risk
marketing

and total 3 Dag processor hosts are running.

Now, at time t + x, DagP_1 recieves the latest list, which omits bundle risk. Now, this DagP correctly marks the dags with bundle risk as stale. At time t+y, (x ~= y), the DagP_2 is still working on stale list, it will mark the Dags coming from bundle risk as non-stale. Though, eventually DagP_2 will recieve the correct list, and correctly marks Dags as stale, or even a different DagP host can do the same. Thus, we establish this contract for anyone who intend to use

A similar scenario can happen with the bundles list too

For example:

1. DagP_A still sees [sales].
2. DagP_B sees [] and marks sales inactive.
3. DagP_A runs later and marks sales active again. DagP_A then stops permanently (due to downscale etc).

Thus, each processor also reconciles the bundle list with database on every check. This repeated reconciliation makes the database converge.

A bundle name identifies stable construction settings. If settings such as the bundle class, repository, branch, connection, or refresh interval change, the provider must use a new bundle name. The provider must continue to resolve the previous name while retained Dag runs can still need it.

The default configuration provider can be used by filtered Dag processors that have different local bundle lists. In that case, absence from one processor's partial list does not deactivate a bundle. A dynamic provider supplies the complete active list, so a filtered processor can still reconcile every bundle while it parses only its assigned bundles.

This PR depends on #73209. So once that code is landed, this will be landed

Related alternative: #71111.
Though, the PR stack solves this differently by providing relevant abstractions.

Test Plan

This was tested on local by creating a temporary FileDagBundleProvider which reads from a file: active_bundles.json.

Initially, only bundle-a was present:
["bundle-a"]

Logs:

Then, without restarting the service bundle-b was added:

Then, bundle-a was removed

Service logs during removal:

DAG bundle bundle-a ... has been disabled
Refreshed Dag bundle configuration: added=[], removed=['bundle-a']
Deactivating Dag bundle_a_reload_test...
Deactivated 1 DAGs...

Validation:

• Ruff formatting and lint checks passed.
• Airflow core mypy passed.
• Fast and manual prek checks passed.

---

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

• Yes — Codex (GPT-5.6)

Generated-by: Codex (GPT-5.6) following the guidelines

---

Drafted-by: Codex (GPT-5.6); reviewed by @samraj2k before posting

[samraj2k]

This builds on #73209, so the question there applies here too: which deployment needs runtime bundle reloads, and how does this relate to the other open design in #71111? Converting to draft until that's settled.

Hey @kaxil replied on the parent PR: #73209 (comment)