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 runner takeover for scoped snapshot workers

Browse Source
This commit is contained in:
Nick Sweeting 2026-06-07 06:40:40 -07:00
parent 0816435b68
commit 2659f20dc4
No known key found for this signature in database
7 changed files with 108 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
archivebox/cli/archivebox_run.py
View File

@ -431,11 +431,19 @@ 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.shutdown_util import foreground_parent_watchdog, foreground_shutdown_signals
from archivebox.machine.models import Process
from archivebox.core.models import Snapshot
from archivebox.services.runner import run_due_snapshot
from django.utils import timezone
current = Process.current()
if not enter_single_runner_gate(current, data_dir=CONSTANTS.DATA_DIR):
current.mark_exited()
return 0
snapshot = None
try:
with foreground_shutdown_signals(), foreground_parent_watchdog():
@ -465,6 +473,10 @@ def run_snapshot_worker(snapshot_id: str) -> int:
traceback.print_exc()
return 1
finally:
current.refresh_from_db()
if current.status != Process.StatusChoices.EXITED:
current.mark_exited()
if __name__ == "__main__":

16
archivebox/cli/archivebox_update.py
View File

@ -235,6 +235,14 @@ def update(
command = current_command(Process.TypeChoices.UPDATE, data_dir=CONSTANTS.DATA_DIR)
def still_owns_foreground_runner() -> bool:
from django.db import connections
try:
return command_owns_foreground_runner(command, data_dir=CONSTANTS.DATA_DIR)
finally:
connections.close_all()
def wait_for_turn() -> None:
raise_if_shutdown_requested()
standby_until_foreground_runner_needed(command, data_dir=CONSTANTS.DATA_DIR)
@ -245,7 +253,11 @@ def update(
wait_for_turn()
if ensure_daemon_reason:
ensure_daemon_stack(reason=ensure_daemon_reason)
exit_code = run_runner_worker(list(args), name=f"worker_runner_update_{os.getpid()}")
exit_code = run_runner_worker(
list(args),
name=f"worker_runner_update_{os.getpid()}",
keep_running=still_owns_foreground_runner,
)
if exit_code == 0:
return
if not command_owns_foreground_runner(command, data_dir=CONSTANTS.DATA_DIR):
@ -479,7 +491,7 @@ def update(
pass
elif not runner_work_queued:
pass
elif touched_snapshot_ids:
elif touched_snapshot_ids and is_filtered_update:
if not touched_snapshot_ids:
print("[*] No matching snapshots queued work for the runner.")
for snapshot_id in sorted(touched_snapshot_ids):

36
archivebox/services/runner.py
View File

@ -1089,12 +1089,7 @@ class CrawlRunner:
return
snapshot_selected_plugins = remaining_queued_plugins
plugins = filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True)
# Queued ArchiveResult rows select plugin work, not a mid-plugin
# resume cursor. Rerun the full selected plugin lifecycle so
# ordered prerequisite hooks like chrome_tab run before barriers
# like chrome_wait, while search_backend_* maintenance remains
# targeted to only those selected plugins.
selected_hooks_by_plugin = None
selected_hooks_by_plugin = include_background_prerequisite_hooks(selected_hooks_by_plugin, plugins)
abx_snapshot = AbxSnapshot(
id=snapshot["id"],
url=snapshot["url"],
@ -1363,6 +1358,35 @@ def fail_unavailable_queued_hooks(
)
def include_background_prerequisite_hooks(
selected_hooks_by_plugin: dict[str, set[str] | None],
plugins: dict[str, Plugin],
) -> dict[str, set[str] | None]:
expanded: dict[str, set[str] | None] = {}
for plugin_name, selected_hook_names in selected_hooks_by_plugin.items():
if selected_hook_names is None or plugin_name not in plugins:
expanded[plugin_name] = selected_hook_names
continue
plugin_hooks = sorted(plugins[plugin_name].filter_hooks("Snapshot"), key=lambda hook: hook.sort_key)
selected_sort_keys = [
hook.sort_key for hook in plugin_hooks if hook.name in selected_hook_names or Path(hook.name).stem in selected_hook_names
]
if not selected_sort_keys:
expanded[plugin_name] = set(selected_hook_names)
continue
first_selected_sort_key = min(selected_sort_keys)
expanded_hook_names = set(selected_hook_names)
# Earlier background hooks publish live resources (e.g. Chrome tabs)
# needed by later foreground hooks, but completed foreground hooks stay
# final and are not rerun during hook-level resume.
for hook in plugin_hooks:
if hook.is_background and hook.sort_key < first_selected_sort_key:
expanded_hook_names.add(hook.name)
expanded_hook_names.add(Path(hook.name).stem)
expanded[plugin_name] = expanded_hook_names
return expanded
def snapshot_hooks_for_pending_archiveresults(snapshot) -> list[tuple[str, str]]:
from archivebox.config.common import get_config

2
archivebox/tests/test_cli_run.py
View File

@ -1468,7 +1468,7 @@ class TestRecoverOrchestratorState:
@pytest.mark.django_db(transaction=True)
@pytest.mark.timeout(300)
@pytest.mark.parametrize("chrome_isolation", ["crawl", "snapshot"])
def test_resume_queued_chrome_wait_reruns_full_chrome_plugin_lifecycle(
def test_resume_queued_chrome_wait_reruns_background_prerequisites(
self,
initialized_archive,
recursive_test_site,

2
archivebox/tests/test_config_DELETE_AFTER.py
View File

@ -206,7 +206,7 @@ def test_delete_after_real_add_page_and_rest_create_paths(client):
"depth": "0",
"max_urls": "1",
"crawl_max_size": "0",
"crawl_timeout": "0",
"crawl_timeout": "60",
"snapshot_max_size": "0",
"delete_after": "2h",
"crawl_max_concurrent_snapshots": "1",

4
archivebox/tests/test_takeover_util.py
View File

@ -345,7 +345,7 @@ def test_live_server_keeps_http_runtime_while_update_runs_real_sqlite_indexer(tm
env = cli_env(
live=True,
PLUGINS="wget,parse_html_urls,search_backend_sqlite",
PLUGINS="wget,parse_html_urls,search_backend_sqlite,search_backend_sonic",
SEARCH_BACKEND_ENGINE="sqlite",
SEARCH_BACKEND_SONIC_PORT=str(get_free_port()),
)
@ -435,7 +435,7 @@ def test_live_update_yields_to_server_then_reclaims_real_sqlite_indexing(tmp_pat
env = cli_env(
live=True,
PLUGINS="wget,parse_html_urls,search_backend_sqlite",
PLUGINS="wget,parse_html_urls,search_backend_sqlite,search_backend_sonic",
SEARCH_BACKEND_ENGINE="sqlite",
SEARCH_BACKEND_SONIC_PORT=str(get_free_port()),
)

51
archivebox/workers/supervisord_util.py
View File

@ -4,6 +4,8 @@ import sys
import time
import socket
import os
import csv
import json
import psutil
import shutil
import subprocess
@ -344,10 +346,36 @@ files = {WORKERS_DIR}/*.conf
(WORKERS_DIR / "initial_startup.conf").write_text("") # hides error about "no files found to include" when supervisord starts
def _worker_environment_value(daemon: dict[str, str], key: str) -> str | None:
environment = daemon.get("environment")
if not environment:
return None
try:
fields = next(csv.reader([environment], skipinitialspace=True))
except csv.Error:
fields = str(environment).split(",")
for field in fields:
name, separator, value = field.partition("=")
if separator and name == key:
try:
return str(json.loads(value))
except json.JSONDecodeError:
return value.strip('"')
return None
def _worker_log_base_dir(daemon: dict[str, str]) -> Path:
data_dir = _worker_environment_value(daemon, "DATA_DIR")
return Path(data_dir) if data_dir else CONSTANTS.DATA_DIR
def create_worker_config(daemon):
"""Create a supervisord worker config file for a given daemon"""
SOCK_FILE = get_sock_file()
WORKERS_DIR = SOCK_FILE.parent / WORKERS_DIR_NAME
log_base_dir = _worker_log_base_dir(daemon)
Path.mkdir(WORKERS_DIR, exist_ok=True, parents=True)
for logfile_key in ("stdout_logfile", "stderr_logfile"):
@ -356,7 +384,7 @@ def create_worker_config(daemon):
continue
logfile_path = Path(logfile)
if not logfile_path.is_absolute():
logfile_path = CONSTANTS.DATA_DIR / logfile_path
logfile_path = log_base_dir / logfile_path
logfile_path.parent.mkdir(parents=True, exist_ok=True)
name = daemon["name"]
@ -371,7 +399,7 @@ def create_worker_config(daemon):
if key in ("stdout_logfile", "stderr_logfile"):
logfile_path = Path(value)
if not logfile_path.is_absolute():
value = str(CONSTANTS.DATA_DIR / logfile_path)
value = str(log_base_dir / logfile_path)
worker_str += f"{key}={value}\n"
worker_str += "\n"
@ -850,10 +878,19 @@ def get_or_create_supervisord_process(daemonize=False):
def start_worker(supervisor, daemon, lazy=False):
existing = get_worker(supervisor, daemon["name"])
if isinstance(existing, dict) and existing.get("statename") in ("STARTING", "RUNNING"):
return existing
return sync_supervisord_workers(supervisor, [(daemon, lazy)], prune=False).get(daemon["name"])
def run_runner_worker(args: list[str], *, name: str = "worker_runner_once", interactive_interrupts: bool = False) -> int:
def run_runner_worker(
args: list[str],
*,
name: str = "worker_runner_once",
interactive_interrupts: bool = False,
keep_running=None,
) -> int:
supervisor = get_or_create_supervisord_process(daemonize=False)
worker = RUNNER_ONCE_WORKER(args, name=name)
log_path = Path(worker["stdout_logfile"])
@ -869,6 +906,14 @@ def run_runner_worker(args: list[str], *, name: str = "worker_runner_once", inte
try:
while True:
try:
if keep_running is not None and not keep_running():
try:
proc = get_worker(supervisor, name)
if proc is not None and proc.get("statename") not in final_states:
supervisor.stopProcess(name, False)
except Fault:
pass
return 1
while True:
line = log_handle.readline()
if not line: