From 2cd0c88cd097b4c4ffa6ecf0cf17398c2ce67cc9 Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Sun, 19 Jul 2026 22:57:52 -0700 Subject: [PATCH] Finalize parser output before sealing snapshots --- archivebox/machine/models.py | 4 +- archivebox/services/runner.py | 56 ++----------------------- archivebox/services/snapshot_service.py | 35 ++++++++++++++++ archivebox/tests/test_api_v1_cli_add.py | 14 ------- archivebox/tests/test_cli_run.py | 46 ++++++++++++++++++++ archivebox/tests/test_crawl_runner.py | 1 + archivebox/tests/test_machine_models.py | 1 + 7 files changed, 89 insertions(+), 68 deletions(-) diff --git a/archivebox/machine/models.py b/archivebox/machine/models.py index 41c268d9..253dfc2e 100755 --- a/archivebox/machine/models.py +++ b/archivebox/machine/models.py @@ -1575,7 +1575,7 @@ class Process(ModelWithDeleteAfter, models.Model): Processes are stale if: - Status is RUNNING but OS process no longer exists - - Status is RUNNING but exceeded its timeout plus a small grace margin + - A bounded HOOK or BINARY process exceeded its timeout plus a small grace margin - Status is RUNNING but started_at is older than PID_REUSE_WINDOW Returns count of processes cleaned up. @@ -1595,7 +1595,7 @@ class Process(ModelWithDeleteAfter, models.Model): is_stale = False - if proc.started_at: + if proc.started_at and proc.process_type in (cls.TypeChoices.HOOK, cls.TypeChoices.BINARY): timeout_seconds = max(int(proc.timeout or 0), 0) timeout_deadline = proc.started_at + timedelta(seconds=timeout_seconds) + PROCESS_TIMEOUT_GRACE if timeout_seconds > 0 and timezone.now() >= timeout_deadline: diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 4d5ebb9c..603badcf 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -70,7 +70,7 @@ from .binary_service import ArchiveBoxBinaryService, ArchiveBoxDBBinaryCacheBack from .crawl_service import CrawlService from .machine_service import MachineService from .process_service import ProcessService as PersistedProcessService -from .snapshot_service import SnapshotService, finalize_completed_snapshot +from .snapshot_service import SnapshotService, finalize_completed_snapshot, project_discovered_snapshots from .tag_service import TagService @@ -829,55 +829,7 @@ class CrawlRunner: } async def enqueue_discovered_snapshots_from_outputs(self, snapshot_payload: dict[str, Any]) -> None: - from archivebox.core.models import Snapshot - from archivebox.config.common import get_config - from archivebox.plugins.hooks import collect_urls_from_plugins - - await sync_to_async(self.crawl.refresh_from_db, thread_sensitive=True)() - if self.crawl.is_paused and not self.allow_maintenance_on_inactive_crawl: - return - if int(snapshot_payload["depth"]) >= self.crawl.max_depth: - return - - discovered_urls = await sync_to_async(collect_urls_from_plugins, thread_sensitive=True)(Path(snapshot_payload["output_dir"])) - if not discovered_urls: - return - - if self.crawl.status == self.crawl.StatusChoices.SEALED: - # Snapshot completion projectors can observe the root snapshot seal - # before the runner has consumed parser urls.jsonl output. A sealed - # crawl must not block those freshly discovered child snapshots; the - # runner is still inside the same crawl lifecycle and will seal it - # again after the discovered queue is empty. - await sync_to_async(self.crawl.update_and_requeue, thread_sensitive=True)( - status=self.crawl.StatusChoices.STARTED, - retry_at=timezone.now(), - ) - - parent_snapshot = await sync_to_async( - lambda: Snapshot.objects.select_related("crawl", "crawl__created_by").filter(id=snapshot_payload["id"]).first(), - thread_sensitive=True, - )() - if parent_snapshot is None: - return - config = await sync_to_async( - lambda: get_config(crawl=self.crawl, snapshot=parent_snapshot).for_crawl_runtime( - crawl=self.crawl, - snapshot=parent_snapshot, - persona=self.crawl.resolve_persona(), - crawl_output_dir=self.crawl.output_dir, - snapshot_output_dir=parent_snapshot.output_dir, - ), - thread_sensitive=True, - )() - if CrawlLimitState.from_config(config).get_stop_reason() in ("crawl_max_size", "crawl_timeout"): - return - - await sync_to_async(self.crawl.create_discovered_snapshots, thread_sensitive=True)( - parent_snapshot, - discovered_urls, - depth=parent_snapshot.depth + 1, - ) + await sync_to_async(project_discovered_snapshots, thread_sensitive=True)(snapshot_payload["id"]) if self.process_discovered_snapshots_inline and isinstance(get_current_event(), CrawlStartEvent): await self.enqueue_pending_snapshots_from_projection() @@ -1811,7 +1763,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo if snapshot.status == Snapshot.StatusChoices.QUEUED: has_archiveresults = snapshot.archiveresult_set.exists() if has_archiveresults and snapshot.is_finished_processing(): - snapshot.sm.tick() + finalize_completed_snapshot(str(snapshot.id), output_dir=Path(snapshot.output_dir)) snapshot.refresh_from_db() if snapshot.status == Snapshot.StatusChoices.SEALED: _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot, status="SEALED") @@ -1827,7 +1779,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot, status="SEALED") return True if snapshot.status == Snapshot.StatusChoices.STARTED and snapshot.archiveresult_set.exists() and snapshot.is_finished_processing(): - snapshot.sm.tick() + finalize_completed_snapshot(str(snapshot.id), output_dir=Path(snapshot.output_dir)) snapshot.refresh_from_db() if snapshot.status == Snapshot.StatusChoices.SEALED: _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot, status="SEALED") diff --git a/archivebox/services/snapshot_service.py b/archivebox/services/snapshot_service.py index 8ae63c82..1934596d 100644 --- a/archivebox/services/snapshot_service.py +++ b/archivebox/services/snapshot_service.py @@ -1,6 +1,7 @@ from __future__ import annotations import sys +from pathlib import Path from asgiref.sync import sync_to_async from django.utils import timezone @@ -11,6 +12,35 @@ from abx_dl.limits import CrawlLimitState from abx_dl.services.base import BaseService +def project_discovered_snapshots(snapshot_id: str) -> list: + """Persist parser output before the parent Snapshot can enter a final state.""" + from archivebox.config.common import get_config + from archivebox.core.models import Snapshot + from archivebox.plugins.hooks import collect_urls_from_plugins + + snapshot = Snapshot.objects.select_related("crawl", "crawl__created_by", "crawl__persona").filter(id=snapshot_id).first() + if snapshot is None: + return [] + crawl = snapshot.crawl + if crawl.status not in crawl.RUNNABLE_STATES or crawl.is_paused or snapshot.depth >= crawl.max_depth: + return [] + + discovered_urls = collect_urls_from_plugins(Path(snapshot.output_dir)) + if not discovered_urls: + return [] + + config = get_config(crawl=crawl, snapshot=snapshot).for_crawl_runtime( + crawl=crawl, + snapshot=snapshot, + persona=crawl.resolve_persona(), + crawl_output_dir=crawl.output_dir, + snapshot_output_dir=snapshot.output_dir, + ) + if CrawlLimitState.from_config(config).get_stop_reason() in ("crawl_max_size", "crawl_timeout"): + return [] + return crawl.create_discovered_snapshots(snapshot, discovered_urls, depth=snapshot.depth + 1) + + def finalize_completed_snapshot( snapshot_id: str, *, @@ -23,6 +53,11 @@ def finalize_completed_snapshot( if snapshot is None: return + # urls.jsonl is durable hook output. Project it while the Snapshot/Crawl are + # still runnable so an interruption cannot seal a parser root after its + # ArchiveResults finish but before its discovered child rows are persisted. + project_discovered_snapshots(str(snapshot.id)) + if snapshot.downloaded_at is None: snapshot.downloaded_at = timezone.now() snapshot.save(update_fields=["downloaded_at", "modified_at"]) diff --git a/archivebox/tests/test_api_v1_cli_add.py b/archivebox/tests/test_api_v1_cli_add.py index 2f0ee72c..486f6ab5 100644 --- a/archivebox/tests/test_api_v1_cli_add.py +++ b/archivebox/tests/test_api_v1_cli_add.py @@ -147,19 +147,6 @@ IMPORT_FORMAT_ENV = { } -def wait_for_import_processing(cwd: Path, expected_urls: set[str], *, timeout: float = 120.0) -> None: - import time - - deadline = time.time() + timeout - while time.time() < deadline: - with use_archivebox_db(cwd): - snapshot_started = Snapshot.objects.filter(url__in=expected_urls).exists() - if snapshot_started: - return - time.sleep(1) - raise AssertionError("timed out waiting for import crawl processing to start") - - def wait_for_expected_import_snapshots( cwd: Path, expected_urls: set[str], @@ -372,7 +359,6 @@ def test_api_cli_add_import_text_formats_preserve_metadata_and_crawl_inner_urls( break time.sleep(1) assert root_counts and all(count == 1 for count in root_counts.values()), root_counts - wait_for_import_processing(tmp_path, expected_urls) with use_archivebox_db(tmp_path): for crawl in Crawl.objects.all(): root_snapshot = crawl.snapshot_set.get(url=Snapshot.INTERNAL_INPUT_URL) diff --git a/archivebox/tests/test_cli_run.py b/archivebox/tests/test_cli_run.py index b1f0507b..41e763bb 100644 --- a/archivebox/tests/test_cli_run.py +++ b/archivebox/tests/test_cli_run.py @@ -1911,6 +1911,52 @@ class TestRunDueCrawlState: assert finished.output_str == "keep me" assert finished.output_files == {"favicon.ico": {"size": 1}} + def test_finished_parser_result_projects_children_before_resume_seals_snapshot(self): + import json + + 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="Plain text import containing https://example.org/\n", + max_depth=1, + created_by_id=get_or_create_system_user_pk(), + status=Crawl.StatusChoices.STARTED, + retry_at=timezone.now(), + ) + root = Snapshot.objects.create( + url=Snapshot.INTERNAL_INPUT_URL, + crawl=crawl, + depth=0, + status=Snapshot.StatusChoices.STARTED, + retry_at=timezone.now(), + ) + parser_dir = root.output_dir / "parse_txt_urls" + parser_dir.mkdir(parents=True, exist_ok=True) + (parser_dir / "urls.jsonl").write_text( + json.dumps({"type": "Snapshot", "url": "https://example.org/"}) + "\n", + encoding="utf-8", + ) + ArchiveResult.objects.create( + snapshot=root, + plugin="parse_txt_urls", + hook_name="on_Snapshot__71_parse_txt_urls", + status=ArchiveResult.StatusChoices.SUCCEEDED, + output_files={"urls.jsonl": {"size": (parser_dir / "urls.jsonl").stat().st_size}}, + ) + + assert run_due_snapshot(root, lock_seconds=60) + + root.refresh_from_db() + child = Snapshot.objects.get(crawl=crawl, url="https://example.org/") + assert root.status == Snapshot.StatusChoices.SEALED + assert child.parent_snapshot_id == root.id + assert child.status == Snapshot.StatusChoices.QUEUED + def test_due_started_snapshot_with_live_child_extends_lease_without_reset(self): import os from datetime import datetime diff --git a/archivebox/tests/test_crawl_runner.py b/archivebox/tests/test_crawl_runner.py index ee355e86..160e23b8 100644 --- a/archivebox/tests/test_crawl_runner.py +++ b/archivebox/tests/test_crawl_runner.py @@ -167,6 +167,7 @@ def test_ensure_background_runner_skips_with_real_running_orchestrator_record(): status=Process.StatusChoices.RUNNING, pid=os.getpid(), started_at=datetime.fromtimestamp(os_proc.create_time(), tz=timezone.get_current_timezone()), + timeout=1, ) assert ensure_background_runner(allow_under_pytest=True) is False diff --git a/archivebox/tests/test_machine_models.py b/archivebox/tests/test_machine_models.py index 421b460d..ea6068ea 100644 --- a/archivebox/tests/test_machine_models.py +++ b/archivebox/tests/test_machine_models.py @@ -898,6 +898,7 @@ class TestProcessClassMethods: """cleanup_stale_running should retire RUNNING rows that exceed timeout + grace.""" stale = Process.objects.create( machine=self.machine, + process_type=Process.TypeChoices.HOOK, status=Process.StatusChoices.RUNNING, pid=999998, timeout=5,