mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-14 11:06:13 +05:00
Enforce the sealed search maintenance boundary
This commit is contained in:
Nick Sweeting
parent
aab13e2d9f
commit
ee0e3eb43c
@ -46,7 +46,6 @@ def process_archiveresult_by_id(archiveresult_id: str) -> int:
|
||||
through the shared crawl runner with the corresponding plugin selected.
|
||||
"""
|
||||
from rich import print as rprint
|
||||
from django.utils import timezone
|
||||
from archivebox.core.models import ArchiveResult
|
||||
from archivebox.api.v1_core import _uuid_ref_query
|
||||
from archivebox.services.runner import run_crawl
|
||||
@ -60,22 +59,9 @@ def process_archiveresult_by_id(archiveresult_id: str) -> int:
|
||||
rprint(f"[blue]Extracting {archiveresult.plugin} for {archiveresult.snapshot.url}[/blue]", file=sys.stderr)
|
||||
|
||||
try:
|
||||
was_paused = archiveresult.snapshot.is_paused
|
||||
archiveresult.reset_for_retry()
|
||||
snapshot = archiveresult.snapshot
|
||||
if not was_paused:
|
||||
snapshot.queue_for_extraction()
|
||||
else:
|
||||
# A paused snapshot may still accept explicit maintenance for one
|
||||
# ArchiveResult, but this path must not transition it back to
|
||||
# queued/startable work. Guard: only set retry_at while the row is
|
||||
# still paused — concurrent resume would otherwise see a stale
|
||||
# retry_at marker.
|
||||
snapshot.safe_update(
|
||||
{"retry_at": timezone.now()},
|
||||
refresh=False,
|
||||
extra_filter={"status": snapshot.StatusChoices.PAUSED},
|
||||
)
|
||||
snapshot.queue_for_extraction()
|
||||
crawl = snapshot.crawl
|
||||
if not crawl.claim_processing_lock(lock_seconds=10):
|
||||
rprint(
|
||||
@ -84,11 +70,7 @@ def process_archiveresult_by_id(archiveresult_id: str) -> int:
|
||||
)
|
||||
return 1
|
||||
|
||||
try:
|
||||
run_crawl(str(snapshot.crawl_id), snapshot_ids=[str(snapshot.id)], selected_plugins=[archiveresult.plugin])
|
||||
finally:
|
||||
if was_paused:
|
||||
snapshot.restore_paused_scheduler_marker()
|
||||
run_crawl(str(snapshot.crawl_id), snapshot_ids=[str(snapshot.id)], selected_plugins=[archiveresult.plugin])
|
||||
archiveresult.refresh_from_db()
|
||||
|
||||
if archiveresult.status == ArchiveResult.StatusChoices.SUCCEEDED:
|
||||
@ -290,12 +272,9 @@ def run_plugins(
|
||||
queue_at = timezone.now()
|
||||
if existing_snapshot_ids:
|
||||
if requested_rows:
|
||||
# Targeted ArchiveResult retries use retry_at as the scheduling
|
||||
# signal and keep sealed snapshots sealed so extractors are not
|
||||
# re-run outside the explicitly queued plugin rows. Paused snapshots
|
||||
# also keep status=paused here: `retry_at` only asks the orchestrator
|
||||
# to process the queued plugin rows, and run_due_snapshot restores
|
||||
# retry_at=MAX afterward instead of resuming the snapshot lifecycle.
|
||||
# Search indexing on a sealed Snapshot is the only targeted hook
|
||||
# allowed to bypass the normal lifecycle. Every other requested
|
||||
# plugin requeues its Snapshot through the unified state machine.
|
||||
affected_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in rows_to_queue}
|
||||
if preserve_queued and queued_rows:
|
||||
queued_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in queued_rows}
|
||||
@ -312,14 +291,30 @@ def run_plugins(
|
||||
flat=True,
|
||||
)
|
||||
)
|
||||
requested_plugins_by_id: dict[str, set[str]] = defaultdict(set)
|
||||
for snapshot_id, plugin_name, _hook_name in requested_rows:
|
||||
requested_plugins_by_id[snapshot_id].add(plugin_name)
|
||||
for snapshot in Snapshot.objects.filter(id__in=affected_snapshot_ids).only("id", "status", "modified_at"):
|
||||
# Guard the read-time status so we never bump retry_at on a
|
||||
# row that's been re-queued / started by a concurrent runner.
|
||||
snapshot.safe_update(
|
||||
{"retry_at": queue_at, "modified_at": queue_at},
|
||||
refresh=False,
|
||||
extra_filter={"status": snapshot.status},
|
||||
plugin_names = requested_plugins_by_id.get(str(snapshot.id), set())
|
||||
sealed_search_backfill = (
|
||||
snapshot.status == Snapshot.StatusChoices.SEALED
|
||||
and plugin_names
|
||||
and all(plugin_name.startswith("search_backend_") for plugin_name in plugin_names)
|
||||
)
|
||||
if sealed_search_backfill:
|
||||
snapshot.safe_update(
|
||||
{"retry_at": queue_at, "modified_at": queue_at},
|
||||
refresh=False,
|
||||
extra_filter={"status": snapshot.status},
|
||||
)
|
||||
else:
|
||||
snapshot.update_and_requeue(
|
||||
status=Snapshot.StatusChoices.QUEUED,
|
||||
retry_at=queue_at,
|
||||
current_step=0,
|
||||
)
|
||||
else:
|
||||
# No plugin rows were requested, so this is a full snapshot retry.
|
||||
for snapshot in Snapshot.objects.filter(id__in=existing_snapshot_ids).only("id", "status", "retry_at", "modified_at"):
|
||||
|
||||
@ -348,7 +348,7 @@ def run_runner(
|
||||
@click.option("--crawl-id", help="Run the crawl runner for a specific crawl only")
|
||||
@click.option("--snapshot-id", help="Run one snapshot through its crawl")
|
||||
@click.option("--binary-id", help="Run one queued binary install directly on the bus")
|
||||
@click.option("--maintenance-only", is_flag=True, help="Only process due maintenance ticks on sealed/paused snapshots")
|
||||
@click.option("--maintenance-only", is_flag=True, help="Only process sealed Snapshot maintenance and search-index backfills")
|
||||
@click.option(
|
||||
"--maintenance-batch-size",
|
||||
type=int,
|
||||
|
||||
@ -76,9 +76,14 @@ def reindex_snapshots(
|
||||
wait_for_turn=None,
|
||||
) -> dict[str, Any]:
|
||||
from archivebox.cli.archivebox_extract import run_plugins
|
||||
from archivebox.core.models import ArchiveResult
|
||||
from archivebox.core.models import ArchiveResult, Snapshot
|
||||
from abx_dl.models import discover_plugins
|
||||
|
||||
# Search backfill is the one maintenance hook allowed to execute without
|
||||
# reopening a Snapshot. Restrict that exception to already-sealed rows;
|
||||
# every open lifecycle state remains owned by the normal state machine.
|
||||
snapshots = snapshots.filter(status=Snapshot.StatusChoices.SEALED)
|
||||
|
||||
stats: dict[str, Any] = {"processed": 0, "requested": 0, "queued": 0, "skipped_queued": 0, "reindexed": 0, "snapshot_ids": []}
|
||||
records: list[dict[str, str]] = []
|
||||
plugins_by_name = discover_plugins(runtime="archivebox")
|
||||
@ -365,7 +370,7 @@ def update(
|
||||
# Snapshot rows: Snapshot.save() moves archive/<ts> to
|
||||
# the current output_dir and preserves the lifecycle
|
||||
# status. Drain those retry_at ticks before queuing
|
||||
# search backfill below. Otherwise the sealed/paused
|
||||
# search backfill below. Otherwise the sealed search
|
||||
# runner branch correctly sees queued ArchiveResult
|
||||
# rows first, runs the targeted plugins, and may leave
|
||||
# the fs_version maintenance tick hidden behind that
|
||||
@ -430,9 +435,7 @@ def update(
|
||||
now = timezone.now()
|
||||
queued_result_for_snapshot = queued_index_results.filter(snapshot_id=OuterRef("pk"))
|
||||
snapshots_to_wake = (
|
||||
Snapshot.objects.filter(
|
||||
status__in=(Snapshot.StatusChoices.SEALED, Snapshot.StatusChoices.PAUSED),
|
||||
)
|
||||
Snapshot.objects.filter(status=Snapshot.StatusChoices.SEALED)
|
||||
.annotate(
|
||||
has_queued_index_result=Exists(queued_result_for_snapshot),
|
||||
)
|
||||
@ -475,8 +478,8 @@ def update(
|
||||
|
||||
if do_run_until_idle and (do_index or not ran_post_migrate_runner):
|
||||
# Search/index backfill intentionally queues targeted
|
||||
# ArchiveResult rows without reopening sealed/paused
|
||||
# snapshots. This second runner pass drains those plugin
|
||||
# ArchiveResult rows without reopening sealed snapshots.
|
||||
# This second runner pass drains those plugin
|
||||
# rows after filesystem maintenance has had its own turn.
|
||||
# For a normal unfiltered `archivebox update`, keep the
|
||||
# historical final pass broad enough to resume genuinely
|
||||
|
||||
@ -695,12 +695,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
|
||||
|
||||
def restore_paused_scheduler_marker(self) -> None:
|
||||
"""
|
||||
Keep explicit maintenance from accidentally resuming paused snapshots.
|
||||
|
||||
Targeted jobs such as `archivebox update --index-only` may bump
|
||||
retry_at so the orchestrator can run only queued search ArchiveResult
|
||||
rows. After that maintenance pass, the lifecycle must remain PAUSED and
|
||||
retry_at must go back to MAX until a real resume transition happens.
|
||||
Restore the indefinite scheduler marker owned by the PAUSED lifecycle.
|
||||
"""
|
||||
type(self).objects.filter(
|
||||
pk=self.pk,
|
||||
|
||||
@ -317,16 +317,17 @@ class CrawlRunner:
|
||||
|
||||
@property
|
||||
def allow_maintenance_on_inactive_crawl(self) -> bool:
|
||||
"""Run the requested hooks on a snapshot whose parent crawl is paused or sealed.
|
||||
"""Run targeted search indexing on an already-sealed snapshot.
|
||||
|
||||
Maintenance entry paths — direct ``snapshot_ids + selected_plugins`` invocations
|
||||
for search backend backfill, fs migration, plugin-targeted updates — are
|
||||
legitimately allowed to operate on finished/paused crawls. Without this gate,
|
||||
``crawl_is_cancelled`` would treat a SEALED parent as a cancellation signal
|
||||
and short-circuit every guard before any hook ran, leaving the queued
|
||||
ArchiveResult rows stuck and the orchestrator looping on them.
|
||||
This is the singular hook-execution exception to the unified lifecycle.
|
||||
All other work must queue the Snapshot and Crawl normally.
|
||||
"""
|
||||
return bool(self.initial_snapshot_ids and self.selected_plugins)
|
||||
return bool(
|
||||
self.initial_snapshot_ids
|
||||
and self.selected_plugins
|
||||
and self.crawl.status == self.crawl.StatusChoices.SEALED
|
||||
and all(plugin.startswith("search_backend_") for plugin in self.selected_plugins),
|
||||
)
|
||||
|
||||
async def run(self) -> None:
|
||||
heartbeat = CrawlHeartbeat(
|
||||
@ -594,13 +595,12 @@ class CrawlRunner:
|
||||
hook.parent.name for event_name in runtime_events for hook in discover_hooks(event_name, config=self.base_config)
|
||||
}
|
||||
self.selected_plugins = sorted(runtime_plugins) or None
|
||||
if self.initial_snapshot_ids:
|
||||
# Direct snapshot maintenance paths are allowed to name paused
|
||||
# snapshots explicitly. The runner still requires selected_plugins
|
||||
# later, so this does not restart the crawl lifecycle.
|
||||
return [str(snapshot_id) for snapshot_id in self.initial_snapshot_ids]
|
||||
if self.crawl.is_paused:
|
||||
return []
|
||||
if self.initial_snapshot_ids:
|
||||
# Explicit ids select normal runnable work, except for the one
|
||||
# sealed-search backfill admitted by allow_maintenance_on_inactive_crawl.
|
||||
return [str(snapshot_id) for snapshot_id in self.initial_snapshot_ids]
|
||||
pending_snapshots = list(
|
||||
self.crawl.snapshot_set.filter(status__in=Snapshot.RUNNABLE_STATES)
|
||||
.filter(retry_at__lte=timezone.now())
|
||||
@ -1603,10 +1603,8 @@ def run_snapshot_maintenance(snapshot_id: str, *, output_dir: Path | None = None
|
||||
# queued ArchiveResult rows, so run it whenever this helper is called.
|
||||
# The only thing queued rows change is the next scheduler value:
|
||||
# - no queued rows left: clear retry_at because maintenance is done
|
||||
# - queued rows remain: leave the Snapshot due so the sealed/paused runner
|
||||
# branch can process those targeted plugin rows on the next tick
|
||||
# This avoids reopening final/paused snapshots while also avoiding stranded
|
||||
# queued ArchiveResults that have no independent scheduler.
|
||||
# - queued rows remain on a sealed Snapshot: leave it due so the search
|
||||
# backfill exception can process them on the next tick
|
||||
current_retry_at = snapshot.retry_at
|
||||
next_retry_at = timezone.now() if has_queued_results else None
|
||||
snapshot.retry_at = next_retry_at
|
||||
@ -1731,62 +1729,12 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo
|
||||
return parent_reconciled
|
||||
|
||||
if snapshot.is_paused:
|
||||
selected_plugins = queued_plugins_for_snapshot(str(snapshot.id))
|
||||
if snapshot.fs_migration_needed and Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds):
|
||||
_runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot)
|
||||
run_snapshot_maintenance(str(snapshot.id))
|
||||
if not selected_plugins:
|
||||
# No targeted plugin rows remain, so put paused snapshots back
|
||||
# behind the indefinite retry_at marker. If queued plugin rows
|
||||
# remain, continue into the targeted plugin path below and let
|
||||
# its finally block restore the paused marker after completion.
|
||||
snapshot.restore_paused_scheduler_marker()
|
||||
return True
|
||||
snapshot.refresh_from_db()
|
||||
if not selected_plugins:
|
||||
# Paused is a real lifecycle state; retry_at=MAX is only the
|
||||
# orchestrator selection marker. If a direct maintenance/update
|
||||
# command bumps retry_at on a paused snapshot but there are no
|
||||
# targeted ArchiveResult rows to run, restore the scheduler marker
|
||||
# without changing status.
|
||||
snapshot.restore_paused_scheduler_marker()
|
||||
return True
|
||||
if not Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds):
|
||||
return False
|
||||
try:
|
||||
_runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot)
|
||||
# Explicit maintenance, e.g. `archivebox update --index-only`, may
|
||||
# need to run search/index hooks for a paused snapshot. That should
|
||||
# not resume the crawl or make unrelated queued work runnable. The
|
||||
# queued ArchiveResult rows are the durable maintenance request, so
|
||||
# run that exact plugin set even when the paused crawl's normal
|
||||
# PLUGINS config names a different extractor surface.
|
||||
run_crawl(
|
||||
str(snapshot.crawl_id),
|
||||
snapshot_ids=[str(snapshot.id)],
|
||||
selected_plugins=selected_plugins,
|
||||
process_discovered_snapshots_inline=True,
|
||||
interactive_interrupts=interactive_interrupts,
|
||||
config_overrides=config_overrides_for_queued_plugins(selected_plugins),
|
||||
selected_plugins_are_explicit=False,
|
||||
)
|
||||
finally:
|
||||
# Targeted plugin rows can complete while the Snapshot remains
|
||||
# paused. Put retry_at back at MAX only after the queued rows are
|
||||
# gone; if a hook was interrupted before projection, keep the
|
||||
# paused row due so the next runner can retry that targeted work
|
||||
# without a user-visible resume transition.
|
||||
if queued_plugins_for_snapshot(str(snapshot.id)):
|
||||
now = timezone.now()
|
||||
type(snapshot).objects.filter(
|
||||
pk=snapshot.pk,
|
||||
status=snapshot.StatusChoices.PAUSED,
|
||||
).update(
|
||||
retry_at=now,
|
||||
modified_at=now,
|
||||
)
|
||||
else:
|
||||
snapshot.restore_paused_scheduler_marker()
|
||||
# Paused work never executes out of band. Preserve the lifecycle marker
|
||||
# until an explicit resume moves it through the normal state machine.
|
||||
from archivebox.core.models import ArchiveResult
|
||||
|
||||
ArchiveResult.pause_queryset(snapshot.archiveresult_set.all())
|
||||
snapshot.restore_paused_scheduler_marker()
|
||||
return True
|
||||
if snapshot.status == Snapshot.StatusChoices.SEALED:
|
||||
if not Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds):
|
||||
@ -1805,6 +1753,13 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo
|
||||
selected_plugins = queued_plugins_for_snapshot(str(snapshot.id))
|
||||
if selected_plugins:
|
||||
search_only_plugins = all(plugin.startswith("search_backend_") for plugin in selected_plugins)
|
||||
if not search_only_plugins:
|
||||
snapshot.update_and_requeue(
|
||||
status=Snapshot.StatusChoices.QUEUED,
|
||||
retry_at=timezone.now(),
|
||||
current_step=0,
|
||||
)
|
||||
return True
|
||||
_runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot)
|
||||
run_crawl(
|
||||
str(snapshot.crawl_id),
|
||||
@ -1853,18 +1808,14 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo
|
||||
return False
|
||||
snapshot.refresh_from_db()
|
||||
if snapshot.status == Snapshot.StatusChoices.QUEUED:
|
||||
has_results = snapshot.archiveresult_set.exists()
|
||||
has_extraction_results = snapshot.archiveresult_set.exclude(plugin__startswith="search_backend_").exists()
|
||||
if has_results and has_extraction_results and snapshot.is_finished_processing():
|
||||
if snapshot.archiveresult_set.exists() and snapshot.is_finished_processing():
|
||||
snapshot.sm.tick()
|
||||
snapshot.refresh_from_db()
|
||||
if snapshot.status == Snapshot.StatusChoices.SEALED:
|
||||
_runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot, status="SEALED")
|
||||
return True
|
||||
# The runner owns queued Snapshot setup. Create missing enabled hook
|
||||
# rows before ticking when the only existing rows are search
|
||||
# maintenance. Otherwise a search backfill on a paused Snapshot can
|
||||
# make queued -> sealed skip the real extraction work after resume.
|
||||
# rows before ticking so queued lifecycle work has a durable hook set.
|
||||
snapshot.create_pending_archiveresults(hooks=snapshot_hooks_for_pending_archiveresults(snapshot))
|
||||
snapshot.sm.tick()
|
||||
snapshot.refresh_from_db()
|
||||
@ -2112,7 +2063,7 @@ def _run_due_queued_plugin_result(
|
||||
status=ArchiveResult.StatusChoices.QUEUED,
|
||||
plugin__in=plugin_names,
|
||||
snapshot__retry_at__lte=now,
|
||||
snapshot__status__in=(Snapshot.StatusChoices.SEALED, Snapshot.StatusChoices.PAUSED),
|
||||
snapshot__status=Snapshot.StatusChoices.SEALED,
|
||||
)
|
||||
.filter(**({"snapshot__crawl_id": crawl_id} if crawl_id else {}))
|
||||
.values("snapshot_id", "snapshot__crawl_id")[:1]
|
||||
@ -2122,15 +2073,6 @@ def _run_due_queued_plugin_result(
|
||||
return False
|
||||
root_crawl_id = str(first_due_results[0]["snapshot__crawl_id"])
|
||||
|
||||
first_due_snapshot = Snapshot.objects.filter(pk=first_due_results[0]["snapshot_id"]).first()
|
||||
if first_due_snapshot and first_due_snapshot.status == Snapshot.StatusChoices.PAUSED:
|
||||
return run_due_snapshot(
|
||||
first_due_snapshot,
|
||||
lock_seconds=lock_seconds,
|
||||
interactive_interrupts=interactive_interrupts,
|
||||
runtime_config=runtime_config,
|
||||
)
|
||||
|
||||
due_snapshots = Snapshot.objects.filter(
|
||||
retry_at__lte=now,
|
||||
status=Snapshot.StatusChoices.SEALED,
|
||||
@ -2284,18 +2226,11 @@ def run_pending_crawls(
|
||||
from archivebox.config.common import get_config
|
||||
from archivebox.crawls.models import Crawl, CrawlSchedule
|
||||
from archivebox.core.models import ArchiveResult, Snapshot
|
||||
from archivebox.plugins.discovery import discover_plugin_configs
|
||||
from archivebox.plugins.hooks import discover_hooks
|
||||
from archivebox.machine.models import Process
|
||||
|
||||
crawl_claim_lock_seconds = 10
|
||||
runtime_config = get_config()
|
||||
plugin_configs = discover_plugin_configs()
|
||||
download_plugin_names = frozenset(
|
||||
plugin_name
|
||||
for plugin_name, plugin_config in plugin_configs.items()
|
||||
if plugin_config.get("output_mimetypes") and not plugin_name.startswith("search_backend_")
|
||||
)
|
||||
last_recovery_at = 0.0
|
||||
last_retention_at = 0.0
|
||||
last_retention_repair_at = 0.0
|
||||
@ -2323,19 +2258,6 @@ def run_pending_crawls(
|
||||
if schedule.is_due(now):
|
||||
schedule.enqueue(queued_at=now)
|
||||
|
||||
# Final-state download rows are always first: they have no parent crawl
|
||||
# scheduler of their own, and leaving them behind makes the global
|
||||
# counters report stale queued work while new crawls continue.
|
||||
if _run_due_queued_plugin_result(
|
||||
download_plugin_names,
|
||||
crawl_id=crawl_id,
|
||||
lock_seconds=60,
|
||||
interactive_interrupts=interactive_interrupts,
|
||||
runtime_config=runtime_config,
|
||||
batch_size=maintenance_batch_size,
|
||||
):
|
||||
continue
|
||||
|
||||
if _fast_forward_same_path_snapshot_fs_versions():
|
||||
continue
|
||||
|
||||
@ -2411,21 +2333,20 @@ def run_pending_crawls(
|
||||
# Final active-state fallback uses only the retry_at scheduler index and
|
||||
# selects an id first. Keep final SEALED rows out of this broad path so
|
||||
# large filesystem/index backfills cannot starve newly queued crawls.
|
||||
due_snapshots = Snapshot.objects.filter(
|
||||
retry_at__lte=timezone.now(),
|
||||
status__in=Snapshot.OPEN_STATES,
|
||||
)
|
||||
if maintenance_only:
|
||||
due_snapshots = due_snapshots.filter(status=Snapshot.StatusChoices.PAUSED)
|
||||
if crawl_id:
|
||||
due_snapshots = due_snapshots.filter(crawl_id=crawl_id)
|
||||
if _run_due_snapshot_query(
|
||||
due_snapshots,
|
||||
lock_seconds=60,
|
||||
interactive_interrupts=interactive_interrupts,
|
||||
runtime_config=runtime_config,
|
||||
):
|
||||
continue
|
||||
if not maintenance_only:
|
||||
due_snapshots = Snapshot.objects.filter(
|
||||
retry_at__lte=timezone.now(),
|
||||
status__in=Snapshot.OPEN_STATES,
|
||||
)
|
||||
if crawl_id:
|
||||
due_snapshots = due_snapshots.filter(crawl_id=crawl_id)
|
||||
if _run_due_snapshot_query(
|
||||
due_snapshots,
|
||||
lock_seconds=60,
|
||||
interactive_interrupts=interactive_interrupts,
|
||||
runtime_config=runtime_config,
|
||||
):
|
||||
continue
|
||||
|
||||
# Search backend selection is live crawl-execution config, not an
|
||||
# installed-plugin list. Old queued rows for a backend that is disabled
|
||||
|
||||
@ -130,22 +130,6 @@ def wait_for_crawl_wget_success_or_sealed(cwd, crawl_id, timeout=240):
|
||||
raise AssertionError(f"timed out waiting for crawl resume completion for crawl {crawl_id}: {latest_state}")
|
||||
|
||||
|
||||
def wait_for_sqlite_index_result(cwd, crawl_id, timeout=45):
|
||||
deadline = time.time() + timeout
|
||||
latest_state = None
|
||||
while time.time() < deadline:
|
||||
latest_state = get_crawl_runtime_state(cwd, crawl_id)
|
||||
final_results = [
|
||||
result
|
||||
for result in latest_state["results"]
|
||||
if result["plugin"] == "search_backend_sqlite" and result["status"] not in {"queued", "started", "paused"}
|
||||
]
|
||||
if final_results:
|
||||
return latest_state
|
||||
time.sleep(0.2)
|
||||
raise AssertionError(f"timed out waiting for sqlite index result for crawl {crawl_id}: {latest_state}")
|
||||
|
||||
|
||||
def seed_paused_crawl(client, cwd: Path, api_token: str, url: str, tag: str) -> tuple[str, str]:
|
||||
from archivebox.services.runner import run_due_snapshot
|
||||
|
||||
@ -434,7 +418,7 @@ def test_crawl_pause_resume_api_survives_server_restart_and_processes_after_resu
|
||||
|
||||
|
||||
@pytest.mark.timeout(420)
|
||||
def test_update_index_only_runs_paused_search_rows_and_resume_later_runs_crawl(client, tmp_path, recursive_test_site):
|
||||
def test_update_index_only_leaves_paused_snapshot_on_normal_lifecycle_path(client, tmp_path, recursive_test_site):
|
||||
init_archive(tmp_path)
|
||||
|
||||
port = get_free_port()
|
||||
@ -464,14 +448,13 @@ def test_update_index_only_runs_paused_search_rows_and_resume_later_runs_crawl(c
|
||||
)
|
||||
assert update_process.returncode == 0, update_process.stderr
|
||||
|
||||
indexed_state = wait_for_sqlite_index_result(tmp_path, crawl_id)
|
||||
indexed_state = get_crawl_runtime_state(tmp_path, crawl_id)
|
||||
assert indexed_state["crawl_status"] == "paused"
|
||||
assert indexed_state["crawl_retry_at"] == indexed_state["retry_at_max"]
|
||||
assert indexed_state["snapshots"][0]["status"] == "paused"
|
||||
assert indexed_state["snapshots"][0]["retry_at"] == indexed_state["retry_at_max"]
|
||||
search_results = [result for result in indexed_state["results"] if result["plugin"] == "search_backend_sqlite"]
|
||||
assert search_results
|
||||
assert any(result["status"] not in {"queued", "started", "paused"} for result in search_results)
|
||||
assert search_results == []
|
||||
|
||||
try:
|
||||
start_archivebox_server(tmp_path, env=env, port=port)
|
||||
|
||||
@ -12,6 +12,84 @@ from archivebox.tests.test_orm_helpers import use_archivebox_db
|
||||
pytestmark = pytest.mark.django_db(transaction=True)
|
||||
|
||||
|
||||
def test_only_sealed_search_backfill_bypasses_snapshot_lifecycle():
|
||||
from archivebox.base_models.models import get_or_create_system_user_pk
|
||||
from archivebox.cli.archivebox_extract import run_plugins
|
||||
from archivebox.core.models import ArchiveResult
|
||||
from archivebox.crawls.models import Crawl
|
||||
|
||||
search_crawl = Crawl.objects.create(
|
||||
urls="https://example.com/search",
|
||||
created_by_id=get_or_create_system_user_pk(),
|
||||
status=Crawl.StatusChoices.SEALED,
|
||||
)
|
||||
search_snapshot = Snapshot.objects.create(
|
||||
url="https://example.com/search",
|
||||
crawl=search_crawl,
|
||||
status=Snapshot.StatusChoices.SEALED,
|
||||
)
|
||||
extract_crawl = Crawl.objects.create(
|
||||
urls="https://example.com/extract",
|
||||
created_by_id=get_or_create_system_user_pk(),
|
||||
status=Crawl.StatusChoices.SEALED,
|
||||
)
|
||||
extract_snapshot = Snapshot.objects.create(
|
||||
url="https://example.com/extract",
|
||||
crawl=extract_crawl,
|
||||
status=Snapshot.StatusChoices.SEALED,
|
||||
)
|
||||
|
||||
assert (
|
||||
run_plugins(
|
||||
args=(),
|
||||
records=[
|
||||
{
|
||||
"type": "ArchiveResult",
|
||||
"snapshot_id": str(search_snapshot.id),
|
||||
"plugin": "search_backend_sqlite",
|
||||
},
|
||||
],
|
||||
wait=False,
|
||||
emit_results=False,
|
||||
show_progress=False,
|
||||
)
|
||||
== 0
|
||||
)
|
||||
search_crawl.refresh_from_db()
|
||||
search_snapshot.refresh_from_db()
|
||||
assert search_crawl.status == Crawl.StatusChoices.SEALED
|
||||
assert search_snapshot.status == Snapshot.StatusChoices.SEALED
|
||||
assert search_snapshot.archiveresult_set.filter(
|
||||
plugin="search_backend_sqlite",
|
||||
status=ArchiveResult.StatusChoices.QUEUED,
|
||||
).exists()
|
||||
|
||||
assert (
|
||||
run_plugins(
|
||||
args=(),
|
||||
records=[
|
||||
{
|
||||
"type": "ArchiveResult",
|
||||
"snapshot_id": str(extract_snapshot.id),
|
||||
"plugin": "wget",
|
||||
},
|
||||
],
|
||||
wait=False,
|
||||
emit_results=False,
|
||||
show_progress=False,
|
||||
)
|
||||
== 0
|
||||
)
|
||||
extract_crawl.refresh_from_db()
|
||||
extract_snapshot.refresh_from_db()
|
||||
assert extract_crawl.status == Crawl.StatusChoices.QUEUED
|
||||
assert extract_snapshot.status == Snapshot.StatusChoices.QUEUED
|
||||
assert extract_snapshot.archiveresult_set.filter(
|
||||
plugin="wget",
|
||||
status=ArchiveResult.StatusChoices.QUEUED,
|
||||
).exists()
|
||||
|
||||
|
||||
def test_update_imports_orphaned_snapshots(tmp_path, initialized_archive):
|
||||
"""Test that archivebox update imports real legacy archive directories."""
|
||||
env = cli_env(disable_extractors=True)
|
||||
@ -67,6 +145,11 @@ def test_reindex_snapshots_resets_existing_search_results_and_reruns_requested_p
|
||||
crawl=crawl,
|
||||
status=Snapshot.StatusChoices.SEALED,
|
||||
)
|
||||
paused_snapshot = Snapshot.objects.create(
|
||||
url="https://example.com/paused",
|
||||
crawl=crawl,
|
||||
status=Snapshot.StatusChoices.PAUSED,
|
||||
)
|
||||
result = ArchiveResult.objects.create(
|
||||
snapshot=snapshot,
|
||||
plugin="search_backend_sqlite",
|
||||
@ -87,7 +170,7 @@ def test_reindex_snapshots_resets_existing_search_results_and_reruns_requested_p
|
||||
os.environ["SEARCH_BACKEND_ENGINE"] = "sqlite"
|
||||
try:
|
||||
stats = reindex_snapshots(
|
||||
Snapshot.objects.filter(id=snapshot.id),
|
||||
Snapshot.objects.filter(id__in=(snapshot.id, paused_snapshot.id)),
|
||||
search_plugins=["search_backend_sqlite"],
|
||||
batch_size=10,
|
||||
)
|
||||
@ -105,6 +188,7 @@ def test_reindex_snapshots_resets_existing_search_results_and_reruns_requested_p
|
||||
assert result.status == ArchiveResult.StatusChoices.QUEUED
|
||||
assert result.output_str == ""
|
||||
assert result.output_json is None
|
||||
assert not paused_snapshot.archiveresult_set.exists()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
|
||||
Loading…
Reference in New Issue
Block a user