The_Great_Archive/ArchiveBox
1
0
Fork 0
mirror of https://github.com/ArchiveBox/ArchiveBox.git synced 2026-09-12 19:50:57 +05:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

fix: simplify cross-machine runner handling

Browse Source
This commit is contained in:
Nick Sweeting 2026-08-31 23:43:33 -07:00
parent 402c983f61
commit fe967c8a96
No known key found for this signature in database
4 changed files with 93 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
archivebox/cli/archivebox_run.py
View File

@ -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:

103
archivebox/core/takeover_util.py
View File

@ -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

6
archivebox/tests/test_cli_run.py
View File

@ -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

38
archivebox/tests/test_takeover_util.py
View File

@ -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