Make worker liveness heartbeat-based so executions stop being falsely finalized

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
 

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,
+            ),
+        ),
+    ]

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,

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/<pid>/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: