dag_processing: initialize versioned bundles for callbacks (#52040)

[nailo2c]

Closes: #52040

Why

1. Original issue:
   When using GitDagBundle with supports_versioning=True, Airflow DAG Processor fails to execute DAG callbacks because it cannot find the materialized code under versions/<commit_hash>/.
2. Additional concern:
   The DAG processor should honor stale_bundle_cleanup_* to avoid accumulating versioned bundles.

How

1. Initialize versioned bundles in _add_callback_to_queue.
2. Wrap callback execution in BundleVersionLock to update usage and avoid cleanup races.
3. Add _cleanup_stale_bundle_versions in DagFileProcessorManager._run_parsing_loop to periodically check and remove stale bundle versions.

What

Before the fixes, _execute_dag_callbacks raised errors and the callback logs did not appear in the DAG processor logs.

Here is my remote dag repo: https://github.com/nailo2c/my_repo/blob/main/example_git_bundle_success_dag.py

After the fixes, _execute_dag_callbacks completes without errors, and callbacks continue to work after committing new versions.

aaron.chen@Aarons-MacBook-Air-3 airflow_2 % kubectl exec -n airflow airflow-dag-processor-fd9f96f64-nzc7h -c dag-processor -- grep -R "SUCCESS CALLBACK" /opt/airflow/logs/dag_processor
/opt/airflow/logs/dag_processor/2026-01-17/git_bundle/example_git_bundle_success_dag.py.log:{"timestamp":"2026-01-17T18:47:18.871227Z","level":"warning","event":">>> SUCCESS CALLBACK TRIGGERED VERSION 2 <<<","logger":"root","filename":"example_git_bundle_success_dag.py","lineno":7}
/opt/airflow/logs/dag_processor/2026-01-17/git_bundle/example_git_bundle_success_dag.py.log:{"timestamp":"2026-01-17T19:32:06.676855Z","level":"warning","event":">>> SUCCESS CALLBACK TRIGGERED VERSION 3 <<<","logger":"root","filename":"example_git_bundle_success_dag.py","lineno":7}

Both on_success_callback and on_failure_callback behave as expected.

---

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

• Yes (please specify the tool below)
  GPT 5.2

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[SameerMesiah97]

Looks good to me. The placement of the lock is appropriate and the test coverage is solid.

[Copilot]

Pull request overview

This PR fixes an issue where GitDagBundle with versioning support fails during callback execution because versioned code bundles are not initialized. The changes ensure that versioned bundles are properly initialized before callback execution and that stale bundle versions are cleaned up periodically.

Changes:

• Initialize versioned bundles when callbacks are queued
• Wrap callback execution in BundleVersionLock to track usage and prevent cleanup races
• Add periodic cleanup of stale bundle versions in the DAG processing loop

Reviewed changes

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

File	Description
airflow-core/src/airflow/dag_processing/manager.py	Added bundle initialization in _add_callback_to_queue and periodic stale bundle cleanup logic
airflow-core/src/airflow/dag_processing/processor.py	Wrapped callback execution in BundleVersionLock context manager
airflow-core/tests/unit/dag_processing/test_manager.py	Added tests for bundle initialization and stale bundle cleanup
airflow-core/tests/unit/dag_processing/test_processor.py	Added test for BundleVersionLock usage during callback execution

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[jedcunningham]

@nailo2c thanks for the PR! Copilot has suggested a couple extra paths to test, could you take a look at adding those?

[nailo2c]

Thanks for the review! Let me take a look at it.

[amoghrajesh]

Just some nits, tried with my use case:

cat callback.py.log | grep -i succeed
{"timestamp":"2026-01-29T07:21:32.422700Z","level":"info","event":"DAG succeeded!","logger":"task.stdout"}

And it works fine, thanks for looking into this @nailo2c

[ephraimbuddy]

@nailo2c , I will like to include this in 3.1.7. Can you address the reviews?

[nailo2c]

Hi @ephraimbuddy, thanks for the thorough review, I appreciate it. I'll check it today :)

[nailo2c]

Hi all, thanks for all the reviews, they were very helpful!

I've addressed all the feedback. If you notice anything else I can improve, please let me know :)

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker cac9c0b v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.