Extracted bundle_cleanup_main() function from local scope to avoid subprocess errors on OSes using spawn()

[bluczko]

closes: #62512

This PR slightly modifies the _run_stale_bundle_cleanup() context manager:

• Nested function bundle_cleanup_main() has been extracted out of _run_stale_bundle_cleanup() to be accessible from module level scope and renamed to _bundle_cleanup_main() to match the convention
• check_interval parameter was added to _bundle_cleanup_main(), which previously was accessible from outer function scope; the param is applied to the function when subprocess is created

This change allows to run Airflow Celery worker on macOS and possibly other OSes, which use spawn() instead of fork() to create subprocesses. The stale bundle cleanup Process target function lifecycle is now handled properly on these systems.

---

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

• Yes (please specify the tool below)

---

• 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.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[Copilot]

Pull request overview

Adjusts the Celery worker’s stale DAG bundle cleanup subprocess to be compatible with OSes using multiprocessing spawn() by ensuring the subprocess target is picklable (module-level).

Changes:

• Extracts the stale bundle cleanup loop into a module-level _bundle_cleanup_main() function.
• Passes check_interval into the subprocess via Process(..., args=(check_interval,)) instead of capturing it from an outer scope.

Comments suppressed due to low confidence (2)

providers/celery/src/airflow/providers/celery/cli/celery_command.py:132

• In the cleanup subprocess target, check_interval is untyped and the function has no return type annotation. Most helpers in this module are annotated, and adding check_interval: int (or float) plus -> None would improve readability/static checking (and makes it clear the value must be picklable for spawn).

def _bundle_cleanup_main(check_interval):
    from airflow.dag_processing.bundles.base import BundleUsageTrackingManager

    mgr = BundleUsageTrackingManager()

    while True:
        time.sleep(check_interval)
        mgr.remove_stale_bundle_versions()

providers/celery/src/airflow/providers/celery/cli/celery_command.py:162

• sub_proc.terminate() can raise AttributeError: 'NoneType' object has no attribute 'terminate' if sub_proc.start() fails (the process exists but _popen is still None), which was also visible in the original issue trace. Consider guarding with if sub_proc and sub_proc.is_alive(): (and optionally sub_proc.join() after terminate) so the cleanup path never masks the original exception.

    finally:
        if sub_proc:
            sub_proc.terminate()