feat(companion): Shut down the manager from the arbiter

Arbiter.stop now signals the companion manager alongside the workers. It sends
the same SIGTERM (graceful) or SIGQUIT (immediate), waits the graceful_timeout
for both the workers and the manager to exit, then SIGKILLs whatever remains.
A graceful SIGTERM lets the manager stop its own companions before exiting.

stop_companion_manager(sig) signals the manager pid when it is running and
clears the pid on ESRCH; the SIGCHLD reaper clears it on a normal exit.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

docs/design/companion-process-manager.md

@@ -689,7 +689,7 @@ No per-companion logic in Arbiter.
 - [x] Implement `restart`.
 - [x] Implement transactional `reread`.
 - [x] Add manager spawn/reap logic in Arbiter.
-- [ ] Add manager shutdown handling in Arbiter.
+- [x] Add manager shutdown handling in Arbiter.
 - [ ] Wire Gunicorn reload to manager `reread` or restart.
 - [ ] Close Gunicorn-only fds in manager child.
 - [ ] Close manager-only fds in companion child.

gunicorn/arbiter.py

@@ -396,13 +396,15 @@ class Arbiter:
         if not graceful:
             sig = signal.SIGQUIT
         limit = time.time() + self.cfg.graceful_timeout
-        # instruct the workers to exit
+        # instruct the workers and the companion manager to exit
         self.kill_workers(sig)
+        self.stop_companion_manager(sig)
         # wait until the graceful timeout
-        while self.WORKERS and time.time() < limit:
+        while (self.WORKERS or self.companion_manager_pid) and time.time() < limit:
             time.sleep(0.1)
 
         self.kill_workers(signal.SIGKILL)
+        self.stop_companion_manager(signal.SIGKILL)
 
     def reexec(self):
         """\
@@ -703,6 +705,23 @@ class Arbiter:
             self.log.exception("Exception in companion manager process")
             sys.exit(-1)
 
+    def stop_companion_manager(self, sig):
+        """Signal the companion manager to exit, if it is running.
+
+        A graceful SIGTERM lets the manager stop its own companions before it
+        exits; SIGKILL forces it down. The reaper clears the pid once it dies,
+        so a manager that is already gone is a no-op here.
+        """
+        if self.companion_manager_pid == 0:
+            return
+        try:
+            os.kill(self.companion_manager_pid, sig)
+        except OSError as e:
+            if e.errno == errno.ESRCH:
+                self.companion_manager_pid = 0
+                return
+            raise
+
     def kill_workers(self, sig):
         """\
         Kill all workers with the signal `sig`

tests/test_arbiter.py

@@ -2,7 +2,9 @@
 # This file is part of gunicorn released under the MIT license.
 # See the NOTICE for more information.
 
+import errno
 import os
+import signal
 from unittest import mock
 
 import gunicorn.app.base
@@ -180,6 +182,39 @@ def test_arbiter_reap_clears_companion_manager_pid(mock_os_waitpid):
     assert arbiter.companion_manager_pid == 0
 
 
+def test_stop_companion_manager_signals_running():
+    arbiter = gunicorn.arbiter.Arbiter(DummyApplication())
+    arbiter.companion_manager_pid = 4242
+    with mock.patch("os.kill") as kill:
+        arbiter.stop_companion_manager(signal.SIGTERM)
+    kill.assert_called_once_with(4242, signal.SIGTERM)
+
+
+def test_stop_companion_manager_noop_when_not_running():
+    arbiter = gunicorn.arbiter.Arbiter(DummyApplication())
+    with mock.patch("os.kill") as kill:
+        arbiter.stop_companion_manager(signal.SIGTERM)
+    kill.assert_not_called()
+
+
+def test_stop_companion_manager_clears_pid_when_already_gone():
+    arbiter = gunicorn.arbiter.Arbiter(DummyApplication())
+    arbiter.companion_manager_pid = 4242
+    with mock.patch("os.kill", side_effect=OSError(errno.ESRCH, "no such process")):
+        arbiter.stop_companion_manager(signal.SIGTERM)
+    assert arbiter.companion_manager_pid == 0
+
+
+@mock.patch('gunicorn.sock.close_sockets')
+def test_arbiter_stop_signals_companion_manager(close_sockets):
+    arbiter = gunicorn.arbiter.Arbiter(DummyApplication())
+    arbiter.stop_companion_manager = mock.Mock()
+    arbiter.stop()
+    signals = [call.args[0] for call in arbiter.stop_companion_manager.call_args_list]
+    assert signal.SIGTERM in signals
+    assert signal.SIGKILL in signals
+
+
 class PreloadedAppWithEnvSettings(DummyApplication):
     """
     Simple application that makes use of the 'preload' feature to