From 1b19736b2f8a394d188561785fcdcbb3731876bf Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Fri, 5 Jun 2026 03:25:38 -0700 Subject: [PATCH] Recover interrupted hook work by hook identity --- archivebox/cli/archivebox_run.py | 1 + archivebox/core/models.py | 19 ++- archivebox/core/recovery_util.py | 72 +++++++++++ archivebox/core/views.py | 2 +- archivebox/machine/models.py | 23 +++- archivebox/services/archive_result_service.py | 22 +++- archivebox/services/runner.py | 116 ++++++++++++------ archivebox/tests/test_opencode_agent.py | 6 +- archivebox/tests/test_takeover_util.py | 34 +++-- archivebox/tests/test_urls.py | 16 +-- 10 files changed, 226 insertions(+), 85 deletions(-) diff --git a/archivebox/cli/archivebox_run.py b/archivebox/cli/archivebox_run.py index 4f20548c..7c3c5eb3 100644 --- a/archivebox/cli/archivebox_run.py +++ b/archivebox/cli/archivebox_run.py @@ -243,6 +243,7 @@ def process_stdin_records() -> int: crawl_id, snapshot_ids=None if crawl_id in full_crawl_ids else sorted(snapshot_ids_by_crawl[crawl_id]), selected_plugins=None if crawl_id in run_all_plugins_for_crawl else sorted(plugin_names_by_crawl[crawl_id]), + selected_plugins_are_explicit=False, ) return 0 diff --git a/archivebox/core/models.py b/archivebox/core/models.py index 31afba9a..f06bd4a7 100755 --- a/archivebox/core/models.py +++ b/archivebox/core/models.py @@ -2737,7 +2737,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW return snapshot - def create_pending_archiveresults(self) -> list["ArchiveResult"]: + def create_pending_archiveresults(self, hooks: Iterable[tuple[str, str]] | None = None) -> list["ArchiveResult"]: """ Create ArchiveResult records for all enabled hooks. @@ -2748,18 +2748,17 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW Creates one ArchiveResult per hook (not per plugin), with hook_name set. This enables step-based execution where all hooks in a step can run in parallel. """ - from archivebox.plugins.hooks import discover_hooks - from archivebox.config.common import get_config + if hooks is None: + from archivebox.plugins.hooks import discover_hooks + from archivebox.config.common import get_config - # Get merged config with crawl-specific PLUGINS filter - config = get_config(crawl=self.crawl, snapshot=self) - hooks = discover_hooks("Snapshot", config=config) + # Compatibility path for direct model callers. The runner passes its + # abx-dl hook inventory explicitly so queued rows match execution. + config = get_config(crawl=self.crawl, snapshot=self) + hooks = ((hook_path.parent.name, hook_path.stem) for hook_path in discover_hooks("Snapshot", config=config)) archiveresults = [] - for hook_path in hooks: - hook_name = hook_path.stem # e.g., 'on_Snapshot__50_wget' - plugin = hook_path.parent.name # e.g., 'wget' - + for plugin, hook_name in hooks: # ArchiveResult output is one filesystem directory per plugin hook, so # retries must update this row in place instead of creating siblings. archiveresult, _created = ArchiveResult.objects.get_or_create( diff --git a/archivebox/core/recovery_util.py b/archivebox/core/recovery_util.py index aeb3b6ce..eb96d2b4 100644 --- a/archivebox/core/recovery_util.py +++ b/archivebox/core/recovery_util.py @@ -1,12 +1,19 @@ from __future__ import annotations +from pathlib import Path + from django.utils import timezone from rich.console import Console +def _is_signal_interrupted_exit(exit_code: int | None) -> bool: + return exit_code is not None and (exit_code < 0 or exit_code >= 128) + + def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int]: from archivebox.crawls.models import Crawl from archivebox.core.models import ArchiveResult, Snapshot + from archivebox.services.archive_result_service import _collect_output_metadata from archivebox.machine.models import Process from django.db.models import Exists, OuterRef, Q, Subquery, Value from django.db.models.functions import Coalesce @@ -22,6 +29,7 @@ def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int "archiveresults_backoff": 0, "snapshots_queued_plugin_rows_waiting_on_stale_lease": 0, "archiveresults_started_without_running_process": 0, + "archiveresults_missing_for_orphaned_hook_processes": 0, "snapshots_started_without_running_results": 0, "crawls_started_with_due_snapshots": 0, "crawls_started_waiting_on_future_snapshots": 0, @@ -100,6 +108,70 @@ def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int process=None, modified_at=now, ) + orphaned_hook_processes = Process.objects.filter( + process_type=Process.TypeChoices.HOOK, + archiveresult__isnull=True, + ).exclude(status=Process.StatusChoices.RUNNING) + for process in orphaned_hook_processes.only("id", "pwd", "cmd", "process_type", "status"): + hook_script_name = process.hook_script_name + if not hook_script_name or not process.pwd: + continue + plugin_dir = Path(process.pwd) + snapshot = Snapshot.objects.filter(id=plugin_dir.parent.name).first() + if snapshot is None: + continue + result, created = ArchiveResult.objects.get_or_create( + snapshot=snapshot, + plugin=plugin_dir.name, + hook_name=Path(hook_script_name).stem, + defaults={ + "status": ArchiveResult.StatusChoices.QUEUED, + }, + ) + if result.status == ArchiveResult.StatusChoices.QUEUED: + requeue_snapshot = False + # A runner can die after the hook Process exits but before the + # ProcessCompletedEvent projector links/finalizes ArchiveResult. + # Reconstruct only that exact hook row from the durable Process row. + output_files, output_size, output_mimetypes = _collect_output_metadata(plugin_dir) + result.process = process + if _is_signal_interrupted_exit(process.exit_code): + # The owning runner died or was asked to stop while the hook was + # still active. Keep the work item queued so takeover retries the + # same hook; treating an unknown signal exit as success would + # silently skip unfinished side effects. + result.output_files = {} + result.output_size = 0 + result.output_mimetypes = "" + result.output_str = "" + result.status = ArchiveResult.StatusChoices.QUEUED + requeue_snapshot = True + else: + result.output_files = output_files + result.output_size = output_size + result.output_mimetypes = output_mimetypes + result.output_str = process.stderr if process.exit_code not in (0, None) else "" + result.status = ( + ArchiveResult.StatusChoices.FAILED + if process.exit_code not in (0, None) + else (ArchiveResult.StatusChoices.SUCCEEDED if output_files else ArchiveResult.StatusChoices.NORESULTS) + ) + result.save( + update_fields=[ + "process", + "output_files", + "output_size", + "output_mimetypes", + "output_str", + "status", + "modified_at", + ], + ) + if requeue_snapshot: + Snapshot.objects.filter(id=snapshot.id).update(retry_at=now, modified_at=now) + if created: + cleaned["archiveresults_missing_for_orphaned_hook_processes"] += 1 + Snapshot.objects.filter(id=snapshot.id).update(retry_at=now, modified_at=now) started_snapshots = Snapshot.objects.filter(status=Snapshot.StatusChoices.STARTED).filter( Q(retry_at__isnull=True) | Q(retry_at__gt=now), ) diff --git a/archivebox/core/views.py b/archivebox/core/views.py index b2fa7db0..1baadf1e 100644 --- a/archivebox/core/views.py +++ b/archivebox/core/views.py @@ -329,7 +329,7 @@ class SnapshotView(View): hidden_card_plugins = {"archivedotorg", "favicon", "title"} outputs = [ out - for out in snapshot.discover_outputs(include_filesystem_fallback=False) + for out in snapshot.discover_outputs(include_filesystem_fallback=True) if (out.get("size") or 0) > 0 and out.get("name") not in hidden_card_plugins ] archiveresults = {out["name"]: out for out in outputs} diff --git a/archivebox/machine/models.py b/archivebox/machine/models.py index 56c1cb09..265e1b82 100755 --- a/archivebox/machine/models.py +++ b/archivebox/machine/models.py @@ -3,6 +3,7 @@ from __future__ import annotations __package__ = "archivebox.machine" import os +import signal import sys import uuid import socket @@ -55,6 +56,13 @@ PROCESS_TIMEOUT_GRACE = timedelta(seconds=30) # Extra margin before force-clean START_TIME_TOLERANCE = 5.0 # Seconds tolerance for start time matching +def _default_exit_code_for_unowned_process(process_type: str) -> int: + # Hooks are externally visible work items. If their owning runner disappeared + # before recording the real exit code, retrying is safer than converting an + # unknown interrupted extraction into a durable success/no-result row. + return 128 + signal.SIGTERM if process_type == Process.TypeChoices.HOOK else 0 + + def _find_existing_binary_for_reference(machine: Machine, reference: str) -> Binary | None: reference = str(reference or "").strip() if not reference: @@ -1608,7 +1616,9 @@ class Process(ModelWithDeleteAfter, models.Model): is_stale = True # Process no longer exists if is_stale: - proc.mark_exited(exit_code=proc.exit_code if proc.exit_code is not None else 0) + proc.mark_exited( + exit_code=proc.exit_code if proc.exit_code is not None else _default_exit_code_for_unowned_process(proc.process_type), + ) cleaned += 1 return cleaned @@ -2138,8 +2148,7 @@ class Process(ModelWithDeleteAfter, models.Model): # TODO: Uncomment to cleanup (keeping for debugging for now) # self.stderr_file.unlink(missing_ok=True) - # Try to get exit code from proc or default to unknown - self.exit_code = self.exit_code if self.exit_code is not None else 0 + self.exit_code = self.exit_code if self.exit_code is not None else _default_exit_code_for_unowned_process(self.process_type) if self.exit_code == -1: self.exit_code = 137 self.ended_at = timezone.now() @@ -2503,7 +2512,9 @@ class Process(ModelWithDeleteAfter, models.Model): # orphaned Process backlog cannot be materialized in memory at once. for proc in running_children.iterator(chunk_size=100): if not proc.is_running: - proc.mark_exited(exit_code=proc.exit_code if proc.exit_code is not None else 0) + proc.mark_exited( + exit_code=proc.exit_code if proc.exit_code is not None else _default_exit_code_for_unowned_process(proc.process_type), + ) cleaned += 1 continue @@ -2527,7 +2538,9 @@ class Process(ModelWithDeleteAfter, models.Model): ): continue - proc.mark_exited(exit_code=proc.exit_code if proc.exit_code is not None else 0) + proc.mark_exited( + exit_code=proc.exit_code if proc.exit_code is not None else _default_exit_code_for_unowned_process(proc.process_type), + ) cleaned += 1 if cleaned: diff --git a/archivebox/services/archive_result_service.py b/archivebox/services/archive_result_service.py index 520c803b..e0a9c258 100644 --- a/archivebox/services/archive_result_service.py +++ b/archivebox/services/archive_result_service.py @@ -240,6 +240,10 @@ def _has_content_files(output_files: Any) -> bool: return any(Path(path).suffix not in {".log", ".pid", ".sh"} for path in _normalize_output_files(output_files)) +def _is_signal_interrupted_exit(exit_code: int) -> bool: + return exit_code < 0 or (exit_code >= 128 and exit_code != PROCESS_EXIT_SKIPPED) + + def _iter_archiveresult_records(stdout: str) -> list[dict]: records: list[dict] = [] for raw_line in stdout.splitlines(): @@ -352,6 +356,13 @@ def _save_archiveresult_event_to_db( with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.result_update"): result.save(update_fields=[*update_fields, "modified_at"]) + if result.status == ArchiveResult.StatusChoices.QUEUED: + # ArchiveResult has no retry_at column. If a shutdown/takeover projects + # a killed hook back to QUEUED, wake the parent Snapshot/Crawl so the + # next runner retries that exact hook instead of waiting on a stale + # active-state lease. + snapshot.update_and_requeue(retry_at=timezone.now()) + if result.status in (ArchiveResult.StatusChoices.SUCCEEDED, ArchiveResult.StatusChoices.NORESULTS): with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.title_update"): title_output_str = result.output_str if result.status == ArchiveResult.StatusChoices.SUCCEEDED else "" @@ -435,14 +446,21 @@ class ArchiveResultService(BaseService): # TODO: consider moving this fallback derivation into abx-dl itself. # First try both patterns: if the whole abx-dl process crashes, restarting # the snapshot may be enough, but don't guess before validating it. - process_failed = event.exit_code not in (0, PROCESS_EXIT_SKIPPED) + process_interrupted = _is_signal_interrupted_exit(event.exit_code) + process_failed = event.exit_code not in (0, PROCESS_EXIT_SKIPPED) and not process_interrupted with _perf_span("archivebox.ArchiveResultService.on_ProcessCompletedEvent.emit_archive_result_fallback"): await event.emit( ArchiveResultEvent( snapshot_id=snapshot_event.snapshot_id, plugin=event.plugin_name, hook_name=event.hook_name, - status="failed" if process_failed else ("succeeded" if _has_content_files(event.output_files) else "noresult"), + status=( + "queued" + if process_interrupted + else "failed" + if process_failed + else ("succeeded" if _has_content_files(event.output_files) else "noresult") + ), output_str=event.stderr if process_failed else "", output_files=event.output_files, start_ts=event.start_ts, diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 363faf17..b1032c2d 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -10,7 +10,6 @@ import threading import time from contextlib import nullcontext from datetime import timedelta -from functools import lru_cache from pathlib import Path from tempfile import TemporaryDirectory from typing import Any @@ -214,6 +213,7 @@ class CrawlRunner: show_progress: bool = True, interactive_interrupts: bool = False, config_overrides: dict[str, Any] | None = None, + selected_plugins_are_explicit: bool = True, ): self.crawl = crawl self.bus = create_bus(name=_bus_name("ArchiveBox", str(crawl.id)), total_timeout=3600.0) @@ -242,6 +242,7 @@ class CrawlRunner: ) ArchiveResultService(self.bus) self.selected_plugins = selected_plugins + self.selected_plugins_from_args = selected_plugins is not None and selected_plugins_are_explicit self.initial_snapshot_ids = snapshot_ids self.snapshot_tasks: dict[str, asyncio.Task[None]] = {} self.snapshot_semaphore = asyncio.Semaphore(1) @@ -1015,6 +1016,7 @@ class CrawlRunner: await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id) return snapshot_selected_plugins = self.selected_plugins + selected_hooks_by_plugin = None if snapshot["status"] == "started": _reset_count, running_count = await sync_to_async(snapshot["_snapshot"].reset_abandoned_results, thread_sensitive=True)() if running_count: @@ -1025,10 +1027,24 @@ class CrawlRunner: thread_sensitive=True, )() return - snapshot_selected_plugins = snapshot_selected_plugins or await sync_to_async( - queued_plugins_for_snapshot, - thread_sensitive=True, - )(snapshot["id"]) + if await sync_to_async(snapshot["_snapshot"].is_finished_processing, thread_sensitive=True)(): + await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)( + snapshot["id"], + output_dir=Path(snapshot["output_dir"]), + ) + return + if not self.selected_plugins_from_args: + queued_plugins, selected_hooks_by_plugin = await sync_to_async( + queued_plugins_and_hooks_for_snapshot, + thread_sensitive=True, + )(snapshot["id"]) + if queued_plugins: + if snapshot_selected_plugins: + queued_plugins = [plugin for plugin in queued_plugins if plugin in snapshot_selected_plugins] + selected_hooks_by_plugin = { + plugin: hooks for plugin, hooks in (selected_hooks_by_plugin or {}).items() if plugin in queued_plugins + } + snapshot_selected_plugins = queued_plugins if snapshot["depth"] > 0 and CrawlLimitState.from_config(snapshot["config"]).get_stop_reason() in ( "crawl_max_size", "crawl_timeout", @@ -1043,6 +1059,18 @@ class CrawlRunner: if snapshot_selected_plugins else self.plugins ) + if selected_hooks_by_plugin is not None: + filtered_plugins = {} + for plugin_name, plugin in plugins.items(): + selected_hook_names = selected_hooks_by_plugin.get(plugin_name) + if selected_hook_names is None: + filtered_plugins[plugin_name] = plugin + continue + filtered_hooks = [ + hook for hook in plugin.hooks if hook.name in selected_hook_names or Path(hook.name).stem in selected_hook_names + ] + filtered_plugins[plugin_name] = plugin.model_copy(update={"hooks": filtered_hooks}) + plugins = filtered_plugins abx_snapshot = AbxSnapshot( id=snapshot["id"], url=snapshot["url"], @@ -1062,6 +1090,7 @@ class CrawlRunner: snapshot_cleanup_enabled=True, snapshot_cleanup_phase_timeout=snapshot_phase_timeout, abort_requested=self.crawl_is_cancelled, + selected_hooks_by_plugin=selected_hooks_by_plugin, ) try: snapshot_event = SnapshotEvent( @@ -1137,6 +1166,7 @@ def run_crawl( show_progress: bool = True, interactive_interrupts: bool = False, config_overrides: dict[str, Any] | None = None, + selected_plugins_are_explicit: bool = True, ) -> None: from archivebox.crawls.models import Crawl from django.db import close_old_connections @@ -1154,6 +1184,7 @@ def run_crawl( show_progress=show_progress, interactive_interrupts=interactive_interrupts, config_overrides=config_overrides, + selected_plugins_are_explicit=selected_plugins_are_explicit, ).run(), ) finally: @@ -1241,14 +1272,7 @@ def run_binary(binary_id: str) -> None: asyncio.run(_run_binary(binary_id)) -@lru_cache(maxsize=1) -def _snapshot_hook_names_by_plugin() -> dict[str, frozenset[str]]: - return { - plugin.name: frozenset(hook.name for hook in plugin.filter_hooks("Snapshot")) for plugin in _discover_archivebox_plugins().values() - } - - -def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None: +def queued_plugins_and_hooks_for_snapshot(snapshot_id: str) -> tuple[list[str] | None, dict[str, set[str] | None] | None]: from archivebox.core.models import ArchiveResult queued_results = list( @@ -1259,30 +1283,39 @@ def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None: .exclude(plugin="") .only("id", "plugin", "hook_name"), ) - hooks_by_plugin = _snapshot_hook_names_by_plugin() - obsolete_result_ids = [ - result.id - for result in queued_results - if result.hook_name and result.hook_name not in hooks_by_plugin.get(result.plugin, frozenset()) - ] - if obsolete_result_ids: - # Hook names are the scheduler identity for ArchiveResults. If an old - # queued row names a hook that the current plugin model cannot run, hard - # fail only that row so the scheduler drains without hiding stale/broken - # plugin state as an intentional skip. - ArchiveResult.objects.filter( - id__in=obsolete_result_ids, - status=ArchiveResult.StatusChoices.QUEUED, - ).update( - status=ArchiveResult.StatusChoices.FAILED, - output_str="Hook no longer exists in the current plugin set.", - modified_at=timezone.now(), - ) - queued_plugins = sorted({result.plugin for result in queued_results if result.id not in obsolete_result_ids}) + selected_hooks_by_plugin: dict[str, set[str] | None] = {} + queued_plugins = sorted({result.plugin for result in queued_results}) + for result in queued_results: + # hook_name is the modern scheduler identity. Empty hook_name rows are + # legacy plugin-level work and must keep running the whole plugin. + if not result.hook_name: + selected_hooks_by_plugin[result.plugin] = None + elif result.plugin not in selected_hooks_by_plugin: + selected_hooks_by_plugin[result.plugin] = {result.hook_name} + elif selected_hooks_by_plugin[result.plugin] is not None: + selected_hooks_by_plugin[result.plugin].add(result.hook_name) if queued_plugins: - return queued_plugins - return None + return queued_plugins, selected_hooks_by_plugin + return None, None + + +def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None: + queued_plugins, _selected_hooks_by_plugin = queued_plugins_and_hooks_for_snapshot(snapshot_id) + return queued_plugins + + +def snapshot_hooks_for_pending_archiveresults(snapshot) -> list[tuple[str, str]]: + from archivebox.config.common import get_config + + config = get_config(crawl=snapshot.crawl, snapshot=snapshot) + plugin_names = [name.strip() for name in str(config.PLUGINS or "").split(",") if name.strip()] + plugins = ( + filter_plugins(_discover_archivebox_plugins(), plugin_names, include_providers=True) + if plugin_names + else _discover_archivebox_plugins() + ) + return sorted((plugin.name, hook.name) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")) def run_snapshot_maintenance(snapshot_id: str, *, output_dir: Path | None = None) -> bool: @@ -1461,6 +1494,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo selected_plugins=selected_plugins, process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts, + selected_plugins_are_explicit=False, ) finally: # Targeted plugin rows can complete while the Snapshot remains @@ -1492,6 +1526,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo selected_plugins=selected_plugins, process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts, + selected_plugins_are_explicit=False, ) if search_only_plugins: from archivebox.core.models import ArchiveResult @@ -1533,7 +1568,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo # rows before ticking so maintenance-only final rows, e.g. search # backfill on a paused snapshot, cannot make queued -> sealed skip the # real extraction work after resume. - snapshot.create_pending_archiveresults() + snapshot.create_pending_archiveresults(hooks=snapshot_hooks_for_pending_archiveresults(snapshot)) snapshot.sm.tick() snapshot.refresh_from_db() if snapshot.status == Snapshot.StatusChoices.SEALED: @@ -1552,7 +1587,15 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo selected_plugins=queued_plugins_for_snapshot(str(snapshot.id)), process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts, + selected_plugins_are_explicit=False, ) + snapshot.refresh_from_db() + if queued_plugins_for_snapshot(str(snapshot.id)): + # Hook-level resume work is tracked by queued ArchiveResult rows, not by + # the Snapshot lease. If a partial pass returns with rows still queued, + # wake the Snapshot immediately so takeover does not wait out a stale + # active-state lock before running the remaining hooks. + snapshot.update_and_requeue(retry_at=timezone.now()) return True @@ -1826,6 +1869,7 @@ def _run_due_queued_plugin_result( config_overrides={ "CRAWL_MAX_CONCURRENT_SNAPSHOTS": QUEUED_PLUGIN_RESULT_BATCH_SIZE, }, + selected_plugins_are_explicit=False, ) if all(plugin.startswith("search_backend_") for plugin in selected_plugins): queued_results = ArchiveResult.objects.filter( diff --git a/archivebox/tests/test_opencode_agent.py b/archivebox/tests/test_opencode_agent.py index b403805b..f7ee1113 100644 --- a/archivebox/tests/test_opencode_agent.py +++ b/archivebox/tests/test_opencode_agent.py @@ -48,7 +48,6 @@ def _set_archivebox_config(data_dir: Path, *values: str, env: dict[str, str] | N @pytest.fixture def opencode_archive_config(initialized_archive): port = _free_port() - lib_dir = initialized_archive / "lib" state_dir = initialized_archive / "opencode" env = os.environ.copy() env.update( @@ -57,8 +56,6 @@ def opencode_archive_config(initialized_archive): "ABXPKG_MIN_RELEASE_AGE": "0", "ABX_RUNTIME": "archivebox", "ARCHIVEBOX_ALLOW_NO_UNIX_SOCKETS": "true", - "LIB_DIR": str(lib_dir), - "ABXPKG_LIB_DIR": str(lib_dir), "OPENCODE_ENABLED": "True", "OPENCODE_HOST": "127.0.0.1", "OPENCODE_PORT": str(port), @@ -75,10 +72,9 @@ def opencode_archive_config(initialized_archive): f"OPENCODE_WORKDIR={initialized_archive}", f"OPENCODE_STATE_DIR={state_dir}", "OPENCODE_TIMEOUT=60", - f"LIB_DIR={lib_dir}", env=env, ) - return SimpleNamespace(data_dir=initialized_archive, lib_dir=lib_dir, port=port, state_dir=state_dir, env=env) + return SimpleNamespace(data_dir=initialized_archive, port=port, state_dir=state_dir, env=env) @pytest.fixture diff --git a/archivebox/tests/test_takeover_util.py b/archivebox/tests/test_takeover_util.py index 2d0f0f22..a47ef4a3 100644 --- a/archivebox/tests/test_takeover_util.py +++ b/archivebox/tests/test_takeover_util.py @@ -379,7 +379,7 @@ def test_live_server_keeps_http_runtime_while_update_runs_real_sqlite_indexer(tm ) assert "Stopping older ArchiveBox runner process" in update_stdout - deadline = time.time() + 90 + deadline = time.time() + 180 runner_pid_after = runner_pid_before while time.time() < deadline: with use_archivebox_db(tmp_path): @@ -578,7 +578,7 @@ def test_live_repeated_server_startups_take_over_cleanly(tmp_path, initialized_a @pytest.mark.timeout(420) -def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initialized_archive): +def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initialized_archive, recursive_test_site): plugins_root = tmp_path / "runtime_plugins" marker_dir = tmp_path / "slow-plugin-markers" plugin_dir = plugins_root / "slow_exit" @@ -658,8 +658,8 @@ def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initi "--max-urls=2", "--crawl-max-size=50mb", "--plugins=wget,parse_html_urls,slow_exit", - "https://example.com", - "https://blog.sweeting.me", + recursive_test_site["root_url"], + recursive_test_site["child_urls"][0], ], cwd=tmp_path, env=env, @@ -696,12 +696,15 @@ def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initi assert add_proc.poll() is None, "foreground add should keep owning its crawl after the server exits" assert "Got SIGTERM" in server_log.read_text(encoding="utf-8", errors="replace") + stop_archivebox_process(add_proc, signal.SIGKILL, timeout=30) + add_output = add_log.read_text(encoding="utf-8", errors="replace") + assert "Runner error" not in add_output + kill_processes_for_data_dir(tmp_path) + assert_no_processes_for_data_dir(tmp_path, timeout=12) + server2 = start_archivebox_server(tmp_path, port=port, log_name="server-add-owner-2.log", env=env) _server2_log = server2.log_path (marker_dir / "allow-finish").touch() - stop_archivebox_process(add_proc, signal.SIGTERM, timeout=30) - add_output = add_log.read_text(encoding="utf-8", errors="replace") - assert "Runner error" not in add_output deadline = time.time() + 90 crawls = [] @@ -751,17 +754,12 @@ def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initi assert len(counter_runs) == len(set(counter_runs)) assert not (marker_dir / "counter-duplicates.txt").exists() - # TODO: improve abx-dl's ability to explicitly resume from a given plugin / hook and skip ones before that - # current behavior: on retry, earlier sealed results are left untouched; the interrupted result is marked skipped - # and may have partial output saved to fs - # assertions that enforce the current behavior (uncommented): previous results are not run twice + interrupted - # result is marked skipped - assert bad_results == [("slow_exit", ArchiveResult.StatusChoices.SKIPPED, "")] - - # desired future behavior: earlier sealed results are left untouched, interrupted result is retried and cleanly - # overwrites on top of any previous partial output - # assert not bad_results - # assert (marker_dir / "hook-finished").exists() + # The interrupted hook should be retried directly without rerunning the + # previous hook in the same plugin. That keeps plugin-level shell hooks + # idempotent across runner takeover instead of depending on each hook to + # detect partial prior work itself. + assert not bad_results + assert (marker_dir / "hook-finished").exists() finally: for proc in (add_proc, add_proc2, server, server2, server3): if proc is not None and proc.poll() is None: diff --git a/archivebox/tests/test_urls.py b/archivebox/tests/test_urls.py index 6911d0c9..7b32f125 100644 --- a/archivebox/tests/test_urls.py +++ b/archivebox/tests/test_urls.py @@ -895,19 +895,19 @@ class TestUrlRouting: assert f"http://{public_host}/static/archive.png" in live_html assert "?preview=1" in live_html assert "function createMainFrame(previousFrame)" in live_html - assert "function activateCardPreview(card, link)" in live_html - assert "ensureMainFrame(true)" in live_html + assert "function activateCardPreview(card, link, updateHash=true)" in live_html + assert "ensureMainFrame(currentSrc !== nextSrcAbs)" in live_html assert "previousFrame.parentNode.replaceChild(frame, previousFrame)" in live_html assert "previousFrame.src = 'about:blank'" in live_html assert "event.stopImmediatePropagation()" in live_html - assert "const matchingLink = [...document.querySelectorAll('a[target=preview]')].find" in live_html + assert "const matchingLink = findPreviewLinkForHash(selectedPreviewHash)" in live_html assert "jQuery(link).click()" not in live_html assert "searchParams.delete('preview')" in live_html assert "doc.body.style.flexDirection = 'column'" in live_html assert "doc.body.style.alignItems = 'center'" in live_html assert "img.style.margin = '0 auto'" in live_html assert "window.location.hash = getPreviewHashValueFromHref(rawTarget)" in live_html - assert "const selectedPreviewHash = decodeURIComponent(window.location.hash.slice(1)).toLowerCase()" in live_html + assert "const selectedPreviewHash = window.location.hash ? decodeURIComponent(window.location.hash.slice(1)).toLowerCase() : ''" in live_html assert "pointer-events: none;" in live_html assert "pointer-events: auto;" in live_html assert 'class="thumbnail-click-overlay"' in live_html @@ -921,19 +921,19 @@ class TestUrlRouting: assert f"http://{public_host}/static/archive.png" in static_html assert "?preview=1" in static_html assert "function createMainFrame(previousFrame)" in static_html - assert "function activateCardPreview(card, link)" in static_html - assert "ensureMainFrame(true)" in static_html + assert "function activateCardPreview(card, link, updateHash=true)" in static_html + assert "ensureMainFrame(currentSrc !== nextSrcAbs)" in static_html assert "previousFrame.parentNode.replaceChild(frame, previousFrame)" in static_html assert "previousFrame.src = 'about:blank'" in static_html assert "event.stopImmediatePropagation()" in static_html - assert "const matchingLink = [...document.querySelectorAll('a[target=preview]')].find" in static_html + assert "const matchingLink = findPreviewLinkForHash(selectedPreviewHash)" in static_html assert "jQuery(link).click()" not in static_html assert "searchParams.delete('preview')" in static_html assert "doc.body.style.flexDirection = 'column'" in static_html assert "doc.body.style.alignItems = 'center'" in static_html assert "img.style.margin = '0 auto'" in static_html assert "window.location.hash = getPreviewHashValueFromHref(rawTarget)" in static_html - assert "const selectedPreviewHash = decodeURIComponent(window.location.hash.slice(1)).toLowerCase()" in static_html + assert "const selectedPreviewHash = window.location.hash ? decodeURIComponent(window.location.hash.slice(1)).toLowerCase() : ''" in static_html assert "pointer-events: none;" in static_html assert "pointer-events: auto;" in static_html assert 'class="thumbnail-click-overlay"' in static_html