diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 9f2980b2..9dedfac6 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -1146,7 +1146,14 @@ class CrawlRunner: )(snapshot["id"]) if queued_plugins: if snapshot_selected_plugins: + original_queued_plugins = queued_plugins queued_plugins = queued_plugins_selected_by_config(queued_plugins) + disabled_queued_plugins = sorted(set(original_queued_plugins) - set(queued_plugins)) + if disabled_queued_plugins: + await sync_to_async(skip_disabled_queued_plugins, thread_sensitive=True)( + snapshot["id"], + disabled_queued_plugins, + ) selected_hooks_by_plugin = { plugin: hooks for plugin, hooks in (selected_hooks_by_plugin or {}).items() if plugin in queued_plugins } @@ -1158,7 +1165,14 @@ class CrawlRunner: )(snapshot["id"]) if queued_plugins: if snapshot_selected_plugins: + original_queued_plugins = queued_plugins queued_plugins = queued_plugins_selected_by_config(queued_plugins) + disabled_queued_plugins = sorted(set(original_queued_plugins) - set(queued_plugins)) + if disabled_queued_plugins: + await sync_to_async(skip_disabled_queued_plugins, thread_sensitive=True)( + snapshot["id"], + disabled_queued_plugins, + ) selected_hooks_by_plugin = { plugin: hooks for plugin, hooks in (selected_hooks_by_plugin or {}).items() if plugin in queued_plugins } @@ -1189,7 +1203,13 @@ class CrawlRunner: if snapshot_selected_plugins and remaining_queued_plugins: remaining_queued_plugins = queued_plugins_selected_by_config(remaining_queued_plugins) if not remaining_queued_plugins: - await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id, output_dir=output_dir) + if snapshot["status"] in ("queued", "started"): + await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)( + snapshot_id, + output_dir=output_dir, + ) + else: + await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id, output_dir=output_dir) return snapshot_selected_plugins = remaining_queued_plugins plugins = filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True) @@ -1498,6 +1518,28 @@ def fail_unavailable_queued_hooks( ) +def skip_disabled_queued_plugins(snapshot_id: str, plugin_names: list[str]) -> None: + from archivebox.core.models import ArchiveResult + + if not plugin_names: + return + now = timezone.now() + # Queued ArchiveResult rows are durable scheduler state and can outlive a + # config change or deploy. If a plugin is no longer selected for this + # Snapshot/Crawl, leaving its old row queued keeps the Snapshot STARTED + # forever even though there is no runnable work left. + ArchiveResult.objects.filter( + snapshot_id=snapshot_id, + plugin__in=plugin_names, + status=ArchiveResult.StatusChoices.QUEUED, + ).update( + status=ArchiveResult.StatusChoices.SKIPPED, + start_ts=now, + end_ts=now, + output_str="Queued plugin is disabled by this Snapshot/Crawl config", + ) + + def include_background_prerequisite_hooks( selected_hooks_by_plugin: dict[str, set[str] | None], plugins: dict[str, Plugin], @@ -1530,12 +1572,13 @@ def include_background_prerequisite_hooks( def snapshot_hooks_for_pending_archiveresults(snapshot) -> list[tuple[str, str]]: from archivebox.config.common import get_config from archivebox.core.models import Snapshot + from archivebox.plugins.discovery import get_enabled_plugins config = get_config(crawl=snapshot.crawl, snapshot=snapshot) snapshot_plugin_names = [name.strip() for name in str((snapshot.config or {}).get("PLUGINS") or "").split(",") if name.strip()] crawl_plugin_names = [name.strip() for name in str((snapshot.crawl.config or {}).get("PLUGINS") or "").split(",") if name.strip()] config_plugin_names = [name.strip() for name in str(config.PLUGINS or "").split(",") if name.strip()] - plugin_names = snapshot_plugin_names or crawl_plugin_names or config_plugin_names + plugin_names = snapshot_plugin_names or crawl_plugin_names or config_plugin_names or get_enabled_plugins(config=config) plugins = ( filter_plugins(_discover_archivebox_plugins(), plugin_names, include_providers=True) if plugin_names diff --git a/archivebox/tests/test_cli_run.py b/archivebox/tests/test_cli_run.py index 244a6300..8717d14e 100644 --- a/archivebox/tests/test_cli_run.py +++ b/archivebox/tests/test_cli_run.py @@ -991,6 +991,42 @@ class TestRecoverOrchestratorState: assert hook_names assert all(not hook_name.endswith((".py", ".js", ".sh")) for hook_name in hook_names) + def test_snapshot_hooks_for_pending_archiveresults_respects_disabled_plugins_when_plugins_empty(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 Snapshot + from archivebox.services.runner import snapshot_hooks_for_pending_archiveresults + + crawl = Crawl.objects.create( + urls="https://example.com", + created_by_id=get_or_create_system_user_pk(), + config={ + "CLAUDECHROME_ENABLED": False, + "CLAUDECODEEXTRACT_ENABLED": False, + "CLAUDECODECLEANUP_ENABLED": False, + "SEARCH_BACKEND_SQLITE_ENABLED": False, + }, + status=Crawl.StatusChoices.STARTED, + retry_at=timezone.now(), + ) + snapshot = Snapshot.objects.create( + url="https://example.com", + crawl=crawl, + status=Snapshot.StatusChoices.QUEUED, + retry_at=timezone.now(), + ) + + hooks = snapshot_hooks_for_pending_archiveresults(snapshot) + queued_plugins = {plugin for plugin, _hook_name in hooks} + + assert "title" in queued_plugins + assert "claudechrome" not in queued_plugins + assert "claudecodeextract" not in queued_plugins + assert "claudecodecleanup" not in queued_plugins + assert "search_backend_sqlite" not in queued_plugins + def test_run_due_snapshot_pauses_child_when_parent_is_paused(self): from django.utils import timezone @@ -1467,6 +1503,73 @@ class TestRecoverOrchestratorState: assert result.status == ArchiveResult.StatusChoices.FAILED assert snapshot.retry_at is None + @pytest.mark.django_db(transaction=True) + def test_run_due_snapshot_skips_disabled_queued_plugins_and_seals_started_snapshot(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.runner import run_due_snapshot + + crawl = Crawl.objects.create( + urls="https://example.com", + created_by_id=get_or_create_system_user_pk(), + config={ + "CLAUDECHROME_ENABLED": False, + "CLAUDECODEEXTRACT_ENABLED": False, + "CLAUDECODECLEANUP_ENABLED": False, + "SEARCH_BACKEND_SQLITE_ENABLED": False, + }, + status=Crawl.StatusChoices.STARTED, + retry_at=timezone.now(), + ) + snapshot = Snapshot.objects.create( + url="https://example.com", + crawl=crawl, + status=Snapshot.StatusChoices.STARTED, + retry_at=timezone.now(), + downloaded_at=timezone.now(), + ) + ArchiveResult.objects.create( + snapshot=snapshot, + plugin="title", + hook_name="on_Snapshot__01_title", + status=ArchiveResult.StatusChoices.SUCCEEDED, + output_str="Example Domain", + ) + stale_results = [ + ArchiveResult.objects.create( + snapshot=snapshot, + plugin=plugin, + hook_name=hook_name, + status=ArchiveResult.StatusChoices.QUEUED, + ) + for plugin, hook_name in ( + ("claudechrome", "on_Snapshot__47_claudechrome"), + ("claudecodeextract", "on_Snapshot__58_claudecodeextract"), + ("claudecodecleanup", "on_Snapshot__92_claudecodecleanup"), + ("search_backend_sqlite", "on_Snapshot__90_index_sqlite"), + ) + ] + stale_result_ids = [result.id for result in stale_results] + stale_plugins = [result.plugin for result in stale_results] + + assert run_due_snapshot(snapshot, lock_seconds=60) is True + + snapshot.refresh_from_db() + assert snapshot.status == Snapshot.StatusChoices.SEALED + assert snapshot.retry_at is None + assert not snapshot.archiveresult_set.filter( + plugin__in=stale_plugins, + status=ArchiveResult.StatusChoices.QUEUED, + ).exists() + for result in ArchiveResult.objects.filter(id__in=stale_result_ids): + assert result.status == ArchiveResult.StatusChoices.SKIPPED + assert result.start_ts is not None + assert result.end_ts is not None + assert "disabled by this Snapshot/Crawl config" in result.output_str + @pytest.mark.django_db(transaction=True) @pytest.mark.timeout(300) @pytest.mark.parametrize("chrome_isolation", ["crawl", "snapshot"])