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 diff --git a/etc/package.json b/etc/package.json index 40c87aee..b208b747 100644 --- a/etc/package.json +++ b/etc/package.json @@ -1,6 +1,6 @@ { "name": "archivebox", - "version": "0.9.35rc352", + "version": "0.9.35rc356", "repository": "github:ArchiveBox/ArchiveBox", "license": "MIT", "dependencies": { diff --git a/pyproject.toml b/pyproject.toml index 0eb0bd90..e1998683 100755 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "archivebox" -version = "0.9.35rc352" +version = "0.9.35rc356" requires-python = ">=3.13" description = "Self-hosted internet archiving solution." authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}] @@ -81,9 +81,9 @@ dependencies = [ ### Extractor dependencies (runtime binaries resolved through abxpkg) ### Binary/Package Management "abxbus==2.5.56", # direct imports only; version constrained by abx-dl -> abx-plugins - "abxpkg==1.12.109", # direct imports only; version constrained by abx-dl -> abx-plugins - "abx-plugins==1.12.192", # direct imports only; version constrained by abx-dl - "abx-dl==1.12.218", # shared ArchiveBox downloader package + "abxpkg==1.12.110", # direct imports only; version constrained by abx-dl -> abx-plugins + "abx-plugins==1.12.194", # direct imports only; version constrained by abx-dl + "abx-dl==1.12.221", # shared ArchiveBox downloader package ### UUID7 backport for Python <3.14 "uuid7>=0.1.0; python_version < '3.14'", # provides the uuid_extensions module on Python 3.13 ] @@ -344,7 +344,7 @@ Donate = "https://github.com/ArchiveBox/ArchiveBox/wiki/Donations" [tool.bumpver] -current_version = "v0.9.35rc352" +current_version = "v0.9.35rc356" version_pattern = "vMAJOR.MINOR.PATCH[PYTAGNUM]" commit_message = "bump version {old_version} -> {new_version}" tag_message = "{new_version}" diff --git a/uv.lock b/uv.lock index 7113e2f3..e9005b6d 100644 --- a/uv.lock +++ b/uv.lock @@ -13,18 +13,18 @@ supported-markers = [ ] [options] -exclude-newer = "2026-08-24T02:15:26.974926946Z" +exclude-newer = "2026-08-26T02:23:48.072440255Z" exclude-newer-span = "P5D" [options.exclude-newer-package] -abxbus = { timestamp = "2026-08-29T02:15:25.974939359Z", span = "PT1S" } +abxbus = { timestamp = "2026-08-31T02:23:47.072457601Z", span = "PT1S" } abx-plugins = "2100-01-01T00:00:00Z" abx-dl = "2100-01-01T00:00:00Z" abxpkg = "2100-01-01T00:00:00Z" [[package]] name = "abx-dl" -version = "1.12.218" +version = "1.12.221" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "abx-plugins", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, @@ -38,14 +38,14 @@ dependencies = [ { name = "rich", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, { name = "rich-click", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/80/71/7f316e610cf7f8421cc5b5fe3887ec2eb60ef52fe21741721681c9cda910/abx_dl-1.12.218.tar.gz", hash = "sha256:064c543943da3a5185184e9b261d9d909783d83ccbd141f74a404369b63045bc", size = 84041 } +sdist = { url = "https://files.pythonhosted.org/packages/36/be/2843cae923a47edb4760c2ac7a0a1ce2cfc86891ec4d6d40c9fa1b84afda/abx_dl-1.12.221.tar.gz", hash = "sha256:070ec8a589331f5f5bfa7bdf05482c1be5dbb885915c05ecf0fa3a648715d4bd", size = 84188 } wheels = [ - { url = "https://files.pythonhosted.org/packages/26/2e/233a861d901356b5220ab543356ec8ca2f3d63388327bded4cac5f59c487/abx_dl-1.12.218-py3-none-any.whl", hash = "sha256:1ef7a7fb2f0204d6fd9dcaec45f48f0cfaee03502175d209498b381752c6bb70", size = 87728 }, + { url = "https://files.pythonhosted.org/packages/8d/19/452a6d45c095aff7e8f2ae92321681e695223cea17a18a4466473ea5063f/abx_dl-1.12.221-py3-none-any.whl", hash = "sha256:8a8d0ce77565144c3858d923cdc39fff9afd260667075eab8943cb16eb726e96", size = 87897 }, ] [[package]] name = "abx-plugins" -version = "1.12.192" +version = "1.12.194" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "abxbus", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, @@ -54,9 +54,9 @@ dependencies = [ { name = "rich-click", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, { name = "uv", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/7f/13/412ee8b46d7fbb977c1d8607d521ba3ea0dae61feb5bf4fb8b176830b65c/abx_plugins-1.12.192.tar.gz", hash = "sha256:4edcc841a85f231bb2ab01c89cdaab6e7873b40af04f459449907348bcb19513", size = 255242, upload-time = "2026-08-28T22:24:40.73Z" } +sdist = { url = "https://files.pythonhosted.org/packages/dc/a0/439b2dc6dd52f0c68aff026ab938ad0f4e4e42c81276db10d77ac86eb60e/abx_plugins-1.12.194.tar.gz", hash = "sha256:baa1e5225619000ab6cdd7d426783734ac2d63939a655e5acdf512d36a790b30", size = 256164, upload-time = "2026-08-31T01:27:36.17Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/17/3e/2504344111dd0a769d2430c76c131d4a6cc5f58418f296c547660e9a7f1f/abx_plugins-1.12.192-py3-none-any.whl", hash = "sha256:0b3c6ee3fc6600e08266206ae7eca3c5a98df2f6855ec772120703cfc3d6ff4f", size = 406098, upload-time = "2026-08-28T22:24:42.035Z" }, + { url = "https://files.pythonhosted.org/packages/68/8c/a209995e1e7043dcf014353ec4781d40df0af418082d4814d8188f0caa84/abx_plugins-1.12.194-py3-none-any.whl", hash = "sha256:ce2e3b63d908a7ea0a8d837dad8fd5407effc09064ece321859d8cf9459c6fab", size = 407200, upload-time = "2026-08-31T01:27:34.666Z" }, ] [[package]] @@ -75,7 +75,7 @@ wheels = [ [[package]] name = "abxpkg" -version = "1.12.109" +version = "1.12.110" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "platformdirs", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, @@ -83,9 +83,9 @@ dependencies = [ { name = "rich-click", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, { name = "typing-extensions", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a9/70/1cabf0920b425191f91cbefde3b144ceb478c647dfecc49676f01a239f30/abxpkg-1.12.109.tar.gz", hash = "sha256:c20f3b28bff82a83984e940e9165d2f7b2fccd1dbe84ba04ff5414db2ade29e0", size = 239416, upload-time = "2026-08-28T22:19:19.002Z" } +sdist = { url = "https://files.pythonhosted.org/packages/dc/25/152a0e2dd37494cab5116629b4a3f26232dda6641e7b79f8eb7d1e9bfd75/abxpkg-1.12.110.tar.gz", hash = "sha256:94777e6ea76944b6577c46aaf4d845cdc68cfbdfe4c58ca24cd02071e367a660", size = 239950, upload-time = "2026-08-31T01:25:16.393Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/71/90/37a9df601259ce300bdfa840b5cab82d1566a39d8ddddde6283623caa17e/abxpkg-1.12.109-py3-none-any.whl", hash = "sha256:c8cbf3b6ea66e7ed7dfc4bdf2c277a99bea4728293c39840cf6be8fc349c01dd", size = 253436, upload-time = "2026-08-28T22:19:17.472Z" }, + { url = "https://files.pythonhosted.org/packages/05/95/49a113ceccf1d1ac33a90a0f8735abd43919f6cf2543917542acbdd61ea3/abxpkg-1.12.110-py3-none-any.whl", hash = "sha256:01e3e95c0af1a8fe775dee07eb16fa540a4e7007c3961a52e11d57bc2eae2430", size = 254023, upload-time = "2026-08-31T01:25:14.833Z" }, ] [[package]] @@ -120,7 +120,7 @@ wheels = [ [[package]] name = "archivebox" -version = "0.9.35rc352" +version = "0.9.35rc356" source = { editable = "." } dependencies = [ { name = "abx-dl", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, @@ -220,10 +220,10 @@ dev = [ [package.metadata] requires-dist = [ - { name = "abx-dl", specifier = "==1.12.218" }, - { name = "abx-plugins", specifier = "==1.12.192" }, + { name = "abx-dl", specifier = "==1.12.221" }, + { name = "abx-plugins", specifier = "==1.12.194" }, { name = "abxbus", specifier = "==2.5.56" }, - { name = "abxpkg", specifier = "==1.12.109" }, + { name = "abxpkg", specifier = "==1.12.110" }, { name = "archivebox", extras = ["sonic", "ldap", "debug"], marker = "extra == 'all'" }, { name = "atomicwrites", specifier = "==1.4.1" }, { name = "base32-crockford", specifier = ">=0.3.0" },