mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-12 19:50:57 +05:00
Fix stuck snapshots with disabled queued plugins
This commit is contained in:
Nick Sweeting
parent
e9cb936c81
commit
83188f70e6
@ -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
|
||||
|
||||
@ -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"])
|
||||
|
||||
Loading…
Reference in New Issue
Block a user