From 62fb2c8659b43b679ba4ee526cbcc5f0677b0470 Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Mon, 1 Jun 2026 17:36:52 -0700 Subject: [PATCH] release: archivebox 0.9.34rc30 --- archivebox/config/common.py | 23 +- archivebox/services/archive_result_service.py | 41 ++ archivebox/services/process_service.py | 42 ++ archivebox/services/runner.py | 499 ++++++++++-------- archivebox/services/snapshot_service.py | 40 ++ etc/package.json | 2 +- pyproject.toml | 8 +- 7 files changed, 431 insertions(+), 224 deletions(-) diff --git a/archivebox/config/common.py b/archivebox/config/common.py index 0ab5553c..ff62c342 100644 --- a/archivebox/config/common.py +++ b/archivebox/config/common.py @@ -7,9 +7,10 @@ import os import re import secrets import sys +import time import shutil import inspect -from functools import lru_cache +from functools import lru_cache, wraps from collections.abc import Mapping from datetime import timedelta from typing import Any, ClassVar, cast @@ -41,6 +42,24 @@ _STDERR_CONSOLE = Console(stderr=True) _WARNED_ARCHIVING_CONFIGS: set[tuple[int, bool]] = set() +def _perf_trace(label): + def decorator(func): + @wraps(func) + def wrapper(*args, **kwargs): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + return func(*args, **kwargs) + started_at = time.perf_counter() + try: + return func(*args, **kwargs) + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + return wrapper + + return decorator + + def _legacy_bool(value: object) -> bool | None: if value is None: return None @@ -662,6 +681,7 @@ class ArchiveBoxBaseConfig( frozen.pop(key, None) return frozen + @_perf_trace("archivebox.config.for_crawl_runtime") def for_crawl_runtime( self, *, @@ -946,6 +966,7 @@ def get_request_config(request: Any, *, resolve_plugins: bool = False) -> Archiv return request_config +@_perf_trace("archivebox.config.get_config") def get_config( defaults: ConfigOverrides | None = None, overrides: ConfigOverrides | None = None, diff --git a/archivebox/services/archive_result_service.py b/archivebox/services/archive_result_service.py index 1a2cf798..b60a7cb5 100644 --- a/archivebox/services/archive_result_service.py +++ b/archivebox/services/archive_result_service.py @@ -2,8 +2,12 @@ from __future__ import annotations import asyncio import json +import os +import sys +import time from collections import defaultdict from collections.abc import Iterable +from functools import wraps from pathlib import Path from typing import Any, Protocol, runtime_checkable @@ -18,11 +22,45 @@ from abx_dl.services.base import BaseService from .process_service import parse_event_datetime +def _perf_trace(label): + def decorator(func): + if asyncio.iscoroutinefunction(func): + + @wraps(func) + async def async_wrapper(*args, **kwargs): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + return await func(*args, **kwargs) + started_at = time.perf_counter() + try: + return await func(*args, **kwargs) + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + return async_wrapper + + @wraps(func) + def sync_wrapper(*args, **kwargs): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + return func(*args, **kwargs) + started_at = time.perf_counter() + try: + return func(*args, **kwargs) + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + return sync_wrapper + + return decorator + + @runtime_checkable class ModelDumpable(Protocol): def model_dump(self) -> dict[str, Any]: ... +@_perf_trace("archivebox.ArchiveResultService._collect_output_metadata") def _collect_output_metadata(plugin_dir: Path) -> tuple[dict[str, dict], int, str]: exclude_names = {"stdout.log", "stderr.log", "process.pid", "hook.pid", "listener.pid"} output_files: dict[str, dict] = {} @@ -135,6 +173,7 @@ def _summarize_output_files(output_files: dict[str, dict]) -> tuple[int, str]: return total_size, output_mimetypes +@_perf_trace("archivebox.ArchiveResultService._resolve_output_metadata") def _resolve_output_metadata(raw_output_files: Any, plugin_dir: Path) -> tuple[dict[str, dict], int, str]: normalized_output_files = _normalize_output_files(raw_output_files) if normalized_output_files and _has_structured_output_metadata(normalized_output_files): @@ -215,6 +254,7 @@ class ArchiveResultService(BaseService): self.bus.on(ArchiveResultEvent, self.on_ArchiveResultEvent__save_to_db) self.bus.on(ProcessCompletedEvent, self.on_ProcessCompletedEvent__save_to_db) + @_perf_trace("archivebox.ArchiveResultService.on_ArchiveResultEvent__save_to_db") async def on_ArchiveResultEvent__save_to_db(self, event: ArchiveResultEvent) -> None: from archivebox.core.models import ArchiveResult, Snapshot from archivebox.machine.models import Process @@ -307,6 +347,7 @@ class ArchiveResultService(BaseService): snapshot.title = next_title await snapshot.asave(update_fields=["title", "modified_at"]) + @_perf_trace("archivebox.ArchiveResultService.on_ProcessCompletedEvent__save_to_db") async def on_ProcessCompletedEvent__save_to_db(self, event: ProcessCompletedEvent) -> None: if event.event_id in self._completed_process_event_ids: return diff --git a/archivebox/services/process_service.py b/archivebox/services/process_service.py index 3cd9e26c..0c7640b7 100644 --- a/archivebox/services/process_service.py +++ b/archivebox/services/process_service.py @@ -1,7 +1,12 @@ from __future__ import annotations import asyncio +import os +import sys +import time +from contextlib import contextmanager from datetime import datetime +from functools import wraps from typing import ClassVar from asgiref.sync import sync_to_async @@ -12,6 +17,41 @@ from abx_dl.events import CrawlCleanupEvent, CrawlCompletedEvent, ProcessComplet from abx_dl.services.base import BaseService +def _perf_trace(label): + def decorator(func): + if asyncio.iscoroutinefunction(func): + + @wraps(func) + async def async_wrapper(*args, **kwargs): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + return await func(*args, **kwargs) + started_at = time.perf_counter() + try: + return await func(*args, **kwargs) + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + return async_wrapper + + return func + + return decorator + + +@contextmanager +def _perf_span(label: str): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + yield + return + started_at = time.perf_counter() + try: + yield + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + def parse_event_datetime(value: str | None): if not value: return None @@ -55,6 +95,7 @@ class ProcessService(BaseService): self._iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)() return self._iface + @_perf_trace("archivebox.ProcessService.on_ProcessStartedEvent__save_to_db") async def on_ProcessStartedEvent__save_to_db(self, event: ProcessStartedEvent) -> None: from archivebox.machine.models import Process @@ -153,6 +194,7 @@ class ProcessService(BaseService): async def on_CrawlCompletedEvent__flush_completed(self, event: CrawlCompletedEvent) -> None: await self.flush_completed() + @_perf_trace("archivebox.ProcessService._save_completed_process_to_db") async def _save_completed_process_to_db(self, event: ProcessCompletedEvent) -> None: from archivebox.machine.models import Process diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index c27d2296..3355f255 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -8,7 +8,7 @@ import shutil import sys import threading import time -from contextlib import nullcontext +from contextlib import contextmanager, nullcontext from datetime import timedelta from functools import lru_cache, wraps from pathlib import Path @@ -107,6 +107,19 @@ def _perf_trace(label): return decorator +@contextmanager +def _perf_span(label: str): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + yield + return + started_at = time.perf_counter() + try: + yield + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + def _bus_name(prefix: str, identifier: str) -> str: normalized = "".join(ch if ch.isalnum() else "_" for ch in identifier) return f"{prefix}_{normalized}" @@ -167,6 +180,7 @@ def _is_external_task_cancelled(error: asyncio.CancelledError) -> bool: return not isinstance(error, (EventHandlerAbortedError, EventHandlerCancelledError)) +@_perf_trace("runner._emit_machine_config") async def _emit_machine_config( bus, *, @@ -194,10 +208,14 @@ async def _emit_machine_config( await bus.emit(derived_event).now() +@_perf_trace("runner._run_event_now") async def _run_event_now(event, timeout: float | None = None): - await event.now(timeout=timeout) - await event.wait(timeout=timeout) - await event.event_results_list() + with _perf_span("runner._run_event_now.now"): + await event.now(timeout=timeout) + with _perf_span("runner._run_event_now.wait"): + await event.wait(timeout=timeout) + with _perf_span("runner._run_event_now.results"): + await event.event_results_list() return event @@ -352,6 +370,7 @@ class CrawlRunner: """ return bool(self.initial_snapshot_ids and self.selected_plugins) + @_perf_trace("runner.CrawlRunner.run") async def run(self) -> None: heartbeat = CrawlHeartbeat( Path(self.crawl_output_dir), @@ -726,33 +745,44 @@ class CrawlRunner: ) return live_ui + @_perf_trace("runner.CrawlRunner.load_snapshot_payload") def load_snapshot_payload(self, snapshot_id: str) -> dict[str, Any]: from archivebox.config.common import get_config from archivebox.core.models import Snapshot - snapshot = Snapshot.objects.select_related("crawl", "crawl__created_by").get(id=snapshot_id) + with _perf_span("runner.CrawlRunner.load_snapshot_payload.snapshot_get"): + snapshot = Snapshot.objects.select_related("crawl", "crawl__created_by").get(id=snapshot_id) self.crawl = snapshot.crawl - self.persona = snapshot.crawl.resolve_persona() - self.base_config = get_config(crawl=snapshot.crawl) + with _perf_span("runner.CrawlRunner.load_snapshot_payload.resolve_persona"): + self.persona = snapshot.crawl.resolve_persona() + with _perf_span("runner.CrawlRunner.load_snapshot_payload.get_config"): + self.base_config = get_config(crawl=snapshot.crawl) if self.persona: - self.base_config.update( - self.persona.prepare_runtime_for_crawl( - snapshot.crawl, - chrome_binary=self.base_config["CHROME_BINARY"], - ), + with _perf_span("runner.CrawlRunner.load_snapshot_payload.persona_runtime"): + self.base_config.update( + self.persona.prepare_runtime_for_crawl( + snapshot.crawl, + chrome_binary=self.base_config["CHROME_BINARY"], + ), + ) + with _perf_span("runner.CrawlRunner.load_snapshot_payload.runtime_dirs"): + self.base_config.update(self.config_overrides) + self.crawl_output_dir = str(snapshot.crawl.output_dir) + runtime_chrome_overrides = { + key: self.base_config[key] for key in ("CHROME_USER_DATA_DIR", "CHROME_DOWNLOADS_DIR") if self.base_config.get(key) + } + snapshot_output_dir = str(snapshot.output_dir) + with _perf_span("runner.CrawlRunner.load_snapshot_payload.for_crawl_runtime"): + config = self.base_config.for_crawl_runtime( + crawl=snapshot.crawl, + snapshot=snapshot, + persona=self.persona, + runtime_overrides=runtime_chrome_overrides, + extra_context={"snapshot_id": str(snapshot.id), "snapshot_depth": snapshot.depth}, ) - self.base_config.update(self.config_overrides) - self.crawl_output_dir = str(snapshot.crawl.output_dir) - runtime_chrome_overrides = { - key: self.base_config[key] for key in ("CHROME_USER_DATA_DIR", "CHROME_DOWNLOADS_DIR") if self.base_config.get(key) - } - config = self.base_config.for_crawl_runtime( - crawl=snapshot.crawl, - snapshot=snapshot, - persona=self.persona, - runtime_overrides=runtime_chrome_overrides, - extra_context={"snapshot_id": str(snapshot.id), "snapshot_depth": snapshot.depth}, - ) + with _perf_span("runner.CrawlRunner.load_snapshot_payload.serialize"): + tags = snapshot.tags_str() + normalized_config = normalize_runtime_config(config) return { "id": str(snapshot.id), "url": snapshot.url, @@ -760,11 +790,11 @@ class CrawlRunner: "timestamp": snapshot.timestamp, "bookmarked_at": snapshot.bookmarked_at.isoformat() if snapshot.bookmarked_at else "", "created_at": snapshot.created_at.isoformat() if snapshot.created_at else "", - "tags": snapshot.tags_str(), + "tags": tags, "depth": snapshot.depth, "status": snapshot.status, - "output_dir": str(snapshot.output_dir), - "config": normalize_runtime_config(config), + "output_dir": snapshot_output_dir, + "config": normalized_config, "_snapshot": snapshot, } @@ -805,33 +835,37 @@ class CrawlRunner: if self.process_discovered_snapshots_inline and isinstance(get_current_event(), CrawlStartEvent): await self.enqueue_pending_snapshots_from_projection() + @_perf_trace("runner.CrawlRunner.run_crawl") async def run_crawl(self, root_snapshot_id: str, snapshot_ids: list[str]) -> None: - snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(root_snapshot_id) - config = normalize_runtime_config(snapshot["config"]) - derived_config = normalize_runtime_config(self.derived_config) - output_dir = Path(self.crawl_output_dir) - plugins = self.runtime_plugins() - abx_snapshot = AbxSnapshot( - id=snapshot["id"], - url=snapshot["url"], - depth=int(snapshot["depth"]), - crawl_id=str(self.crawl.id), - ) - setup_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("CrawlSetup")] - crawl_setup_phase_timeout = compute_phase_timeout(setup_hooks, config) - install_phase_timeout = compute_install_phase_timeout(get_install_plugins(plugins), config) - snapshot_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")] - max_snapshot_count = max(1, int(config.get("CRAWL_MAX_URLS") or len(snapshot_ids) or 1)) - snapshot_phase_timeout = compute_phase_timeout(snapshot_hooks, config) * max_snapshot_count - crawl_cleanup_phase_timeout = crawl_setup_phase_timeout - crawl_lifecycle_timeout = ( - crawl_setup_phase_timeout - + snapshot_phase_timeout - + crawl_cleanup_phase_timeout - + CrawlCompletedEvent.model_fields["event_timeout"].default - + 30.0 - ) - await _emit_machine_config(self.bus, config=config, derived_config=derived_config) + with _perf_span("runner.CrawlRunner.run_crawl.load_root_payload"): + snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(root_snapshot_id) + with _perf_span("runner.CrawlRunner.run_crawl.prepare_runtime"): + config = normalize_runtime_config(snapshot["config"]) + derived_config = normalize_runtime_config(self.derived_config) + output_dir = Path(self.crawl_output_dir) + plugins = self.runtime_plugins() + abx_snapshot = AbxSnapshot( + id=snapshot["id"], + url=snapshot["url"], + depth=int(snapshot["depth"]), + crawl_id=str(self.crawl.id), + ) + setup_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("CrawlSetup")] + crawl_setup_phase_timeout = compute_phase_timeout(setup_hooks, config) + install_phase_timeout = compute_install_phase_timeout(get_install_plugins(plugins), config) + snapshot_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")] + max_snapshot_count = max(1, int(config.get("CRAWL_MAX_URLS") or len(snapshot_ids) or 1)) + snapshot_phase_timeout = compute_phase_timeout(snapshot_hooks, config) * max_snapshot_count + crawl_cleanup_phase_timeout = crawl_setup_phase_timeout + crawl_lifecycle_timeout = ( + crawl_setup_phase_timeout + + snapshot_phase_timeout + + crawl_cleanup_phase_timeout + + CrawlCompletedEvent.model_fields["event_timeout"].default + + 30.0 + ) + with _perf_span("runner.CrawlRunner.run_crawl.emit_machine_config"): + await _emit_machine_config(self.bus, config=config, derived_config=derived_config) install_cancel_watcher: asyncio.Task[None] | None = None install_event = self.bus.emit( InstallEvent( @@ -851,38 +885,40 @@ class CrawlRunner: on_archivebox_InstallEvent.__name__ = "on_archivebox_InstallEvent__cancel_watcher" self.bus.on(InstallEvent, on_archivebox_InstallEvent) - setup_abx_services( - self.bus, - plugins=plugins, - url=snapshot["url"], - snapshot=abx_snapshot, - output_dir=output_dir, - install_enabled=False, - crawl_setup_enabled=True, - crawl_event_enabled=False, - crawl_start_enabled=False, - snapshot_cleanup_enabled=False, - crawl_cleanup_enabled=True, - crawl_completed_enabled=False, - crawl_setup_phase_timeout=crawl_setup_phase_timeout, - snapshot_phase_timeout=crawl_setup_phase_timeout, - snapshot_cleanup_phase_timeout=crawl_setup_phase_timeout, - crawl_cleanup_phase_timeout=crawl_setup_phase_timeout, - persist_derived=False, - auto_install=True, - emit_jsonl=False, - abort_requested=self.crawl_is_cancelled, - MachineService=None, - PluginBinariesService=PluginBinariesService, - BinaryCacheService=None, - BinaryService=None, - ProcessService=None, - ArchiveResultService=None, - TagService=None, - SnapshotService=None, - ) + with _perf_span("runner.CrawlRunner.run_crawl.setup_abx_services"): + setup_abx_services( + self.bus, + plugins=plugins, + url=snapshot["url"], + snapshot=abx_snapshot, + output_dir=output_dir, + install_enabled=False, + crawl_setup_enabled=True, + crawl_event_enabled=False, + crawl_start_enabled=False, + snapshot_cleanup_enabled=False, + crawl_cleanup_enabled=True, + crawl_completed_enabled=False, + crawl_setup_phase_timeout=crawl_setup_phase_timeout, + snapshot_phase_timeout=crawl_setup_phase_timeout, + snapshot_cleanup_phase_timeout=crawl_setup_phase_timeout, + crawl_cleanup_phase_timeout=crawl_setup_phase_timeout, + persist_derived=False, + auto_install=True, + emit_jsonl=False, + abort_requested=self.crawl_is_cancelled, + MachineService=None, + PluginBinariesService=PluginBinariesService, + BinaryCacheService=None, + BinaryService=None, + ProcessService=None, + ArchiveResultService=None, + TagService=None, + SnapshotService=None, + ) try: - await _run_event_now(install_event, install_phase_timeout) + with _perf_span("runner.CrawlRunner.run_crawl.install_event"): + await _run_event_now(install_event, install_phase_timeout) finally: if install_cancel_watcher is not None: install_cancel_watcher.cancel() @@ -987,7 +1023,8 @@ class CrawlRunner: event_handler_slow_timeout=slow_warning_timeout(crawl_lifecycle_timeout), ) self.root_crawl_event_id = crawl_event.event_id - await _run_event_now(self.bus.emit(crawl_event), None) + with _perf_span("runner.CrawlRunner.run_crawl.crawl_event"): + await _run_event_now(self.bus.emit(crawl_event), None) if await self.crawl_is_cancelled(): self._skip_wait_until_idle = True return @@ -1020,12 +1057,14 @@ class CrawlRunner: if completed_process.status == "failed": raise RuntimeError(f"Crawl setup hook {plugin.name}:{hook.name} failed") + @_perf_trace("runner.CrawlRunner.run_snapshot") async def run_snapshot(self, snapshot_id: str, crawl_start_event: CrawlStartEvent | None = None) -> None: async with self.snapshot_semaphore: crawl_start_event = crawl_start_event or get_current_event() if not isinstance(crawl_start_event, CrawlStartEvent): raise RuntimeError("Snapshot events must be emitted from a CrawlStartEvent handler") - snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(snapshot_id) + with _perf_span("runner.CrawlRunner.run_snapshot.load_payload"): + snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(snapshot_id) if snapshot["status"] == "sealed" and not self.selected_plugins: await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id) return @@ -1050,34 +1089,37 @@ class CrawlRunner: ): await sync_to_async(self.seal_snapshot_due_to_limit, thread_sensitive=True)(snapshot_id) return - config = normalize_runtime_config(snapshot["config"]) - derived_config = normalize_runtime_config(self.derived_config) - output_dir = Path(snapshot["output_dir"]) - plugins = ( - filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True) - if snapshot_selected_plugins - else self.plugins - ) - abx_snapshot = AbxSnapshot( - id=snapshot["id"], - url=snapshot["url"], - depth=int(snapshot["depth"]), - crawl_id=str(self.crawl.id), - ) - snapshot_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")] - snapshot_phase_timeout = compute_phase_timeout(snapshot_hooks, config) - await _emit_machine_config(self.bus, config=config, derived_config=derived_config, parent_event=crawl_start_event) - snapshot_service = HookSnapshotService( - self.bus, - url=snapshot["url"], - snapshot=abx_snapshot, - output_dir=output_dir, - plugins=plugins, - snapshot_phase_timeout=snapshot_phase_timeout, - snapshot_cleanup_enabled=True, - snapshot_cleanup_phase_timeout=snapshot_phase_timeout, - abort_requested=self.crawl_is_cancelled, - ) + with _perf_span("runner.CrawlRunner.run_snapshot.prepare_runtime"): + config = normalize_runtime_config(snapshot["config"]) + derived_config = normalize_runtime_config(self.derived_config) + output_dir = Path(snapshot["output_dir"]) + plugins = ( + filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True) + if snapshot_selected_plugins + else self.plugins + ) + abx_snapshot = AbxSnapshot( + id=snapshot["id"], + url=snapshot["url"], + depth=int(snapshot["depth"]), + crawl_id=str(self.crawl.id), + ) + snapshot_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")] + snapshot_phase_timeout = compute_phase_timeout(snapshot_hooks, config) + with _perf_span("runner.CrawlRunner.run_snapshot.emit_machine_config"): + await _emit_machine_config(self.bus, config=config, derived_config=derived_config, parent_event=crawl_start_event) + with _perf_span("runner.CrawlRunner.run_snapshot.init_snapshot_service"): + snapshot_service = HookSnapshotService( + self.bus, + url=snapshot["url"], + snapshot=abx_snapshot, + output_dir=output_dir, + plugins=plugins, + snapshot_phase_timeout=snapshot_phase_timeout, + snapshot_cleanup_enabled=True, + snapshot_cleanup_phase_timeout=snapshot_phase_timeout, + abort_requested=self.crawl_is_cancelled, + ) try: snapshot_event = SnapshotEvent( url=snapshot["url"], @@ -1089,42 +1131,49 @@ class CrawlRunner: ) snapshot_event.event_parent_id = crawl_start_event.event_id emitted_snapshot_event = self.bus.emit(snapshot_event) - await _run_event_now(emitted_snapshot_event, snapshot_phase_timeout) - completed_snapshot = await self.bus.find( - SnapshotCompletedEvent, - child_of=emitted_snapshot_event, - past=True, - future=snapshot_phase_timeout, - ) + with _perf_span("runner.CrawlRunner.run_snapshot.snapshot_event"): + await _run_event_now(emitted_snapshot_event, snapshot_phase_timeout) + with _perf_span("runner.CrawlRunner.run_snapshot.find_completed"): + completed_snapshot = await self.bus.find( + SnapshotCompletedEvent, + child_of=emitted_snapshot_event, + past=True, + future=snapshot_phase_timeout, + ) if completed_snapshot is None: raise RuntimeError(f"Snapshot {snapshot_id} did not complete") - await completed_snapshot.wait(timeout=snapshot_phase_timeout) - await completed_snapshot.event_results_list() + with _perf_span("runner.CrawlRunner.run_snapshot.wait_completed"): + await completed_snapshot.wait(timeout=snapshot_phase_timeout) + await completed_snapshot.event_results_list() # SnapshotCompletedEvent is the normal projection path, but the # runner is the scheduler owner. Finalize idempotently here too # so a completed snapshot cannot remain STARTED if the event was # observed before its DB projector advanced the state machine. crawl_limit_stop_reason = CrawlLimitState.from_config(config).get_stop_reason() - await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)( - snapshot_id, - output_dir=output_dir, - crawl_limit_stop_reason=crawl_limit_stop_reason, - ) + with _perf_span("runner.CrawlRunner.run_snapshot.finalize_completed_snapshot"): + await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)( + snapshot_id, + output_dir=output_dir, + crawl_limit_stop_reason=crawl_limit_stop_reason, + ) if snapshot["status"] == "sealed": - await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id, output_dir=output_dir) + with _perf_span("runner.CrawlRunner.run_snapshot.run_snapshot_maintenance"): + await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id, output_dir=output_dir) return - await self.enqueue_discovered_snapshots_from_outputs(snapshot) - await sync_to_async( - lambda: ( - self.crawl.sm.seal() - if self.crawl.status == self.crawl.StatusChoices.STARTED - and not self.crawl.snapshot_set.filter( - status__in=self.crawl.snapshot_set.model.OPEN_STATES, - ).exists() - else None - ), - thread_sensitive=True, - )() + with _perf_span("runner.CrawlRunner.run_snapshot.enqueue_discovered"): + await self.enqueue_discovered_snapshots_from_outputs(snapshot) + with _perf_span("runner.CrawlRunner.run_snapshot.maybe_seal_crawl"): + await sync_to_async( + lambda: ( + self.crawl.sm.seal() + if self.crawl.status == self.crawl.StatusChoices.STARTED + and not self.crawl.snapshot_set.filter( + status__in=self.crawl.snapshot_set.model.OPEN_STATES, + ).exists() + else None + ), + thread_sensitive=True, + )() finally: snapshot_service.close() @@ -1748,6 +1797,7 @@ def _run_due_snapshot_id(snapshot_id, *, lock_seconds: int, interactive_interrup return True +@_perf_trace("runner._run_due_queued_plugin_result") def _run_due_queued_plugin_result( plugin_names: frozenset[str], *, @@ -1761,100 +1811,113 @@ def _run_due_queued_plugin_result( if not plugin_names: return False - now = timezone.now() - queued_results = ArchiveResult.objects.filter( - snapshot_id=OuterRef("pk"), - status=ArchiveResult.StatusChoices.QUEUED, - plugin__in=plugin_names, - ) - first_due_results = list( - ArchiveResult.objects.filter( + with _perf_span("runner._run_due_queued_plugin_result.prepare_queries"): + now = timezone.now() + queued_results = ArchiveResult.objects.filter( + snapshot_id=OuterRef("pk"), status=ArchiveResult.StatusChoices.QUEUED, plugin__in=plugin_names, - snapshot__retry_at__lte=now, - snapshot__status=Snapshot.StatusChoices.SEALED, ) - .filter(**({"snapshot__crawl_id": crawl_id} if crawl_id else {})) - .values("snapshot_id", "snapshot__crawl_id")[:1], - ) + first_due_query = ( + ArchiveResult.objects.filter( + status=ArchiveResult.StatusChoices.QUEUED, + plugin__in=plugin_names, + snapshot__retry_at__lte=now, + snapshot__status=Snapshot.StatusChoices.SEALED, + ) + .filter(**({"snapshot__crawl_id": crawl_id} if crawl_id else {})) + .values("snapshot_id", "snapshot__crawl_id")[:1] + ) + with _perf_span("runner._run_due_queued_plugin_result.first_due_query"): + first_due_results = list(first_due_query) if not first_due_results: return False root_crawl_id = str(first_due_results[0]["snapshot__crawl_id"]) - due_snapshots = Snapshot.objects.filter( - retry_at__lte=now, - status=Snapshot.StatusChoices.SEALED, - ).filter(Exists(queued_results)) - if crawl_id: - due_snapshots = due_snapshots.filter(crawl_id=crawl_id) - - batch_candidates = list( - # The crawl picker above starts from enabled queued ArchiveResult rows - # and uses a sliced LIMIT 1. Do not use QuerySet.first() here: it adds - # ordering and can turn this hot scheduler check into a temp-sort over - # hundreds of thousands of plugin rows. Once a crawl is selected, - # sibling order is irrelevant; the crawl_id/status index can fetch this - # small local batch directly while EXISTS proves the enabled queued - # plugin rows via the existing ArchiveResult unique index. - due_snapshots.filter(crawl_id=root_crawl_id).order_by()[:QUEUED_PLUGIN_RESULT_BATCH_SIZE], - ) + with _perf_span("runner._run_due_queued_plugin_result.batch_query"): + due_snapshots = Snapshot.objects.filter( + retry_at__lte=now, + status=Snapshot.StatusChoices.SEALED, + ).filter(Exists(queued_results)) + if crawl_id: + due_snapshots = due_snapshots.filter(crawl_id=crawl_id) + batch_candidates = list( + # The crawl picker above starts from enabled queued ArchiveResult rows + # and uses a sliced LIMIT 1. Do not use QuerySet.first() here: it adds + # ordering and can turn this hot scheduler check into a temp-sort over + # hundreds of thousands of plugin rows. Once a crawl is selected, + # sibling order is irrelevant; the crawl_id/status index can fetch this + # small local batch directly while EXISTS proves the enabled queued + # plugin rows via the existing ArchiveResult unique index. + due_snapshots.filter(crawl_id=root_crawl_id).order_by()[:QUEUED_PLUGIN_RESULT_BATCH_SIZE], + ) if not batch_candidates: return False selected_plugins: list[str] | None = None claimed_snapshot_ids: list[str] = [] - for snapshot in batch_candidates: - snapshot_selected_plugins = [ - plugin_name for plugin_name in (queued_plugins_for_snapshot(str(snapshot.id)) or []) if plugin_name in plugin_names - ] - if not snapshot_selected_plugins: - continue - if selected_plugins is None: - selected_plugins = snapshot_selected_plugins - if snapshot_selected_plugins != selected_plugins: - continue - if not Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds): - continue - snapshot.refresh_from_db() - snapshot.finalize_completed_upload_results() - if snapshot.fs_migration_needed: - run_snapshot_maintenance(str(snapshot.id)) - snapshot.refresh_from_db() - if snapshot.status != Snapshot.StatusChoices.SEALED: - continue - claimed_snapshot_ids.append(str(snapshot.id)) - _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot) + with _perf_span("runner._run_due_queued_plugin_result.claim_loop"): + for snapshot in batch_candidates: + with _perf_span("runner._run_due_queued_plugin_result.claim_loop.queued_plugins_for_snapshot"): + snapshot_selected_plugins = [ + plugin_name for plugin_name in (queued_plugins_for_snapshot(str(snapshot.id)) or []) if plugin_name in plugin_names + ] + if not snapshot_selected_plugins: + continue + if selected_plugins is None: + selected_plugins = snapshot_selected_plugins + if snapshot_selected_plugins != selected_plugins: + continue + with _perf_span("runner._run_due_queued_plugin_result.claim_loop.claim_for_worker"): + claimed = Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds) + if not claimed: + continue + with _perf_span("runner._run_due_queued_plugin_result.claim_loop.refresh_from_db"): + snapshot.refresh_from_db() + with _perf_span("runner._run_due_queued_plugin_result.claim_loop.finalize_uploads"): + snapshot.finalize_completed_upload_results() + if snapshot.fs_migration_needed: + with _perf_span("runner._run_due_queued_plugin_result.claim_loop.fs_migration"): + run_snapshot_maintenance(str(snapshot.id)) + snapshot.refresh_from_db() + if snapshot.status != Snapshot.StatusChoices.SEALED: + continue + claimed_snapshot_ids.append(str(snapshot.id)) + with _perf_span("runner._run_due_queued_plugin_result.claim_loop.console_line"): + _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot) if not claimed_snapshot_ids or selected_plugins is None: return True - run_crawl( - root_crawl_id, - snapshot_ids=claimed_snapshot_ids, - selected_plugins=selected_plugins, - process_discovered_snapshots_inline=True, - interactive_interrupts=interactive_interrupts, - config_overrides={ - "CRAWL_MAX_CONCURRENT_SNAPSHOTS": QUEUED_PLUGIN_RESULT_BATCH_SIZE, - }, - ) + with _perf_span("runner._run_due_queued_plugin_result.run_crawl"): + run_crawl( + root_crawl_id, + snapshot_ids=claimed_snapshot_ids, + selected_plugins=selected_plugins, + process_discovered_snapshots_inline=True, + interactive_interrupts=interactive_interrupts, + config_overrides={ + "CRAWL_MAX_CONCURRENT_SNAPSHOTS": QUEUED_PLUGIN_RESULT_BATCH_SIZE, + }, + ) if all(plugin.startswith("search_backend_") for plugin in selected_plugins): - queued_results = ArchiveResult.objects.filter( - snapshot_id=OuterRef("pk"), - status=ArchiveResult.StatusChoices.QUEUED, - plugin__in=selected_plugins, - ) - Snapshot.objects.filter( - id__in=claimed_snapshot_ids, - status=Snapshot.StatusChoices.SEALED, - ).annotate( - has_queued_results=Exists(queued_results), - ).filter( - has_queued_results=False, - ).update( - retry_at=None, - modified_at=timezone.now(), - ) + with _perf_span("runner._run_due_queued_plugin_result.clear_completed_search"): + queued_results = ArchiveResult.objects.filter( + snapshot_id=OuterRef("pk"), + status=ArchiveResult.StatusChoices.QUEUED, + plugin__in=selected_plugins, + ) + Snapshot.objects.filter( + id__in=claimed_snapshot_ids, + status=Snapshot.StatusChoices.SEALED, + ).annotate( + has_queued_results=Exists(queued_results), + ).filter( + has_queued_results=False, + ).update( + retry_at=None, + modified_at=timezone.now(), + ) return True diff --git a/archivebox/services/snapshot_service.py b/archivebox/services/snapshot_service.py index 8ae63c82..1e11c822 100644 --- a/archivebox/services/snapshot_service.py +++ b/archivebox/services/snapshot_service.py @@ -1,6 +1,10 @@ from __future__ import annotations +import asyncio import sys +import os +import time +from functools import wraps from asgiref.sync import sync_to_async from django.utils import timezone @@ -11,6 +15,40 @@ from abx_dl.limits import CrawlLimitState from abx_dl.services.base import BaseService +def _perf_trace(label): + def decorator(func): + if asyncio.iscoroutinefunction(func): + + @wraps(func) + async def async_wrapper(*args, **kwargs): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + return await func(*args, **kwargs) + started_at = time.perf_counter() + try: + return await func(*args, **kwargs) + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + return async_wrapper + + @wraps(func) + def sync_wrapper(*args, **kwargs): + if os.environ.get("ARCHIVEBOX_PERF_TRACE") != "1": + return func(*args, **kwargs) + started_at = time.perf_counter() + try: + return func(*args, **kwargs) + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + print(f"PERF_TRACE label={label} ms={elapsed_ms:.3f}", file=sys.stderr, flush=True) + + return sync_wrapper + + return decorator + + +@_perf_trace("archivebox.SnapshotService.finalize_completed_snapshot") def finalize_completed_snapshot( snapshot_id: str, *, @@ -70,6 +108,7 @@ class SnapshotService(BaseService): self.bus.on(SnapshotEvent, self.on_SnapshotEvent) self.bus.on(SnapshotCompletedEvent, self.on_SnapshotCompletedEvent) + @_perf_trace("archivebox.SnapshotService.on_SnapshotEvent") async def on_SnapshotEvent(self, event: SnapshotEvent) -> None: from archivebox.core.models import Snapshot @@ -101,5 +140,6 @@ class SnapshotService(BaseService): return await sync_to_async(snapshot.ensure_crawl_symlink, thread_sensitive=True)() + @_perf_trace("archivebox.SnapshotService.on_SnapshotCompletedEvent") async def on_SnapshotCompletedEvent(self, event: SnapshotCompletedEvent) -> None: await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)(event.snapshot_id) diff --git a/etc/package.json b/etc/package.json index 2b1f4c3f..9b037d22 100644 --- a/etc/package.json +++ b/etc/package.json @@ -1,6 +1,6 @@ { "name": "archivebox", - "version": "0.9.34rc29", + "version": "0.9.34rc30", "repository": "github:ArchiveBox/ArchiveBox", "license": "MIT", "dependencies": { diff --git a/pyproject.toml b/pyproject.toml index 6b9e29fe..e96d0865 100755 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "archivebox" -version = "0.9.34rc29" +version = "0.9.34rc30" requires-python = ">=3.13" description = "Self-hosted internet archiving solution." authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}] @@ -79,9 +79,9 @@ dependencies = [ ### Extractor dependencies (optional binaries detected at runtime via shutil.which) ### Binary/Package Management "abxbus==2.5.9", # EventBus API - "abxpkg>=1.11.143", # for: detecting, versioning, and installing binaries via apt/brew/pip/npm - "abx-plugins>=1.11.146", # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring - "abx-dl>=1.11.146", # shared ArchiveBox downloader package with blocking install preflight + "abxpkg>=1.11.144", # for: detecting, versioning, and installing binaries via apt/brew/pip/npm + "abx-plugins>=1.11.147", # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring + "abx-dl>=1.11.147", # shared ArchiveBox downloader package with blocking install preflight ### UUID7 backport for Python <3.14 "uuid7>=0.1.0; python_version < '3.14'", # provides the uuid_extensions module on Python 3.13 ]