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]