From 99fdb77c586da3fee0fd14ae7a8d7a1ccd29f261 Mon Sep 17 00:00:00 2001 From: PhiLily <252857470+PhiLily@users.noreply.github.com> Date: Tue, 15 Sep 2026 18:34:02 +0300 Subject: [PATCH] Keep a stop signal from hanging an idle worker ox_worker's SIGTERM and SIGINT handler logged and called worker.request_stop(), which sets a threading.Event. Python runs signal handlers on the main thread, and an idle worker's main thread spends its time in Event.wait() on that same Event. A signal that arrived while Event.wait() held the Event's lock ran a handler that waited for that lock on the thread holding it, and the worker hung instead of draining. The handler now only records the signal on a SimpleQueue, which is documented as safe to call from a signal handler, and a second signal calls os._exit(130) with nothing before it. A daemon thread requests the stop and then logs. The handlers are installed before that thread starts, and if it cannot start the previous handlers come back and a signal already queued stops the worker. The supervisor's handler also only records. Its run loop logs, stops, forwards and escalates, including between child starts and while it waits for children to exit after the loop has ended, so a second or third signal still reaches SIGKILL after an error. The supervisor sets OX_SUPERVISOR_PID in each child's environment. The child removes it, arms PR_SET_PDEATHSIG after its handlers are in place, and compares its parent with that pid before every poll, so a child whose supervisor died while it was starting drains having claimed nothing. A child running an older release ignores the variable. The regression test raises SIGTERM from inside Condition.__exit__ on the worker's stop Event, where the lock is still held, so it fails on the old code every time rather than by timing. --- CHANGELOG.md | 18 + docs/llms-full.txt | 27 +- docs/production.md | 9 +- .../management/commands/ox_worker.py | 143 +++- src/django_ox/supervisor.py | 102 ++- tests/test_stop_signals.py | 624 ++++++++++++++++++ tests/test_supervisor.py | 127 +++- 7 files changed, 976 insertions(+), 74 deletions(-) create mode 100644 tests/test_stop_signals.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e9aa903..7982d2d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/llms-full.txt b/docs/llms-full.txt index ae3281b..e1e57f1 100644 --- a/docs/llms-full.txt +++ b/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 diff --git a/docs/production.md b/docs/production.md index a7eba75..20eb18f 100644 --- a/docs/production.md +++ b/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 diff --git a/src/django_ox/management/commands/ox_worker.py b/src/django_ox/management/commands/ox_worker.py index 6cf74de..0393659 100644 --- a/src/django_ox/management/commands/ox_worker.py +++ b/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,12 +157,84 @@ 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 @@ -168,7 +242,12 @@ def _die_with_parent() -> None: 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 diff --git a/src/django_ox/supervisor.py b/src/django_ox/supervisor.py index c25f35a..5248834 100644 --- a/src/django_ox/supervisor.py +++ b/src/django_ox/supervisor.py @@ -21,6 +21,7 @@ import logging import os +import queue import signal import subprocess import sys @@ -79,6 +80,13 @@ FORCE_SIGNAL = getattr(signal, "SIGKILL", signal.SIGTERM) +# Set in each child's environment to the supervisor's pid, read before the +# child existed, so a child whose supervisor is already gone can tell. An +# environment variable rather than a flag: a child running an older release, +# as after a downgrade in place, ignores it instead of refusing to start. +SUPERVISOR_PID_ENV = "OX_SUPERVISOR_PID" + + def child_command( worker_args: list[str], index: int, argv0: str | None = None ) -> list[str]: @@ -115,6 +123,7 @@ def _child_env() -> dict[str, str]: module = getattr(settings, "SETTINGS_MODULE", None) if module: env["DJANGO_SETTINGS_MODULE"] = module + env[SUPERVISOR_PID_ENV] = str(os.getpid()) return env @@ -162,11 +171,12 @@ def __init__( self._cap_tripped = False self._signals_seen = 0 self._kill_at: float | None = None + # What handle_signal records and the run loop acts on. + self._signals: queue.SimpleQueue[int] = queue.SimpleQueue() # Guards the stop flag and the restart decision together, so a stop - # requested between the two never starts a child that nobody will - # signal. Signal handlers run on this same thread, between two - # bytecodes of whatever it was doing, so the lock has to be - # re-entrant for them. + # requested from another thread between the two never starts a + # child that nobody will signal. Re-entrant because _start_due holds + # it while it calls _start, which takes it too. self._lock = threading.RLock() # -- children ---------------------------------------------------------- @@ -191,8 +201,8 @@ def _start(self, index: int) -> None: self._exit_codes.pop(index, None) self._restart_due.pop(index, None) if self._stopping: - # A stop request landed while Popen was running; the handler - # could not see this child yet. + # A stop requested from another thread landed while Popen was + # running, before this child was in _children to be signalled. with suppress(ProcessLookupError): proc.send_signal(signal.SIGTERM) @@ -342,28 +352,44 @@ def _kill_overdue(self) -> None: # -- signals ----------------------------------------------------------- def handle_signal(self, signum: int, frame: Any) -> None: - self._signals_seen += 1 - name = signal.Signals(signum).name - if self._signals_seen == 1: - logger.info( - "Received %s; stopping %d worker process(es). " - "Signal again to force exit.", - name, - len(self._children), - ) - elif self._kill_at is None: - logger.error( - "Second signal received; forcing worker exit, SIGKILL in %.0fs.", - self.kill_grace, - ) - self._kill_at = time.monotonic() + self.kill_grace - else: - # A third signal: the operator has waited long enough. - self._kill_at = time.monotonic() - self.request_stop() - # Children treat SIGTERM and SIGINT the same way, and their second - # signal is the force-exit, so the supervisor's count is theirs. - self._signal_children(signal.SIGTERM) + # Records the signal and nothing else; the run loop acts on it + # within POLL_INTERVAL. Python runs a handler on the main thread in + # the middle of whatever that thread was doing, which may be a log + # write or a held lock, and a handler that logs or locks there can + # fail or block. SimpleQueue.put is documented as safe to call from + # a signal handler. + self._signals.put(signum) + + def _process_signals(self) -> None: + """Act on the signals handle_signal recorded, oldest first.""" + while True: + try: + signum = self._signals.get_nowait() + except queue.Empty: + return + self._signals_seen += 1 + name = signal.Signals(signum).name + if self._signals_seen == 1: + logger.info( + "Received %s; stopping %d worker process(es). " + "Signal again to force exit.", + name, + len(self._children), + ) + elif self._kill_at is None: + logger.error( + "Second signal received; forcing worker exit, SIGKILL in %.0fs.", + self.kill_grace, + ) + self._kill_at = time.monotonic() + self.kill_grace + else: + # A third signal: the operator has waited long enough. + self._kill_at = time.monotonic() + self.request_stop() + # Children treat SIGTERM and SIGINT the same way, and their + # second signal is the force-exit, so the supervisor's count is + # theirs. + self._signal_children(signal.SIGTERM) def request_stop(self) -> None: with self._lock: @@ -385,9 +411,16 @@ def run(self) -> int: extra={"event": "supervisor_started", "processes": self.processes}, ) for index in range(self.processes): + # Between starts, so a stop that lands while children are being + # started ends the starting too, and one that arrived before + # run() starts nothing. + self._process_signals() + if self._stopping: + break self._start(index) try: while self._children or (self._restart_due and not self._stopping): + self._process_signals() self._reap_exited() self._start_due() self._kill_overdue() @@ -397,9 +430,16 @@ def run(self) -> int: # exception), leave no child behind. self.request_stop() self._signal_children(signal.SIGTERM) - for index, proc in self._children.items(): - self._record_exit(index, proc.wait()) - self._children.clear() + # Polled rather than waited on: the signals only this loop acts + # on include the second and third, which are what turn a child + # that will not exit into a SIGKILL. + while True: + self._process_signals() + self._kill_overdue() + self._reap_exited() + if not self._children: + break + time.sleep(POLL_INTERVAL) # A recycle that lands during a stop is a worker that finished the # job it was asked to do, not a failure to report upwards. failures = [ diff --git a/tests/test_stop_signals.py b/tests/test_stop_signals.py new file mode 100644 index 0000000..87d3140 --- /dev/null +++ b/tests/test_stop_signals.py @@ -0,0 +1,624 @@ +""" +A stop signal is recorded by its handler and acted on by ordinary code. + +Python runs a signal handler on the main thread, between two bytecodes of +whatever that thread was doing. When that is the inside of ``Event.wait()``, +the Event's lock is held, and a handler that sets the Event waits forever on +a lock its own thread owns. These tests put the handler exactly there rather +than hoping a timed signal lands in a window a few microseconds wide. +""" + +import builtins +import logging +import os +import signal +import subprocess +import sys +import textwrap +import threading +import time +import types + +import pytest +from django.core.management import call_command + +from django_ox.management.commands import ox_worker +from django_ox.models import OxTask +from django_ox.supervisor import Supervisor +from django_ox.worker import Worker + +from .conftest import wait_for +from .tasks import add, slow +from .test_supervisor import ( + REPO, + child_env, + in_process_env, + slot_pid, + start_worker, + wait_for_workers, +) + +# Spelled out rather than imported, so this file still collects against a +# release without the variable and its tests fail there for the real reason. +SUPERVISOR_PID_ENV = "OX_SUPERVISOR_PID" + +pytestmark = pytest.mark.skipif( + os.name != "posix", reason="the handlers under test are POSIX signal handlers" +) + +# Runs the real ox_worker command. Condition.__exit__ is wrapped so that the +# first time the main thread leaves Event.wait() on the worker's stop Event, +# with that Event's lock still held, SIGTERM is raised. raise_signal runs the +# Python handler before it returns, so the handler runs inside the lock. +_SIGNAL_INSIDE_EVENT_WAIT = textwrap.dedent( + """ + import signal + import threading + + import django + + django.setup() + + from django.core.management import call_command + + from django_ox.worker import Worker + + target = {} + real_run = Worker.run + real_exit = threading.Condition.__exit__ + + def run(self): + target["cond"] = self._stop._cond + return real_run(self) + + def exit_holding_the_lock(self, *args): + if ( + self is target.get("cond") + and threading.current_thread() is threading.main_thread() + and not target.get("raised") + ): + target["raised"] = True + signal.raise_signal(signal.SIGTERM) + return real_exit(self, *args) + + Worker.run = run + threading.Condition.__exit__ = exit_holding_the_lock + try: + call_command("ox_worker", "--interval", "0.05") + finally: + print("RAISED=%s" % bool(target.get("raised")), flush=True) + """ +) + + +@pytest.mark.django_db(transaction=True) +def test_a_signal_inside_the_idle_wait_drains_the_worker(): + try: + done = subprocess.run( # noqa: S603 + [sys.executable, "-c", _SIGNAL_INSIDE_EVENT_WAIT], + cwd=REPO, + env=child_env(), + capture_output=True, + text=True, + # A drain with nothing in flight takes well under a second. + timeout=15, + check=False, + ) + except subprocess.TimeoutExpired as hung: + # subprocess.run has already killed it. + pytest.fail( + "the worker hung after a signal inside Event.wait(); output:\n" + f"{hung.stdout!r}\n{hung.stderr!r}" + ) + output = done.stdout + done.stderr + # The signal did land inside the lock; without this a run that never + # reached the wait would pass. + assert "RAISED=True" in done.stdout, output + assert done.returncode == 0, output + assert "received SIGTERM; draining" in output, output + assert "stopped" in output, output + + +@pytest.fixture +def restore_signal_handlers(): + saved = { + signum: signal.getsignal(signum) for signum in (signal.SIGTERM, signal.SIGINT) + } + yield + for signum, handler in saved.items(): + signal.signal(signum, handler) + + +def test_the_worker_handler_only_hands_the_signal_on(restore_signal_handlers): + """ + The handler returns without calling request_stop(), and the helper + thread does the stop. The handler is called by the thread that holds + the stop Event's lock, which is the hang itself, so a handler that still + set the Event would never return. That thread is a daemon with a bounded + join, so such a handler fails this test rather than hanging the suite. + """ + worker = Worker(poll_interval=0.05) + retire = ox_worker.install_stop_handlers(worker) + handler = signal.getsignal(signal.SIGTERM) + assert callable(handler) + seen: dict[str, bool] = {} + + def signal_while_holding_the_lock(): + with worker._stop._cond: + handler(signal.SIGTERM, None) + # The helper cannot set the Event until this lock is released. + seen["stopping_inside"] = worker.stopping + + holder = threading.Thread(target=signal_while_holding_the_lock, daemon=True) + try: + holder.start() + holder.join(timeout=5) + assert not holder.is_alive(), "the handler blocked on the stop Event" + assert seen == {"stopping_inside": False} + assert wait_for(lambda: worker.stopping, timeout=5) + finally: + retire() + assert not any(t.name == "ox-signal" for t in threading.enumerate()) + + +class _Exited(Exception): + pass + + +def test_the_worker_handler_writes_nothing(restore_signal_handlers, monkeypatch): + """ + Neither signal makes the handler log, write or look anything up: the + first only queues, the second only calls os._exit(130). Logging takes + locks, and a write to a stderr pipe nobody reads blocks, so either + could stop the handler from returning or the forced exit from + happening. Every module the handler could reach them through is + replaced by one that records what the handler's thread touches: every + module the command module imports, its logger, print and open. + """ + worker = Worker(poll_interval=0.05) + handler_thread = threading.get_ident() + in_handler = False + touched: list[str] = [] + exits: list[int] = [] + + def fake_exit(code): + exits.append(code) + raise _Exited + + class Recording: + def __init__(self, name, real): + self._name = name + self._real = real + + def __getattr__(self, attr): + if in_handler and threading.get_ident() == handler_thread: + touched.append(f"{self._name}.{attr}") + if self._name == "os" and attr == "_exit": + return fake_exit + return getattr(self._real, attr) + + watched = [ + name + for name, value in vars(ox_worker).items() + if isinstance(value, types.ModuleType) or name == "logger" + ] + assert {"logger", "logging", "os", "sys", "signal", "threading"} <= set(watched) + for name in watched: + monkeypatch.setattr(ox_worker, name, Recording(name, getattr(ox_worker, name))) + for name in ("print", "open"): + real = getattr(builtins, name) + + def recorded(*args, _name=name, _real=real, **kwargs): + if in_handler and threading.get_ident() == handler_thread: + touched.append(_name) + return _real(*args, **kwargs) + + monkeypatch.setattr(builtins, name, recorded) + retire = ox_worker.install_stop_handlers(worker) + handler = signal.getsignal(signal.SIGTERM) + assert callable(handler) + try: + in_handler = True + handler(signal.SIGTERM, None) + in_handler = False + assert touched == [] + assert wait_for(lambda: worker.stopping, timeout=5) + + in_handler = True + with pytest.raises(_Exited): + handler(signal.SIGTERM, None) + in_handler = False + finally: + in_handler = False + retire() + assert touched == ["os._exit"] + assert exits == [130] + + +def test_the_helper_stops_the_worker_before_it_logs( + restore_signal_handlers, monkeypatch +): + """A log handler that blocks must not hold up the drain.""" + worker = Worker(poll_interval=0.05) + release = threading.Event() + + class BlockingLogger: + def __getattr__(self, attr): + return lambda *args, **kwargs: release.wait(10) + + monkeypatch.setattr(ox_worker, "logger", BlockingLogger()) + retire = ox_worker.install_stop_handlers(worker) + try: + signal.getsignal(signal.SIGTERM)(signal.SIGTERM, None) + stopped = wait_for(lambda: worker.stopping, timeout=2) + finally: + release.set() + retire() + assert stopped + + +def test_a_signal_before_the_helper_starts_is_kept( + restore_signal_handlers, monkeypatch +): + """ + The handlers are installed before the helper thread starts, and a + signal in between waits in the queue for it. + """ + worker = Worker(poll_interval=0.05) + real_start = threading.Thread.start + installed: list[bool] = [] + + def start(self): + if self.name == "ox-signal": + handler = signal.getsignal(signal.SIGTERM) + installed.append(callable(handler)) + if callable(handler): + handler(signal.SIGTERM, None) + real_start(self) + + with monkeypatch.context() as patch: + patch.setattr(threading.Thread, "start", start) + retire = ox_worker.install_stop_handlers(worker) + try: + assert installed == [True] + assert wait_for(lambda: worker.stopping, timeout=5) + finally: + retire() + + +def test_handlers_are_restored_when_the_helper_cannot_start( + restore_signal_handlers, monkeypatch +): + before = [signal.getsignal(s) for s in (signal.SIGTERM, signal.SIGINT)] + + def start(self): + raise RuntimeError("can't start new thread") + + with monkeypatch.context() as patch: + patch.setattr(threading.Thread, "start", start) + with pytest.raises(RuntimeError): + ox_worker.install_stop_handlers(Worker(poll_interval=0.05)) + assert [signal.getsignal(s) for s in (signal.SIGTERM, signal.SIGINT)] == before + + +def test_a_signal_queued_before_a_failed_helper_start_still_stops( + restore_signal_handlers, monkeypatch +): + """The queue that held it has no reader, so the start-up code acts on it.""" + worker = Worker(poll_interval=0.05) + + def start(self): + signal.getsignal(signal.SIGTERM)(signal.SIGTERM, None) + raise RuntimeError("can't start new thread") + + with monkeypatch.context() as patch: + patch.setattr(threading.Thread, "start", start) + with pytest.raises(RuntimeError): + ox_worker.install_stop_handlers(worker) + assert worker.stopping + + +@pytest.mark.django_db(transaction=True) +def test_a_second_signal_exits_130_without_waiting_for_the_task(tmp_path): + """ + The real command, mid-task: the first SIGTERM starts the drain, the + second ends the process at once with 130 and leaves the task where it + was. The row stays RUNNING for the reaper, as a crash would leave it. + """ + result = slow.enqueue(30) + log = tmp_path / "worker.log" + proc = start_worker(tmp_path, "--interval", "0.05") + try: + assert wait_for( + lambda: OxTask.objects.get(id=result.id).status == OxTask.Status.RUNNING, + timeout=30, + ), log.read_text() + proc.send_signal(signal.SIGTERM) + # Logged by the helper once the first signal has been acted on, so + # the next one cannot be merged into it. + assert wait_for( + lambda: "received SIGTERM; draining" in log.read_text(), timeout=10 + ), log.read_text() + proc.send_signal(signal.SIGTERM) + try: + code = proc.wait(timeout=10) + except subprocess.TimeoutExpired: + pytest.fail(f"the second signal did not end the worker:\n{log.read_text()}") + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + assert code == 130, log.read_text() + assert OxTask.objects.get(id=result.id).status == OxTask.Status.RUNNING + assert "stopped" not in log.read_text() + + +class TestSupervisorHandler: + def test_the_handler_neither_logs_nor_locks(self, caplog, monkeypatch): + """ + With the supervisor's lock held elsewhere and every log record + blocking, the handler still returns at once: it takes neither. The + loop's processing step then does what the handler used to. + """ + caplog.set_level(logging.INFO, logger="django_ox") + supervisor = Supervisor(processes=1, worker_args=[], kill_grace=5.0) + sent: list[int] = [] + monkeypatch.setattr(supervisor, "_signal_children", sent.append) + + release = threading.Event() + locked = threading.Event() + + class BlockingHandler(logging.Handler): + def emit(self, record): + release.wait() + + def hold_the_lock(): + with supervisor._lock: + locked.set() + release.wait() + + blocking = BlockingHandler() + logging.getLogger("django_ox").addHandler(blocking) + holder = threading.Thread(target=hold_the_lock, daemon=True) + holder.start() + caller = threading.Thread( + target=lambda: [ + supervisor.handle_signal(signal.SIGTERM, None) for _ in range(3) + ], + daemon=True, + ) + try: + assert locked.wait(5) + caller.start() + caller.join(timeout=2) + returned = not caller.is_alive() + finally: + release.set() + logging.getLogger("django_ox").removeHandler(blocking) + holder.join(timeout=5) + caller.join(timeout=5) + + assert returned, "handle_signal blocked on a log handler or the lock" + assert sent == [] + assert not supervisor.stopping + + supervisor._process_signals() + + assert supervisor.stopping + assert sent == [signal.SIGTERM] * 3 + messages = [r.getMessage() for r in caplog.records] + assert "Received SIGTERM; stopping 0 worker process(es)." in messages[0] + assert messages[1].startswith("Second signal received; forcing worker exit") + # The third signal brought the SIGKILL forward to now. + assert supervisor._kill_at is not None + assert supervisor._kill_at <= time.monotonic() + + @pytest.mark.django_db(transaction=True) + def test_escalation_still_works_after_the_loop_has_failed( + self, caplog, monkeypatch, tmp_path + ): + """ + The run loop's clean-up waits for every child. The second and third + signals are acted on only by a loop, so that wait has to be one: a + child that will not exit must still be SIGKILLed. + """ + caplog.set_level(logging.INFO, logger="django_ox") + in_process_env(monkeypatch, tmp_path) + supervisor = Supervisor( + processes=1, + worker_args=["--interval", "0.05", "--verbosity", "0"], + kill_grace=0.5, + ) + errors: list[BaseException] = [] + + def run(): + try: + supervisor.run() + except RuntimeError as exc: + errors.append(exc) + + thread = threading.Thread(target=run, daemon=True) + thread.start() + stuck = None + try: + assert wait_for_workers(tmp_path, 1) + assert slot_pid(supervisor, 0) is not None + # The Popen, not its pid: once the supervisor has reaped the child + # the number can belong to another process. + stuck = supervisor._children[0] + stuck.send_signal(signal.SIGSTOP) + + def boom(): + raise RuntimeError("the loop failed") + + monkeypatch.setattr(supervisor, "_start_due", boom) + assert wait_for(lambda: supervisor.stopping, timeout=5) + time.sleep(0.3) + assert thread.is_alive() + supervisor.handle_signal(signal.SIGTERM, None) + supervisor.handle_signal(signal.SIGTERM, None) + thread.join(timeout=10) + finished = not thread.is_alive() + finally: + if thread.is_alive(): + # An assertion above failed: escalate so the supervisor thread + # and its child do not outlive the test. + for _ in range(3): + supervisor.handle_signal(signal.SIGTERM, None) + if stuck is not None and stuck.poll() is None: + stuck.kill() + thread.join(timeout=10) + + assert finished, "the clean-up wait never acted on the second signal" + assert [str(e) for e in errors] == ["the loop failed"] + events = [getattr(r, "event", None) for r in caplog.records] + assert events.count("supervisor_killed_workers") == 1 + assert supervisor._exit_codes == {0: -signal.SIGKILL} + assert stuck.poll() == -signal.SIGKILL + + +class RecordingWorker(Worker): + """Records what was already true when run() was entered.""" + + seen: dict[str, object] = {} + + def run(self): + RecordingWorker.seen["parent_pid"] = self.parent_pid + RecordingWorker.seen["worker_id"] = self.worker_id + # What a task that starts a process of its own would pass on. + RecordingWorker.seen["env"] = os.environ.get(SUPERVISOR_PID_ENV) + + +def new_threads(before: set[threading.Thread], name: str) -> list[threading.Thread]: + return [ + t + for t in threading.enumerate() + if t not in before and t.name == name and t.is_alive() + ] + + +class TestExpectedParent: + @pytest.fixture + def recording(self, settings, monkeypatch, restore_signal_handlers): + settings.TASKS = { + "default": { + "BACKEND": "django_ox.backend.OxBackend", + "OPTIONS": {"WORKER_CLASS": "tests.test_stop_signals.RecordingWorker"}, + } + } + RecordingWorker.seen = {} + # Restored afterwards, whatever the command did with it. + monkeypatch.delenv(SUPERVISOR_PID_ENV, raising=False) + armed: dict[str, object] = {} + + def die_with_parent(): + # What is in place at the moment the parent-death signal is armed. + armed["handler"] = signal.getsignal(signal.SIGTERM) + armed["helper"] = any(t.name == "ox-signal" for t in threading.enumerate()) + + monkeypatch.setattr(ox_worker, "_die_with_parent", die_with_parent) + return armed + + def test_the_supervisor_pid_comes_from_the_environment( + self, recording, monkeypatch + ): + gone = os.getppid() + 100000 + monkeypatch.setenv(SUPERVISOR_PID_ENV, str(gone)) + with pytest.raises(SystemExit) as excinfo: + call_command("ox_worker", "--worker-index=3", verbosity=0) + assert excinfo.value.code == 0 + assert RecordingWorker.seen["parent_pid"] == gone + assert str(RecordingWorker.seen["worker_id"]).endswith("-3") + # Gone before the worker ran, so nothing it starts inherits it. + assert RecordingWorker.seen["env"] is None + assert SUPERVISOR_PID_ENV not in os.environ + # The handler and the thread that acts on it existed before arming. + assert callable(recording["handler"]) + assert recording["handler"] not in (signal.SIG_DFL, signal.SIG_IGN) + assert recording["helper"] is True + + def test_without_the_variable_the_parent_is_read_at_start_up(self, recording): + with pytest.raises(SystemExit): + call_command("ox_worker", "--worker-index=0", verbosity=0) + assert RecordingWorker.seen["parent_pid"] == os.getppid() + + def test_an_unreadable_variable_falls_back_to_the_parent( + self, recording, monkeypatch + ): + monkeypatch.setenv(SUPERVISOR_PID_ENV, "not-a-pid") + with pytest.raises(SystemExit): + call_command("ox_worker", "--worker-index=0", verbosity=0) + assert RecordingWorker.seen["parent_pid"] == os.getppid() + assert SUPERVISOR_PID_ENV not in os.environ + + def test_a_worker_that_is_not_a_child_ignores_the_variable( + self, recording, monkeypatch + ): + """A task's own ox_worker, say, run under a leftover variable.""" + monkeypatch.setenv(SUPERVISOR_PID_ENV, str(os.getppid() + 100000)) + with pytest.raises(SystemExit): + call_command("ox_worker", verbosity=0) + assert RecordingWorker.seen["parent_pid"] is None + assert RecordingWorker.seen["env"] is None + + def test_the_command_ends_its_helper_thread(self, recording): + before = set(threading.enumerate()) + with pytest.raises(SystemExit): + call_command("ox_worker", verbosity=0) + assert wait_for(lambda: not new_threads(before, "ox-signal"), timeout=2) + + def test_off_the_main_thread_nothing_is_left_behind(self, recording): + """ + signal.signal() refuses to run off the main thread. The command + fails there, as it always has, and leaves no helper thread behind. + """ + before = set(threading.enumerate()) + errors: list[BaseException] = [] + + def run(): + try: + call_command("ox_worker", verbosity=0) + except BaseException as exc: + errors.append(exc) + + caller = threading.Thread(target=run) + caller.start() + caller.join(timeout=10) + assert [type(e) for e in errors] == [ValueError] + assert RecordingWorker.seen == {} + assert new_threads(before, "ox-signal") == [] + + @pytest.mark.django_db(transaction=True) + def test_an_orphaned_child_claims_nothing(self): + result = add.enqueue(1, 2) + gone = os.getpid() + 100000 + env = child_env() + env[SUPERVISOR_PID_ENV] = str(gone) + done = subprocess.run( + [ + sys.executable, + "-m", + "django", + "ox_worker", + "--interval", + "0.05", + "--worker-index", + "0", + ], + cwd=REPO, + env=env, + capture_output=True, + text=True, + timeout=60, + check=False, + ) + output = done.stdout + done.stderr + assert done.returncode == 0, output + lost = f"lost its supervisor (pid {gone}); draining" + assert output.count(lost) == 1, output + # Checked once, by the run loop, after the worker says it is starting. + assert output.index("starting: queues") < output.index(lost), output + assert "stopped" in output, output + row = OxTask.objects.get(id=result.id) + assert row.status == OxTask.Status.READY + assert row.attempts == 0 diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index d165e62..71c4319 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -14,14 +14,21 @@ import sys import threading import time +import types from pathlib import Path import pytest from django.conf import settings from django.db import connection +from django_ox import supervisor as supervisor_module from django_ox.models import OxTask -from django_ox.supervisor import Supervisor, child_command +from django_ox.supervisor import ( + SUPERVISOR_PID_ENV, + Supervisor, + _child_env, + child_command, +) from django_ox.worker import Worker from .conftest import start_worker_thread, wait_for @@ -581,6 +588,17 @@ def test_child_command_reinvokes_a_single_worker(): ] +def test_each_child_is_told_the_supervisors_pid_in_its_environment(): + """ + The child compares its parent with this pid rather than with a getppid() + of its own, which names the adopter if the supervisor died first. It is + not a flag: a child running an older release would refuse the flag and + never start. + """ + assert _child_env()[SUPERVISOR_PID_ENV] == str(os.getpid()) + assert SUPERVISOR_PID_ENV == "OX_SUPERVISOR_PID" + + def test_child_command_reuses_the_script_by_absolute_path(tmp_path, monkeypatch): script = tmp_path / "manage.py" script.write_text("") @@ -700,9 +718,111 @@ def test_second_signal_kills_a_stuck_child_after_the_grace( assert not alive(stuck) +class FakeChild: + """A child process that runs until it is signalled, then drains.""" + + def __init__(self) -> None: + self.returncode: int | None = None + self.signals: list[int] = [] + self.pid = 0 + + def poll(self) -> int | None: + return self.returncode + + def send_signal(self, signum: int) -> None: + self.signals.append(signum) + if self.returncode is None: + self.returncode = 0 + + +def fake_children(monkeypatch, on_start=None) -> list[FakeChild]: + """Make the supervisor start FakeChild objects instead of processes.""" + started: list[FakeChild] = [] + + def popen(argv, **kwargs): + child = FakeChild() + started.append(child) + if on_start is not None: + on_start(child, len(started)) + return child + + monkeypatch.setattr(supervisor_module.subprocess, "Popen", popen) + return started + + +class TestStopWhileStarting: + def test_a_signal_before_run_starts_nothing(self, monkeypatch): + started = fake_children(monkeypatch) + supervisor = Supervisor(processes=2, worker_args=[]) + supervisor.handle_signal(signal.SIGTERM, None) + assert supervisor.run() == 0 + assert started == [] + + def test_a_signal_between_starts_ends_the_starting(self, monkeypatch): + """ + The handler runs while the first child is being started; the start + loop acts on it before the next start, as it did when the handler + set the stop itself. + """ + supervisor = Supervisor(processes=8, worker_args=[]) + + def on_start(child, count): + if count == 1: + supervisor.handle_signal(signal.SIGTERM, None) + + started = fake_children(monkeypatch, on_start) + assert supervisor.run() == 0 + assert len(started) == 1 + assert started[0].signals == [signal.SIGTERM] + + class TestStopRestartRace: + def test_a_recorded_signal_is_acted_on_before_a_restart(self, monkeypatch): + """ + A slot has died and its restart falls due while the loop sleeps, and + a signal is recorded in the same sleep, where the handler most often + runs. The next pass acts on the signal before it restarts anything, + so nothing starts. + """ + + def on_start(child, count): + if count == 1: + child.returncode = -signal.SIGKILL + + started = fake_children(monkeypatch, on_start) + supervisor = Supervisor(processes=1, worker_args=[], restart_delay=3600) + real_sleep = time.sleep + recorded = False + + def sleep(seconds): + nonlocal recorded + if supervisor._restart_due and not recorded: + recorded = True + for index in list(supervisor._restart_due): + supervisor._restart_due[index] = 0.0 + supervisor.handle_signal(signal.SIGTERM, None) + real_sleep(seconds) + + monkeypatch.setattr( + supervisor_module, + "time", + types.SimpleNamespace(monotonic=time.monotonic, sleep=sleep), + ) + thread, result = run_in_thread(supervisor) + thread.join(timeout=10) + if thread.is_alive(): + supervisor.request_stop() + supervisor._signal_children(signal.SIGTERM) + thread.join(timeout=10) + assert recorded + assert len(started) == 1 + assert result == [128 + signal.SIGKILL] + def test_a_stop_just_before_the_restart_starts_nothing(self, monkeypatch): - """A stop that lands between the snapshot and Popen must start nothing.""" + """ + A stop requested between the restart decision and Popen, as another + thread embedding the supervisor could, must start nothing. + """ in_process_env(monkeypatch) supervisor = Supervisor( processes=1, @@ -716,7 +836,7 @@ def patched(index): nonlocal starts starts += 1 if starts == 2: - supervisor.handle_signal(signal.SIGTERM, None) + supervisor.request_stop() original(index) monkeypatch.setattr(supervisor, "_start", patched) @@ -741,6 +861,7 @@ def test_a_second_signal_forwards_again(monkeypatch): supervisor.handle_signal(signal.SIGINT, None) supervisor.handle_signal(signal.SIGINT, None) + supervisor._process_signals() assert supervisor.stopping assert sent == [signal.SIGTERM, signal.SIGTERM]