The_Great_Archive/ArchiveBox
1
0
Fork 0
mirror of https://github.com/ArchiveBox/ArchiveBox.git synced 2026-09-14 11:06:13 +05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Finalize parser output before sealing snapshots

Browse Source
This commit is contained in:
Nick Sweeting 2026-07-19 22:57:52 -07:00
parent 5bc028eb76
commit 2cd0c88cd0
No known key found for this signature in database
7 changed files with 89 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
archivebox/machine/models.py
View File

@ -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:

56
archivebox/services/runner.py
View File

@ -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")

35
archivebox/services/snapshot_service.py
View File

@ -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"])

14
archivebox/tests/test_api_v1_cli_add.py
View File

@ -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)

46
archivebox/tests/test_cli_run.py
View File

@ -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

1
archivebox/tests/test_crawl_runner.py
View File

@ -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

1
archivebox/tests/test_machine_models.py
View File

@ -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,