From fe967c8a964a4b9ab05309e9a48372974aa895f8 Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Mon, 31 Aug 2026 23:43:33 -0700 Subject: [PATCH] fix: simplify cross-machine runner handling --- archivebox/cli/archivebox_run.py | 89 +++++++++++---------- archivebox/core/takeover_util.py | 103 +++++++------------------ archivebox/tests/test_cli_run.py | 6 +- archivebox/tests/test_takeover_util.py | 38 +++++---- 4 files changed, 93 insertions(+), 143 deletions(-) diff --git a/archivebox/cli/archivebox_run.py b/archivebox/cli/archivebox_run.py index 432483d8..320e0e62 100644 --- a/archivebox/cli/archivebox_run.py +++ b/archivebox/cli/archivebox_run.py @@ -270,7 +270,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, maintain_runner_lease, standby_until_foreground_runner_needed + from archivebox.core.takeover_util import enter_single_runner_gate, standby_until_foreground_runner_needed from archivebox.core.recovery_util import recover_orchestrator_state from archivebox.services.runner import run_pending_crawls @@ -292,40 +292,40 @@ def run_runner( current.mark_exited() return 0 - try: - 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 + 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 {}), - ) + 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 {}), + ) return 0 except KeyboardInterrupt: return 0 @@ -432,7 +432,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, maintain_runner_lease + from archivebox.core.takeover_util import enter_single_runner_gate from archivebox.core.shutdown_util import foreground_parent_watchdog, foreground_shutdown_signals from archivebox.machine.models import Process from archivebox.core.models import Snapshot @@ -446,16 +446,15 @@ def run_snapshot_worker(snapshot_id: str) -> int: snapshot = None try: - 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 + 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 9862aa33..295efbec 100644 --- a/archivebox/core/takeover_util.py +++ b/archivebox/core/takeover_util.py @@ -2,13 +2,10 @@ 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 DatabaseError, IntegrityError, close_old_connections +from django.db import IntegrityError from django.utils import timezone from archivebox.config import CONSTANTS from archivebox.config.common import rprint @@ -16,8 +13,6 @@ 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(): @@ -216,76 +211,43 @@ def live_runner_processes(*, data_dir: str | Path, exclude_id=None): machine = Machine.current() Process.cleanup_stale_running(machine=machine) qs = Process.objects.filter( - machine=machine, status=Process.StatusChoices.RUNNING, process_type=Process.TypeChoices.ORCHESTRATOR, worker_type__in=RUNNER_GATE_WORKER_TYPES, pwd=str(data_dir), ) + foreign_machine_exists = qs.exclude(machine=machine).exists() + qs = qs.filter(machine=machine) if exclude_id is not None: qs = qs.exclude(id=exclude_id) - now = timezone.now() + live = [] + foreign_namespace_ids = [] 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 not process.shares_pid_namespace: + foreign_namespace_ids.append(process.id) + continue if process.is_running: live.append(process) + if foreign_machine_exists or foreign_namespace_ids: + rprint( + "[bold yellow]WARNING: Multiple orchestrators sharing a single collection is not officially supported! " + "Corruption may occur if you run two ArchiveBox workers on the same collection at once.[/bold yellow]", + file=sys.stderr, + soft_wrap=True, + ) + if foreign_namespace_ids: + now = timezone.now() + Process.objects.filter(id__in=foreign_namespace_ids, status=Process.StatusChoices.RUNNING).update( + status=Process.StatusChoices.EXITED, + exit_code=0, + ended_at=now, + retry_at=None, + modified_at=now, + ) 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: """ Admit exactly one active runner for this DATA_DIR using Process rows. @@ -304,12 +266,7 @@ 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() @@ -332,16 +289,8 @@ def enter_single_runner_gate(command, *, data_dir: str | Path, graceful_timeout: older_runners = [process for process in runners if process.id != command.id] if older_runners: for process in older_runners: - if process.shares_pid_namespace: - rprint(f"[yellow][*] Stopping older ArchiveBox runner process (pid={process.pid})...[/yellow]", file=sys.stderr) - process.kill_tree(graceful_timeout=graceful_timeout) - else: - 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) + rprint(f"[yellow][*] Stopping older ArchiveBox runner process (pid={process.pid})...[/yellow]", file=sys.stderr) + process.kill_tree(graceful_timeout=graceful_timeout) time.sleep(0.1) continue diff --git a/archivebox/tests/test_cli_run.py b/archivebox/tests/test_cli_run.py index cdda4e93..821cde5a 100644 --- a/archivebox/tests/test_cli_run.py +++ b/archivebox/tests/test_cli_run.py @@ -10,7 +10,6 @@ Tests cover: import os import signal import subprocess -from datetime import timedelta import psutil import pytest @@ -734,7 +733,7 @@ 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): + def test_run_daemon_retires_runner_from_previous_pid_namespace(self, initialized_archive, db): from django.utils import timezone from archivebox.core.takeover_util import RUNNER_ACTIVE_WORKER_TYPE @@ -753,8 +752,6 @@ class TestRunDaemonMode: 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 @@ -774,6 +771,7 @@ class TestRunDaemonMode: try: wait_for_log(daemon_log, "[Crawl#", timeout=10) + assert "Multiple orchestrators sharing a single collection is not officially supported" in daemon_log.read_text() with use_archivebox_db(initialized_archive): stopped_runner.refresh_from_db() assert stopped_runner.status == Process.StatusChoices.EXITED diff --git a/archivebox/tests/test_takeover_util.py b/archivebox/tests/test_takeover_util.py index 6246e377..8d075459 100644 --- a/archivebox/tests/test_takeover_util.py +++ b/archivebox/tests/test_takeover_util.py @@ -7,7 +7,6 @@ import re import signal import subprocess import sys -import time from pathlib import Path import pytest @@ -963,27 +962,32 @@ def test_runtime_stack_owner_allows_top_level_runner_when_no_parent_command_exis _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 +def test_foreign_machine_runner_only_warns(tmp_path, initialized_archive, capsys): + from archivebox.core.takeover_util import RUNNER_ACTIVE_WORKER_TYPE, live_runner_processes from archivebox.machine.models import Machine - runner = Process.objects.create( - machine=Machine.current(), + foreign_machine = Machine.objects.create( + guid="foreign-machine", + hostname="foreign-host", + hw_manufacturer="Test", + hw_product="Test", + hw_uuid="foreign-hardware", + os_arch="x86_64", + os_family="linux", + os_platform="linux", + os_release="test", + os_kernel="test", + ) + foreign_runner = Process.objects.create( + machine=foreign_machine, process_type=Process.TypeChoices.ORCHESTRATOR, worker_type=RUNNER_ACTIVE_WORKER_TYPE, pwd=str(tmp_path), - pid=os.getpid(), - started_at=timezone.now(), + pid=1, 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 + assert live_runner_processes(data_dir=tmp_path) == [] + foreign_runner.refresh_from_db() + assert foreign_runner.status == Process.StatusChoices.RUNNING + assert "Multiple orchestrators sharing a single collection is not officially supported" in capsys.readouterr().err