diff --git a/archivebox/cli/archivebox_run.py b/archivebox/cli/archivebox_run.py index 0f20acdc..5037175c 100644 --- a/archivebox/cli/archivebox_run.py +++ b/archivebox/cli/archivebox_run.py @@ -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__": diff --git a/archivebox/cli/archivebox_update.py b/archivebox/cli/archivebox_update.py index 010e63c3..c978f2a1 100644 --- a/archivebox/cli/archivebox_update.py +++ b/archivebox/cli/archivebox_update.py @@ -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): diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 69c1f60c..9a9a51a5 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -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 diff --git a/archivebox/tests/test_cli_run.py b/archivebox/tests/test_cli_run.py index f11e0cb5..99928adf 100644 --- a/archivebox/tests/test_cli_run.py +++ b/archivebox/tests/test_cli_run.py @@ -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, diff --git a/archivebox/tests/test_config_DELETE_AFTER.py b/archivebox/tests/test_config_DELETE_AFTER.py index f75af8e2..d01997da 100644 --- a/archivebox/tests/test_config_DELETE_AFTER.py +++ b/archivebox/tests/test_config_DELETE_AFTER.py @@ -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", diff --git a/archivebox/tests/test_takeover_util.py b/archivebox/tests/test_takeover_util.py index d0e24226..5e7fc9f7 100644 --- a/archivebox/tests/test_takeover_util.py +++ b/archivebox/tests/test_takeover_util.py @@ -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()), ) diff --git a/archivebox/workers/supervisord_util.py b/archivebox/workers/supervisord_util.py index debf75fb..6213015e 100644 --- a/archivebox/workers/supervisord_util.py +++ b/archivebox/workers/supervisord_util.py @@ -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: