Merge branch 'dev' into fix/static-archive-install-cache

This commit is contained in:
Nick Sweeting 2026-08-31 22:11:02 -07:00 committed by GitHub
commit 72955a08af
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 225 additions and 70 deletions

View File

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

View File

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

View File

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

View File

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

View File

@ -1,6 +1,6 @@
{
"name": "archivebox",
"version": "0.9.35rc352",
"version": "0.9.35rc356",
"repository": "github:ArchiveBox/ArchiveBox",
"license": "MIT",
"dependencies": {

View File

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

30
uv.lock
View File

@ -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" },