From b28a4e1a555426730b7926b67a0e93a0d3fb4afc Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Mon, 31 Aug 2026 21:06:06 -0700 Subject: [PATCH] Fix runner recovery across PID namespaces --- archivebox/cli/archivebox_run.py | 88 +++++++++++++------------- archivebox/core/takeover_util.py | 83 ++++++++++++++++++++++-- archivebox/tests/test_cli_run.py | 55 ++++++++++++++++ archivebox/tests/test_takeover_util.py | 27 ++++++++ 4 files changed, 204 insertions(+), 49 deletions(-) diff --git a/archivebox/cli/archivebox_run.py b/archivebox/cli/archivebox_run.py index 8dec58f9..3d2a65e4 100644 --- a/archivebox/cli/archivebox_run.py +++ b/archivebox/cli/archivebox_run.py @@ -265,7 +265,7 @@ def run_runner( from archivebox.config import CONSTANTS from archivebox.core.shutdown_util import foreground_parent_watchdog, foreground_shutdown_signals from archivebox.machine.models import Machine, Process - from archivebox.core.takeover_util import enter_single_runner_gate, standby_until_foreground_runner_needed + from archivebox.core.takeover_util import enter_single_runner_gate, maintain_runner_lease, standby_until_foreground_runner_needed from archivebox.core.recovery_util import recover_orchestrator_state from archivebox.services.runner import run_pending_crawls @@ -287,39 +287,40 @@ def run_runner( current.mark_exited() return 0 - recover_orchestrator_state(include_chrome=crawl_id is None, crawl_id=crawl_id) - if crawl_id: - from django.utils import timezone - from archivebox.crawls.models import Crawl - - crawl = Crawl.objects.filter(id=crawl_id, status__in=Crawl.RUNNABLE_STATES).first() - now = timezone.now() - # Winning the single-runner gate terminates and waits for every older - # runner. Requeue the explicitly requested crawl so an abandoned - # future ownership lease cannot hide it from the unified scheduler. - if crawl is not None: - crawl.update_and_requeue(retry_at=now, refresh=False) - # Only a foreground `archivebox add` gets the interactive "abort current - # hook, continue/retry, second Ctrl+C exits" flow. Server/update/run owned - # orchestrators should shut down immediately and cleanly on the first signal. - interactive_interrupts = current.root.process_type == Process.TypeChoices.ADD - if daemon: - os.environ[RUNNER_DAEMON_ENV] = "1" try: - with ( - foreground_shutdown_signals( - on_signal=_exit_daemon_runner_on_signal if daemon else None, - raise_on_first_signal=not daemon, - ), - foreground_parent_watchdog(enabled=not daemon), - ): - run_pending_crawls( - daemon=daemon, - crawl_id=crawl_id, - maintenance_only=maintenance_only, - interactive_interrupts=interactive_interrupts, - **({"maintenance_batch_size": maintenance_batch_size} if maintenance_batch_size else {}), - ) + with maintain_runner_lease(current): + recover_orchestrator_state(include_chrome=crawl_id is None, crawl_id=crawl_id) + if crawl_id: + from django.utils import timezone + from archivebox.crawls.models import Crawl + + crawl = Crawl.objects.filter(id=crawl_id, status__in=Crawl.RUNNABLE_STATES).first() + now = timezone.now() + # Winning the single-runner gate terminates and waits for every older + # runner. Requeue the explicitly requested crawl so an abandoned + # future ownership lease cannot hide it from the unified scheduler. + if crawl is not None: + crawl.update_and_requeue(retry_at=now, refresh=False) + # Only a foreground `archivebox add` gets the interactive "abort current + # hook, continue/retry, second Ctrl+C exits" flow. Server/update/run owned + # orchestrators should shut down immediately and cleanly on the first signal. + interactive_interrupts = current.root.process_type == Process.TypeChoices.ADD + if daemon: + os.environ[RUNNER_DAEMON_ENV] = "1" + with ( + foreground_shutdown_signals( + on_signal=_exit_daemon_runner_on_signal if daemon else None, + raise_on_first_signal=not daemon, + ), + foreground_parent_watchdog(enabled=not daemon), + ): + run_pending_crawls( + daemon=daemon, + crawl_id=crawl_id, + maintenance_only=maintenance_only, + interactive_interrupts=interactive_interrupts, + **({"maintenance_batch_size": maintenance_batch_size} if maintenance_batch_size else {}), + ) return 0 except KeyboardInterrupt: return 0 @@ -426,7 +427,7 @@ def main( def run_snapshot_worker(snapshot_id: str) -> int: from archivebox.config import CONSTANTS - from archivebox.core.takeover_util import enter_single_runner_gate + from archivebox.core.takeover_util import enter_single_runner_gate, maintain_runner_lease from archivebox.core.shutdown_util import foreground_parent_watchdog, foreground_shutdown_signals from archivebox.machine.models import Process from archivebox.core.models import Snapshot @@ -440,15 +441,16 @@ def run_snapshot_worker(snapshot_id: str) -> int: snapshot = None try: - with foreground_shutdown_signals(), foreground_parent_watchdog(): - for _ in range(10): - snapshot = Snapshot.objects.select_related("crawl").get(id=snapshot_id) - if snapshot.retry_at is None: - snapshot.update_and_requeue(retry_at=timezone.now()) - elif snapshot.retry_at > timezone.now(): - break - if not run_due_snapshot(snapshot, lock_seconds=60): - break + with maintain_runner_lease(current): + with foreground_shutdown_signals(), foreground_parent_watchdog(): + for _ in range(10): + snapshot = Snapshot.objects.select_related("crawl").get(id=snapshot_id) + if snapshot.retry_at is None: + snapshot.update_and_requeue(retry_at=timezone.now()) + elif snapshot.retry_at > timezone.now(): + break + if not run_due_snapshot(snapshot, lock_seconds=60): + break return 0 except KeyboardInterrupt: try: diff --git a/archivebox/core/takeover_util.py b/archivebox/core/takeover_util.py index 3013dee2..9862aa33 100644 --- a/archivebox/core/takeover_util.py +++ b/archivebox/core/takeover_util.py @@ -2,10 +2,13 @@ from __future__ import annotations import time import sys +import threading from collections.abc import Callable +from contextlib import contextmanager +from datetime import timedelta from pathlib import Path -from django.db import IntegrityError +from django.db import DatabaseError, IntegrityError, close_old_connections from django.utils import timezone from archivebox.config import CONSTANTS from archivebox.config.common import rprint @@ -13,6 +16,8 @@ from archivebox.config.common import rprint RUNNER_ACTIVE_WORKER_TYPE = "worker_runner" RUNNER_WAITING_WORKER_TYPE = "runner_waiting" RUNNER_GATE_WORKER_TYPES = (RUNNER_ACTIVE_WORKER_TYPE, RUNNER_WAITING_WORKER_TYPE, "") +RUNNER_LEASE_HEARTBEAT_SECONDS = 5.0 +RUNNER_LEASE_TIMEOUT = timedelta(seconds=30) def runtime_stack_owner_types(): @@ -219,7 +224,66 @@ def live_runner_processes(*, data_dir: str | Path, exclude_id=None): ) if exclude_id is not None: qs = qs.exclude(id=exclude_id) - return [process for process in qs.order_by("started_at", "created_at").iterator(chunk_size=20) if process.is_running] + now = timezone.now() + live = [] + for process in qs.order_by("started_at", "created_at").iterator(chunk_size=20): + if not process.shares_pid_namespace and process.modified_at < now - RUNNER_LEASE_TIMEOUT: + # A PID from another container cannot be inspected or signaled + # safely. Its DB heartbeat is the only cross-namespace liveness + # proof, so release the runner gate only after that lease expires. + exited = Process.objects.filter( + pk=process.pk, + status=Process.StatusChoices.RUNNING, + modified_at=process.modified_at, + ).update( + status=Process.StatusChoices.EXITED, + exit_code=process.exit_code if process.exit_code is not None else 0, + ended_at=now, + retry_at=None, + modified_at=now, + ) + if exited: + continue + process.refresh_from_db() + if process.is_running: + live.append(process) + return live + + +@contextmanager +def maintain_runner_lease(command, *, interval: float = RUNNER_LEASE_HEARTBEAT_SECONDS): + """Keep the active runner row fresh for takeover across PID namespaces.""" + from archivebox.machine.models import Process + + stopped = threading.Event() + + def heartbeat() -> None: + close_old_connections() + try: + while not stopped.wait(interval): + try: + updated = Process.objects.filter( + pk=command.pk, + status=Process.StatusChoices.RUNNING, + worker_type=RUNNER_ACTIVE_WORKER_TYPE, + ).update(modified_at=timezone.now()) + if not updated: + return + except DatabaseError: + # A transient SQLite lock must not terminate the runner. + # The timeout allows several missed heartbeats before a + # cross-namespace contender may reclaim the gate. + continue + finally: + close_old_connections() + + heartbeat_thread = threading.Thread(target=heartbeat, name="archivebox-runner-lease", daemon=True) + heartbeat_thread.start() + try: + yield + finally: + stopped.set() + heartbeat_thread.join(timeout=interval + 1.0) def enter_single_runner_gate(command, *, data_dir: str | Path, graceful_timeout: float = 5.0) -> bool: @@ -240,7 +304,12 @@ def enter_single_runner_gate(command, *, data_dir: str | Path, graceful_timeout: pwd=str(data_dir), timeout=CONSTANTS.MAX_HOOK_RUNTIME_SECONDS, ) + last_waiting_heartbeat = time.monotonic() + announced_cross_namespace_runners = set() while True: + if time.monotonic() - last_waiting_heartbeat >= RUNNER_LEASE_HEARTBEAT_SECONDS: + command.heartbeat() + last_waiting_heartbeat = time.monotonic() runners = live_runner_processes(data_dir=data_dir) if all(process.id != command.id for process in runners): command.refresh_from_db() @@ -267,10 +336,12 @@ def enter_single_runner_gate(command, *, data_dir: str | Path, graceful_timeout: rprint(f"[yellow][*] Stopping older ArchiveBox runner process (pid={process.pid})...[/yellow]", file=sys.stderr) process.kill_tree(graceful_timeout=graceful_timeout) else: - rprint( - "[yellow][*] Waiting for older ArchiveBox runner in another PID namespace to stop...[/yellow]", - file=sys.stderr, - ) + if process.id not in announced_cross_namespace_runners: + rprint( + "[yellow][*] Waiting for older ArchiveBox runner in another PID namespace to stop...[/yellow]", + file=sys.stderr, + ) + announced_cross_namespace_runners.add(process.id) time.sleep(0.1) continue diff --git a/archivebox/tests/test_cli_run.py b/archivebox/tests/test_cli_run.py index cd3a10af..cdda4e93 100644 --- a/archivebox/tests/test_cli_run.py +++ b/archivebox/tests/test_cli_run.py @@ -10,6 +10,7 @@ Tests cover: import os import signal import subprocess +from datetime import timedelta import psutil import pytest @@ -733,6 +734,60 @@ class TestRunDaemonMode: cleanup_process_group(proc.pid) proc.wait(timeout=15) + def test_run_daemon_recovers_stopped_runner_from_previous_pid_namespace(self, initialized_archive, db): + from django.utils import timezone + + from archivebox.core.takeover_util import RUNNER_ACTIVE_WORKER_TYPE + from archivebox.machine.models import Machine, PROCESS_PID_NAMESPACE_KEY, Process, get_current_pid_namespace + from archivebox.tests.test_orm_helpers import use_archivebox_db + + env = cli_env(PLUGINS="__archivebox_test_no_plugins__") + with use_archivebox_db(initialized_archive): + stopped_runner = Process.objects.create( + machine=Machine.current(), + process_type=Process.TypeChoices.ORCHESTRATOR, + worker_type=RUNNER_ACTIVE_WORKER_TYPE, + status=Process.StatusChoices.RUNNING, + pwd=str(initialized_archive), + pid=1, + started_at=timezone.now(), + env={PROCESS_PID_NAMESPACE_KEY: f"{get_current_pid_namespace()}-stopped-container"}, + ) + Process.objects.filter(pk=stopped_runner.pk).update(modified_at=timezone.now() - timedelta(minutes=5)) + + queued = run_archivebox_cmd(["crawl", "create", create_test_url()], cwd=initialized_archive, env=env, timeout=60) + assert queued.returncode == 0, queued.stderr or queued.stdout + + daemon_log = initialized_archive / "run-daemon-stopped-container.log" + daemon_log_handle = daemon_log.open("w", encoding="utf-8") + replacement = run_archivebox_cmd( + ["run", "--daemon"], + cwd=initialized_archive, + env=env, + stdin=subprocess.DEVNULL, + stdout=daemon_log_handle, + stderr=subprocess.STDOUT, + start_new_session=True, + wait=False, + ) + daemon_log_handle.close() + + try: + wait_for_log(daemon_log, "[Crawl#", timeout=10) + with use_archivebox_db(initialized_archive): + stopped_runner.refresh_from_db() + assert stopped_runner.status == Process.StatusChoices.EXITED + active_runner = Process.objects.get( + process_type=Process.TypeChoices.ORCHESTRATOR, + worker_type=RUNNER_ACTIVE_WORKER_TYPE, + status=Process.StatusChoices.RUNNING, + pwd=str(initialized_archive), + ) + assert active_runner.pid == replacement.pid + finally: + cleanup_process_group(replacement.pid) + replacement.wait(timeout=15) + @pytest.mark.django_db class TestRecoverOrchestratorState: diff --git a/archivebox/tests/test_takeover_util.py b/archivebox/tests/test_takeover_util.py index 951b61a0..1a9d46aa 100644 --- a/archivebox/tests/test_takeover_util.py +++ b/archivebox/tests/test_takeover_util.py @@ -7,6 +7,7 @@ import re import signal import subprocess import sys +import time from pathlib import Path import pytest @@ -955,3 +956,29 @@ def test_runtime_stack_owner_allows_top_level_runner_when_no_parent_command_exis assert owner.id == runner_row.id finally: _stop_archivebox_shells([proc]) + + +def test_active_runner_lease_heartbeat_updates_persisted_process(tmp_path, initialized_archive): + from django.utils import timezone + + from archivebox.core.takeover_util import RUNNER_ACTIVE_WORKER_TYPE, maintain_runner_lease + from archivebox.machine.models import Machine + + runner = Process.objects.create( + machine=Machine.current(), + process_type=Process.TypeChoices.ORCHESTRATOR, + worker_type=RUNNER_ACTIVE_WORKER_TYPE, + pwd=str(tmp_path), + pid=os.getpid(), + started_at=timezone.now(), + status=Process.StatusChoices.RUNNING, + ) + initial_heartbeat = runner.modified_at + + with maintain_runner_lease(runner, interval=0.01): + deadline = time.monotonic() + 2.0 + while runner.modified_at <= initial_heartbeat and time.monotonic() < deadline: + time.sleep(0.01) + runner.refresh_from_db() + + assert runner.modified_at > initial_heartbeat