release: archivebox 0.9.34rc30

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,

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

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
 

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
 
 

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)

etc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "archivebox",
-  "version": "0.9.34rc29",
+  "version": "0.9.34rc30",
   "repository": "github:ArchiveBox/ArchiveBox",
   "license": "MIT",
   "dependencies": {

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
 ]