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

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

@@ -122,6 +122,21 @@ def _serve_logs(skip_serve_logs: bool = False):
             sub_proc.terminate()
 
 
+def _bundle_cleanup_main(check_interval: int) -> None:
+    """
+    Main function for stale bundle cleanup process.
+    Must be picklable for process start method.
+    """
+
+    from airflow.dag_processing.bundles.base import BundleUsageTrackingManager
+
+    mgr = BundleUsageTrackingManager()
+
+    while True:
+        time.sleep(check_interval)
+        mgr.remove_stale_bundle_versions()
+
+
 @contextmanager
 def _run_stale_bundle_cleanup():
     """Start stale bundle cleanup sub-process."""
@@ -136,19 +151,15 @@ def _run_stale_bundle_cleanup():
         with suppress(BaseException):
             yield
         return
-    from airflow.dag_processing.bundles.base import BundleUsageTrackingManager
 
     log.info("starting stale bundle cleanup process")
     sub_proc = None
 
-    def bundle_cleanup_main():
-        mgr = BundleUsageTrackingManager()
-        while True:
-            time.sleep(check_interval)
-            mgr.remove_stale_bundle_versions()
-
     try:
-        sub_proc = Process(target=bundle_cleanup_main)
+        sub_proc = Process(
+            target=_bundle_cleanup_main,
+            args=(check_interval,),
+        )
         sub_proc.start()
         yield
     finally:

providers/celery/tests/unit/celery/cli/test_celery_command.py

@@ -632,7 +632,7 @@ def test_stale_bundle_cleanup(mock_process):
     calls = mock_process.call_args_list
     assert len(calls) == 1
     actual = [x.kwargs["target"] for x in calls]
-    assert actual[0].__name__ == "bundle_cleanup_main"
+    assert actual[0].__name__ == "_bundle_cleanup_main"
 
 
 class TestLoggerSetupHandler: