diff --git a/docs/guides/stopping-commands.md b/docs/guides/stopping-commands.md index be7b485..442b0be 100644 --- a/docs/guides/stopping-commands.md +++ b/docs/guides/stopping-commands.md @@ -34,7 +34,7 @@ hidden once a graceful stop has been requested. | Backend | Graceful stop | Force stop | Mechanism | |---|---|---|---| | `celery` | ✅ | ✅ | stop flag + `revoke(terminate=True, SIGTERM/SIGKILL)` | -| `django-q2` | ✅ | ✅ | stop flag + PID-targeted SIGTERM/SIGKILL (PID reuse guarded via `/proc`) | +| `django-q2` | ✅ | ✅ | stop flag + PID-targeted SIGTERM/SIGKILL (only when the worker ran on the same host; PID reuse guarded via `/proc`) | | `django` (django.tasks) | ✅ | ❌ | stop flag only | | `sync` | ✅ (limited) | ❌ | runs in the request thread — stop flag only | @@ -43,8 +43,14 @@ Notes: - A command blocked in a long C call or network I/O without producing output only responds to **Force Stop** — and on `sync` / `django` it cannot be interrupted at all. -- django-q2 tasks are enqueued with retries disabled, and `execute_command` - refuses duplicate attempts: a force-killed task is never re-run. +- When the worker ran on a *different host* (e.g. another Kubernetes + pod — a PID namespace the admin process cannot see), the PID-targeted + signal is skipped for safety and the cooperative stop flag is the + only mechanism; a Force Stop finalizes the row while the worker (if + it is still alive there) runs on until it next checks the flag. +- django-q2 tasks are enqueued with `ack_failure` so saved failures are + not redelivered, and `execute_command` refuses duplicate attempts: + a force-killed task is never re-run. ## Permissions @@ -65,14 +71,22 @@ forever. Two mechanisms handle this: the limit counts as **Cancelled** instead. - **Stale-run sweeper** (lazy, no background infra): when an admin page or output poll is served, executions that are still Running while - their recorded worker process no longer exists are finalized — - **Timed out** when the backend's records attribute the kill to a - timeout (django-q2 Task record, Celery result state), otherwise - **Failed** with a "worker lost" note. Legacy rows without a recorded - PID are swept after `ADMIN_RUNNER_STALE_AFTER` seconds (default 1 h). - The sweep runs at most once per `ADMIN_RUNNER_SWEEP_INTERVAL` seconds - (default 30) per web process and never races a finishing worker - (conditional updates). + their worker is gone are finalized — **Timed out** when the backend's + records attribute the kill to a timeout (django-q2 Task record, Celery + result state), otherwise **Failed** with a "worker lost" note. + + Liveness is heartbeat-based: workers refresh `last_heartbeat_at` + every `ADMIN_RUNNER_HEARTBEAT_INTERVAL` seconds while running, and a + RUNNING execution whose heartbeat is older than + `ADMIN_RUNNER_HEARTBEAT_STALE_AFTER` is swept. The heartbeat flows + through the database, so it stays meaningful across container/pod + boundaries, where the worker's OS PID is invisible to the web + processes running the sweep. Rows recorded before heartbeats existed + fall back to the legacy PID rules (process gone or PID reused past + the flush grace; PID-less rows after `ADMIN_RUNNER_STALE_AFTER` + seconds, default 1 h). The sweep runs at most once per + `ADMIN_RUNNER_SWEEP_INTERVAL` seconds (default 30) per web process + and never races a finishing worker (conditional updates). Keep backend timeouts well above your longest command, and prefer soft limits where available — they produce the cleanest result. @@ -82,7 +96,9 @@ Settings: | Setting | Default | Meaning | |---|---|---| | `ADMIN_RUNNER_SWEEP_INTERVAL` | `30` | Min. seconds between sweeps per web process | -| `ADMIN_RUNNER_STALE_AFTER` | `3600` | Age before PID-less Running rows are swept | +| `ADMIN_RUNNER_STALE_AFTER` | `3600` | Age before PID-less Running rows (no heartbeat) are swept | +| `ADMIN_RUNNER_HEARTBEAT_INTERVAL` | `30` | Seconds between worker liveness heartbeat updates | +| `ADMIN_RUNNER_HEARTBEAT_STALE_AFTER` | `180` | Age after which a stale heartbeat means the worker is lost | ## Command contract diff --git a/src/django_admin_runner/migrations/0012_commandexecution_worker_host_worker_started_at_and_more.py b/src/django_admin_runner/migrations/0012_commandexecution_worker_host_worker_started_at_and_more.py new file mode 100644 index 0000000..7b37056 --- /dev/null +++ b/src/django_admin_runner/migrations/0012_commandexecution_worker_host_worker_started_at_and_more.py @@ -0,0 +1,52 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ( + "django_admin_runner", + "0011_scheduledcommand_django_admin_runner_scheduledcommand_source_valid", + ), + ] + + operations = [ + migrations.AddField( + model_name="commandexecution", + name="worker_host", + field=models.CharField( + blank=True, + default="", + help_text=( + "Host (machine or container) running the command, recorded at start. " + "PIDs are only meaningful inside their own PID namespace, so stop " + "signals are only sent when this matches the local host." + ), + max_length=255, + ), + ), + migrations.AddField( + model_name="commandexecution", + name="worker_started_at", + field=models.DateTimeField( + blank=True, + help_text=( + "Start time of the worker OS process, recorded at start. Compared " + "against the process currently holding the PID to detect reuse." + ), + null=True, + ), + ), + migrations.AddField( + model_name="commandexecution", + name="last_heartbeat_at", + field=models.DateTimeField( + blank=True, + help_text=( + "Liveness heartbeat refreshed periodically by the worker while " + "running. The stale-run sweeper finalizes RUNNING executions whose " + "heartbeat went stale." + ), + null=True, + ), + ), + ] diff --git a/src/django_admin_runner/models.py b/src/django_admin_runner/models.py index 2479394..5bda4a1 100644 --- a/src/django_admin_runner/models.py +++ b/src/django_admin_runner/models.py @@ -54,6 +54,26 @@ class Status(models.TextChoices): blank=True, help_text="OS PID of the process running the command, recorded at start.", ) + worker_host = models.CharField( + max_length=255, + blank=True, + default="", # type: ignore[assignment] + help_text="Host (machine or container) running the command, recorded at start. " + "PIDs are only meaningful inside their own PID namespace, so stop signals " + "are only sent when this matches the local host.", + ) + worker_started_at = models.DateTimeField( + null=True, + blank=True, + help_text="Start time of the worker OS process, recorded at start. " + "Compared against the process currently holding the PID to detect reuse.", + ) + last_heartbeat_at = models.DateTimeField( + null=True, + blank=True, + help_text="Liveness heartbeat refreshed periodically by the worker while running. " + "The stale-run sweeper finalizes RUNNING executions whose heartbeat went stale.", + ) schedule = models.ForeignKey( "ScheduledCommand", null=True, diff --git a/src/django_admin_runner/runners/django_q2.py b/src/django_admin_runner/runners/django_q2.py index 4a39e04..e853697 100644 --- a/src/django_admin_runner/runners/django_q2.py +++ b/src/django_admin_runner/runners/django_q2.py @@ -27,21 +27,34 @@ class DjangoQ2CommandRunner(BaseCommandRunner): def _signal_worker(self, execution, sig: int) -> bool: """Send *sig* to the worker process running *execution*. - Guards against PID reuse: the signal is only sent when the target - process started at or after the execution's ``started_at`` (its - ``/proc//stat`` starttime, converted via the boot time from - ``/proc/stat``). Best effort — on any read failure the check is - skipped and the signal is sent if the PID exists. + A pid is only meaningful inside its own PID namespace: when the + execution was recorded on another host (e.g. another pod in + Kubernetes), the signal is *not* sent — the cooperative stop flag + stays the only cross-host stop mechanism. Within the local + namespace, PID reuse is guarded by comparing the target process's + start time with the recorded worker start time (falling back to + the execution start for rows recorded before it existed). + + Best effort — on any read failure the check is skipped and the + signal is sent if the PID exists locally. """ import os + import socket + + from django_admin_runner.tasks import _proc_starttime pid = execution.worker_pid if not pid: return False - if execution.started_at is not None: - started = self._proc_starttime(pid) - if started is not None and started < execution.started_at: + host = getattr(execution, "worker_host", "") or "" + if host and host != socket.gethostname(): + return False # another container/pod: the pid is meaningless here + + reference = execution.worker_started_at or execution.started_at + if reference is not None: + started = _proc_starttime(pid) + if started is not None and started < reference: return False # PID predates the execution → reused try: @@ -50,27 +63,13 @@ def _signal_worker(self, execution, sig: int) -> bool: return False return True - @staticmethod - def _proc_starttime(pid: int): - """Start time of *pid* as a timezone-aware datetime, or ``None``.""" - try: - with open(f"/proc/{pid}/stat", "rb") as fh: - # Field 22 (starttime, 1-based) — after the comm field, - # which may contain spaces, so split after its ')'. - fields = fh.read().rsplit(b")", 1)[1].split() - starttime_ticks = int(fields[19]) # (22) minus the two consumed - with open("/proc/stat", "rb") as fh: - btime = next(int(line.split()[1]) for line in fh if line.startswith(b"btime")) - except (OSError, ValueError, IndexError, StopIteration): - return None - from django.utils import timezone - - return timezone.datetime.fromtimestamp( - btime + starttime_ticks // 100, tz=timezone.get_current_timezone() - ) - def stop(self, execution) -> None: - """Graceful stop: stop flag (heartbeat) + SIGTERM to the worker PID.""" + """Graceful stop: stop flag (heartbeat) + SIGTERM to the worker PID. + + The signal only fires when the worker ran on this host (see + ``_signal_worker``); the cooperative stop flag set by + ``super().stop()`` is the cross-host mechanism. + """ super().stop(execution) import signal @@ -96,7 +95,9 @@ def finalize_stale(self, execution): try: from django_q.models import Task - task = Task.objects.filter(name=execution.task_id).first() + # ``task_id`` stores the django-q2 task *id* (a UUID), not the + # human-readable ``Task.name`` (a humanhash) — match on ``pk``. + task = Task.objects.filter(pk=execution.task_id).first() if execution.task_id else None except Exception: # noqa: BLE001 - attribution must never break the sweep return None if task is None or task.success or not task.stopped: @@ -140,11 +141,15 @@ def run(self, command_name, kwargs, triggered_by, execution) -> RunResult: q_options = {} if timeout is not None: q_options["timeout"] = timeout - # Never retry admin-runner tasks: a force-killed worker leaves the - # task unacknowledged, and a retry would run the command twice - # (the PENDING→RUNNING idempotency guard in execute_command also - # backs this up). -1 = disable retries in django-q2. - q_options["retry"] = -1 + # Ack failed tasks immediately so a failed/killed task is not + # redelivered after the broker's retry window. A *hard-killed* + # worker cannot ack anything — such tasks redeliver after + # Q_CLUSTER["retry"] — but the PENDING→RUNNING idempotency guard + # in execute_command turns any redelivery into a no-op. Per-task + # ``retry`` is deliberately not set: django-q2 drops unknown + # q_options keys silently, and the ORM broker locks deliveries + # with the global retry anyway (retry must exceed timeout). + q_options["ack_failure"] = True try: from django_q.tasks import async_task @@ -155,6 +160,9 @@ def run(self, command_name, kwargs, triggered_by, execution) -> RunResult: command_name, kwargs, execution.pk, + # Readable audit rows in the backend's task list (the task + # *name* is a humanhash; the id stays in execution.task_id). + task_name=f"admin-runner:{execution.pk}", **task_kwargs, ) except Exception as exc: diff --git a/src/django_admin_runner/tasks.py b/src/django_admin_runner/tasks.py index b67fd4d..952b2c7 100644 --- a/src/django_admin_runner/tasks.py +++ b/src/django_admin_runner/tasks.py @@ -5,6 +5,7 @@ import threading import time import traceback +from datetime import timedelta from django.conf import settings from django.core.management import call_command @@ -32,6 +33,52 @@ def _terminal_size() -> tuple[int, int]: return max(20, min(cols, 500)), max(5, min(rows, 200)) +def _heartbeat_interval() -> float: + """Seconds between worker liveness heartbeats. + + ``ADMIN_RUNNER_HEARTBEAT_INTERVAL`` (default 30 s). While a command + runs, a worker thread refreshes ``last_heartbeat_at`` on this cadence; + the stale-run sweeper treats a RUNNING execution whose heartbeat is + older than ``ADMIN_RUNNER_HEARTBEAT_STALE_AFTER`` as lost. + """ + return float(getattr(settings, "ADMIN_RUNNER_HEARTBEAT_INTERVAL", 30.0)) + + +def _heartbeat_stale_after() -> int: + """Seconds after which a stale heartbeat means the worker is gone. + + ``ADMIN_RUNNER_HEARTBEAT_STALE_AFTER`` (default 180 s — six missed + heartbeats at the default interval). Keep it comfortably above the + heartbeat interval so a single slow or failed update cannot falsely + finalize a live worker. + """ + return int(getattr(settings, "ADMIN_RUNNER_HEARTBEAT_STALE_AFTER", 180)) + + +def _proc_starttime(pid: int): + """Start time of *pid* as a timezone-aware datetime, or ``None``. + + Reads ``/proc//stat`` (Linux only; returns ``None`` elsewhere). + Used to detect PID reuse: a process occupying a recorded PID whose + start time predates the execution cannot be the original worker. + """ + try: + with open(f"/proc/{pid}/stat", "rb") as fh: + # Field 22 (starttime, 1-based) — after the comm field, which + # may contain spaces, so split after the final ')'. + fields = fh.read().rsplit(b")", 1)[1].split() + starttime_ticks = int(fields[19]) # (22) minus the two consumed + with open("/proc/stat", "rb") as fh: + btime = next(int(line.split()[1]) for line in fh if line.startswith(b"btime")) + except (OSError, ValueError, IndexError, StopIteration): + return None + from django.utils import timezone + + return timezone.datetime.fromtimestamp( + btime + starttime_ticks // 100, tz=timezone.get_current_timezone() + ) + + class WorkerStoppedError(Exception): """Raised inside a command when the worker receives SIGTERM.""" @@ -71,28 +118,75 @@ def _is_timeout_exception(exc: BaseException) -> bool: return isinstance(exc, SoftTimeLimitExceeded) +def _worker_is_dead( + execution, + sweep_from, + heartbeat_stale: timedelta, + grace: timedelta, + stale_after: timedelta, +) -> bool: + """Whether *execution*'s worker is (very likely) gone. + + The heartbeat verdict is authoritative and host-independent: it works + across containers because the worker writes its liveness through the + shared database, never through the OS. Rows recorded before + heartbeats existed (no ``last_heartbeat_at``) fall back to the legacy + pid rules — those are only meaningful inside the sweeping process's + own PID namespace and are deliberately conservative: + + - inside the grace period → alive; + - pid recorded and the process no longer exists (or is a *reused* + pid whose process predates the execution) → dead; + - no pid recorded and older than ``ADMIN_RUNNER_STALE_AFTER`` → dead. + """ + import errno + import os + + if execution.last_heartbeat_at is not None: + return execution.last_heartbeat_at <= sweep_from - heartbeat_stale + + # --- legacy rows (recorded by pre-heartbeat workers) --- + if execution.started_at is None or execution.started_at > sweep_from - grace: + return False + pid = execution.worker_pid + if not pid: + # No pid recorded: age is the only available signal. + return execution.started_at <= sweep_from - stale_after + try: + os.kill(pid, 0) + except OSError as exc: + return exc.errno == errno.ESRCH # e.g. EPERM: treat as alive + except (ValueError, OverflowError, TypeError): + return True # bogus pid: treat as dead + # Something is living at this pid — but on a restarted host/pod that + # may be an unrelated, *reused* pid. Only trust it when its start + # time matches the execution (unknown start time → assume alive). + started = _proc_starttime(pid) + return bool( + started is not None and execution.started_at is not None and started < execution.started_at + ) + + def sweep_stale_executions() -> int: """Finalize RUNNING executions whose worker died without finalizing. A hard kill (django-q2 ``timeout``, Celery ``time_limit``, crash, OOM, dead cluster) leaves the row RUNNING: the only writer — the worker — is gone. This sweeper runs lazily from admin request handlers and - finalizes such rows: - - - ``worker_pid`` recorded and the process no longer exists (grace - period past), or - - legacy rows without ``worker_pid`` older than - ``ADMIN_RUNNER_STALE_AFTER`` seconds (default 1 h). + finalizes such rows. Liveness is heartbeat-based (see + :func:`_worker_is_dead`): workers refresh ``last_heartbeat_at`` every + ``ADMIN_RUNNER_HEARTBEAT_INTERVAL`` seconds, so a RUNNING execution + whose heartbeat is older than ``ADMIN_RUNNER_HEARTBEAT_STALE_AFTER`` + is finalized — a check that works across containers, where PIDs of + another pod are invisible. Rows without a heartbeat (recorded by + older versions) keep the legacy pid rules. The runner's ``finalize_stale()`` may attribute the cause (backend records → TIMEOUT); otherwise the row becomes FAILED with a "worker lost" note. All updates are conditional on ``status=RUNNING`` so a concurrently finishing worker always wins. Never raises. """ - import errno import logging - import os - from datetime import timedelta from django.utils.timezone import now @@ -101,6 +195,7 @@ def sweep_stale_executions() -> int: logger = logging.getLogger(__name__) grace = timedelta(seconds=max(10, int(2 * _default_flush_interval()))) + heartbeat_stale = timedelta(seconds=_heartbeat_stale_after()) stale_after = timedelta( seconds=int(getattr(settings, "ADMIN_RUNNER_STALE_AFTER", 3600)), ) @@ -109,26 +204,16 @@ def sweep_stale_executions() -> int: runner = None swept = 0 try: - candidates = CommandExecution.objects.filter( - status=CommandExecution.Status.RUNNING, - started_at__lt=sweep_from - grace, - ) + candidates = CommandExecution.objects.filter(status=CommandExecution.Status.RUNNING) for execution in candidates: - if execution.worker_pid: - try: - os.kill(execution.worker_pid, 0) - except OSError as exc: - if exc.errno != errno.ESRCH: - continue # e.g. EPERM: treat as alive - except (ValueError, OverflowError, TypeError): - pass # bogus pid: treat as dead - else: - continue # alive: never sweep, regardless of age - elif execution.started_at and execution.started_at > sweep_from - stale_after: - continue # legacy row (no pid): not old enough yet + if not _worker_is_dead(execution, sweep_from, heartbeat_stale, grace, stale_after): + continue status = CommandExecution.Status.FAILED - note = "Worker process no longer exists (worker lost)." + if execution.last_heartbeat_at is not None: + note = "Worker heartbeat went stale (worker lost)." + else: + note = "Worker process no longer exists (worker lost)." try: if runner is None: runner = get_runner() @@ -210,6 +295,41 @@ def _restore_sigterm_handler(installed) -> None: pass +def _run_heartbeat(execution_pk: int, interval: float, stop_event: threading.Event) -> None: + """Refresh ``last_heartbeat_at`` every *interval* until *stop_event*. + + Runs on a daemon thread started by :func:`execute_command`. The + heartbeat — not the OS pid — is the liveness signal the stale-run + sweeper trusts: it flows through the shared database, so it stays + meaningful across container/pod boundaries where PIDs are invisible. + + A failed update (e.g. a brief database outage) is logged and retried + on the next tick — the heartbeat must never kill the command it + keeps alive. + """ + import logging + + from django.db import connection + + from .models import CommandExecution + + logger = logging.getLogger(__name__) + try: + while not stop_event.wait(interval): + try: + CommandExecution.objects.filter( + pk=execution_pk, status=CommandExecution.Status.RUNNING + ).update(last_heartbeat_at=now()) + except Exception: # noqa: BLE001 + logger.debug( + "Heartbeat update failed for execution %s", execution_pk, exc_info=True + ) + finally: + # Long-lived helper thread: release its DB connection so a pooled + # backend (or SQLite's file lock) is not held open for nothing. + connection.close() + + def _append_output(execution, field: str, chunk: str) -> None: """Append *chunk* to the active output part, sealing at ``PART_SIZE``. @@ -686,18 +806,36 @@ def execute_command(command_name: str, kwargs: dict, execution_pk: int) -> None: # Idempotency guard: claim the execution by transitioning PENDING → # RUNNING with this process's PID. A duplicate attempt (e.g. a backend # retry after a force-killed worker re-delivers the task) finds the row - # in a non-PENDING state and exits without touching it. + # in a non-PENDING state and exits without touching it. Host, process + # start time and the first heartbeat are recorded here: the host makes + # signals namespace-safe, the start time guards against PID reuse, and + # the heartbeat is the liveness signal for the stale-run sweeper. + import socket + updated = CommandExecution.objects.filter( pk=execution_pk, status=CommandExecution.Status.PENDING ).update( status=CommandExecution.Status.RUNNING, started_at=now(), worker_pid=os.getpid(), + worker_host=socket.gethostname(), + worker_started_at=_proc_starttime(os.getpid()), + last_heartbeat_at=now(), ) if not updated: return execution = CommandExecution.objects.get(pk=execution_pk) + # Liveness heartbeat (daemon thread; stopped in the finally below). + heartbeat_stop = threading.Event() + heartbeat_thread = threading.Thread( + target=_run_heartbeat, + args=(execution.pk, _heartbeat_interval(), heartbeat_stop), + name=f"dar-heartbeat-{execution.pk}", + daemon=True, + ) + heartbeat_thread.start() + # Fail gracefully when the worker is asked to stop (qcluster shutdown # sends SIGTERM to its workers). Without this, a stopped cluster leaves # the execution (and any app-level run guards) stuck in a running state @@ -754,6 +892,12 @@ def execute_command(command_name: str, kwargs: dict, execution_pk: int) -> None: import sys old_stdout, old_stderr = sys.stdout, sys.stderr + old_stdin = sys.stdin + # Commands run from the admin must never block on interactive + # prompts (input(), click.prompt, …): give them an EOF, non-tty + # stdin so prompts fail fast instead of hanging the worker + # forever with nobody there to answer. + sys.stdin = io.StringIO() sys.stdout, sys.stderr = stdout_buf, stderr_buf # Export the configured terminal size so commands (rich et al.) # lay out progress bars deterministically; restore afterwards. @@ -778,6 +922,7 @@ def execute_command(command_name: str, kwargs: dict, execution_pk: int) -> None: os.environ.pop(key, None) else: os.environ[key] = old_value + sys.stdin = old_stdin sys.stdout, sys.stderr = old_stdout, old_stderr execution.status = CommandExecution.Status.SUCCESS except CommandCancelledError: @@ -881,4 +1026,6 @@ def execute_command(command_name: str, kwargs: dict, execution_pk: int) -> None: if command_exc is not None: raise command_exc finally: + heartbeat_stop.set() + heartbeat_thread.join(timeout=2) _clear_execution_context() diff --git a/tests/test_cancellation.py b/tests/test_cancellation.py index 3832c7e..0f57dac 100644 --- a/tests/test_cancellation.py +++ b/tests/test_cancellation.py @@ -193,7 +193,11 @@ def fake_async_task(func, *args, **kwargs): ex = _make_execution(superuser) DjangoQ2CommandRunner().run("simple_command", {}, superuser, ex) - assert captured["q_options"]["retry"] == -1 + # Failed tasks are acked immediately (no broker redelivery); the + # per-task "retry" key is not set at all (django-q2 drops unknown + # q_options keys silently). + assert captured["q_options"]["ack_failure"] is True + assert "retry" not in captured["q_options"] def test_q2_pid_reuse_guard(self, superuser, monkeypatch): from django.utils.timezone import now as tz_now @@ -211,7 +215,8 @@ def test_q2_pid_reuse_guard(self, superuser, monkeypatch): from datetime import timedelta monkeypatch.setattr( - runner, "_proc_starttime", staticmethod(lambda pid: tz_now() - timedelta(hours=1)) + "django_admin_runner.tasks._proc_starttime", + lambda pid: tz_now() - timedelta(hours=1), ) killed = [] monkeypatch.setattr("os.kill", lambda pid, sig: killed.append((pid, sig))) diff --git a/tests/test_heartbeat_liveness.py b/tests/test_heartbeat_liveness.py new file mode 100644 index 0000000..9d3b78c --- /dev/null +++ b/tests/test_heartbeat_liveness.py @@ -0,0 +1,229 @@ +"""Heartbeat-based worker liveness. + +Covers the stale-run sweeper's heartbeat verdict (authoritative across +hosts/containers), the legacy pid rules for pre-heartbeat rows (including +the PID-reuse guard), the heartbeat recorded by ``execute_command``, the +worker-side heartbeat thread, and the stdin-EOF guard against hanging +interactive prompts. +""" + +import errno +import socket +import time +from datetime import timedelta + +import pytest +from django.test import override_settings +from django.utils.timezone import now + +import django_admin_runner.tasks as tasks_mod +from django_admin_runner.models import CommandExecution +from django_admin_runner.tasks import execute_command, sweep_stale_executions + + +def _make_execution(superuser, **kwargs) -> CommandExecution: + defaults = {"command_name": "simple_command", "triggered_by": superuser} + defaults.update(kwargs) + return CommandExecution.objects.create(**defaults) + + +def _dead_pid(monkeypatch) -> int: + """Make os.kill(pid, 0) report ESRCH (process gone) for any pid.""" + + def fake_kill(pid, sig): + if sig == 0: + raise OSError(errno.ESRCH, "No such process") + return None + + monkeypatch.setattr("os.kill", fake_kill) + return 12345 + + +# --------------------------------------------------------------------------- +# Sweeper — heartbeat verdict +# --------------------------------------------------------------------------- + + +@pytest.mark.django_db +class TestHeartbeatSweep: + def test_fresh_heartbeat_prevents_sweep(self, superuser, monkeypatch): + """A live heartbeat wins even when the pid is invisible (other pod).""" + _dead_pid(monkeypatch) + ex = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=12345, + worker_host="some-other-pod", + worker_started_at=now() - timedelta(minutes=10), + started_at=now() - timedelta(minutes=10), + last_heartbeat_at=now() - timedelta(seconds=10), + ) + assert sweep_stale_executions() == 0 + ex.refresh_from_db() + assert ex.status == CommandExecution.Status.RUNNING + + def test_stale_heartbeat_swept_with_note(self, superuser, monkeypatch): + _dead_pid(monkeypatch) + ex = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=12345, + worker_host="some-other-pod", + worker_started_at=now() - timedelta(minutes=10), + started_at=now() - timedelta(minutes=10), + last_heartbeat_at=now() - timedelta(seconds=600), + ) + swept = sweep_stale_executions() + ex.refresh_from_db() + assert swept == 1 + assert ex.status == CommandExecution.Status.FAILED + assert ex.finished_at is not None + assert "worker lost" in ex.output_text("stderr") + assert "heartbeat" in ex.output_text("stderr") + + def test_heartbeat_staleness_threshold_configurable(self, superuser, monkeypatch): + _dead_pid(monkeypatch) + ex = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=12345, + worker_host="some-other-pod", + last_heartbeat_at=now() - timedelta(seconds=60), + ) + with override_settings(ADMIN_RUNNER_HEARTBEAT_STALE_AFTER=30): + assert sweep_stale_executions() == 1 + ex.refresh_from_db() + assert ex.status == CommandExecution.Status.FAILED + + +# --------------------------------------------------------------------------- +# Sweeper — legacy pid rules (rows without heartbeat) +# --------------------------------------------------------------------------- + + +@pytest.mark.django_db +class TestLegacyPidSweep: + def test_reused_pid_is_dead(self, superuser, monkeypatch): + """PID occupied by an older process → original worker is gone. + + Regression test for rows hanging RUNNING forever after a pod + restart recycled the pid (the pre-heartbeat sweep treated any + living pid as a live worker). + """ + monkeypatch.setattr("os.kill", lambda pid, sig: None) # pid alive + ex = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=17, + started_at=now() - timedelta(minutes=10), + ) + started = ex.started_at + # The process holding the pid started an hour before the execution. + monkeypatch.setattr(tasks_mod, "_proc_starttime", lambda pid: started - timedelta(hours=1)) + assert sweep_stale_executions() == 1 + ex.refresh_from_db() + assert ex.status == CommandExecution.Status.FAILED + + def test_matching_pid_starttime_is_alive(self, superuser, monkeypatch): + monkeypatch.setattr("os.kill", lambda pid, sig: None) # pid alive + ex = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=17, + started_at=now() - timedelta(minutes=10), + ) + started = ex.started_at + monkeypatch.setattr(tasks_mod, "_proc_starttime", lambda pid: started) + assert sweep_stale_executions() == 0 + ex.refresh_from_db() + assert ex.status == CommandExecution.Status.RUNNING + + def test_unknown_pid_starttime_is_alive(self, superuser, monkeypatch): + """Cannot read the process start time → assume alive (conservative).""" + monkeypatch.setattr("os.kill", lambda pid, sig: None) # pid alive + monkeypatch.setattr(tasks_mod, "_proc_starttime", lambda pid: None) + ex = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=17, + started_at=now() - timedelta(minutes=10), + ) + assert sweep_stale_executions() == 0 + ex.refresh_from_db() + assert ex.status == CommandExecution.Status.RUNNING + + def test_pidless_row_uses_stale_after(self, superuser, monkeypatch): + monkeypatch.setattr("os.kill", lambda pid, sig: None) + young = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=None, + started_at=now() - timedelta(minutes=10), + ) + assert sweep_stale_executions() == 0 + old = _make_execution( + superuser, + status=CommandExecution.Status.RUNNING, + worker_pid=None, + started_at=now() - timedelta(hours=3), + ) + assert sweep_stale_executions() == 1 + old.refresh_from_db() + young.refresh_from_db() + assert old.status == CommandExecution.Status.FAILED + assert young.status == CommandExecution.Status.RUNNING + + +# --------------------------------------------------------------------------- +# execute_command — recorded liveness data + heartbeat thread +# --------------------------------------------------------------------------- + + +@pytest.mark.django_db(transaction=True) +class TestExecuteCommandLiveness: + def test_claim_records_host_and_heartbeat(self, superuser, monkeypatch): + ex = _make_execution(superuser) + execute_command("simple_command", {}, ex.pk) + ex.refresh_from_db() + assert ex.status == CommandExecution.Status.SUCCESS + assert ex.worker_host == socket.gethostname() + assert ex.last_heartbeat_at is not None + + def test_heartbeat_thread_refreshes_while_running(self, superuser, monkeypatch): + ex = _make_execution(superuser) + + def slow_command(*args, **kwargs): + time.sleep(0.4) + + monkeypatch.setattr(tasks_mod, "call_command", slow_command) + with override_settings(ADMIN_RUNNER_HEARTBEAT_INTERVAL=0.05): + execute_command("simple_command", {}, ex.pk) + ex.refresh_from_db() + # The worker thread must have refreshed the heartbeat several + # times while the command was running (transaction=True so the + # side thread's writes are visible to this connection). + assert ex.last_heartbeat_at is not None + assert ex.last_heartbeat_at > ex.started_at + timedelta(seconds=0.1) + + +# --------------------------------------------------------------------------- +# stdin EOF guard (no hanging interactive prompts) +# --------------------------------------------------------------------------- + + +@pytest.mark.django_db +def test_interactive_prompt_gets_eof(superuser, monkeypatch): + """A command calling input() must fail fast, not hang the worker.""" + ex = _make_execution(superuser) + + def blocked_command(*args, **kwargs): + input("Limit of entries to add [100000]: ") + + monkeypatch.setattr(tasks_mod, "call_command", blocked_command) + with pytest.raises(EOFError): + # execute_command re-raises the command's exception so the task + # backend records the failure — the EOF must surface, not hang. + execute_command("simple_command", {}, ex.pk) + ex.refresh_from_db() + assert ex.status == CommandExecution.Status.FAILED + assert "EOFError" in ex.output_text("stderr") diff --git a/tests/test_runners.py b/tests/test_runners.py index c84740b..207abed 100644 --- a/tests/test_runners.py +++ b/tests/test_runners.py @@ -1,6 +1,9 @@ +import signal +import socket import sys import unittest.mock from contextlib import contextmanager +from datetime import UTC, datetime import pytest from django.contrib.auth import get_user_model @@ -240,6 +243,92 @@ def test_enqueue_failure_marks_failed(self, user, execution): assert result.is_async is False assert result.task_id == "" + def test_enqueue_options(self, user, execution): + """Failed tasks are acked (no redelivery) and rows stay auditable.""" + from django_admin_runner.runners.django_q2 import DjangoQ2CommandRunner + + runner = DjangoQ2CommandRunner() + with _mock_django_q(return_value="abc-123") as mock_async: + runner.run("simple_command", {}, user, execution) + kwargs = mock_async.call_args.kwargs + assert kwargs["task_name"] == f"admin-runner:{execution.pk}" + assert kwargs["q_options"]["ack_failure"] is True + # django-q2 drops unknown q_options keys: a per-task "retry" would + # be silently ignored, so it must not be set. + assert "retry" not in kwargs["q_options"] + + def test_finalize_stale_matches_task_by_id(self, execution): + """Attribution looks up the q2 Task by *id*, not by humanhash name.""" + from django_admin_runner.runners.django_q2 import DjangoQ2CommandRunner + + execution.task_id = "9a3b052a" # q2 Task.id (a UUID hex) + task = unittest.mock.MagicMock(success=False, stopped=True, result="execution timed out") + manager = unittest.mock.MagicMock() + manager.filter.return_value.first.return_value = task + fake_models = unittest.mock.MagicMock(Task=unittest.mock.MagicMock(objects=manager)) + with _mock_django_q_models(fake_models): + status, note = DjangoQ2CommandRunner().finalize_stale(execution) + manager.filter.assert_called_once_with(pk="9a3b052a") + assert status == CommandExecution.Status.TIMEOUT + assert "timed out" in note + + def test_finalize_stale_no_task_is_none(self, execution): + from django_admin_runner.runners.django_q2 import DjangoQ2CommandRunner + + execution.task_id = "missing-id" + manager = unittest.mock.MagicMock() + manager.filter.return_value.first.return_value = None + fake_models = unittest.mock.MagicMock(Task=unittest.mock.MagicMock(objects=manager)) + with _mock_django_q_models(fake_models): + assert DjangoQ2CommandRunner().finalize_stale(execution) is None + + +# --------------------------------------------------------------------------- +# Signal delivery (host gating / PID-reuse guard) +# --------------------------------------------------------------------------- + + +@pytest.mark.django_db +class TestSignalWorker: + def _runner(self): + from django_admin_runner.runners.django_q2 import DjangoQ2CommandRunner + + return DjangoQ2CommandRunner() + + def test_signal_skipped_for_other_host(self, execution, monkeypatch): + """PIDs from another pod/container are meaningless here: no signal.""" + sent: list = [] + monkeypatch.setattr("os.kill", lambda pid, sig: sent.append((pid, sig))) + execution.worker_pid = 4242 + execution.worker_host = "some-other-pod" + assert self._runner()._signal_worker(execution, signal.SIGKILL) is False + assert sent == [] + + def test_signal_sent_for_local_host_with_matching_start(self, execution, monkeypatch): + sent: list = [] + monkeypatch.setattr("os.kill", lambda pid, sig: sent.append(pid)) + worker_started = datetime(2026, 10, 6, 22, 18, 4, tzinfo=UTC) + monkeypatch.setattr("django_admin_runner.tasks._proc_starttime", lambda pid: worker_started) + execution.worker_pid = 25 + execution.worker_host = socket.gethostname() + execution.worker_started_at = worker_started + assert self._runner()._signal_worker(execution, signal.SIGKILL) is True + assert sent == [25] + + def test_signal_skipped_on_reused_pid(self, execution, monkeypatch): + """PID now held by an older process → reused, do not signal.""" + sent: list = [] + monkeypatch.setattr("os.kill", lambda pid, sig: sent.append(pid)) + monkeypatch.setattr( + "django_admin_runner.tasks._proc_starttime", + lambda pid: datetime(2026, 10, 6, 20, 0, 0, tzinfo=UTC), + ) + execution.worker_pid = 25 + execution.worker_host = socket.gethostname() + execution.worker_started_at = datetime(2026, 10, 6, 22, 18, 4, tzinfo=UTC) + assert self._runner()._signal_worker(execution, signal.SIGKILL) is False + assert sent == [] + # --------------------------------------------------------------------------- # get_runner factory — django-q2 @@ -281,3 +370,24 @@ def _mock_django_q(**kwargs): del sys.modules[key] else: sys.modules[key] = orig + + +@contextmanager +def _mock_django_q_models(models_module): + """Context manager that installs a fake ``django_q.models`` module.""" + mock_django_q = unittest.mock.MagicMock(models=models_module) + saved = {} + for key, val in [ + ("django_q", mock_django_q), + ("django_q.models", models_module), + ]: + saved[key] = sys.modules.get(key) + sys.modules[key] = val + try: + yield + finally: + for key, orig in saved.items(): + if orig is None: + del sys.modules[key] + else: + sys.modules[key] = orig