diff --git a/archivebox/cli/archivebox_extract.py b/archivebox/cli/archivebox_extract.py index aec80e60..f1dd0bb3 100644 --- a/archivebox/cli/archivebox_extract.py +++ b/archivebox/cli/archivebox_extract.py @@ -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"): diff --git a/archivebox/cli/archivebox_run.py b/archivebox/cli/archivebox_run.py index c14beb9d..ce40ce7d 100644 --- a/archivebox/cli/archivebox_run.py +++ b/archivebox/cli/archivebox_run.py @@ -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, diff --git a/archivebox/cli/archivebox_update.py b/archivebox/cli/archivebox_update.py index c978f2a1..f1b60fe5 100644 --- a/archivebox/cli/archivebox_update.py +++ b/archivebox/cli/archivebox_update.py @@ -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/ 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 diff --git a/archivebox/core/models.py b/archivebox/core/models.py index f1d40eba..7f6673f5 100755 --- a/archivebox/core/models.py +++ b/archivebox/core/models.py @@ -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, diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 226bd6ae..16c78cf1 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -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 diff --git a/archivebox/tests/test_api_v1_crawls_crawl_crawl_id.py b/archivebox/tests/test_api_v1_crawls_crawl_crawl_id.py index 0f399e81..5317bc54 100644 --- a/archivebox/tests/test_api_v1_crawls_crawl_crawl_id.py +++ b/archivebox/tests/test_api_v1_crawls_crawl_crawl_id.py @@ -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) diff --git a/archivebox/tests/test_cli_update_reindex_snapshots.py b/archivebox/tests/test_cli_update_reindex_snapshots.py index 411e9be2..07e89dfa 100644 --- a/archivebox/tests/test_cli_update_reindex_snapshots.py +++ b/archivebox/tests/test_cli_update_reindex_snapshots.py @@ -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