Fix task supervisor hang by resetting SIGCHLD in CeleryExecutor

[hooiv]

What type of change is this?

Bug fix.

What does this PR do?

Fixes a task supervisor hang and memory leak in the CeleryExecutor where supervisors stay alive indefinitely after their subprocess has exited.

The Root Cause:
As discussed in #70117, psutil.Process.wait(timeout=0) was returning None because the child's wait status was already being consumed by something else before the supervisor could retrieve it.

The culprit is Celery's billiard library. When the CeleryExecutor runs, it executes the task supervisor directly inside the Celery worker process. billiard installs a global SIGCHLD signal handler that performs an out-of-band waitpid(-1, WNOHANG) loop. When the Airflow task subprocess exits, billiard's signal handler wakes up and indiscriminately reaps it. This steals the exit status, causing the Task SDK's psutil.wait() to return None and bypass the socket cleanup logic.

The Fix:
This PR explicitly resets SIGCHLD to SIG_DFL in celery_executor_utils.py right before the supervisor process drops into setproctitle and executes the task. By restoring the default signal disposition, billiard no longer steals the exit status, and the Airflow supervisor can properly reap its own children and gracefully exit.

Closes #70117

Note: An AI coding assistant was used to help write pr description and trace the Celery/billiard signal handler interaction .

[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
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

[hooiv]

Updated the patch and description to match the confirmed failure path in #70117. psutil.Process.wait(timeout=0) can return None after the child wait status has already been consumed; _exit_code then stayed unset, so the supervisor loop never entered socket cleanup. The regression now covers that exact None result and maps it to the existing unknown exit code (-1). The earlier dumb-init/NoSuchProcess explanation was removed because it was not confirmed. Focused Linux test and Ruff checks pass.

[hooiv]

I've pushed a completely new approach that fixes the actual root cause!

In #70117, @Andrushika correctly pointed out that dumb-init couldn't be reaping the process while the supervisor was alive, but proved something was stealing the exit status.

It turns out it was Celery's billiard library. Because the supervisor runs inside the Celery worker process, it inherits billiard's aggressive global SIGCHLD handler. When the Airflow task subprocess exits, the billiard handler intercepts the signal and calls waitpid(-1, WNOHANG), silently eating the exit status before our psutil.wait() can see it!

I've force-pushed a fix that simply resets SIGCHLD to SIG_DFL inside celery_executor_utils.py right before the supervisor takes over. This prevents billiard from interfering and perfectly resolves the memory leak.

The PR description has been updated .

[Andrushika]

Thanks for figuring it out! Would it be possible to give a minimal reproducible example for how "Celery's billiard library" triggers this bug? Please also include the necessary tests in your fix.
BTW, you can turn this PR into "open" when it's ready for review. It's draft currently.

[hooiv]

Thanks for the review! I have pushed a commit that adds a unit test for this fix in tests/providers/celery/executors/test_celery_executor.py.

Regarding how billiard triggers this bug, Celery's multiprocessing fork (billiard) installs a global SIGCHLD handler upon initialization. This handler loops and calls waitpid(-1, os.WNOHANG) to aggressively reap any child process that exits, which intercepts the exit statuses of child processes spawned by the Airflow Task SDK.

Here is a minimal reproducible example demonstrating the issue:

`python
import os
import time
import psutil
from billiard.pool import Pool

if name == "main":
# 1. Starting a billiard pool installs a global SIGCHLD handler
# that aggressively calls waitpid(-1, WNOHANG) to reap ANY exiting children
pool = Pool(1)

# 2. Airflow Task SDK forks a child process to run the task
pid = os.fork()
if pid == 0:
    # Task subprocess does its work and exits
    os._exit(42)
    
proc = psutil.Process(pid)

# Give the child time to exit and billiard's signal handler time to wake up and reap it
time.sleep(0.2)

# 3. Airflow Task SDK tries to check the exit status, but wait() returns None 
# because billiard already stole the exit status!
print(f"Child PID {pid}")
print(f"psutil.wait(timeout=0) returned: {proc.wait(timeout=0)}")

pool.close()
pool.join()

`

If you run this script, proc.wait(timeout=0) will return None (meaning the status has been consumed) instead of the actual exit code 42. By resetting SIGCHLD to SIG_DFL before executing the workload, we prevent billiard from stealing the exit status of the Airflow tasks.