diff --git a/archivebox/core/recovery_util.py b/archivebox/core/recovery_util.py index eb96d2b4..1fa32c06 100644 --- a/archivebox/core/recovery_util.py +++ b/archivebox/core/recovery_util.py @@ -15,6 +15,7 @@ def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int 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.core.exceptions import ValidationError from django.db.models import Exists, OuterRef, Q, Subquery, Value from django.db.models.functions import Coalesce @@ -117,7 +118,13 @@ def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int 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() + try: + # Old or synthetic hook Process rows can point at arbitrary paths. + # Only paths whose parent directory is a valid Snapshot id can be + # reconstructed into ArchiveResult rows. + snapshot = Snapshot.objects.filter(id=plugin_dir.parent.name).first() + except ValidationError: + continue if snapshot is None: continue result, created = ArchiveResult.objects.get_or_create( diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index b1032c2d..7452f164 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -1045,6 +1045,18 @@ class CrawlRunner: plugin: hooks for plugin, hooks in (selected_hooks_by_plugin or {}).items() if plugin in queued_plugins } snapshot_selected_plugins = queued_plugins + elif 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", @@ -1060,6 +1072,11 @@ class CrawlRunner: else self.plugins ) if selected_hooks_by_plugin is not None: + await sync_to_async(fail_unavailable_queued_hooks, thread_sensitive=True)( + snapshot["id"], + selected_hooks_by_plugin, + plugins, + ) filtered_plugins = {} for plugin_name, plugin in plugins.items(): selected_hook_names = selected_hooks_by_plugin.get(plugin_name) @@ -1305,6 +1322,39 @@ def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None: return queued_plugins +def fail_unavailable_queued_hooks( + snapshot_id: str, + selected_hooks_by_plugin: dict[str, set[str] | None], + plugins: dict[str, Plugin], +) -> None: + from archivebox.core.models import ArchiveResult + + now = timezone.now() + for plugin_name, selected_hook_names in selected_hooks_by_plugin.items(): + if selected_hook_names is None or plugin_name not in plugins: + continue + available_hook_names = { + name for hook in plugins[plugin_name].filter_hooks("Snapshot") for name in (hook.name, Path(hook.name).stem) + } + missing_hook_names = [hook_name for hook_name in selected_hook_names if hook_name not in available_hook_names] + if not missing_hook_names: + continue + # Hook-level resume rows are durable scheduler state. If a plugin is + # installed but no longer exposes a queued hook, mark that row failed so + # the snapshot is not retried forever with no hook left to execute. + ArchiveResult.objects.filter( + snapshot_id=snapshot_id, + plugin=plugin_name, + hook_name__in=missing_hook_names, + status=ArchiveResult.StatusChoices.QUEUED, + ).update( + status=ArchiveResult.StatusChoices.FAILED, + start_ts=now, + end_ts=now, + output_str="Queued hook is no longer available in the installed plugin", + ) + + 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 4eb5686d..5dbc234a 100644 --- a/archivebox/tests/test_cli_run.py +++ b/archivebox/tests/test_cli_run.py @@ -1390,13 +1390,14 @@ class TestRecoverOrchestratorState: assert result.status == ArchiveResult.StatusChoices.SUCCEEDED assert snapshot.retry_at is None - def test_run_due_snapshot_runs_queued_plugin_after_fs_migration(self, monkeypatch): + @pytest.mark.django_db(transaction=True) + def test_run_due_snapshot_runs_queued_plugin_after_fs_migration(self): from django.utils import timezone from archivebox.base_models.models import get_or_create_system_user_pk from archivebox.crawls.models import Crawl from archivebox.core.models import ArchiveResult, Snapshot - from archivebox.services import runner + from archivebox.services.runner import run_due_snapshot crawl = Crawl.objects.create( urls="https://example.com", @@ -1412,45 +1413,28 @@ class TestRecoverOrchestratorState: ) Snapshot.objects.filter(pk=snapshot.pk).update(fs_version="0.9.0") snapshot.refresh_from_db() + snapshot.output_dir.mkdir(parents=True, exist_ok=True) + title_dir = snapshot.output_dir / "title" + title_dir.mkdir(parents=True, exist_ok=True) + (title_dir / "title.txt").write_text("Example Domain\n", encoding="utf-8") result = ArchiveResult.objects.create( snapshot=snapshot, - plugin="search_backend_sonic", - hook_name="on_Snapshot__91_index_sonic", + plugin="search_backend_sqlite", + hook_name="on_Snapshot__90_index_sqlite", status=ArchiveResult.StatusChoices.QUEUED, ) - calls = [] - def fake_run_crawl(crawl_id, *, snapshot_ids=None, selected_plugins=None, **kwargs): - calls.append((crawl_id, snapshot_ids, selected_plugins, kwargs)) - ArchiveResult.objects.filter(pk=result.pk).update( - status=ArchiveResult.StatusChoices.NORESULTS, - start_ts=timezone.now(), - end_ts=timezone.now(), - output_str="No indexable content", - ) - - monkeypatch.setattr( - runner, - "_snapshot_hook_names_by_plugin", - lambda: {"search_backend_sonic": frozenset({"on_Snapshot__91_index_sonic"})}, - ) - monkeypatch.setattr(runner, "run_crawl", fake_run_crawl) - - assert runner.run_due_snapshot(snapshot, lock_seconds=60) is True + assert run_due_snapshot(snapshot, lock_seconds=60) is True snapshot.refresh_from_db() result.refresh_from_db() assert snapshot.fs_version == Snapshot._fs_current_version() - assert result.status == ArchiveResult.StatusChoices.NORESULTS - assert calls == [ - ( - str(crawl.id), - [str(snapshot.id)], - ["search_backend_sonic"], - {"process_discovered_snapshots_inline": True, "interactive_interrupts": False}, - ), - ] + assert result.status in ArchiveResult.FINAL_STATES + assert result.status != ArchiveResult.StatusChoices.QUEUED + assert result.start_ts is not None + assert result.end_ts is not None + @pytest.mark.django_db(transaction=True) def test_run_due_snapshot_fails_obsolete_queued_hook_name(self): from django.utils import timezone diff --git a/archivebox/tests/test_machine_models.py b/archivebox/tests/test_machine_models.py index fa013d17..eb53e8cc 100644 --- a/archivebox/tests/test_machine_models.py +++ b/archivebox/tests/test_machine_models.py @@ -929,7 +929,7 @@ class TestProcessClassMethods: child.refresh_from_db() assert child.status == Process.StatusChoices.EXITED assert child.ended_at is not None - assert child.exit_code == 0 + assert child.exit_code == 143 class TestProcessStateMachine: