From 72a67bd511fb16c04333a6d093fef903f0f28d5c Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Mon, 1 Jun 2026 01:59:40 -0700 Subject: [PATCH] Project abxpkg binary events --- archivebox/cli/archivebox_pluginmap.py | 16 +++- archivebox/machine/models.py | 2 +- archivebox/plugins/hooks.py | 18 ++-- archivebox/progressmonitor/views.py | 2 - archivebox/services/__init__.py | 5 +- archivebox/services/binary_service.py | 93 ++++++++++++++++++- archivebox/services/process_service.py | 8 +- archivebox/services/runner.py | 25 ++--- .../tests/test_archive_result_service.py | 17 ++-- archivebox/tests/test_hooks.py | 49 ++++------ 10 files changed, 155 insertions(+), 80 deletions(-) diff --git a/archivebox/cli/archivebox_pluginmap.py b/archivebox/cli/archivebox_pluginmap.py index a0143b5a..e63c1272 100644 --- a/archivebox/cli/archivebox_pluginmap.py +++ b/archivebox/cli/archivebox_pluginmap.py @@ -16,8 +16,9 @@ EVENT_FLOW_DIAGRAM = """ │ InstallEvent │ │ └─ config.json > required_binaries │ │ └─ BinaryRequestEvent │ -│ └─ on_BinaryRequest__* │ +│ └─ abxpkg BinaryService builtin providers │ │ └─ BinaryEvent │ +│ └─ BinaryCacheService / project cache backend │ │ │ │ CrawlEvent │ │ └─ CrawlSetupEvent │ @@ -75,12 +76,14 @@ def pluginmap( "emits": ["BinaryRequestEvent", "BinaryEvent", "ProcessEvent"], }, "BinaryRequestEvent": { - "description": "Provider phase. on_BinaryRequest hooks resolve or install requested binaries.", + "description": "Binary resolution phase. abxpkg BinaryService resolves or installs requested binaries using built-in providers.", "emits": ["BinaryEvent", "ProcessEvent"], + "direct_hooks": False, }, "BinaryEvent": { - "description": "Resolved binary metadata event. Projected into the DB binary cache.", + "description": "Resolved binary metadata event. ArchiveBoxBinaryService projects it into the ArchiveBox DB binary cache.", "emits": [], + "direct_hooks": False, }, "CrawlEvent": { "description": "Root crawl lifecycle event emitted by the runner.", @@ -140,7 +143,8 @@ def pluginmap( for event_name, info in event_phases.items(): hook_event = normalize_hook_event_name(event_name) - hooks = discover_hooks(event_name, filter_disabled=not show_disabled) + has_direct_hooks = info.get("direct_hooks", True) + hooks = discover_hooks(event_name, filter_disabled=not show_disabled) if has_direct_hooks else [] hook_infos = [] for hook_path in hooks: @@ -194,8 +198,10 @@ def pluginmap( prnt(f"[dim]{info['description']}[/dim]") if info["emits"]: prnt(f"[dim]Emits: {', '.join(info['emits'])}[/dim]") - if not hook_infos: + if not hook_infos and has_direct_hooks: prnt(f"[dim]No direct on_{hook_event}__* scripts are currently defined for this event family.[/dim]") + elif not has_direct_hooks: + prnt("[dim]No direct plugin hook family. This event is handled by services.[/dim]") prnt() if not quiet: diff --git a/archivebox/machine/models.py b/archivebox/machine/models.py index 713983eb..a457ecfb 100755 --- a/archivebox/machine/models.py +++ b/archivebox/machine/models.py @@ -691,7 +691,7 @@ class Binary(ModelWithHealthStats, ModelWithStateMachine): ) return binary - # Case 3: From on_BinaryRequest__ hook output - update with installation results + # Case 3: From Binary output - update with installation results if abspath and version: binary, _ = Binary.objects.update_or_create( machine=machine, diff --git a/archivebox/plugins/hooks.py b/archivebox/plugins/hooks.py index 104c11c9..c3118557 100644 --- a/archivebox/plugins/hooks.py +++ b/archivebox/plugins/hooks.py @@ -9,7 +9,6 @@ ArchiveBox no longer drives plugin execution itself during normal crawls. - parses hook stdout JSONL records into ArchiveBox models when needed Hook-backed event families are discovered from filenames like: - on_BinaryRequest__* on_CrawlSetup__* on_Snapshot__* @@ -114,7 +113,6 @@ def normalize_hook_event_name(event_name: str) -> str | None: Normalize a hook event family or event class name to its on_* prefix. Examples: - BinaryRequestEvent -> BinaryRequest CrawlSetupEvent -> CrawlSetup SnapshotEvent -> Snapshot BinaryEvent -> Binary @@ -168,7 +166,7 @@ def discover_hooks( Args: event_name: Hook event family or event class name. - Examples: 'BinaryRequestEvent', 'Snapshot'. + Examples: 'CrawlSetupEvent', 'Snapshot'. Event names are normalized by stripping a trailing `Event`. If no matching `on_{EventFamily}__*` scripts exist, returns []. filter_disabled: If True, skip hooks from disabled plugins (default: True) @@ -196,6 +194,8 @@ def discover_hooks( hook_event_name = normalize_hook_event_name(event_name) if not hook_event_name: return [] + if hook_event_name == "BinaryRequest": + return [] hooks = [] @@ -212,11 +212,7 @@ def discover_hooks( pattern_direct = f"on_{hook_event_name}__*.{ext}" hooks.extend(base_dir.glob(pattern_direct)) - # Binary provider hooks are not end-user extractors. They - # self-filter via `binproviders`, so applying the PLUGINS whitelist here - # can hide the very installer needed by a selected plugin (e.g. - # `--plugins=singlefile` still needs the `npm` BinaryRequest hook). - if filter_disabled and hook_event_name != "BinaryRequest": + if filter_disabled: # Get merged config if not provided (lazy import to avoid circular dependency) if config is None: from archivebox.config.common import get_config @@ -576,8 +572,8 @@ def process_hook_records(records: list[dict[str, Any]], overrides: dict[str, Any """ Process JSONL records emitted by hook stdout. - This handles hook-emitted record types such as Snapshot, Tag, BinaryRequest, - and Binary. It does not process internal bus lifecycle events, since those + This handles hook-emitted record types such as Snapshot, Tag, and Binary. + It does not process internal bus lifecycle events, since those are not emitted as JSONL records by hook subprocesses. Args: @@ -630,7 +626,7 @@ def process_hook_records(records: list[dict[str, Any]], overrides: dict[str, Any if obj: stats["Tag"] = stats.get("Tag", 0) + 1 - elif record_type in {"BinaryRequest", "Binary"}: + elif record_type == "Binary": from archivebox.machine.models import Binary obj = Binary.from_json(record.copy(), overrides) diff --git a/archivebox/progressmonitor/views.py b/archivebox/progressmonitor/views.py index b6c3eff9..c6a3086d 100644 --- a/archivebox/progressmonitor/views.py +++ b/archivebox/progressmonitor/views.py @@ -118,8 +118,6 @@ def live_progress_view(request): phase = "crawl" elif normalized_hook_name.startswith("on_Snapshot__"): phase = "snapshot" - elif normalized_hook_name.startswith("on_BinaryRequest__"): - phase = "binary" label = normalized_hook_name if "__" in normalized_hook_name: diff --git a/archivebox/services/__init__.py b/archivebox/services/__init__.py index a4f1d711..7709e39a 100644 --- a/archivebox/services/__init__.py +++ b/archivebox/services/__init__.py @@ -1,5 +1,5 @@ from .archive_result_service import ArchiveResultService -from .binary_service import ArchiveBoxBinaryCacheBackend +from .binary_service import ArchiveBoxBinaryService, ArchiveBoxDBBinaryCacheBackend from .crawl_service import CrawlService from .machine_service import MachineService from .process_service import ProcessService @@ -9,7 +9,8 @@ from .tag_service import TagService __all__ = [ "ArchiveResultService", - "ArchiveBoxBinaryCacheBackend", + "ArchiveBoxBinaryService", + "ArchiveBoxDBBinaryCacheBackend", "CrawlService", "MachineService", "ProcessService", diff --git a/archivebox/services/binary_service.py b/archivebox/services/binary_service.py index d11cd17f..8e62f4e8 100644 --- a/archivebox/services/binary_service.py +++ b/archivebox/services/binary_service.py @@ -1,13 +1,17 @@ from __future__ import annotations +import json from pathlib import Path from typing import Any from asgiref.sync import sync_to_async +from django.utils import timezone from abxpkg import Binary as AbxBinary from abxpkg import BinProvider, PROVIDER_CLASS_BY_NAME -from abxpkg.binary_service import BinaryRequestEvent +from abxpkg.binary_service import BinaryEvent, BinaryRequestEvent +from abxbus import BaseEvent, EventBus +from abx_dl.services.base import BaseService _LIB_DIR_MANAGED_PROVIDERS = { @@ -16,6 +20,7 @@ _LIB_DIR_MANAGED_PROVIDERS = { "deno", "gem", "goget", + "chromewebstore", "nix", "npm", "pip", @@ -24,7 +29,7 @@ _LIB_DIR_MANAGED_PROVIDERS = { } -class ArchiveBoxBinaryCacheBackend: +class ArchiveBoxDBBinaryCacheBackend: """ArchiveBox machine.Binary projection backend for abxpkg BinaryCacheService.""" async def get(self, request: BinaryRequestEvent) -> AbxBinary | None: @@ -175,6 +180,90 @@ class ArchiveBoxBinaryCacheBackend: await installed.asave(update_fields=["status", "retry_at", "modified_at"]) +class ArchiveBoxBinaryService(BaseService): + """Preserve ArchiveBox's legacy Binary Process rows around abxpkg requests.""" + + LISTENS_TO = [BinaryRequestEvent] + EMITS: list[type[BaseEvent]] = [] + + def __init__(self, bus: EventBus): + super().__init__(bus) + self.bus.on(BinaryRequestEvent, self.on_BinaryRequestEvent__project_process) + + async def on_BinaryRequestEvent__project_process(self, request: BinaryRequestEvent) -> None: + from archivebox.machine.models import Binary, Machine, Process, _canonical_binary_name + from archivebox.services.process_service import current_network_interface_with_machine + + machine = await sync_to_async(Machine.current, thread_sensitive=True)() + binary_name = _canonical_binary_name(request.name) + if not binary_name: + return + binary = await Binary.objects.filter(machine=machine, name=binary_name).order_by("-modified_at").afirst() + if binary is None: + return + + binary_event = await self.bus.find( + BinaryEvent, + child_of=request, + past=True, + future=False, + name=request.name, + where=lambda candidate: bool(candidate.abspath), + ) + iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)() + now = timezone.now() + success = isinstance(binary_event, BinaryEvent) + output_dir = self._process_output_dir(binary, request) + output_dir.mkdir(parents=True, exist_ok=True) + cmd = [ + "abxpkg", + "install", + f"--name={request.name}", + f"--binproviders={_binproviders_to_str(request.binproviders)}", + ] + if request.overrides: + cmd.append(f"--overrides={json.dumps(request.overrides, sort_keys=True)}") + stdout = json.dumps(binary.to_json()) + "\n" if success else "" + stderr = "" if success else f"Binary request did not resolve: {request.name}" + process = await Process.objects.acreate( + machine=iface.machine, + iface=iface, + process_type=Process.TypeChoices.BINARY, + worker_type="", + pwd=str(output_dir), + cmd=cmd, + env={}, + timeout=int(request.event_timeout or request.install_timeout or 600), + pid=None, + url=None, + started_at=now, + ended_at=now, + stdout=stdout, + stderr=stderr, + exit_code=0 if success else 1, + status=Process.StatusChoices.EXITED, + retry_at=None, + binary=binary, + ) + self._write_binary_index(binary, process, output_dir) + + def _process_output_dir(self, binary, request: BinaryRequestEvent) -> Path: + raw_output_dir = str(request.extra_context.get("output_dir") or "").strip() + if raw_output_dir: + output_dir = Path(raw_output_dir).expanduser() + if output_dir.name == str(binary.id): + return output_dir.parent + return output_dir + return binary.output_dir.parent + + def _write_binary_index(self, binary, process, output_dir: Path) -> None: + output_dir.mkdir(parents=True, exist_ok=True) + index_path = output_dir / "index.jsonl" + with index_path.open("w", encoding="utf-8") as f: + f.write(json.dumps(binary.to_json()) + "\n") + f.write(json.dumps(process.to_json()) + "\n") + + def _provider_names(binproviders: str | list[str] | None) -> list[str]: if isinstance(binproviders, str): raw_names = [part.strip() for part in binproviders.split(",")] diff --git a/archivebox/services/process_service.py b/archivebox/services/process_service.py index f44c459a..3cd9e26c 100644 --- a/archivebox/services/process_service.py +++ b/archivebox/services/process_service.py @@ -59,9 +59,7 @@ class ProcessService(BaseService): from archivebox.machine.models import Process iface = await self.current_iface() - process_type = event.process_type or ( - Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK - ) + process_type = event.process_type or Process.TypeChoices.HOOK worker_type = event.worker_type or "" started_at = parse_event_datetime(event.start_ts) if started_at is None: @@ -159,9 +157,7 @@ class ProcessService(BaseService): from archivebox.machine.models import Process iface = await self.current_iface() - process_type = event.process_type or ( - Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK - ) + process_type = event.process_type or Process.TypeChoices.HOOK worker_type = event.worker_type or "" started_at = parse_event_datetime(event.start_ts) if started_at is None: diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index b4717900..123e118b 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -51,7 +51,7 @@ from abx_dl.orchestrator import ( setup_services as setup_abx_services, ) from abx_dl.services.process_service import ProcessService as HookProcessService -from abx_dl.services.binary_service import PluginBinariesService as HookPluginBinariesService +from abx_dl.services.binary_service import PluginBinariesService from abx_dl.services.snapshot_service import SnapshotService as HookSnapshotService from abx_dl.cli import LiveBusUI from abxbus import BaseEvent @@ -66,7 +66,7 @@ from archivebox.search.sonic_daemon import register_sonic_daemon_event_handler from archivebox.workers.models import ACTIVE_STATE_LEASE_SECONDS from .archive_result_service import ArchiveResultService -from .binary_service import ArchiveBoxBinaryCacheBackend +from .binary_service import ArchiveBoxBinaryService, ArchiveBoxDBBinaryCacheBackend from .crawl_service import CrawlService from .machine_service import MachineService from .process_service import ProcessService as PersistedProcessService @@ -142,11 +142,6 @@ def _is_external_task_cancelled(error: asyncio.CancelledError) -> bool: return not isinstance(error, (EventHandlerAbortedError, EventHandlerCancelledError)) -def _register_binary_services(bus) -> None: - BinaryCacheService(bus, backend=ArchiveBoxBinaryCacheBackend()) - BinaryService(bus) - - async def _emit_machine_config( bus, *, @@ -237,7 +232,9 @@ class CrawlRunner: HookProcessService(self.bus, emit_jsonl=False, interactive_tty=interactive_interrupts) register_sonic_daemon_event_handler(self.bus) PersistedProcessService(self.bus) - _register_binary_services(self.bus) + BinaryCacheService(self.bus, backend=ArchiveBoxDBBinaryCacheBackend()) + BinaryService(self.bus) + ArchiveBoxBinaryService(self.bus) TagService(self.bus) CrawlService(self.bus, crawl_id=str(crawl.id)) MachineService(self.bus) @@ -842,7 +839,7 @@ class CrawlRunner: emit_jsonl=False, abort_requested=self.crawl_is_cancelled, MachineService=None, - PluginBinariesService=HookPluginBinariesService, + PluginBinariesService=PluginBinariesService, BinaryCacheService=None, BinaryService=None, ProcessService=None, @@ -1168,7 +1165,9 @@ async def _run_binary(binary_id: str) -> None: config = _normalize_runtime_config(config) bus = create_bus(name=_bus_name("ArchiveBox_binary", str(binary.id)), total_timeout=1800.0) process_service = PersistedProcessService(bus) - _register_binary_services(bus) + BinaryCacheService(bus, backend=ArchiveBoxDBBinaryCacheBackend()) + BinaryService(bus) + ArchiveBoxBinaryService(bus) TagService(bus) ArchiveResultService(bus) MachineService(bus) @@ -1196,7 +1195,7 @@ async def _run_binary(binary_id: str) -> None: overrides=binary.overrides or None, extra_context={ "plugin_name": "archivebox", - "hook_name": "on_BinaryRequest__archivebox_run", + "hook_name": "archivebox_binary_run", "output_dir": str(binary.output_dir), "binary_id": str(binary.id), "machine_id": str(binary.machine_id), @@ -1524,7 +1523,9 @@ async def _run_install(plugin_names: list[str] | None = None) -> None: config = _normalize_runtime_config(config) bus = create_bus(name="ArchiveBox_install", total_timeout=3600.0) PersistedProcessService(bus) - _register_binary_services(bus) + BinaryCacheService(bus, backend=ArchiveBoxDBBinaryCacheBackend()) + BinaryService(bus) + ArchiveBoxBinaryService(bus) TagService(bus) ArchiveResultService(bus) MachineService(bus) diff --git a/archivebox/tests/test_archive_result_service.py b/archivebox/tests/test_archive_result_service.py index 2fd95d6a..cc2f8585 100644 --- a/archivebox/tests/test_archive_result_service.py +++ b/archivebox/tests/test_archive_result_service.py @@ -638,7 +638,8 @@ def test_process_started_uses_node_binary_for_js_hooks_without_plugin_binary(tmp def test_binary_event_reuses_existing_installed_binary_row(): from archivebox.machine.models import Binary, Machine - from archivebox.services.binary_service import BinaryService as ArchiveBoxBinaryService + from archivebox.services.binary_service import ArchiveBoxDBBinaryCacheBackend + from abxpkg.binary_service import BinaryCacheService, BinaryService import asyncio machine = Machine.current() @@ -653,17 +654,21 @@ def test_binary_event_reuses_existing_installed_binary_row(): status=Binary.StatusChoices.INSTALLED, ) - service = ArchiveBoxBinaryService(create_bus(name="test_binary_event_reuses_existing_installed_binary_row")) + bus = create_bus(name="test_binary_event_reuses_existing_installed_binary_row") + BinaryCacheService(bus, backend=ArchiveBoxDBBinaryCacheBackend()) + BinaryService(bus) event = BinaryRequestEvent( name="wget", - plugin_name="wget", - output_dir="/tmp/wget", binproviders=binary.binproviders, + extra_context={ + "plugin_name": "wget", + "output_dir": "/tmp/wget", + }, ) async def run_event(): - await service.bus.emit(event).now() - await service.bus.wait_until_idle() + await bus.emit(event).now() + await bus.wait_until_idle() asyncio.run(run_event()) diff --git a/archivebox/tests/test_hooks.py b/archivebox/tests/test_hooks.py index d3365d1e..b544240b 100755 --- a/archivebox/tests/test_hooks.py +++ b/archivebox/tests/test_hooks.py @@ -33,7 +33,6 @@ def create_test_plugin_structure(plugins_dir: Path) -> None: wget_dir = plugins_dir / "wget" wget_dir.mkdir() (wget_dir / "on_Snapshot__50_wget.py").write_text("# test hook") - (wget_dir / "on_BinaryRequest__10_wget.py").write_text("# binary request hook") chrome_dir = plugins_dir / "chrome" chrome_dir.mkdir(exist_ok=True) @@ -345,15 +344,14 @@ class TestHookDiscovery: assert "chrome" not in hook_names assert "accessibility" not in hook_names - def test_get_plugins_includes_non_snapshot_plugin_dirs(self, tmp_path): - """get_plugins() should include binary-only plugins with standardized metadata.""" + def test_get_plugins_includes_config_only_plugin_dirs(self, tmp_path): + """get_plugins() should include config-only plugins with standardized metadata.""" plugins_dir = tmp_path / "plugins" create_test_plugin_structure(plugins_dir) - env_dir = plugins_dir / "env" - env_dir.mkdir() - (env_dir / "on_BinaryRequest__15_env.py").write_text("# binary hook") - (env_dir / "config.json").write_text('{"type": "object", "properties": {}}') + helper_dir = plugins_dir / "helper" + helper_dir.mkdir() + (helper_dir / "config.json").write_text('{"type": "object", "properties": {}}') plugins = run_plugin_discovery_subprocess( tmp_path, @@ -366,30 +364,13 @@ class TestHookDiscovery: emit(get_plugins()) """, ) - assert "env" in plugins + assert "helper" in plugins - def test_discover_binary_hooks_ignores_plugins_whitelist(self, tmp_path): - """Binary provider hooks should remain discoverable under --plugins filtering.""" + def test_discover_binary_hooks_returns_empty(self, tmp_path): + """Binary provider hooks are owned by abxpkg, not ArchiveBox plugin discovery.""" plugins_dir = tmp_path / "plugins" create_test_plugin_structure(plugins_dir) - singlefile_dir = plugins_dir / "singlefile" - singlefile_dir.mkdir() - (singlefile_dir / "config.json").write_text( - json.dumps( - { - "type": "object", - "required_plugins": ["chrome"], - "properties": {}, - }, - ), - ) - - npm_dir = plugins_dir / "npm" - npm_dir.mkdir() - (npm_dir / "on_BinaryRequest__10_npm.py").write_text("# npm binary hook") - (npm_dir / "config.json").write_text('{"type": "object", "properties": {}}') - hook_names = run_plugin_discovery_subprocess( tmp_path, plugins_dir, @@ -398,16 +379,18 @@ class TestHookDiscovery: from archivebox.plugins.discovery import get_plugins get_plugins.cache_clear() - hooks = hooks_module.discover_hooks("BinaryRequest", config={"PLUGINS": "singlefile"}) + hooks = hooks_module.discover_hooks("BinaryRequest", filter_disabled=False) emit([hook.name for hook in hooks]) """, ) - assert "on_BinaryRequest__10_npm.py" in hook_names + assert hook_names == [] def test_discover_hooks_accepts_event_class_names(self, tmp_path): - """discover_hooks should accept BinaryRequestEvent / SnapshotEvent class names.""" + """discover_hooks should accept CrawlSetupEvent / SnapshotEvent class names.""" plugins_dir = tmp_path / "plugins" create_test_plugin_structure(plugins_dir) + chrome_dir = plugins_dir / "chrome" + (chrome_dir / "on_CrawlSetup__90_chrome_launch.daemon.bg.js").write_text("// crawl hook") hook_names = run_plugin_discovery_subprocess( tmp_path, @@ -417,15 +400,15 @@ class TestHookDiscovery: from archivebox.plugins.discovery import get_plugins get_plugins.cache_clear() - binary_hooks = hooks_module.discover_hooks("BinaryRequestEvent", filter_disabled=False) + crawl_setup_hooks = hooks_module.discover_hooks("CrawlSetupEvent", filter_disabled=False) snapshot_hooks = hooks_module.discover_hooks("SnapshotEvent", filter_disabled=False) emit({ - "binary": [hook.name for hook in binary_hooks], + "crawl_setup": [hook.name for hook in crawl_setup_hooks], "snapshot": [hook.name for hook in snapshot_hooks], }) """, ) - assert "on_BinaryRequest__10_wget.py" in hook_names["binary"] + assert "on_CrawlSetup__90_chrome_launch.daemon.bg.js" in hook_names["crawl_setup"] assert "on_Snapshot__50_wget.py" in hook_names["snapshot"] def test_discover_hooks_returns_empty_for_non_hook_lifecycle_events(self, tmp_path):