Keep a stop signal from hanging an idle worker

CHANGELOG.md

@@ -22,6 +22,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `ox_health --max-age` and `--worker-timeout` accept the duration forms
   `ox_prune --older-than` takes (`7d`, `24h`, `90m`, `45s`). A plain number
   still means seconds, fractions included.
+- A worker ended by a second stop signal exits with code 130 without
+  logging `second signal received; forcing exit.` first; the
+  `--processes` supervisor still logs its own line.
+
+### Fixed
+
+- A stop signal could leave an idle `ox_worker` hung instead of draining.
+  It stayed hung until a second signal or the process manager ended it,
+  or for good when it was a worker process whose supervisor had been
+  killed. A worker that has
+  finished starting now drains on the signal. Present since 0.1.0.
+- A worker process whose supervisor died while the worker was still
+  starting ran on as an orphan. It now drains and exits having claimed
+  nothing. Present since 0.3.0.
+- If the `--processes` supervisor hit an error while running, a second
+  stop signal did not send SIGKILL to a worker process that would not
+  exit, and the supervisor waited for it forever. The second and third
+  signals now escalate as they do in any other stop. Present since 0.3.0.
 
 ## [1.2.0] - 2026-09-12
 

docs/llms-full.txt

@@ -1877,10 +1877,11 @@ runs in its own process group, and why the systemd unit above sets
 A worker whose supervisor dies without signalling it (SIGKILL, an OOM kill)
 does not run on as an orphan. On Linux the kernel sends it SIGTERM the
 moment the supervisor exits (`PR_SET_PDEATHSIG`), so it drains through its
-ordinary signal path. Everywhere else, and on Linux in the window before
-that flag is set, the worker notices within one poll interval that its
-parent pid has changed, logs `worker_orphaned` at WARNING, drains and
-exits.
+ordinary signal path. Everywhere else, the worker notices within one poll
+interval that its parent pid is no longer the supervisor's, logs
+`worker_orphaned` at WARNING, drains and exits. A worker whose supervisor
+is already gone when the worker finishes starting makes the same check
+before its first poll, and exits having claimed nothing.
 
 ## Scaling out
 
@@ -3551,6 +3552,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `ox_health --max-age` and `--worker-timeout` accept the duration forms
   `ox_prune --older-than` takes (`7d`, `24h`, `90m`, `45s`). A plain number
   still means seconds, fractions included.
+- A worker ended by a second stop signal exits with code 130 without
+  logging `second signal received; forcing exit.` first; the
+  `--processes` supervisor still logs its own line.
+
+### Fixed
+
+- A stop signal could leave an idle `ox_worker` hung instead of draining.
+  It stayed hung until a second signal or the process manager ended it,
+  or for good when it was a worker process whose supervisor had been
+  killed. A worker that has
+  finished starting now drains on the signal. Present since 0.1.0.
+- A worker process whose supervisor died while the worker was still
+  starting ran on as an orphan. It now drains and exits having claimed
+  nothing. Present since 0.3.0.
+- If the `--processes` supervisor hit an error while running, a second
+  stop signal did not send SIGKILL to a worker process that would not
+  exit, and the supervisor waited for it forever. The second and third
+  signals now escalate as they do in any other stop. Present since 0.3.0.
 
 ## [1.2.0] - 2026-09-12
 

docs/production.md

@@ -153,10 +153,11 @@ runs in its own process group, and why the systemd unit above sets
 A worker whose supervisor dies without signalling it (SIGKILL, an OOM kill)
 does not run on as an orphan. On Linux the kernel sends it SIGTERM the
 moment the supervisor exits (`PR_SET_PDEATHSIG`), so it drains through its
-ordinary signal path. Everywhere else, and on Linux in the window before
-that flag is set, the worker notices within one poll interval that its
-parent pid has changed, logs `worker_orphaned` at WARNING, drains and
-exits.
+ordinary signal path. Everywhere else, the worker notices within one poll
+interval that its parent pid is no longer the supervisor's, logs
+`worker_orphaned` at WARNING, drains and exits. A worker whose supervisor
+is already gone when the worker finishes starting makes the same check
+before its first poll, and exits having claimed nothing.
 
 ## Scaling out
 

src/django_ox/management/commands/ox_worker.py

@@ -1,17 +1,21 @@
 import argparse
 import logging
 import os
+import queue
 import signal
 import sys
+import threading
+from collections.abc import Callable
+from contextlib import suppress
 from pathlib import Path
 from typing import Any
 
 from django.core.management.base import BaseCommand, CommandError, CommandParser
 
 from django_ox.compat import DEFAULT_TASK_BACKEND_ALIAS
-from django_ox.supervisor import STOP_SIGNALS, Supervisor
+from django_ox.supervisor import STOP_SIGNALS, SUPERVISOR_PID_ENV, Supervisor
 from django_ox.timeouts import RECYCLE_EXIT_CODE
-from django_ox.worker import worker_class
+from django_ox.worker import Worker, worker_class
 
 logger = logging.getLogger("django_ox")
 
@@ -72,6 +76,9 @@ def add_arguments(self, parser: CommandParser) -> None:
         )
 
     def handle(self, *args: Any, **options: Any) -> None:
+        # Removed at once, so a task that starts an ox_worker of its own does
+        # not pass this process's supervisor on to it.
+        supervisor_pid = os.environ.pop(SUPERVISOR_PID_ENV, None)
         if options["processes"] < 1:
             raise CommandError("--processes must be at least 1.")
         if options["verbosity"] > 0 and not logger.handlers:
@@ -100,8 +107,16 @@ def handle(self, *args: Any, **options: Any) -> None:
 
         parent_pid = None
         if options["worker_index"] is not None:
-            parent_pid = os.getppid()
-            _die_with_parent()
+            # A getppid() read here names whoever adopted the child if the
+            # supervisor died first, and the orphan would watch that pid
+            # forever. The snapshot remains for a child started without the
+            # variable, such as one a supervisor still running the previous
+            # release restarts after an upgrade.
+            if supervisor_pid is not None:
+                with suppress(ValueError):
+                    parent_pid = int(supervisor_pid)
+            if parent_pid is None:
+                parent_pid = os.getppid()
 
         queues = (
             [q.strip() for q in options["queues"].split(",") if q.strip()]
@@ -118,31 +133,18 @@ def handle(self, *args: Any, **options: Any) -> None:
             parent_pid=parent_pid,
         )
 
-        signals_seen = 0
-
-        def handle_signal(signum: int, frame: Any) -> None:
-            # Counted here rather than read off worker.stopping: a worker
-            # that is recycling is already stopping, and the operator's
-            # first signal during that drain should not be the force-exit.
-            nonlocal signals_seen
-            signals_seen += 1
-            if signals_seen > 1:
-                logger.error(
-                    "Worker %s: second signal received; forcing exit.", worker.worker_id
-                )
-                os._exit(130)
-            logger.info(
-                "Worker %s received %s; draining in-flight tasks. "
-                "Signal again to force exit.",
-                worker.worker_id,
-                signal.Signals(signum).name,
-            )
-            worker.request_stop()
-
-        signal.signal(signal.SIGTERM, handle_signal)
-        signal.signal(signal.SIGINT, handle_signal)
+        retire_signal_thread = install_stop_handlers(worker)
+        if parent_pid is not None:
+            # After the handlers: a parent-death signal armed before they
+            # exist would kill the child instead of draining it. A supervisor
+            # that died before the arming sends nothing; run() compares the
+            # parent with parent_pid before every poll, its first included.
+            _die_with_parent()
 
-        worker.run()
+        try:
+            worker.run()
+        finally:
+            retire_signal_thread()
         if worker.recycling:
             # A thread the timeout could not stop is still running. A normal
             # exit would wait for it at interpreter shutdown, which is the
@@ -155,20 +157,97 @@ def handle_signal(signum: int, frame: Any) -> None:
         sys.exit(0)
 
 
+def install_stop_handlers(worker: Worker) -> Callable[[], None]:
+    """
+    Make SIGTERM and SIGINT drain ``worker``, and a second one exit at once.
+
+    Returns the function that ends the helper thread once ``run()`` returns.
+    """
+    stop_requests: queue.SimpleQueue[int | None] = queue.SimpleQueue()
+    signals_seen = 0
+
+    def handle_signal(signum: int, frame: Any) -> None:
+        # Only a count, a SimpleQueue.put and os._exit belong here. Python
+        # runs this on the main thread wherever that thread was, including
+        # inside Event.wait() with the stop Event's lock held, so
+        # request_stop(), logging or anything else that takes a lock can
+        # block forever on the lock its own thread holds. SimpleQueue.put is
+        # documented as safe to call from a signal handler.
+        #
+        # Counted here rather than read off worker.stopping: a worker that
+        # is recycling is already stopping, and the operator's first signal
+        # during that drain should not be the force-exit.
+        nonlocal signals_seen
+        signals_seen += 1
+        if signals_seen > 1:
+            # Nothing is written first, not even with os.write: a stderr
+            # pipe that nobody is reading would block the exit.
+            os._exit(130)
+        stop_requests.put(signum)
+
+    def act_on_stop_requests() -> None:
+        while (signum := stop_requests.get()) is not None:
+            # The stop first: a log handler that raises or blocks must not
+            # cost the drain.
+            worker.request_stop()
+            with suppress(Exception):
+                logger.info(
+                    "Worker %s received %s; draining in-flight tasks. "
+                    "Signal again to force exit.",
+                    worker.worker_id,
+                    signal.Signals(signum).name,
+                )
+
+    # The handlers first: off the main thread signal.signal() raises, and
+    # nothing should be left behind when it does. A signal that arrives
+    # before the thread starts waits in the queue. A daemon, so a forced
+    # exit need not wait for it.
+    signums = (signal.SIGTERM, signal.SIGINT)
+    previous = {signum: signal.getsignal(signum) for signum in signums}
+    for signum in signums:
+        signal.signal(signum, handle_signal)
+    thread = threading.Thread(
+        target=act_on_stop_requests, name="ox-signal", daemon=True
+    )
+    try:
+        thread.start()
+    except BaseException:
+        # Handlers that fed a queue nobody reads would swallow a stop, and a
+        # signal already queued is one nobody else will act on. This is
+        # ordinary code, not a handler, so it can stop the worker itself.
+        for signum, handler in previous.items():
+            if handler is not None:
+                signal.signal(signum, handler)
+        if not stop_requests.empty():
+            worker.request_stop()
+        raise
+
+    def retire() -> None:
+        stop_requests.put(None)
+        thread.join(timeout=1.0)
+
+    return retire
+
+
 def _die_with_parent() -> None:
     """
     On Linux, ask the kernel to SIGTERM this process when its parent exits
-    (PR_SET_PDEATHSIG). The worker also polls ``os.getppid()``, which covers
-    every platform and the window before this call; this is the prompt
-    version. Best effort: anything missing or refused is ignored.
+    (PR_SET_PDEATHSIG). The worker also compares ``os.getppid()`` with the
+    supervisor's pid, which covers every platform, a supervisor that died
+    before this call and a refused call; this is the prompt version.
     """
     if not sys.platform.startswith("linux"):
         return
     try:
         import ctypes
 
         libc = ctypes.CDLL(None, use_errno=True)
-        libc.prctl(1, signal.SIGTERM, 0, 0, 0)  # PR_SET_PDEATHSIG
+        if libc.prctl(1, signal.SIGTERM, 0, 0, 0) != 0:  # PR_SET_PDEATHSIG
+            logger.debug(
+                "PR_SET_PDEATHSIG refused (errno %d); the parent pid check "
+                "still applies",
+                ctypes.get_errno(),
+            )
     except (OSError, AttributeError):
         return