From 5559370c8f4eaebb091fad1f10c0862f57aa643f Mon Sep 17 00:00:00 2001 From: "tbxy-agent[bot]" <336722940+tbxy-agent[bot]@users.noreply.github.com> Date: Wed, 7 Oct 2026 09:31:23 +0200 Subject: [PATCH] feat: heartbeat-based worker liveness, host-gated stop signals Liveness for RUNNING executions is now heartbeat-based instead of pid-based. Workers refresh last_heartbeat_at every ADMIN_RUNNER_HEARTBEAT_INTERVAL (default 30 s) on a daemon thread, and the stale-run sweeper finalizes RUNNING rows whose heartbeat is older than ADMIN_RUNNER_HEARTBEAT_STALE_AFTER (default 180 s). Rationale: a pid is only meaningful inside its own PID namespace. In container deployments (web pods serving the admin, a separate task cluster pod) the old sweep checked pids from the wrong namespace, so *every* watched long-running command was falsely finalized as FAILED 'worker lost' minutes into a healthy run; conversely, a recycled pid in the cluster pod kept dead rows RUNNING forever. The heartbeat flows through the shared database and stays meaningful across pods. Rows recorded before this change keep the legacy pid rules, now with a PID-reuse guard (a pid occupied by a process older than the execution counts as dead, not alive). Also: - stop/force_stop signals are only sent when the worker ran on the local host (worker_host recorded at claim); the cooperative stop flag remains the cross-host mechanism. Previously a Force Stop from a web pod silently no-oped - or could signal an unrelated local process on pid collision. - finalize_stale looks up the django-q2 Task by id (pk) instead of name: task_id stores the UUID, Task.name holds a humanhash, so timeout attribution never matched. - django-q2 tasks are enqueued with ack_failure=True so failed tasks are not redelivered after the broker retry window; the per-task retry=-1 option was dropped (django-q2 silently ignores unknown q_options keys) and tasks get a readable 'admin-runner:' name. - sys.stdin is redirected to an EOF, non-tty StringIO while a command runs: interactive prompts (input()/click) now fail fast with EOFError instead of hanging the worker forever (commands ran with a TTY- reporting stdout, so prompts believed a terminal was attached). Adds ADMIN_RUNNER_HEARTBEAT_INTERVAL and ADMIN_RUNNER_HEARTBEAT_STALE_AFTER settings (docs updated). --- docs/guides/stopping-commands.md | 40 ++- ..._worker_host_worker_started_at_and_more.py | 52 ++++ src/django_admin_runner/models.py | 20 ++ src/django_admin_runner/runners/django_q2.py | 76 +++--- src/django_admin_runner/tasks.py | 201 ++++++++++++--- tests/test_cancellation.py | 9 +- tests/test_heartbeat_liveness.py | 229 ++++++++++++++++++ tests/test_runners.py | 110 +++++++++ 8 files changed, 662 insertions(+), 75 deletions(-) create mode 100644 src/django_admin_runner/migrations/0012_commandexecution_worker_host_worker_started_at_and_more.py create mode 100644 tests/test_heartbeat_liveness.py 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